From d16622cd14a41574c0a2cf237279f8d71f7e1274 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Mon, 31 Aug 2026 06:26:11 -0700 Subject: [PATCH] test(buzz): regression coverage for fresh-subscription window, membership-rejection phrasing, and WS periodic discovery - fresh conversation subscribes with no since floor + bounded limit; seeded channels keep since=last_ts-1 (#78429) - all three production CLOSED membership phrasings prune per-subscription without reconnect (#76850 + #97502) - restricted channels are not re-adopted by discovery sweeps - WS periodic discovery subscribes conversations found without a membership event (#93557) and the companion task dies with its connection - watch-all live channel adoption vs explicit-list scoping (#75107) All six new tests sabotage-checked: each fails when its fix is reverted. --- tests/gateway/test_buzz_adapter.py | 28 +++- tests/gateway/test_buzz_websocket.py | 242 +++++++++++++++++++++++++++ 2 files changed, 267 insertions(+), 3 deletions(-) diff --git a/tests/gateway/test_buzz_adapter.py b/tests/gateway/test_buzz_adapter.py index 19be9adf7a..f012a8a2cf 100644 --- a/tests/gateway/test_buzz_adapter.py +++ b/tests/gateway/test_buzz_adapter.py @@ -1184,8 +1184,9 @@ class TestDmClassification: @pytest.mark.asyncio async def test_dm_shaped_channel_discovered_when_dms_list_empty(self): """Fallback discovery: with `dms list` broken (returns []), a - DM-shaped `channels list` entry becomes a DM; real channels not - already watched are left alone.""" + DM-shaped `channels list` entry gets watched. In watch-all mode a + real channel is adopted too (live join, #75107) — but seeded from + history, so nothing is replayed.""" a = _make_adapter() cli = _ScriptedCli() cli.script("dms", "list", []) @@ -1198,8 +1199,29 @@ class TestDmClassification: await a._discover_dms(seed=False) assert a._channel_state[DM_CHANNEL]["chat_type"] == "dm" assert a._may_reclassify_as_dm(DM_CHANNEL) is True - assert CHANNEL not in a._channel_state + # Watch-all mode: the real channel is live-adopted (seeded, never + # reclassified as DM). + assert CHANNEL in a._channel_state + assert a._channel_state[CHANNEL]["chat_type"] == "group" assert a._may_reclassify_as_dm(CHANNEL) is False + # Adoption seeded it via a messages get call (history suppressed). + assert any(c[0][:2] == ["messages", "get"] for c in cli.calls) + + @pytest.mark.asyncio + async def test_explicit_watch_list_blocks_live_channel_adoption(self): + """With an explicit channels: list, discovery must NOT adopt real + channels outside that list — the user chose the watch set (#75107 + scoping).""" + a = _make_adapter(extra={"channels": ["some-other-channel"]}) + cli = _ScriptedCli() + cli.script("dms", "list", []) + cli.script("channels", "list", [ + {"channel_id": CHANNEL, "name": "general", + "description": "General conversation and community updates.", "created_at": 2}, + ]) + a._run_cli = cli + await a._discover_dms(seed=False) + assert CHANNEL not in a._channel_state @pytest.mark.asyncio async def test_dm_metadata_promotes_existing_group_without_recipient_tag(self, adapter): diff --git a/tests/gateway/test_buzz_websocket.py b/tests/gateway/test_buzz_websocket.py index 989390a023..543a55896b 100644 --- a/tests/gateway/test_buzz_websocket.py +++ b/tests/gateway/test_buzz_websocket.py @@ -438,3 +438,245 @@ def test_restricted_channels_skipped_during_subscribe(): assert all(CHANNEL not in ch_list for ch_list in req_channels), ( "restricted channel must not be sent in any REQ frame" ) + + +# ── Fresh-conversation subscription window (#78429) ──────────────────────── + + +@pytest.mark.asyncio +async def test_new_subscription_without_high_water_mark_has_no_since_floor(): + """A conversation adopted mid-run (last_ts == 0) must NOT subscribe with + `since ≈ now` — that drops the message that created the conversation + (#78429). It must request from the beginning with a bounded limit.""" + adapter = _make_adapter() + adapter._channel_state[CHANNEL] = {"chat_type": "dm", "last_ts": 0, "seen": {}} + + class _Ws: + def __init__(self): + self.sent = [] + + async def send(self, raw): + self.sent.append(json.loads(raw)) + + ws = _Ws() + await adapter._send_channel_subscription(ws, "hermes-buzz-dm-1", CHANNEL) + assert len(ws.sent) == 1 + req_filter = ws.sent[0][2] + assert "since" not in req_filter, ( + "fresh conversation must not have a since floor (drops the opening message)" + ) + assert req_filter.get("limit") == _buzz_mod._FETCH_LIMIT + assert req_filter["#h"] == [CHANNEL] + + +@pytest.mark.asyncio +async def test_seeded_subscription_resumes_from_high_water_mark(): + """A channel with a real high-water mark keeps the since-resume contract + (last_ts - 1, same-second overlap de-duped by id).""" + adapter = _make_adapter() + adapter._channel_state[CHANNEL] = {"chat_type": "group", "last_ts": 1_700_000_000, "seen": {}} + + class _Ws: + def __init__(self): + self.sent = [] + + async def send(self, raw): + self.sent.append(json.loads(raw)) + + ws = _Ws() + await adapter._send_channel_subscription(ws, "hermes-buzz-0", CHANNEL) + req_filter = ws.sent[0][2] + assert req_filter["since"] == 1_699_999_999 + assert "limit" not in req_filter + + +# ── Membership-rejection phrasing (#97502 composed into #76850) ──────────── + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "detail", + [ + "restricted: not a channel member", + "not a channel member", + "auth-required: subscription needs auth", + ], +) +async def test_closed_membership_phrases_prune_without_reconnect(detail): + """Every production-observed membership-rejection phrasing (#76850, + #97502) prunes the subscription instead of tearing down the socket.""" + import sys + from unittest.mock import patch, MagicMock + from contextlib import asynccontextmanager + + adapter = _make_adapter(extra={"channels": [CHANNEL]}) + adapter._channel_state[CHANNEL] = {"chat_type": "group", "last_ts": 0, "seen": {}} + + sub_id = "hermes-buzz-0" + messages = [json.dumps(["CLOSED", sub_id, detail])] + idx = 0 + + class _FakeWs: + async def send(self, raw): + pass + + def __aiter__(self): + return self + + async def __anext__(self): + nonlocal idx + if idx < len(messages): + val = messages[idx] + idx += 1 + return val + await asyncio.sleep(10) + raise StopAsyncIteration + + @asynccontextmanager + async def _fake_connect(*_a, **_kw): + yield _FakeWs() + + async def _noop_auth(self_inner, ws): + pass + + async def _noop_subscribe(self_inner, ws): + return {sub_id: CHANNEL} + + fake_ws_mod = MagicMock() + fake_ws_mod.connect = _fake_connect + + with ( + patch.dict(sys.modules, {"websockets": fake_ws_mod}), + patch.object(type(adapter), "_authenticate_websocket", _noop_auth), + patch.object(type(adapter), "_subscribe_websocket", _noop_subscribe), + ): + adapter._ws_ready = asyncio.Event() + adapter._ws_ready.set() + task = asyncio.create_task(adapter._websocket_loop()) + await asyncio.sleep(0.1) + + assert CHANNEL in adapter._restricted_channels + assert CHANNEL not in adapter._channel_state + assert not task.done(), "membership rejection must not reconnect the socket" + + task.cancel() + try: + await task + except (asyncio.CancelledError, Exception): + pass + + +@pytest.mark.asyncio +async def test_restricted_channel_not_readopted_by_discovery(): + """The live-discovery paths must skip _restricted_channels; otherwise the + next sweep silently re-adds the channel and re-triggers the rejection + (the re-adoption hole flagged in the #76850 review).""" + adapter = _make_adapter() + adapter._restricted_channels.add(CHANNEL) + + calls = [] + + async def scripted_cli(args, *, input_text=None): + calls.append(list(args)) + if args[:2] == ["dms", "list"]: + return 0, json.dumps([{"dm_id": CHANNEL}]), "" + if args[:2] == ["channels", "list"]: + return 0, json.dumps([{"channel_id": CHANNEL, "name": "DM", "description": ""}]), "" + return 0, "[]", "" + + adapter._run_cli = scripted_cli + await adapter._discover_dms(seed=False) + assert CHANNEL not in adapter._channel_state, ( + "restricted channel must not be re-adopted by discovery" + ) + + +# ── WS periodic discovery (#93557 / #75107) ──────────────────────────────── + + +@pytest.mark.asyncio +async def test_ws_discovery_loop_subscribes_newly_discovered_conversation(monkeypatch): + """Without a kind-44100 membership event, the WS transport's periodic + sweep must still find and subscribe a conversation opened mid-session + (#93557).""" + adapter = _make_adapter() + adapter.poll_interval = 0.01 + + new_dm = "0f0e0d0c-0b0a-4123-8123-cafecafecafe" + + async def fake_discover(*, seed): + adapter._channel_state.setdefault( + new_dm, {"chat_type": "dm", "last_ts": 0, "seen": {}} + ) + + monkeypatch.setattr(adapter, "_discover_dms", fake_discover) + monkeypatch.setattr(_buzz_mod, "_MIN_POLL_INTERVAL", 0.01) + + class _Ws: + def __init__(self): + self.sent = [] + + async def send(self, raw): + self.sent.append(json.loads(raw)) + + ws = _Ws() + subscriptions = {"hermes-buzz-0": CHANNEL} + task = asyncio.create_task(adapter._ws_discovery_loop(ws, subscriptions)) + try: + deadline = time.monotonic() + 5.0 + while not ws.sent and time.monotonic() < deadline: + await asyncio.sleep(0.02) + finally: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + assert new_dm in subscriptions.values(), "sweep did not subscribe the new conversation" + req = ws.sent[0] + assert req[0] == "REQ" and req[2]["#h"] == [new_dm] + # Fresh conversation: no since floor (#78429 applies here too). + assert "since" not in req[2] + + +@pytest.mark.asyncio +async def test_ws_discovery_task_cancelled_when_connection_exits(monkeypatch): + """The companion discovery task must not outlive its connection.""" + adapter = _make_adapter() + adapter.poll_interval = 10.0 # sweep never fires; we only test lifecycle + adapter._channel_state = {CHANNEL: {"chat_type": "group", "last_ts": 1, "seen": {}}} + + started = [] + real_create_task = asyncio.create_task + + def tracking_create_task(coro, **kw): + t = real_create_task(coro, **kw) + if "_ws_discovery_loop" in repr(coro): + started.append(t) + return t + + monkeypatch.setattr(asyncio, "create_task", tracking_create_task) + + async def closed_anext(): + raise StopAsyncIteration + + sockets = [] + + def fake_connect(*args, **kwargs): + if sockets: + raise asyncio.CancelledError() + ws = _ScriptedWebSocket(closed_anext) + sockets.append(ws) + return ws + + import websockets as _ws_mod + + monkeypatch.setattr(_ws_mod, "connect", fake_connect) + + task = real_create_task(adapter._websocket_loop()) + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, 10.0) + + assert started, "discovery task was never started with the connection" + assert all(t.done() for t in started), "discovery task outlived its connection"