test: bind hass fixture to the running test loop and remove listener registry private loop

tests/conftest.py

@@ -175,7 +175,6 @@ class ListenerRegistry:
         self._hass = hass
         self._listeners: dict[str, list[object]] = {}
         self.reloaded: list[str] = []
-        self._loop = asyncio.new_event_loop()
 
     def add(self, entry_id: str, listener) -> object:
         def _remove():
@@ -184,20 +183,15 @@ class ListenerRegistry:
         self._listeners.setdefault(entry_id, []).append(listener)
         return _remove
 
-    def dispatch_options_update(self, entry) -> None:
-        """Synchronous dispatch for sync tests (drives a private event loop)."""
-        for listener in list(self._listeners.get(entry.entry_id, [])):
-            result = listener(self._hass, entry)
-            if inspect.isawaitable(result):
-                self._loop.run_until_complete(result)
-
-    async def async_dispatch_options_update(self, entry) -> None:
-        """Async dispatch for pytest-asyncio tests (awaits listeners in the running loop)."""
+    async def dispatch_options_update(self, entry) -> None:
+        """Dispatch on the running loop; sync listeners run inline, async ones awaited."""
         for listener in list(self._listeners.get(entry.entry_id, [])):
             result = listener(self._hass, entry)
             if inspect.isawaitable(result):
                 await result
 
+    async_dispatch_options_update = dispatch_options_update
+
     @property
     def size(self) -> int:
         return sum(len(listeners) for listeners in self._listeners.values())
@@ -250,20 +244,19 @@ def make_hass() -> tuple:
 
 
 @pytest.fixture
-def hass():
-    """Provide a standard-shaped hass fixture.
+async def hass():
+    """Provide a standard-shaped hass fixture bound to the running test loop.
 
     Mirrors the surface of Home Assistant's ``hass`` fixture: a ``config``
     namespace, a ``config_entries`` manager wired to a listener registry, a
-    ``data`` dict for integration state, and the running event ``loop``.
+    ``data`` dict for integration state, and a ``loop`` bound to the
+    pytest-asyncio loop the test itself runs on (HA's running-loop contract),
+    so code scheduling on ``hass.loop`` runs on the actual test loop.
     """
     _hass, registry = make_hass()
     _hass.registry = registry
-    _hass.loop = asyncio.new_event_loop()
-    try:
-        yield _hass
-    finally:
-        _hass.loop.close()
+    _hass.loop = asyncio.get_running_loop()
+    yield _hass
 
 
 @pytest.fixture

tests/test_harness.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import asyncio
 import importlib.metadata
+import inspect
 import sys
 from unittest.mock import MagicMock
 
@@ -43,4 +45,42 @@ def test_mock_only_harness_contract(monkeypatch):
     monkeypatch.setattr(conftest, "_find_spec", lambda name: None)
     monkeypatch.setitem(sys.modules, "homeassistant", MagicMock(name="real-module"))
     with pytest.raises(pytest.UsageError, match="mock-only"):
-        _ensure_mock_only()
+        _ensure_mock_only()
+
+
+async def test_hass_and_registry_run_on_the_running_test_loop(hass, make_entry):
+    """hass.loop is the test's own loop and the registry spins no private loop."""
+    # hass.loop is the very loop the async test runs on (HA running-loop
+    # contract): code scheduling on hass.loop lands on the test loop itself.
+    assert hass.loop is asyncio.get_running_loop()
+    ran = []
+
+    async def _scheduled():
+        ran.append(True)
+
+    await hass.loop.create_task(_scheduled())
+    assert ran == [True]
+
+    # ListenerRegistry never spins its own event loop (no resource leak) and
+    # dispatch is a plain coroutine: sync listeners run inline, async awaited.
+    registry = hass.registry
+    assert not hasattr(registry, "_loop")
+    assert inspect.iscoroutinefunction(registry.dispatch_options_update)
+
+    calls = []
+
+    def _sync_listener(h, entry):
+        calls.append(("sync", entry.entry_id, h is hass))
+
+    entry = make_entry(entry_id="disp")
+    registry.add("disp", _sync_listener)
+    await registry.dispatch_options_update(entry)
+    assert calls == [("sync", "disp", True)]
+
+    async def _async_listener(h, entry):
+        calls.append(("async", entry.entry_id))
+
+    registry.add("disp", _async_listener)
+    calls.clear()
+    await registry.async_dispatch_options_update(entry)
+    assert calls == [("sync", "disp", True), ("async", "disp")]

tests/test_options_flow.py

@@ -271,7 +271,7 @@ def test_options_flow_create_entry_data_lands_in_entry_options_and_reloads_once(
     assert result["type"] == "create_entry"
     entry.options = dict(result["data"])
 
-    registry.dispatch_options_update(entry)
+    _run(registry.dispatch_options_update(entry))
     assert hass.config_entries.async_reload.call_count == 1
     assert registry.reloaded == [entry.entry_id]
     assert entry.options == VALID_INPUT