| 1 | import asyncio |
| 2 | import sys |
| 3 | from pathlib import Path |
| 4 | |
| 5 | import pytest |
| 6 | |
| 7 | PROJECT_ROOT = Path(__file__).resolve().parents[1] |
| 8 | if str(PROJECT_ROOT) not in sys.path: |
| 9 | sys.path.insert(0, str(PROJECT_ROOT)) |
| 10 | |
| 11 | |
| 12 | @pytest.mark.asyncio |
| 13 | async def test_state_monitor_per_sid_isolation_independent_snapshots_seq_and_cursors(monkeypatch): |
| 14 | import helpers.state_monitor as state_monitor_module |
| 15 | from helpers.state_monitor import StateMonitor |
| 16 | from helpers.state_snapshot import StateRequestV1 |
| 17 | |
| 18 | snapshot_calls: list[dict[str, object]] = [] |
| 19 | emitted: list[dict[str, object]] = [] |
| 20 | |
| 21 | namespace = "/ws" |
| 22 | |
| 23 | async def fake_build_snapshot_from_request(*, request, include_collections=True): |
| 24 | context = request.context |
| 25 | log_from = request.log_from |
| 26 | notifications_from = request.notifications_from |
| 27 | timezone = request.timezone |
| 28 | snapshot_calls.append( |
| 29 | { |
| 30 | "context": context, |
| 31 | "log_from": log_from, |
| 32 | "notifications_from": notifications_from, |
| 33 | "timezone": timezone, |
| 34 | } |
| 35 | ) |
| 36 | # Return poll-shaped keys that StateMonitor expects to advance cursors from. |
| 37 | return { |
| 38 | "deselect_chat": False, |
| 39 | "context": context or "", |
| 40 | "contexts": [], |
| 41 | "tasks": [], |
| 42 | "logs": [], |
| 43 | "log_guid": "log-guid", |
| 44 | "log_version": int(log_from) + 1, |
| 45 | "log_progress": "", |
| 46 | "log_progress_active": False, |
| 47 | "paused": False, |
| 48 | "notifications": [], |
| 49 | "notifications_guid": "notifications-guid", |
| 50 | "notifications_version": int(notifications_from) + 1, |
| 51 | } |
| 52 | |
| 53 | class FakeManager: |
| 54 | def __init__(self, loop): |
| 55 | self._dispatcher_loop = loop |
| 56 | |
| 57 | async def emit_to(self, namespace, sid, event_type, payload, *, handler_id=None): |
| 58 | emitted.append( |
| 59 | { |
| 60 | "namespace": namespace, |
| 61 | "sid": sid, |
| 62 | "event_type": event_type, |
| 63 | "payload": payload, |
| 64 | "handler_id": handler_id, |
| 65 | } |
| 66 | ) |
| 67 | |
| 68 | monkeypatch.setattr( |
| 69 | state_monitor_module, |
| 70 | "build_snapshot_from_request", |
| 71 | fake_build_snapshot_from_request, |
| 72 | ) |
| 73 | |
| 74 | monitor = StateMonitor(debounce_seconds=60.0) |
| 75 | loop = asyncio.get_running_loop() |
| 76 | monitor.bind_manager(FakeManager(loop), handler_id="test.handler") |
| 77 | |
| 78 | monitor.register_sid(namespace, "sid-a") |
| 79 | monitor.register_sid(namespace, "sid-b") |
| 80 | |
| 81 | monitor.update_projection( |
| 82 | namespace, |
| 83 | "sid-a", |
| 84 | request=StateRequestV1(context="ctx-a", log_from=0, notifications_from=0, timezone="UTC"), |
| 85 | seq_base=10, |
| 86 | ) |
| 87 | monitor.update_projection( |
| 88 | namespace, |
| 89 | "sid-b", |
| 90 | request=StateRequestV1( |
| 91 | context="ctx-b", |
| 92 | log_from=40, |
| 93 | notifications_from=7, |
| 94 | timezone="Europe/Berlin", |
| 95 | ), |
| 96 | seq_base=100, |
| 97 | ) |
| 98 | |
| 99 | # Flush pushes directly to avoid relying on debounce scheduling. |
| 100 | await monitor._flush_push((namespace, "sid-a")) |
| 101 | await monitor._flush_push((namespace, "sid-b")) |
| 102 | |
| 103 | assert snapshot_calls == [ |
| 104 | {"context": "ctx-a", "log_from": 0, "notifications_from": 0, "timezone": "UTC"}, |
| 105 | {"context": "ctx-b", "log_from": 40, "notifications_from": 7, "timezone": "Europe/Berlin"}, |
| 106 | ] |
| 107 | |
| 108 | assert len(emitted) == 2 |
| 109 | assert {entry["sid"] for entry in emitted} == {"sid-a", "sid-b"} |
| 110 | assert all(entry["event_type"] == "state_push" for entry in emitted) |
| 111 | assert all(entry["handler_id"] == "test.handler" for entry in emitted) |
| 112 | assert all(entry["namespace"] == namespace for entry in emitted) |
| 113 | |
| 114 | payload_a = next(entry["payload"] for entry in emitted if entry["sid"] == "sid-a") |
| 115 | payload_b = next(entry["payload"] for entry in emitted if entry["sid"] == "sid-b") |
| 116 | |
| 117 | assert payload_a["seq"] == 11 # seq_base=10 -> first push increments to 11 |
| 118 | assert payload_b["seq"] == 101 # seq_base=100 -> first push increments to 101 |
| 119 | |
| 120 | assert payload_a["snapshot"]["context"] == "ctx-a" |
| 121 | assert payload_b["snapshot"]["context"] == "ctx-b" |
| 122 | |
| 123 | # Verify per-sid cursor advancement is independent. |
| 124 | assert monitor._projections[(namespace, "sid-a")].request.log_from == 1 |
| 125 | assert monitor._projections[(namespace, "sid-a")].request.notifications_from == 1 |
| 126 | assert monitor._projections[(namespace, "sid-b")].request.log_from == 41 |
| 127 | assert monitor._projections[(namespace, "sid-b")].request.notifications_from == 8 |
| 128 | |
| 129 | |
| 130 | @pytest.mark.asyncio |
| 131 | async def test_state_monitor_mark_dirty_for_context_scopes_to_active_context(): |
| 132 | from helpers.state_monitor import StateMonitor |
| 133 | from helpers.state_snapshot import StateRequestV1 |
| 134 | |
| 135 | monitor = StateMonitor(debounce_seconds=60.0) |
| 136 | namespace = "/ws" |
| 137 | monitor.register_sid(namespace, "sid-a") |
| 138 | monitor.register_sid(namespace, "sid-b") |
| 139 | |
| 140 | monitor.update_projection( |
| 141 | namespace, |
| 142 | "sid-a", |
| 143 | request=StateRequestV1(context="ctx-a", log_from=0, notifications_from=0, timezone="UTC"), |
| 144 | seq_base=10, |
| 145 | ) |
| 146 | monitor.update_projection( |
| 147 | namespace, |
| 148 | "sid-b", |
| 149 | request=StateRequestV1(context="ctx-b", log_from=0, notifications_from=0, timezone="UTC"), |
| 150 | seq_base=10, |
| 151 | ) |
| 152 | |
| 153 | monitor.mark_dirty_for_context("ctx-a") |
| 154 | assert (namespace, "sid-a") in monitor._debounce_handles |
| 155 | assert (namespace, "sid-b") not in monitor._debounce_handles |
| 156 | |
| 157 | monitor.unregister_sid(namespace, "sid-a") |
| 158 | monitor.unregister_sid(namespace, "sid-b") |