test: isolate shared listener lock across HA event loops

Author: Maksym Shtembuliak

tests_ha/conftest.py

@@ -12,6 +12,7 @@
 
 from __future__ import annotations
 
+import asyncio
 from pathlib import Path
 import sys
 from typing import Any
@@ -33,6 +34,25 @@
 )
 
 
+@pytest.fixture(autouse=True)
+def isolated_listener_loop(monkeypatch):
+    """Model a fresh HA process's lock for each isolated test event loop.
+
+    Production HA uses one loop. Pytest creates a new one for each case while
+    retaining imported modules. A contended module-level asyncio.Lock otherwise
+    stays bound to a previous case's loop. Keep the real lock and registry
+    behavior; do not clear leaked listeners or suppress task errors.
+    """
+    from custom_components.eybond_local.collector.transport import listener
+
+    assert not listener._LISTENERS, "previous HA test leaked a shared listener"
+    lock = asyncio.Lock()
+    monkeypatch.setattr(listener, "_LISTENERS_LOCK", lock)
+    yield
+    assert not lock.locked(), "HA test leaked the shared-listener lock"
+    assert not listener._LISTENERS, "HA test leaked a shared listener"
+
+
 @pytest.fixture(autouse=True)
 def auto_enable_custom_integrations(enable_custom_integrations):
     """Let Home Assistant load `custom_components/eybond_local` for real."""

tests_ha/test_ha_listener_loop_scope.py

@@ -0,0 +1,28 @@
+"""Each isolated HA test owns its event-loop-bound listener lock."""
+import asyncio
+
+import pytest
+
+from custom_components.eybond_local.collector.transport import listener
+
+
+@pytest.mark.parametrize("separate_ha_loop", [1, 2])
+async def test_listener_lock_contention_is_local_to_each_ha_loop(hass, separate_ha_loop):
+    # Both parametrized cases must actually contend. An uncontended Lock can
+    # appear to work on a different loop, hiding cross-test state until CI.
+    lock = listener._LISTENERS_LOCK
+    queued = asyncio.Event()
+    acquired = asyncio.Event()
+
+    async def contender():
+        queued.set()
+        async with lock:
+            acquired.set()
+
+    async with lock:
+        task = asyncio.create_task(contender())
+        await queued.wait()
+        assert not acquired.is_set()
+    await task
+    assert acquired.is_set()
+    assert not lock.locked()