Revert to using call_soon for event triggers and state changed event trackers (#122735)

tests/components/logbook/test_websocket_api.py

@@ -3,6 +3,7 @@
 import asyncio
 from collections.abc import Callable
 from datetime import timedelta
+from typing import Any
 from unittest.mock import ANY, patch
 
 from freezegun import freeze_time
@@ -31,9 +32,10 @@ from homeassistant.const import (
     STATE_OFF,
     STATE_ON,
 )
-from homeassistant.core import Event, HomeAssistant, State
+from homeassistant.core import Event, HomeAssistant, State, callback
 from homeassistant.helpers import device_registry as dr, entity_registry as er
 from homeassistant.helpers.entityfilter import CONF_ENTITY_GLOBS
+from homeassistant.helpers.event import async_track_state_change_event
 from homeassistant.setup import async_setup_component
 import homeassistant.util.dt as dt_util
 
@@ -2965,3 +2967,79 @@ async def test_subscribe_all_entities_are_continuous_with_device(
     assert listeners_without_writes(
         hass.bus.async_listeners()
     ) == listeners_without_writes(init_listeners)
+
+
+@pytest.mark.parametrize("params", [{"entity_ids": ["binary_sensor.is_light"]}, {}])
+async def test_live_stream_with_changed_state_change(
+    async_setup_recorder_instance: RecorderInstanceGenerator,
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    params: dict[str, Any],
+) -> None:
+    """Test the live logbook stream with chained events."""
+    config = {recorder.CONF_COMMIT_INTERVAL: 0.5}
+    await async_setup_recorder_instance(hass, config)
+    now = dt_util.utcnow()
+    await asyncio.gather(
+        *[
+            async_setup_component(hass, comp, {})
+            for comp in ("homeassistant", "logbook")
+        ]
+    )
+
+    hass.states.async_set("binary_sensor.is_light", "ignored")
+    hass.states.async_set("binary_sensor.is_light", "init")
+    await async_wait_recording_done(hass)
+
+    @callback
+    def auto_off_listener(event):
+        hass.states.async_set("binary_sensor.is_light", STATE_OFF)
+
+    async_track_state_change_event(hass, ["binary_sensor.is_light"], auto_off_listener)
+
+    websocket_client = await hass_ws_client()
+    init_listeners = hass.bus.async_listeners()
+    await websocket_client.send_json(
+        {
+            "id": 7,
+            "type": "logbook/event_stream",
+            "start_time": now.isoformat(),
+            **params,
+        }
+    )
+
+    msg = await asyncio.wait_for(websocket_client.receive_json(), 2)
+    assert msg["id"] == 7
+    assert msg["type"] == TYPE_RESULT
+    assert msg["success"]
+
+    await hass.async_block_till_done()
+    hass.states.async_set("binary_sensor.is_light", STATE_ON)
+
+    recieved_rows = []
+    while len(recieved_rows) < 3:
+        msg = await asyncio.wait_for(websocket_client.receive_json(), 2.5)
+        assert msg["id"] == 7
+        assert msg["type"] == "event"
+        recieved_rows.extend(msg["event"]["events"])
+
+    # Make sure we get rows back in order
+    assert recieved_rows == [
+        {"entity_id": "binary_sensor.is_light", "state": "init", "when": ANY},
+        {"entity_id": "binary_sensor.is_light", "state": "on", "when": ANY},
+        {"entity_id": "binary_sensor.is_light", "state": "off", "when": ANY},
+    ]
+
+    await websocket_client.send_json(
+        {"id": 8, "type": "unsubscribe_events", "subscription": 7}
+    )
+    msg = await asyncio.wait_for(websocket_client.receive_json(), 2)
+
+    assert msg["id"] == 8
+    assert msg["type"] == TYPE_RESULT
+    assert msg["success"]
+
+    # Check our listener got unsubscribed
+    assert listeners_without_writes(
+        hass.bus.async_listeners()
+    ) == listeners_without_writes(init_listeners)