From 549747ee01d7331dbedd3536a10185d91a7c90f7 Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Sun, 23 Aug 2026 19:48:33 +0200 Subject: [PATCH 1/5] Add retry-safe live Slack contract tests Exercise the channel against a real workspace in a globally serialized workflow. Keep all outbound mutations acknowledged, fence phase transitions with fresh probes, discard stale test-only callbacks, and emit structured Socket Mode diagnostics without message content or credentials. --- .github/workflows/slack-live.yml | 91 +++++ docs/testing-slack.md | 56 +++ tests/fake_slack.py | 33 +- tests/slack_live.py | 663 +++++++++++++++++++++++++++++++ tests/test_slack_integration.py | 93 +++++ tests/test_slack_live.py | 346 ++++++++++++++++ tests/test_slack_live_inbound.py | 421 ++++++++++++++++++++ 7 files changed, 1698 insertions(+), 5 deletions(-) create mode 100644 .github/workflows/slack-live.yml create mode 100644 docs/testing-slack.md create mode 100644 tests/slack_live.py create mode 100644 tests/test_slack_live.py create mode 100644 tests/test_slack_live_inbound.py diff --git a/.github/workflows/slack-live.yml b/.github/workflows/slack-live.yml new file mode 100644 index 00000000..22a5cf54 --- /dev/null +++ b/.github/workflows/slack-live.yml @@ -0,0 +1,91 @@ +name: Slack live integration + +# Kept out of ci.yml because it needs a concurrency policy the rest of CI +# must not have. These tests drive one real Slack workspace, and Slack hands +# each event to exactly one of an app's open Socket Mode connections — so two +# runs at once steal each other's messages, and collide on reactions and on +# deleting each other's posts. The workspace is the shared resource, not the +# branch, so the lock below is global rather than per-ref. + +on: + push: + branches: [main] + pull_request: + branches: [main] + workflow_dispatch: + +# One run at a time, across every branch and pull request. +# +# cancel-in-progress is false on purpose, the opposite of ci.yml. Killing a +# run mid-flight leaves envelopes unacked and messages undeleted, and Slack +# redelivers an unacked envelope — so a cancelled run does not just lose its +# own result, it poisons the next one. Queueing is the cheaper failure: the +# suite takes about three minutes. +# +# Note the GitHub behaviour this cannot avoid: only one run may be pending +# per group, so a third run arriving while one is executing and one is queued +# cancels the queued one. It is not lost work, it is a skipped check, and +# re-running the job is enough. +concurrency: + group: slack-live-workspace + cancel-in-progress: false + +permissions: + contents: read + +jobs: + slack-live-tests: + name: Slack live integration + runs-on: ubuntu-latest + timeout-minutes: 20 + # The constraint is forks, not pull requests: a fork cannot read the + # secrets, so the job would have nothing to run against. A pull request + # from a branch in this repository can, and is exactly where the result + # is worth having — waiting until merge means finding out too late. + # + # A clone without the secrets configured skips every test rather than + # failing, so this stays green outside ClickHouse/nerve. + if: >- + github.event_name != 'pull_request' || + github.event.pull_request.head.repo.full_name == github.repository + steps: + - name: Checkout + uses: actions/checkout@v5 + + - name: Set up uv + uses: astral-sh/setup-uv@v6 + with: + # Kept in step with the backend job above; bump both together. + version: "0.12.0" + enable-cache: true + cache-dependency-glob: "uv.lock" + + - name: Install project and test dependencies + run: uv sync --extra test --locked --python "3.13" + + # Inbound first, and in its own process. The outbound module now keeps + # an ack-only socket open around all of its Web API mutations, which + # prevents it from scheduling retries that poison the next run. Keep + # inbound first for retry schedules left by older revisions or an + # interrupted job, and keep the process split so the outbound sink can + # never steal an inbound test event. `if: always()` on outbound means a + # failure in the first step still reports the second. + - name: Run live Slack tests (inbound) + env: + NERVE_SLACK_TEST_BOT_TOKEN: ${{ secrets.NERVE_SLACK_TEST_BOT_TOKEN }} + NERVE_SLACK_TEST_APP_TOKEN: ${{ secrets.NERVE_SLACK_TEST_APP_TOKEN }} + NERVE_SLACK_TEST_CHANNEL: ${{ secrets.NERVE_SLACK_TEST_CHANNEL }} + NERVE_SLACK_TEST_USER_TOKEN: ${{ secrets.NERVE_SLACK_TEST_USER_TOKEN }} + # -s exposes the credential-free `SLACK_LIVE {json}` diagnostics from + # the harness even on successful runs; pytest otherwise captures them. + run: .venv/bin/pytest tests/test_slack_live_inbound.py -v -s + + - name: Run live Slack tests (outbound) + if: always() + env: + NERVE_SLACK_TEST_BOT_TOKEN: ${{ secrets.NERVE_SLACK_TEST_BOT_TOKEN }} + NERVE_SLACK_TEST_APP_TOKEN: ${{ secrets.NERVE_SLACK_TEST_APP_TOKEN }} + NERVE_SLACK_TEST_CHANNEL: ${{ secrets.NERVE_SLACK_TEST_CHANNEL }} + NERVE_SLACK_TEST_USER_TOKEN: ${{ secrets.NERVE_SLACK_TEST_USER_TOKEN }} + NERVE_SLACK_TEST_BOT_TOKEN_NO_EMAIL: ${{ secrets.NERVE_SLACK_TEST_BOT_TOKEN_NO_EMAIL }} + run: .venv/bin/pytest tests/test_slack_live.py -v -s diff --git a/docs/testing-slack.md b/docs/testing-slack.md new file mode 100644 index 00000000..494b9727 --- /dev/null +++ b/docs/testing-slack.md @@ -0,0 +1,56 @@ +# Live Slack tests + +These tests cover behavior a fake cannot verify: Slack API payloads, scopes, +text rewriting, and end-to-end Socket Mode delivery. They are optional and +skip when their credentials are absent. + +Use a throwaway workspace. The suite creates and deletes messages, reactions, +and uploads, but an interrupted run can leave test data behind. + +## Setup + +1. Create the app from the [Slack manifest](config.md#setting-up-the-slack-app). +2. Add the user token scopes `chat:write`, `im:write`, and `reactions:write`. +3. In **App Home → Messages Tab**, enable messages and **Allow users to send + Slash commands and messages**. The manifest cannot set this option. +4. Invite both the bot and the user who installed the app to the test channel. + +Set these test-only environment variables: + +| Variable | Value | Required for | +|---|---|---| +| `NERVE_SLACK_TEST_BOT_TOKEN` | Bot token (`xoxb-…`) | All live tests | +| `NERVE_SLACK_TEST_APP_TOKEN` | App token (`xapp-…`) with `connections:write` | All live tests | +| `NERVE_SLACK_TEST_CHANNEL` | Test channel ID (`C…`) | All live tests | +| `NERVE_SLACK_TEST_USER_TOKEN` | User token (`xoxp-…`) | Inbound tests | +| `NERVE_SLACK_TEST_BOT_TOKEN_NO_EMAIL` | Bot token without `users:read.email` | One scope test | + +These variables do not configure Nerve itself. See [config.md](config.md#slack) +for production configuration. + +## Run + +Ensure no CI job or other developer is using the same test workspace. Then run +the inbound suite first and the outbound suite second, as separate processes: + +```bash +uv run --extra test pytest tests/test_slack_live_inbound.py -v -s +uv run --extra test pytest tests/test_slack_live.py -v -s +``` + +Do not combine the modules in one pytest command. Slack sends each event to +only one active Socket Mode connection. Process isolation prevents the +outbound suite's acknowledgement-only connection from stealing inbound events +and prevents unacknowledged events from being retried into later tests. + +## CI and failures + +The `Slack live integration` workflow uses the same variable names as GitHub +secrets and serializes runs against the shared workspace. + +With `-s`, diagnostic lines prefixed by `SLACK_LIVE` show connection state, +probe latency, retries, event age, and a final summary. They omit tokens, IDs, +and message text, so they are safe to share when investigating a failure. + +Slash commands require manual testing because Slack provides no API for +invoking them. diff --git a/tests/fake_slack.py b/tests/fake_slack.py index ee71f685..653359cf 100644 --- a/tests/fake_slack.py +++ b/tests/fake_slack.py @@ -113,22 +113,45 @@ async def wait_connected(self, timeout: float = 5.0) -> None: """Block until the bot has opened its socket.""" await asyncio.wait_for(self._connected.wait(), timeout) - async def push(self, envelope_type: str, payload: dict[str, Any]) -> str: + async def push( + self, + envelope_type: str, + payload: dict[str, Any], + *, + retry_attempt: int | None = None, + retry_reason: str | None = None, + ) -> str: """Push one Socket Mode envelope at the bot. Returns its envelope id.""" await self.wait_connected() assert self._ws is not None envelope_id = str(uuid.uuid4()) - await self._ws.send_str(json.dumps({ + envelope = { "type": envelope_type, "envelope_id": envelope_id, "payload": payload, "accepts_response_payload": False, - })) + } + if retry_attempt is not None: + envelope["retry_attempt"] = retry_attempt + if retry_reason is not None: + envelope["retry_reason"] = retry_reason + await self._ws.send_str(json.dumps(envelope)) return envelope_id - async def push_event(self, event: dict[str, Any]) -> str: + async def push_event( + self, + event: dict[str, Any], + *, + retry_attempt: int | None = None, + retry_reason: str | None = None, + ) -> str: """Push an Events API event (the common case).""" - return await self.push("events_api", {"event": event}) + return await self.push( + "events_api", + {"event": event}, + retry_attempt=retry_attempt, + retry_reason=retry_reason, + ) # -- the Web API ---------------------------------------------------- # diff --git a/tests/slack_live.py b/tests/slack_live.py new file mode 100644 index 00000000..c8201a8d --- /dev/null +++ b/tests/slack_live.py @@ -0,0 +1,663 @@ +"""Shared setup for the live Slack tests. + +These talk to a real Slack workspace. They exist to settle the questions +:mod:`tests.fake_slack` structurally cannot: the fake answers whatever this +code asks it, so it confirms the client is self-consistent, not that its +beliefs about Slack are true. + +Credentials come from the environment and every test skips when they are +absent, so the ordinary suite and CI on a fork are unaffected. See +:data:`SETUP` for what to provide. + +Two tiers: + +* **Outbound** needs the bot token, the app token, and a channel id. It + covers everything Nerve *sends* — Block Kit validation, emoji short names, + message splitting, streaming edits, uploads. +* **Inbound** additionally needs a user token, used to post as a human so a + real event travels Slack → Socket Mode → the channel → the router. Without + it there is no way to originate a message the bot will react to. +""" + +from __future__ import annotations + +import asyncio +import json +import os +import time +import uuid +from collections import Counter +from dataclasses import dataclass, field +from typing import Any + +import pytest + +SETUP = """\ +Live Slack tests need a scratch workspace and these environment variables. + +Required (outbound): + NERVE_SLACK_TEST_BOT_TOKEN xoxb-… Bot User OAuth Token + NERVE_SLACK_TEST_APP_TOKEN xapp-… App-Level Token, connections:write + NERVE_SLACK_TEST_CHANNEL C… A channel the bot has been invited to + +Also required for the inbound tests: + NERVE_SLACK_TEST_USER_TOKEN xoxp-… User OAuth Token for a human account in + that workspace. Tests post as this user + so the bot receives a genuine event. + Needs chat:write and im:write. + +Optional: + NERVE_SLACK_TEST_BOT_TOKEN_NO_EMAIL + xoxb-… A second bot token installed WITHOUT + users:read.email. Proves Slack answers + users.info successfully while omitting + the email — the premise the deny-list + fail-closed rule rests on. + +The bot app needs the scopes from the manifest in docs/config.md, and its +App Home "Messages Tab" must be on with "Allow users to send Slash commands +and messages" ticked — otherwise the DM conversation is read-only and Slack +refuses the direct-message test with restricted_action_read_only_channel. + +Use a throwaway workspace: these tests post, edit, react, and upload. +""" + +BOT_TOKEN = os.environ.get("NERVE_SLACK_TEST_BOT_TOKEN", "") +APP_TOKEN = os.environ.get("NERVE_SLACK_TEST_APP_TOKEN", "") +TEST_CHANNEL = os.environ.get("NERVE_SLACK_TEST_CHANNEL", "") +USER_TOKEN = os.environ.get("NERVE_SLACK_TEST_USER_TOKEN", "") +NO_EMAIL_BOT_TOKEN = os.environ.get("NERVE_SLACK_TEST_BOT_TOKEN_NO_EMAIL", "") + +HAVE_OUTBOUND = bool(BOT_TOKEN and APP_TOKEN and TEST_CHANNEL) +HAVE_INBOUND = HAVE_OUTBOUND and bool(USER_TOKEN) + +requires_outbound = pytest.mark.skipif( + not HAVE_OUTBOUND, + reason="live Slack outbound creds not set (see tests/slack_live.py SETUP)", +) +requires_inbound = pytest.mark.skipif( + not HAVE_INBOUND, + reason="NERVE_SLACK_TEST_USER_TOKEN not set (see tests/slack_live.py SETUP)", +) +requires_no_email_token = pytest.mark.skipif( + not NO_EMAIL_BOT_TOKEN, + reason="NERVE_SLACK_TEST_BOT_TOKEN_NO_EMAIL not set", +) + +# How long to wait for an event to travel Slack → Socket Mode → the router. +# Slack is usually well under a second; the ceiling is for a slow round trip +# rather than an expected wait, since every helper polls. +EVENT_TIMEOUT = 20.0 + +# How long to let a closed Socket Mode connection disappear before the +# next test opens one. Slack gives each event to exactly one of an app's +# connections, so an overlap silently steals events from the new test. +SOCKET_DRAIN_SECONDS = 2.5 + +# After connecting, wait until the socket has been this quiet before treating +# the next arrival as the test's own. Slack retries events that had no listener +# on a fixed schedule, so a test posting amid a retry burst can otherwise read +# someone else's exhaust. +SOCKET_QUIET_SECONDS = 1.5 + + +def live_diagnostic(event: str, **fields: Any) -> None: + """Emit one machine-readable, credential-free live-test diagnostic.""" + payload = { + "event": event, + "pid": os.getpid(), + "time": round(time.time(), 3), + **fields, + } + print(f"SLACK_LIVE {json.dumps(payload, sort_keys=True)}", flush=True) + + +@dataclass +class SocketDiagnostics: + """Observe a live-test socket without changing its acknowledgement path.""" + + label: str + started_at: float = field(default_factory=time.monotonic) + connections: int = 0 + envelopes: int = 0 + fresh: int = 0 + retry_attempts: Counter[int] = field(default_factory=Counter) + event_types: Counter[str] = field(default_factory=Counter) + delayed_envelopes: int = 0 + max_event_age_seconds: float = 0.0 + _clients: int = 0 + + def emit(self, event: str, **fields: Any) -> None: + live_diagnostic(event, socket=self.label, **fields) + + def attach(self, socket) -> None: + """Attach passive listeners before *socket* connects.""" + self._clients += 1 + client_number = self._clients + + async def observe_message(_client, message, _raw): + kind = message.get("type") + if kind == "hello": + self.connections += 1 + self.emit( + "socket_hello", + client=client_number, + num_connections=message.get("num_connections"), + host=(message.get("debug_info") or {}).get("host"), + ) + async def observe_request(_client, req): + self.envelopes += 1 + payload = req.payload or {} + slack_event = payload.get("event") or {} + event_type = slack_event.get("type") or "none" + subtype = slack_event.get("subtype") + kind = f"{req.type}/{event_type}" + if subtype: + kind += f"/{subtype}" + self.event_types[kind] += 1 + + attempt = req.retry_attempt or 0 + if attempt: + self.retry_attempts[attempt] += 1 + else: + self.fresh += 1 + + event_age = None + timestamp_source = "event_time" + event_ts = payload.get("event_time") + if event_ts is None: + timestamp_source = "event_ts" + event_ts = slack_event.get("event_ts") + if event_ts is None: + timestamp_source = "message_ts" + event_ts = slack_event.get("ts") + try: + event_age = max(0.0, time.time() - float(event_ts)) + self.max_event_age_seconds = max( + self.max_event_age_seconds, event_age, + ) + if event_age > 5.0: + self.delayed_envelopes += 1 + except (TypeError, ValueError): + pass + + if attempt: + self.emit( + "retry_envelope", + attempt=attempt, + reason=req.retry_reason, + request_type=req.type, + event_type=event_type, + event_subtype=subtype, + event_age_seconds=( + round(event_age, 3) if event_age is not None else None + ), + ) + elif event_age is not None and event_age > 5.0: + self.emit( + "delayed_unmarked_envelope", + request_type=req.type, + event_type=event_type, + event_subtype=subtype, + event_age_seconds=round(event_age, 3), + timestamp_source=timestamp_source, + ) + + socket.message_listeners.append(observe_message) + socket.socket_mode_request_listeners.append(observe_request) + socket._live_diagnostics = self + socket._live_diagnostics_request_listener = observe_request + self.emit("socket_client_built", client=client_number) + + def emit_summary(self) -> None: + self.emit( + "socket_summary", + duration_seconds=round(time.monotonic() - self.started_at, 3), + clients=self._clients, + connections=self.connections, + envelopes=self.envelopes, + fresh=self.fresh, + retried=sum(self.retry_attempts.values()), + retry_attempts=dict(sorted(self.retry_attempts.items())), + event_types=dict(sorted(self.event_types.items())), + delayed_envelopes=self.delayed_envelopes, + max_event_age_seconds=round(self.max_event_age_seconds, 3), + ) + + + +def unique_marker() -> str: + """A token no other test or earlier run will carry.""" + return f"nvz{uuid.uuid4().hex[:10]}" + + +def make_client(token: str): + """A Web API client with 429 retry, so a slow test does not go flaky.""" + from slack_sdk.http_retry.builtin_async_handlers import ( + AsyncRateLimitErrorRetryHandler, + ) + from slack_sdk.web.async_client import AsyncWebClient + + client = AsyncWebClient(token=token) + client.retry_handlers.append(AsyncRateLimitErrorRetryHandler(max_retry_count=5)) + return client + + +@dataclass +class Posted: + """Messages the test created, so they can be cleaned up afterwards.""" + + bot: list[tuple[str, str]] = field(default_factory=list) # (channel, ts) + user: list[tuple[str, str]] = field(default_factory=list) + + def note_bot(self, channel: str, ts: str | None) -> None: + if ts: + self.bot.append((channel, ts)) + + def note_user(self, channel: str, ts: str | None) -> None: + if ts: + self.user.append((channel, ts)) + + +class RecordingRouter: + """A ChannelRouter stand-in that records what the channel hands it. + + The live tests are about the transport and the guardrails, so the engine + is not involved: this captures each InboundMessage and optionally posts a + reply through the channel, which is what the real router's stream adapter + would end up doing. + """ + + def __init__(self, reply_text: str | None = None): + self.messages: list[Any] = [] + self.reply_text = reply_text + self.channel: Any = None + self._sessions: dict[str, str] = {} + self._arrived = asyncio.Event() + + async def handle_message(self, msg: Any) -> str: + self._sessions.setdefault(msg.channel_key, f"s{len(self._sessions)}") + if self.reply_text and self.channel is not None: + from nerve.channels.base import OutboundMessage + + await self.channel.send( + OutboundMessage(target=msg.sender_id, text=self.reply_text), + ) + # Recorded last, so a test woken by wait_for_message can rely on the + # reply already being posted. Recording first let a test read the + # thread before the bot had answered it. + self.messages.append(msg) + self._arrived.set() + return self.reply_text or "ok" + + async def get_last_session(self, channel_key: str) -> str | None: + return self._sessions.get(channel_key) + + def _matching(self, marker: str) -> list[Any]: + return [m for m in self.messages if marker in (m.text or "")] + + async def wait_for_message( + self, marker: str, timeout: float = EVENT_TIMEOUT, + ) -> Any: + """Block until a message carrying *marker* is routed, and return it. + + Tests match on their own marker rather than on "something arrived". + A live workspace is not a clean room: Slack redelivers an envelope it + thinks went unacked, so a run that was interrupted can push an old + message into a later run and make an unrelated test fail. + """ + deadline = asyncio.get_running_loop().time() + timeout + while asyncio.get_running_loop().time() < deadline: + found = self._matching(marker) + if found: + return found[-1] + await asyncio.sleep(0.1) + raise AssertionError( + f"no inbound message carrying {marker!r} reached the router " + f"within {timeout}s (saw {[m.text for m in self.messages]})", + ) + + async def expect_no_message( + self, marker: str, settle: float = 6.0, + ) -> None: + """Assert nothing carrying *marker* is routed within *settle* seconds. + + The guardrail tests turn on this: a real Slack event reaches the + socket and must go no further. + """ + await asyncio.sleep(settle) + found = self._matching(marker) + assert not found, ( + f"expected the message to be refused, but the router received " + f"{found[-1].text!r}" + ) + + +async def wait_until_quiet( + channel, quiet_for: float = SOCKET_QUIET_SECONDS, timeout: float = 120.0, +) -> None: + """Block until no envelope has arrived for *quiet_for* seconds. + + Adaptive where a fixed sleep is not: instant on a clean socket, and + patient while scheduled retries are arriving. The timeout is generous + because a retry burst is exactly when this matters, and returning early + leaves it streaming — which is how a test came to read a message posted + by the one before it. + + Raises rather than returning quietly on timeout. Giving up silently + turns a socket that never settles into a confusing assertion three + tests later. + """ + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + started = loop.time() + diagnostics = getattr(channel._client, "_live_diagnostics", None) + while loop.time() < deadline: + if time.monotonic() - channel._last_event_time >= quiet_for: + if diagnostics: + diagnostics.emit( + "socket_quiet", + quiet_for_seconds=quiet_for, + waited_seconds=round(loop.time() - started, 3), + ) + return + await asyncio.sleep(0.2) + if diagnostics: + diagnostics.emit( + "socket_quiet_timeout", + quiet_for_seconds=quiet_for, + waited_seconds=round(loop.time() - started, 3), + ) + raise AssertionError( + f"the Slack socket never went quiet for {quiet_for}s within " + f"{timeout}s — scheduled retries are still arriving", + ) + + +async def start_event_sink( + bot_token: str = BOT_TOKEN, + app_token: str = APP_TOKEN, + diagnostics_label: str | None = None, +): + """Open a Socket Mode client that acknowledges and discards every envelope. + + Slack retries events that have no listener, and a run of the outbound live + tests produces dozens of them. Those retries used to make the next run's + inbound socket intermittently deaf to fresh messages. Keeping this client + beside the outbound tests prevents that retry schedule at its source. + """ + from slack_sdk.socket_mode.aiohttp import SocketModeClient + from slack_sdk.socket_mode.response import SocketModeResponse + + socket = SocketModeClient( + app_token=app_token, + web_client=make_client(bot_token), + auto_reconnect_enabled=True, + ) + + async def acknowledge(client, req): + await client.send_socket_mode_response( + SocketModeResponse(envelope_id=req.envelope_id), + ) + + socket.socket_mode_request_listeners.append(acknowledge) + if diagnostics_label: + SocketDiagnostics(diagnostics_label).attach(socket) + await socket.connect() + diagnostics = getattr(socket, "_live_diagnostics", None) + if diagnostics: + diagnostics.emit("socket_connect_returned") + return socket + + +async def wait_until_receiving( + socket, timeout: float = 180.0, web_client=None, + probe_interval: float = 15.0, +) -> None: + """Block until Slack is delivering *fresh* events to this connection. + + Slack does not hold undelivered events in a queue; it retries them on a + schedule — immediately, then again at +60s and ~+5min, each marked with + a ``retry_attempt``. While retries are outstanding a newly opened socket + is deaf to new events for roughly 20-30 seconds: anything posted in that + window misses the first attempts and only comes back a minute later. + + So readiness cannot be "an envelope arrived". A retried envelope is at + least a minute old and often belongs to an earlier run, and accepting + one declares the socket ready while it is still black-holing. This posts + a probe and waits for *that message's own ts*, unretried, re-probing + until it lands. It also waits for the probe's deletion event, making the + round trip a teardown fence: once this returns, it has not left its own + final mutation unacknowledged behind it. + """ + client = web_client or socket.web_client + loop = asyncio.get_running_loop() + started = loop.time() + deadline = started + timeout + diagnostics = getattr(socket, "_live_diagnostics", None) + seen_any: set[str] = set() + seen_fresh: set[str] = set() + seen_fresh_deletes: set[str] = set() + arrivals: dict[str, tuple[float, int, str | None]] = {} + deletion_arrivals: dict[str, float] = {} + probe_posted: dict[str, float] = {} + probes: list[str] = [] + deleted: set[str] = set() + deletion_sent: dict[str, float] = {} + completed = False + + if diagnostics: + diagnostics.emit("delivery_barrier_started", timeout_seconds=timeout) + + # Wrap the listener list rather than channel._on_request: attribute + # access builds a fresh bound method each time, so an identity check + # against one never matches and the swap silently does nothing. + listeners = socket.socket_mode_request_listeners + installed = list(listeners) + + async def watch(sock, req): + payload = req.payload or {} + event = payload.get("event") or {} + if event.get("ts"): + seen_any.add(event["ts"]) + arrivals.setdefault( + event["ts"], + (loop.time(), req.retry_attempt or 0, req.retry_reason), + ) + if not req.retry_attempt: + if event.get("ts"): + seen_fresh.add(event["ts"]) + if event.get("deleted_ts"): + seen_fresh_deletes.add(event["deleted_ts"]) + deletion_arrivals.setdefault(event["deleted_ts"], loop.time()) + for fn in installed: + await fn(sock, req) + + listeners[:] = [watch] + try: + while loop.time() < deadline: + post_started = loop.time() + probe = await client.chat_postMessage( + channel=TEST_CHANNEL, text="nerve socket readiness probe", + ) + ts = probe["ts"] + probes.append(ts) + probe_posted[ts] = post_started + if diagnostics: + diagnostics.emit("probe_posted", probe=len(probes)) + settle = min(deadline, loop.time() + probe_interval) + while loop.time() < settle: + if ts in seen_fresh: + break + await asyncio.sleep(0.2) + if ts in seen_fresh: + break + if diagnostics: + arrived = arrivals.get(ts) + diagnostics.emit( + "probe_missed", + probe=len(probes), + waited_seconds=round(loop.time() - probe_posted[ts], 3), + arrived=arrived is not None, + retry_attempt=arrived[1] if arrived else None, + retry_reason=arrived[2] if arrived else None, + ) + else: + raise AssertionError( + f"Slack was still not delivering fresh events after {timeout}s. " + "A backlog of retries from an earlier run keeps a new socket " + "deaf for 20-30s; longer than that suggests something else.", + ) + + # A failed probe has already missed Slack's immediate attempts and is + # scheduled to come back at +60s. Closing while that retry is pending + # would make the readiness check itself poison the next run. Keep the + # socket open until every probe has arrived and been acknowledged. + while loop.time() < deadline and not set(probes) <= seen_any: + await asyncio.sleep(0.2) + missing = set(probes) - seen_any + if missing: + raise AssertionError( + f"Slack became ready, but {len(missing)} readiness probe(s) " + f"were still awaiting retry after {timeout}s", + ) + + if diagnostics: + for number, probe_ts in enumerate(probes, start=1): + arrived_at, retry_attempt, retry_reason = arrivals[probe_ts] + diagnostics.emit( + "probe_arrived", + probe=number, + latency_seconds=round( + arrived_at - probe_posted[probe_ts], 3, + ), + retry_attempt=retry_attempt, + retry_reason=retry_reason, + ) + + for probe_ts in probes: + deletion_sent[probe_ts] = loop.time() + await client.chat_delete(channel=TEST_CHANNEL, ts=probe_ts) + deleted.add(probe_ts) + + # Do not close a socket immediately after deleting the fence: the Web + # API response wins the race with its Socket Mode event. That race + # used to create one last retry after otherwise-clean fixture teardown. + while loop.time() < deadline and not deleted <= seen_fresh_deletes: + await asyncio.sleep(0.2) + missing = deleted - seen_fresh_deletes + if missing: + raise AssertionError( + f"Slack did not deliver {len(missing)} readiness-probe " + f"deletion event(s) within {timeout}s", + ) + completed = True + if diagnostics: + deletion_latencies = [ + max(0.0, deletion_arrivals[probe_ts] - deletion_sent[probe_ts]) + for probe_ts in deleted + ] + diagnostics.emit( + "delivery_barrier_complete", + duration_seconds=round(loop.time() - started, 3), + probes=len(probes), + missed_probes=max(0, len(probes) - 1), + max_deletion_latency_seconds=round( + max(deletion_latencies, default=0.0), 3, + ), + ) + return + finally: + for probe_ts in set(probes) - deleted: + try: + await client.chat_delete(channel=TEST_CHANNEL, ts=probe_ts) + except Exception: + pass + listeners[:] = installed + if diagnostics and not completed: + diagnostics.emit( + "delivery_barrier_failed", + duration_seconds=round(loop.time() - started, 3), + probes=len(probes), + probes_arrived=len(set(probes) & seen_any), + deletions_arrived=len(set(probes) & seen_fresh_deletes), + ) + + +def ignore_stale_events(channel) -> None: + """Make *channel* skip retries and events too old for the current test. + + Tests only ever wait for a message they just posted. A retried envelope, + or any Events API callback older than ``EVENT_TIMEOUT``, therefore belongs + to an earlier test or run. Slack has also delivered old callbacks without + ``retry_attempt`` metadata, so checking the documented top-level + ``event_time`` closes the hole that checking retry metadata alone leaves. + + Production does the opposite on purpose: a retry is how a message + survives a restart, so the channel must handle it there. This is a + test-harness concern only. + """ + from slack_sdk.socket_mode.response import SocketModeResponse + + listeners = channel._client.socket_mode_request_listeners + installed = list(listeners) + + async def drop_stale(sock, req): + payload = req.payload or {} + event_time = payload.get("event_time") + try: + too_old = time.time() - float(event_time) > EVENT_TIMEOUT + except (TypeError, ValueError): + too_old = False + + if req.retry_attempt or too_old: + await sock.send_socket_mode_response( + SocketModeResponse(envelope_id=req.envelope_id), + ) + observer = getattr( + sock, "_live_diagnostics_request_listener", None, + ) + if observer: + await observer(sock, req) + return + for fn in installed: + await fn(sock, req) + + listeners[:] = [drop_stale] + + +def build_channel( + router: RecordingRouter, + diagnostics_label: str | None = None, + **slack_kwargs, +): + """A SlackChannel wired to the live workspace with the given guardrails.""" + from nerve.channels.slack import SlackChannel + from nerve.config import NerveConfig, SlackConfig + + cfg = NerveConfig() + cfg.slack = SlackConfig( + enabled=True, + bot_token=BOT_TOKEN, + app_token=APP_TOKEN, + **slack_kwargs, + ) + channel = SlackChannel(lambda: cfg, router) # type: ignore[arg-type] + router.channel = channel + if diagnostics_label: + diagnostics = SocketDiagnostics(diagnostics_label) + build_socket_client = channel._build_socket_client + + def build_instrumented_socket( + *, app_token: str | None = None, web_client=None, + ): + socket = build_socket_client( + app_token=app_token, + web_client=web_client, + ) + diagnostics.attach(socket) + return socket + + channel._build_socket_client = build_instrumented_socket + channel._live_diagnostics = diagnostics + return channel, cfg diff --git a/tests/test_slack_integration.py b/tests/test_slack_integration.py index e1248405..3fe6f140 100644 --- a/tests/test_slack_integration.py +++ b/tests/test_slack_integration.py @@ -9,6 +9,8 @@ from __future__ import annotations import copy +import json +import time from unittest.mock import AsyncMock, MagicMock import pytest @@ -17,6 +19,11 @@ from nerve.channels.slack import SlackChannel from nerve.config import NerveConfig, SlackConfig from tests.fake_slack import FakeSlack +from tests.slack_live import ( + ignore_stale_events, + start_event_sink, + wait_until_receiving, +) def _config(**slack_kwargs) -> NerveConfig: @@ -51,6 +58,92 @@ async def _started(server: FakeSlack, monkeypatch, **slack_kwargs): @pytest.mark.asyncio class TestSocketMode: + async def test_an_ack_only_sink_consumes_events_without_dispatching( + self, slack, monkeypatch, capsys, + ): + slack.patch_client(monkeypatch) + sink = await start_event_sink( + "xoxb-fake", "xapp-fake", diagnostics_label="fake-sink", + ) + routed: list[str] = [] + + async def route(_client, req): + event = (req.payload or {}).get("event") or {} + if event.get("ts"): + routed.append(event["ts"]) + + sink.socket_mode_request_listeners.append(route) + channel = MagicMock() + channel._client = sink + ignore_stale_events(channel) + try: + class ProbeClient: + posts = 0 + + async def chat_postMessage(self, **kwargs): + self.posts += 1 + ts = f"1.{self.posts}" + event = { + "type": "message", "channel": kwargs["channel"], + "channel_type": "channel", "user": "U0BOT", + "ts": ts, "text": kwargs["text"], + } + if self.posts == 2: + # The first probe missed the immediate attempts and + # arrives only as a retry. Readiness must ack it before + # returning, or its next retry poisons a later run. + await slack.push_event( + {**event, "ts": "1.1"}, + retry_attempt=2, + retry_reason="timeout", + ) + await slack.push_event(event) + return {"ok": True, "ts": ts} + + async def chat_delete(self, **kwargs): + await slack.push_event({ + "type": "message", "subtype": "message_deleted", + "channel": kwargs["channel"], "ts": "2.1", + "deleted_ts": kwargs["ts"], + }) + return {"ok": True} + + await wait_until_receiving( + sink, + timeout=2.0, + web_client=ProbeClient(), + probe_interval=0.05, + ) + await slack.push("events_api", { + "event_time": time.time() - 30, + "event": {"type": "message", "ts": "3.1"}, + }) + await slack.settle() + assert len(slack.acks) == 5 + assert "1.1" not in routed, "a marked retry reached the router" + assert "3.1" not in routed, "an unmarked stale event reached the router" + finally: + await sink.close() + sink._live_diagnostics.emit_summary() + + output = capsys.readouterr().out + records = [ + json.loads(line.removeprefix("SLACK_LIVE ")) + for line in output.splitlines() + if line.startswith("SLACK_LIVE ") + ] + events = {record["event"] for record in records} + assert { + "socket_hello", + "retry_envelope", + "delayed_unmarked_envelope", + "probe_missed", + "delivery_barrier_complete", + "socket_summary", + } <= events + assert "xoxb-fake" not in output + assert "nerve socket readiness probe" not in output + async def test_the_channel_connects_and_learns_its_own_id( self, slack, monkeypatch, ): diff --git a/tests/test_slack_live.py b/tests/test_slack_live.py new file mode 100644 index 00000000..55db9590 --- /dev/null +++ b/tests/test_slack_live.py @@ -0,0 +1,346 @@ +"""Live Slack integration tests. + +Everything here skips unless the credentials in :data:`tests.slack_live.SETUP` +are present, so the ordinary suite and CI on a fork are unaffected. + +The unit tests already prove the channel is self-consistent. These exist for +the claims that only Slack can settle: whether a Block Kit payload is +accepted, whether an emoji short name exists, whether ``users.info`` really +withholds an email rather than failing, and whether an event makes the whole +trip from a human's keystroke to an InboundMessage. + +Run just these with:: + + pytest tests/test_slack_live.py -v +""" + +from __future__ import annotations + +import asyncio +from types import SimpleNamespace + +import pytest +import pytest_asyncio + +from nerve.channels.slack import ( + _EMOJI_TO_SLACK, + MAX_MSG_LEN, + _md_to_slack, + build_notification_blocks, + format_target, + is_slack_id, + split_message, + SlackChannel, +) +from tests.slack_live import ( + BOT_TOKEN, + HAVE_OUTBOUND, + NO_EMAIL_BOT_TOKEN, + TEST_CHANNEL, + USER_TOKEN, + Posted, + RecordingRouter, + build_channel, + make_client, + requires_no_email_token, + requires_outbound, + start_event_sink, + wait_until_receiving, +) + +# One event loop for the whole module. The fixtures below hold aiohttp +# sessions, and a per-function loop leaves those bound to a loop that has +# already closed. +pytestmark = pytest.mark.asyncio(loop_scope="module") + + +async def test_the_diagnostics_wrapper_preserves_socket_builder_arguments( + monkeypatch, +): + calls = [] + socket = SimpleNamespace( + message_listeners=[], + socket_mode_request_listeners=[], + ) + + def build_socket(_channel, *, app_token=None, web_client=None): + calls.append((app_token, web_client)) + return socket + + monkeypatch.setattr(SlackChannel, "_build_socket_client", build_socket) + channel, _ = build_channel( + RecordingRouter(), diagnostics_label="wrapper-test", + ) + web_client = object() + + assert channel._build_socket_client( + app_token="xapp-test", web_client=web_client, + ) is socket + assert calls == [("xapp-test", web_client)] + + +# ---------------------------------------------------------------------- # +# Fixtures # +# ---------------------------------------------------------------------- # + + +@pytest_asyncio.fixture(scope="module", loop_scope="module", autouse=True) +async def _ack_outbound_events(): + """Keep this module's Web API mutations from poisoning a later run. + + Each post, edit, reaction, upload, and cleanup can produce a Socket Mode + envelope. With no socket open Slack retries those envelopes at +60s and + again around +5min; the pending retries are what made a later inbound + connection miss fresh events. The sink shares this module's lifecycle, + so it is connected before the first mutation and closes after cleanup. + """ + if not HAVE_OUTBOUND: + yield + return + + sink = await start_event_sink(diagnostics_label="outbound") + try: + # A WebSocket handshake is not sufficient when an older retry schedule + # exists. Prove Slack is routing fresh events here before tests post. + await wait_until_receiving(sink) + yield + # Fence fixture cleanup as well: Posted deletes messages after the + # tests, and closing before those events arrive would recreate the + # exact backlog this fixture exists to prevent. + await wait_until_receiving(sink) + finally: + await sink.close() + sink._live_diagnostics.emit_summary() + + +@pytest_asyncio.fixture(loop_scope="module") +async def bot(): + """A Web API client on the bot token.""" + if not BOT_TOKEN: + pytest.skip("no bot token") + yield make_client(BOT_TOKEN) + + +@pytest_asyncio.fixture(loop_scope="module") +async def human(): + """A Web API client on the user token, for posting as a person.""" + if not USER_TOKEN: + pytest.skip("no user token") + yield make_client(USER_TOKEN) + + +@pytest_asyncio.fixture(loop_scope="module") +async def posted(bot, _ack_outbound_events): + """Track messages the test creates and delete them afterwards.""" + tracker = Posted() + yield tracker + for token, entries in ((BOT_TOKEN, tracker.bot), (USER_TOKEN, tracker.user)): + if not token or not entries: + continue + client = make_client(token) + for channel, ts in entries: + try: + await client.chat_delete(channel=channel, ts=ts) + except Exception: + pass # A test that already deleted it, or a stale ts. + + +# ---------------------------------------------------------------------- # +# Outbound — what Slack accepts # +# ---------------------------------------------------------------------- # + + +@requires_outbound +class TestSlackAcceptsWhatWeSend: + async def test_the_bot_authenticates_and_the_channel_id_is_real(self, bot): + auth = await bot.auth_test() + assert auth["ok"] + assert is_slack_id(auth["user_id"]), auth["user_id"] + info = await bot.conversations_info(channel=TEST_CHANNEL) + assert info["ok"], "the bot must be invited to NERVE_SLACK_TEST_CHANNEL" + + async def test_converted_markdown_survives_a_round_trip(self, bot, posted): + source = ( + "## Heading\n" + "**bold** and *italic* and `code`\n" + "- bullet one\n" + "- bullet two\n" + "[docs](https://example.com/a?x=1&y=2)\n" + "```\nliteral **not bold**\n```\n" + "a < b & c > d" + ) + sent = _md_to_slack(source) + resp = await bot.chat_postMessage(channel=TEST_CHANNEL, text=sent) + posted.note_bot(TEST_CHANNEL, resp["ts"]) + + history = await bot.conversations_history( + channel=TEST_CHANNEL, latest=resp["ts"], inclusive=True, limit=1, + ) + stored = history["messages"][0]["text"] + # Slack stores mrkdwn verbatim. A mismatch means the converter emitted + # something Slack rewrote, which is the bug the fake cannot see. + assert stored == sent + + async def test_a_long_reply_splits_into_messages_slack_accepts( + self, bot, posted, + ): + body = "\n".join(f"line {i} " + "x" * 60 for i in range(200)) + chunks = split_message(body, MAX_MSG_LEN) + assert len(chunks) > 1, "test needs a body that actually splits" + for chunk in chunks: + resp = await bot.chat_postMessage( + channel=TEST_CHANNEL, text=_md_to_slack(chunk), + ) + posted.note_bot(TEST_CHANNEL, resp["ts"]) + assert resp["ok"] + + async def test_a_notification_card_is_valid_block_kit(self, bot, posted): + blocks = build_notification_blocks( + "Deploy to production?", "n-live-1", + [("✅ Approve", "approve"), ("❌ Decline", "decline"), + ("💤 Snooze 24h", "snooze_24h")], + ) + resp = await bot.chat_postMessage( + channel=TEST_CHANNEL, text="Deploy to production?", blocks=blocks, + ) + posted.note_bot(TEST_CHANNEL, resp["ts"]) + assert resp["ok"] + + async def test_a_long_option_list_is_chunked_below_the_actions_limit( + self, bot, posted, + ): + # Slack rejects the whole message with invalid_blocks past 25 + # elements in one actions block. Its validator is the authority here, + # not our arithmetic. + options = [(f"Option {i}", f"v{i}") for i in range(60)] + blocks = build_notification_blocks("Pick one", "n-live-2", options) + resp = await bot.chat_postMessage( + channel=TEST_CHANNEL, text="Pick one", blocks=blocks, + ) + posted.note_bot(TEST_CHANNEL, resp["ts"]) + assert resp["ok"] + + async def test_clearing_blocks_removes_the_buttons(self, bot, posted): + # The notification-expiry path relies on blocks=[] dropping the dead + # buttons rather than being ignored as falsy. + blocks = build_notification_blocks( + "Answer me", "n-live-3", [("Yes", "yes"), ("No", "no")], + ) + resp = await bot.chat_postMessage( + channel=TEST_CHANNEL, text="Answer me", blocks=blocks, + ) + posted.note_bot(TEST_CHANNEL, resp["ts"]) + + await bot.chat_update( + channel=TEST_CHANNEL, ts=resp["ts"], + text="Answer me\n\n⏰ Expired unanswered", blocks=[], + ) + history = await bot.conversations_history( + channel=TEST_CHANNEL, latest=resp["ts"], inclusive=True, limit=1, + ) + message = history["messages"][0] + # Slack does not leave the message block-less: it synthesises a + # rich_text block from the new text. What must be gone is the + # actions block, because that is what carries the dead buttons. + kinds = {b.get("type") for b in message.get("blocks") or []} + assert "actions" not in kinds, f"expired card still has buttons: {kinds}" + assert "Expired" in message["text"] + + async def test_every_mapped_emoji_short_name_exists(self, bot, posted): + """Every entry in the emoji table must be a name Slack knows. + + The table is hand-written, and a wrong short name fails at + ``reactions.add`` with ``invalid_name``, which the production path + swallows — the agent's reaction just never appears. + + Reactions are spread over several anchor messages because Slack caps + the distinct reactions on one message at about two dozen, and one + test reports every bad name at once rather than making you re-run to + find the next. + """ + per_anchor = 15 + items = sorted(set(_EMOJI_TO_SLACK.items()), key=lambda kv: kv[1]) + rejected: list[str] = [] + + for start in range(0, len(items), per_anchor): + anchor = await bot.chat_postMessage( + channel=TEST_CHANNEL, text=f"emoji probe {start}", + ) + posted.note_bot(TEST_CHANNEL, anchor["ts"]) + for emoji, name in items[start:start + per_anchor]: + try: + await bot.reactions_add( + channel=TEST_CHANNEL, timestamp=anchor["ts"], name=name, + ) + except Exception as exc: + if "already_reacted" not in str(exc): + rejected.append(f"{emoji} → :{name}: ({exc})") + await asyncio.sleep(0.3) + + assert not rejected, "Slack rejected these reactions:\n" + "\n".join(rejected) + + async def test_a_file_upload_lands_in_the_conversation( + self, bot, tmp_path, posted, + ): + from nerve.channels.slack import SlackChannel + from nerve.config import NerveConfig, SlackConfig + + path = tmp_path / "report.txt" + path.write_text("nerve live upload\n", encoding="utf-8") + + cfg = NerveConfig() + cfg.slack = SlackConfig(bot_token=BOT_TOKEN, app_token="unused") + channel = SlackChannel(lambda: cfg, router=None) # type: ignore[arg-type] + channel._web = bot + assert await channel.send_file(format_target(TEST_CHANNEL), str(path)) + + async def test_streaming_edits_then_removes_the_placeholder( + self, bot, posted, + ): + placeholder = await bot.chat_postMessage(channel=TEST_CHANNEL, text="⏳") + ts = placeholder["ts"] + for fragment in ("partial one", "partial one and two"): + await bot.chat_update( + channel=TEST_CHANNEL, ts=ts, text=_md_to_slack(fragment), + ) + await asyncio.sleep(1.2) # the per-channel chat.update limit + final = await bot.chat_postMessage(channel=TEST_CHANNEL, text="final answer") + posted.note_bot(TEST_CHANNEL, final["ts"]) + await bot.chat_delete(channel=TEST_CHANNEL, ts=ts) + + history = await bot.conversations_history( + channel=TEST_CHANNEL, latest=ts, inclusive=True, limit=1, + ) + assert not history["messages"] or history["messages"][0]["ts"] != ts + + +# ---------------------------------------------------------------------- # +# Scope behaviour — the premise the deny-list rule rests on # +# ---------------------------------------------------------------------- # + + +@requires_outbound +@requires_no_email_token +class TestScopeOmission: + async def test_users_info_omits_email_without_the_scope_instead_of_failing( + self, bot, + ): + """A token without ``users:read.email`` still gets a 200. + + This is why an email deny rule cannot be trusted on a short response: + the absent field looks exactly like a user who has no email, so the + rule matches nothing and would admit the person it names. The channel + refuses instead, and this test is what says that premise is real. + """ + auth = await bot.auth_test() + with_scope = await bot.users_info(user=auth["user_id"]) + assert with_scope["user"]["profile"].get("email"), ( + "the main bot token needs users:read.email for this comparison" + ) + + limited = make_client(NO_EMAIL_BOT_TOKEN) + response = await limited.users_info(user=auth["user_id"]) + assert response["ok"], "expected a successful response, not an error" + assert not response["user"]["profile"].get("email"), ( + "the no-email token returned an email; it still has the scope" + ) diff --git a/tests/test_slack_live_inbound.py b/tests/test_slack_live_inbound.py new file mode 100644 index 00000000..d4829248 --- /dev/null +++ b/tests/test_slack_live_inbound.py @@ -0,0 +1,421 @@ +"""Live Slack tests for the inbound half — events reaching the router. + +Everything here skips unless the credentials in :data:`tests.slack_live.SETUP` +are present, so the ordinary suite and CI on a fork are unaffected. + +The unit tests already prove the channel is self-consistent. These exist for +the claims that only Slack can settle: whether a Block Kit payload is +accepted, whether an emoji short name exists, whether ``users.info`` really +withholds an email rather than failing, and whether an event makes the whole +trip from a human's keystroke to an InboundMessage. + +These hold a Socket Mode connection, which is why they are kept apart from +the outbound tests. Slack gives each event to exactly one of an app's open +connections. The outbound module uses its own ack-only socket so its Web API +traffic cannot schedule future retries; a separate process ensures that sink +can never steal one of the inbound events asserted here. + +Run with:: + + pytest tests/test_slack_live_inbound.py -v + +If a test here reports "no inbound message reached the router", suspect the +connection rather than the routing, and prefer diagnosing it to muting it — +every time this suite has looked flaky it has been describing something real. +The two causes found so far were a connection per test, which lost events +into the gap where Slack had not yet started routing to the new socket, and +outbound API traffic with no listener, whose scheduled retries made a later +socket deaf. Hence a single shared connection here, and an ack-only connection +around the outbound file. +""" + +from __future__ import annotations + +import asyncio +import time + +import pytest +import pytest_asyncio + +from nerve.channels.slack import format_target +from tests.slack_live import ( + BOT_TOKEN, + EVENT_TIMEOUT, + TEST_CHANNEL, + USER_TOKEN, + Posted, + RecordingRouter, + SOCKET_DRAIN_SECONDS, + build_channel, + make_client, + wait_until_quiet, + ignore_stale_events, + wait_until_receiving, + requires_inbound, + requires_outbound, + unique_marker, +) + +# One event loop for the whole module. The fixtures below hold aiohttp +# sessions, and a per-function loop leaves those bound to a loop that has +# already closed. +pytestmark = pytest.mark.asyncio(loop_scope="module") + + +# ---------------------------------------------------------------------- # +# Fixtures # +# ---------------------------------------------------------------------- # + + +@pytest_asyncio.fixture(loop_scope="module") +async def bot(): + """A Web API client on the bot token.""" + if not BOT_TOKEN: + pytest.skip("no bot token") + yield make_client(BOT_TOKEN) + + +@pytest_asyncio.fixture(loop_scope="module") +async def human(): + """A Web API client on the user token, for posting as a person.""" + if not USER_TOKEN: + pytest.skip("no user token") + yield make_client(USER_TOKEN) + + +@pytest_asyncio.fixture(loop_scope="module") +async def posted(bot, _connected_channel): + """Track messages the test creates and delete them afterwards.""" + tracker = Posted() + yield tracker + for token, entries in ((BOT_TOKEN, tracker.bot), (USER_TOKEN, tracker.user)): + if not token or not entries: + continue + client = make_client(token) + for channel, ts in entries: + try: + await client.chat_delete(channel=channel, ts=ts) + except Exception: + pass # A test that already deleted it, or a stale ts. + + +@pytest_asyncio.fixture(scope="module", loop_scope="module") +async def _connected_channel(): + """One Socket Mode connection for the whole module. + + Slack does not start routing to a freshly-opened connection the instant + the handshake completes, and it hands each event to exactly one of an + app's connections. Opening one per test therefore loses events into the + gap: a message posted by one test would surface in the next test's + router, which is what CI kept catching. + + Sharing one connection removes the churn. It works because the channel + resolves its config per event, so a test can retarget the guardrails + between messages — the same property a config reload depends on. + + This needs the outbound tests to live in another process. Their ack-only + socket would otherwise be a second connection competing for exactly the + inbound events this module asserts. + """ + channel, cfg = build_channel( + RecordingRouter(), diagnostics_label="inbound", + ) + await channel.start() + # Order matters. Old events are dropped first so a readiness retry cannot + # reach the router, then readiness waits for the probe's own ts to prove + # Slack is delivering fresh events to this socket. + ignore_stale_events(channel) + await wait_until_receiving(channel._client) + await wait_until_quiet(channel) + try: + yield channel, cfg + finally: + # The Posted fixture deletes its messages before this module-scoped + # connection tears down. Fence those final mutations so closing the + # socket cannot seed the next run with a retry schedule. + try: + await wait_until_receiving(channel._client) + finally: + await channel.stop() + channel._live_diagnostics.emit_summary() + + +@pytest_asyncio.fixture(loop_scope="module") +async def live_channel(_connected_channel): + """Point the shared channel at this test's router and guardrails.""" + channel, cfg = _connected_channel + + async def _use(router: RecordingRouter, **slack_kwargs): + assert await channel._client.is_connected(), ( + "the shared Socket Mode connection went down before this test" + ) + # Let the previous test's events land before moving the goalposts. + # Retargeting the router and the guardrails while an event is still + # in flight judges it under the wrong policy: a message posted by a + # test that expects a refusal would be admitted by the next test's + # allow list, and surface in the next test's router. CI is slower + # than a laptop, so it saw this where local runs did not. + await wait_until_quiet(channel) + for field, default in ( + ("allow_users", []), ("deny_users", []), + ("allow_channels", []), ("deny_channels", []), + ("reply_in_thread", True), ("commands", None), + ): + setattr(cfg.slack, field, slack_kwargs.get(field, default)) + channel.router = router + router.channel = channel + # Resolved identities are policy-specific, so a later test must not + # inherit a verdict computed under different lists. + channel._name_cache.clear() + return channel, cfg + + return _use + + +# ---------------------------------------------------------------------- # +# Inbound — the whole trip, human keystroke to InboundMessage # +# ---------------------------------------------------------------------- # + + +@requires_inbound +class TestInboundLoop: + async def test_a_mention_reaches_the_router_and_is_answered_in_thread( + self, live_channel, human, bot, posted, + ): + marker = unique_marker() + auth = await human.auth_test() + user_id = auth["user_id"] + router = RecordingRouter(reply_text="**live reply**") + channel, _ = await live_channel(router, allow_users=[user_id]) + + sent = await human.chat_postMessage( + channel=TEST_CHANNEL, + text=f"<@{channel._bot_user_id}> live mention test {marker}", + ) + posted.note_user(TEST_CHANNEL, sent["ts"]) + + msg = await router.wait_for_message(marker) + assert msg.channel_name == "slack" + assert msg.channel_key == f"slack:{TEST_CHANNEL}:{sent['ts']}" + assert msg.text == f"live mention test {marker}", ( + "the bot's own mention should be stripped before the prompt" + ) + + replies = await bot.conversations_replies( + channel=TEST_CHANNEL, ts=sent["ts"], + ) + bot_replies = [ + m for m in replies["messages"] if m.get("user") == channel._bot_user_id + ] + assert bot_replies, "the bot did not reply in the thread" + assert bot_replies[0]["text"] == "*live reply*" + posted.note_bot(TEST_CHANNEL, bot_replies[0]["ts"]) + + async def test_channel_chatter_without_a_mention_is_left_alone( + self, live_channel, human, posted, + ): + marker = unique_marker() + auth = await human.auth_test() + router = RecordingRouter() + channel, _ = await live_channel(router, allow_users=[auth["user_id"]]) + + sent = await human.chat_postMessage( + channel=TEST_CHANNEL, text=f"just talking to my colleagues {marker}", + ) + posted.note_user(TEST_CHANNEL, sent["ts"]) + await router.expect_no_message(marker) + + async def test_a_thread_reply_continues_without_another_mention( + self, live_channel, human, posted, + ): + marker = unique_marker() + auth = await human.auth_test() + user_id = auth["user_id"] + router = RecordingRouter() + channel, _ = await live_channel(router, allow_users=[user_id]) + + reply_marker = unique_marker() + opener = await human.chat_postMessage( + channel=TEST_CHANNEL, text=f"<@{channel._bot_user_id}> start a thread {marker}", + ) + posted.note_user(TEST_CHANNEL, opener["ts"]) + await router.wait_for_message(marker) + + follow = await human.chat_postMessage( + channel=TEST_CHANNEL, thread_ts=opener["ts"], text=f"and then? {reply_marker}", + ) + posted.note_user(TEST_CHANNEL, follow["ts"]) + msg = await router.wait_for_message(reply_marker) + assert msg.text == f"and then? {reply_marker}" + assert msg.channel_key == f"slack:{TEST_CHANNEL}:{opener['ts']}" + + async def test_a_direct_message_reaches_the_router( + self, live_channel, human, posted, + ): + marker = unique_marker() + auth = await human.auth_test() + user_id = auth["user_id"] + router = RecordingRouter() + channel, _ = await live_channel(router, allow_users=[user_id]) + + dm = await human.conversations_open(users=channel._bot_user_id) + dm_id = dm["channel"]["id"] + sent = await human.chat_postMessage(channel=dm_id, text=f"live dm test {marker}") + posted.note_user(dm_id, sent["ts"]) + + msg = await router.wait_for_message(marker) + assert msg.channel_key == f"slack:{dm_id}" + assert msg.text == f"live dm test {marker}" + + +@requires_inbound +class TestGuardrailsAgainstRealSlack: + async def test_a_denied_user_never_reaches_the_router( + self, live_channel, human, posted, + ): + marker = unique_marker() + auth = await human.auth_test() + user_id = auth["user_id"] + router = RecordingRouter() + # Allowed by id, then denied by id: deny must win. + channel, _ = await live_channel( + router, allow_users=[user_id], deny_users=[user_id], + ) + sent = await human.chat_postMessage( + channel=TEST_CHANNEL, text=f"<@{channel._bot_user_id}> let me in {marker}", + ) + posted.note_user(TEST_CHANNEL, sent["ts"]) + await router.expect_no_message(marker) + + async def test_an_unconfigured_policy_refuses_a_real_message( + self, live_channel, human, posted, + ): + marker = unique_marker() + router = RecordingRouter() + channel, _ = await live_channel(router) # no allow lists at all + assert not channel.policy.configured + sent = await human.chat_postMessage( + channel=TEST_CHANNEL, text=f"<@{channel._bot_user_id}> anyone home {marker}", + ) + posted.note_user(TEST_CHANNEL, sent["ts"]) + await router.expect_no_message(marker) + + async def test_allowing_by_handle_resolves_through_users_info( + self, live_channel, human, bot, posted, + ): + marker = unique_marker() + auth = await human.auth_test() + profile = await bot.users_info(user=auth["user_id"]) + handle = profile["user"]["name"] + + router = RecordingRouter() + channel, _ = await live_channel(router, allow_users=[handle]) + sent = await human.chat_postMessage( + channel=TEST_CHANNEL, text=f"<@{channel._bot_user_id}> handle test {marker}", + ) + posted.note_user(TEST_CHANNEL, sent["ts"]) + msg = await router.wait_for_message(marker) + assert msg.text == f"handle test {marker}" + + async def test_allowing_by_channel_name_glob_resolves_the_real_name( + self, live_channel, human, bot, posted, + ): + marker = unique_marker() + info = await bot.conversations_info(channel=TEST_CHANNEL) + name = info["channel"]["name"] + auth = await human.auth_test() + + router = RecordingRouter() + channel, _ = await live_channel( + router, allow_users=[auth["user_id"]], allow_channels=[f"{name[:3]}*"], + ) + sent = await human.chat_postMessage( + channel=TEST_CHANNEL, text=f"<@{channel._bot_user_id}> glob test {marker}", + ) + posted.note_user(TEST_CHANNEL, sent["ts"]) + msg = await router.wait_for_message(marker) + assert msg.text == f"glob test {marker}" + + async def test_a_channel_outside_the_glob_is_refused( + self, live_channel, human, posted, + ): + marker = unique_marker() + auth = await human.auth_test() + router = RecordingRouter() + channel, _ = await live_channel( + router, + allow_users=[auth["user_id"]], + allow_channels=["nerve-no-such-channel-*"], + ) + sent = await human.chat_postMessage( + channel=TEST_CHANNEL, text=f"<@{channel._bot_user_id}> should not pass {marker}", + ) + posted.note_user(TEST_CHANNEL, sent["ts"]) + await router.expect_no_message(marker) + + async def test_a_reaction_from_a_human_is_forwarded( + self, live_channel, human, bot, posted, + ): + marker = unique_marker() + auth = await human.auth_test() + user_id = auth["user_id"] + router = RecordingRouter() + channel, _ = await live_channel(router, allow_users=[user_id]) + + anchor = await bot.chat_postMessage( + channel=TEST_CHANNEL, text=f"react to me {marker}", + ) + posted.note_bot(TEST_CHANNEL, anchor["ts"]) + # The channel only forwards reactions on messages it still has + # context for, which is what the outbound cache is for. + channel._cache_message( + anchor["ts"], format_target(TEST_CHANNEL), f"react to me {marker}", + ) + await human.reactions_add( + channel=TEST_CHANNEL, timestamp=anchor["ts"], name="tada", + ) + msg = await router.wait_for_message(marker) + assert ":tada:" in msg.text + + +@requires_outbound +class TestReconnectWatchdog: + async def test_the_watchdog_restores_a_dropped_socket(self): + """Exercise reconnect only after tests that need exclusive delivery. + + This test owns its connection because it deliberately breaks it. Its + reconnect briefly overlaps the module's shared socket, and Slack may + hand an event to either connection, so running this test earlier can + perturb an otherwise-correct inbound assertion. + """ + import nerve.channels.slack as slack_module + + router = RecordingRouter() + channel, _ = build_channel( + router, + diagnostics_label="watchdog", + allow_users=["U0000000"], + ) + await channel.start() + original = slack_module.WATCHDOG_INTERVAL + slack_module.WATCHDOG_INTERVAL = 1 + try: + await channel._client.disconnect() + await asyncio.sleep(0.5) + assert not await channel._client.is_connected() + + deadline = time.monotonic() + EVENT_TIMEOUT + while time.monotonic() < deadline: + if await channel._client.is_connected(): + break + await asyncio.sleep(0.5) + assert await channel._client.is_connected(), ( + "the socket stayed down; the watchdog did not recover it" + ) + finally: + slack_module.WATCHDOG_INTERVAL = original + await channel.stop() + channel._live_diagnostics.emit_summary() + # Let Slack drop this connection before module fixture cleanup + # relies on the shared socket receiving every deletion event. + await asyncio.sleep(SOCKET_DRAIN_SECONDS) From ae0b9bc081259a96e3ab6c7411bed3a84d85cbfb Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Mon, 24 Aug 2026 13:43:54 +0200 Subject: [PATCH 2/5] Opt live Slack DMs in explicitly --- tests/slack_live.py | 8 ++++++++ tests/test_slack_live.py | 8 ++++++++ tests/test_slack_live_inbound.py | 14 +++++++++++++- 3 files changed, 29 insertions(+), 1 deletion(-) diff --git a/tests/slack_live.py b/tests/slack_live.py index c8201a8d..b94a3d1f 100644 --- a/tests/slack_live.py +++ b/tests/slack_live.py @@ -231,6 +231,14 @@ def unique_marker() -> str: return f"nvz{uuid.uuid4().hex[:10]}" +def direct_message_guardrails(user_id: str) -> dict[str, object]: + """The explicit access settings every live DM contract must use.""" + return { + "allow_users": [user_id], + "allow_direct_messages": True, + } + + def make_client(token: str): """A Web API client with 429 retry, so a slow test does not go flaky.""" from slack_sdk.http_retry.builtin_async_handlers import ( diff --git a/tests/test_slack_live.py b/tests/test_slack_live.py index 55db9590..67c2e85d 100644 --- a/tests/test_slack_live.py +++ b/tests/test_slack_live.py @@ -41,6 +41,7 @@ Posted, RecordingRouter, build_channel, + direct_message_guardrails, make_client, requires_no_email_token, requires_outbound, @@ -79,6 +80,13 @@ def build_socket(_channel, *, app_token=None, web_client=None): assert calls == [("xapp-test", web_client)] +async def test_live_direct_messages_use_the_explicit_guardrail(): + assert direct_message_guardrails("U123") == { + "allow_users": ["U123"], + "allow_direct_messages": True, + } + + # ---------------------------------------------------------------------- # # Fixtures # # ---------------------------------------------------------------------- # diff --git a/tests/test_slack_live_inbound.py b/tests/test_slack_live_inbound.py index d4829248..36e89f40 100644 --- a/tests/test_slack_live_inbound.py +++ b/tests/test_slack_live_inbound.py @@ -32,6 +32,7 @@ from __future__ import annotations import asyncio +import inspect import time import pytest @@ -47,6 +48,7 @@ RecordingRouter, SOCKET_DRAIN_SECONDS, build_channel, + direct_message_guardrails, make_client, wait_until_quiet, ignore_stale_events, @@ -158,6 +160,7 @@ async def _use(router: RecordingRouter, **slack_kwargs): await wait_until_quiet(channel) for field, default in ( ("allow_users", []), ("deny_users", []), + ("allow_direct_messages", False), ("allow_channels", []), ("deny_channels", []), ("reply_in_thread", True), ("commands", None), ): @@ -256,7 +259,9 @@ async def test_a_direct_message_reaches_the_router( auth = await human.auth_test() user_id = auth["user_id"] router = RecordingRouter() - channel, _ = await live_channel(router, allow_users=[user_id]) + channel, _ = await live_channel( + router, **direct_message_guardrails(user_id), + ) dm = await human.conversations_open(users=channel._bot_user_id) dm_id = dm["channel"]["id"] @@ -268,6 +273,13 @@ async def test_a_direct_message_reaches_the_router( assert msg.text == f"live dm test {marker}" +async def test_the_live_dm_contract_uses_the_explicit_guardrail(): + source = inspect.getsource( + TestInboundLoop.test_a_direct_message_reaches_the_router, + ) + assert "direct_message_guardrails(user_id)" in source + + @requires_inbound class TestGuardrailsAgainstRealSlack: async def test_a_denied_user_never_reaches_the_router( From bfb80ab658eedd279ea8f8b5e10df926fca44059 Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Tue, 25 Aug 2026 12:56:56 +0200 Subject: [PATCH 3/5] Align live Slack tests with runtime config ownership --- .github/workflows/slack-live.yml | 15 +++------------ tests/slack_live.py | 2 +- tests/test_slack_live.py | 10 ++++++---- tests/test_slack_live_inbound.py | 11 +++++++---- 4 files changed, 17 insertions(+), 21 deletions(-) diff --git a/.github/workflows/slack-live.yml b/.github/workflows/slack-live.yml index 22a5cf54..c1997ed7 100644 --- a/.github/workflows/slack-live.yml +++ b/.github/workflows/slack-live.yml @@ -14,18 +14,9 @@ on: branches: [main] workflow_dispatch: -# One run at a time, across every branch and pull request. -# -# cancel-in-progress is false on purpose, the opposite of ci.yml. Killing a -# run mid-flight leaves envelopes unacked and messages undeleted, and Slack -# redelivers an unacked envelope — so a cancelled run does not just lose its -# own result, it poisons the next one. Queueing is the cheaper failure: the -# suite takes about three minutes. -# -# Note the GitHub behaviour this cannot avoid: only one run may be pending -# per group, so a third run arriving while one is executing and one is queued -# cancels the queued one. It is not lost work, it is a skipped check, and -# re-running the job is enough. +# Serialize against the shared workspace and let active runs finish so +# unacknowledged envelopes cannot retry into the next job. GitHub keeps only +# one queued run per group; a displaced queued check can simply be rerun. concurrency: group: slack-live-workspace cancel-in-progress: false diff --git a/tests/slack_live.py b/tests/slack_live.py index b94a3d1f..63278b09 100644 --- a/tests/slack_live.py +++ b/tests/slack_live.py @@ -650,7 +650,7 @@ def build_channel( app_token=APP_TOKEN, **slack_kwargs, ) - channel = SlackChannel(lambda: cfg, router) # type: ignore[arg-type] + channel = SlackChannel(cfg, router) router.channel = channel if diagnostics_label: diagnostics = SocketDiagnostics(diagnostics_label) diff --git a/tests/test_slack_live.py b/tests/test_slack_live.py index 67c2e85d..2469b72e 100644 --- a/tests/test_slack_live.py +++ b/tests/test_slack_live.py @@ -23,14 +23,16 @@ import pytest_asyncio from nerve.channels.slack import ( + format_target, + is_slack_id, + SlackChannel, +) +from nerve.channels.slack_presentation import ( _EMOJI_TO_SLACK, MAX_MSG_LEN, _md_to_slack, build_notification_blocks, - format_target, - is_slack_id, split_message, - SlackChannel, ) from tests.slack_live import ( BOT_TOKEN, @@ -298,7 +300,7 @@ async def test_a_file_upload_lands_in_the_conversation( cfg = NerveConfig() cfg.slack = SlackConfig(bot_token=BOT_TOKEN, app_token="unused") - channel = SlackChannel(lambda: cfg, router=None) # type: ignore[arg-type] + channel = SlackChannel(cfg, router=None) # type: ignore[arg-type] channel._web = bot assert await channel.send_file(format_target(TEST_CHANNEL), str(path)) diff --git a/tests/test_slack_live_inbound.py b/tests/test_slack_live_inbound.py index 36e89f40..1308769c 100644 --- a/tests/test_slack_live_inbound.py +++ b/tests/test_slack_live_inbound.py @@ -32,6 +32,7 @@ from __future__ import annotations import asyncio +import copy import inspect import time @@ -145,7 +146,7 @@ async def _connected_channel(): @pytest_asyncio.fixture(loop_scope="module") async def live_channel(_connected_channel): """Point the shared channel at this test's router and guardrails.""" - channel, cfg = _connected_channel + channel, _ = _connected_channel async def _use(router: RecordingRouter, **slack_kwargs): assert await channel._client.is_connected(), ( @@ -158,19 +159,21 @@ async def _use(router: RecordingRouter, **slack_kwargs): # allow list, and surface in the next test's router. CI is slower # than a laptop, so it saw this where local runs did not. await wait_until_quiet(channel) + config = copy.deepcopy(channel.config) for field, default in ( ("allow_users", []), ("deny_users", []), ("allow_direct_messages", False), ("allow_channels", []), ("deny_channels", []), - ("reply_in_thread", True), ("commands", None), + ("commands", None), ): - setattr(cfg.slack, field, slack_kwargs.get(field, default)) + setattr(config.slack, field, slack_kwargs.get(field, default)) + channel.apply_config(config) channel.router = router router.channel = channel # Resolved identities are policy-specific, so a later test must not # inherit a verdict computed under different lists. channel._name_cache.clear() - return channel, cfg + return channel, config return _use From 98b0c478e394f0948f64edd5a22e08cfce46ae20 Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Tue, 25 Aug 2026 15:41:44 +0200 Subject: [PATCH 4/5] Make live refusal tests prove the refusal The four guardrail tests waited and asserted the router stayed empty. An event Slack never delivered, one it gave to another connection, and one the harness dropped as stale all look identical from there, so removing the policy altogether left them green. Wait for the envelope to reach the channel first, then assert nothing came out of it, and say which of the two failed. The settle after arrival covers dispatch rather than delivery, so it no longer sits at under a third of the delivery timeout. The staleness cutoff subtracted a stamp Slack wrote from a reading of the runner's clock. A runner ahead of Slack by more than the cutoff called every fresh event stale and dropped it, which on those same refusal tests reads as a guardrail doing its job. Measure the offset from a probe's own ts and treat an uncalibrated harness as having nothing stale. An upload is not a message, so chat.delete never removed the file the send_file test creates and every run added another to the scratch channel. Record file ids and remove them with files.delete. Track the streaming placeholder before the edits and the bot's reply before its text is checked, so a failure part way through leaves nothing behind. The integration test's probe client returned a counter where Slack returns epoch seconds, which left the cutoff it exercises uncalibrated. It also now asserts the fresh probe arrived, not only that the stale one did not. --- tests/slack_live.py | 131 ++++++++++++++++++++++++++++--- tests/test_slack_integration.py | 20 +++-- tests/test_slack_live.py | 127 ++++++++++++++++++++++++++++++ tests/test_slack_live_inbound.py | 12 +-- 4 files changed, 266 insertions(+), 24 deletions(-) diff --git a/tests/slack_live.py b/tests/slack_live.py index 63278b09..8d838273 100644 --- a/tests/slack_live.py +++ b/tests/slack_live.py @@ -100,6 +100,46 @@ # someone else's exhaust. SOCKET_QUIET_SECONDS = 1.5 +# After the envelope has reached the channel, how long to let dispatch run +# before calling a refusal a refusal. Dispatch starts off the ack path, so +# this covers the policy's own Slack lookups rather than network delivery, +# which the arrival wait has already accounted for. +REFUSAL_SETTLE_SECONDS = 3.0 + +# The runner's clock minus Slack's, measured from a real Slack timestamp. +# Event ages are the difference between a stamp Slack wrote and a reading of +# the local clock, which are two different clocks. A runner that has drifted +# ahead of Slack by more than EVENT_TIMEOUT would call every fresh event +# stale and drop it, and on the refusal tests that reads as a guardrail +# doing its job. WSL2 across a host suspend drifts by far more than that, so +# the offset is measured rather than assumed. +_clock_offset: float | None = None + + +def note_slack_timestamp(slack_ts: Any) -> None: + """Calibrate the local clock against one Slack-stamped timestamp.""" + global _clock_offset + try: + _clock_offset = time.time() - float(slack_ts) + except (TypeError, ValueError): + return + + +def event_age_seconds(event_time: Any) -> float | None: + """Age of an event on Slack's clock, or None if it cannot be told. + + None means the harness has no calibration or no usable stamp. Callers + treat that as "not stale": a stale event that slips through only makes a + test wait for a marker it will not match, while wrongly dropping a fresh + one hides the behavior under test. + """ + if _clock_offset is None: + return None + try: + return (time.time() - _clock_offset) - float(event_time) + except (TypeError, ValueError): + return None + def live_diagnostic(event: str, **fields: Any) -> None: """Emit one machine-readable, credential-free live-test diagnostic.""" @@ -125,11 +165,35 @@ class SocketDiagnostics: event_types: Counter[str] = field(default_factory=Counter) delayed_envelopes: int = 0 max_event_age_seconds: float = 0.0 + # Text of every envelope the harness handed to the channel, and of every + # one it dropped first. A refusal test needs the difference: an event the + # harness dropped as stale never met the policy, so a quiet router says + # nothing about the guardrail. + forwarded_texts: list[str] = field(default_factory=list) + dropped_texts: list[str] = field(default_factory=list) _clients: int = 0 def emit(self, event: str, **fields: Any) -> None: live_diagnostic(event, socket=self.label, **fields) + @staticmethod + def _text_of(req) -> str: + payload = req.payload or {} + return ((payload.get("event") or {}).get("text")) or "" + + def note_forwarded(self, req) -> None: + self.forwarded_texts.append(self._text_of(req)) + + def note_dropped(self, req) -> None: + self.dropped_texts.append(self._text_of(req)) + + def forwarded(self, marker: str) -> bool: + """Whether an envelope carrying *marker* reached the channel.""" + return any(marker in text for text in self.forwarded_texts) + + def dropped(self, marker: str) -> bool: + return any(marker in text for text in self.dropped_texts) + def attach(self, socket) -> None: """Attach passive listeners before *socket* connects.""" self._clients += 1 @@ -253,10 +317,16 @@ def make_client(token: str): @dataclass class Posted: - """Messages the test created, so they can be cleaned up afterwards.""" + """What the test created, so it can be cleaned up afterwards. + + An upload is not a message: chat.delete cannot remove it, so a file + needs its own id recorded and files.delete to take it away. Without + that, every run leaves another copy in the scratch channel for good. + """ bot: list[tuple[str, str]] = field(default_factory=list) # (channel, ts) user: list[tuple[str, str]] = field(default_factory=list) + files: list[str] = field(default_factory=list) # file ids def note_bot(self, channel: str, ts: str | None) -> None: if ts: @@ -266,6 +336,10 @@ def note_user(self, channel: str, ts: str | None) -> None: if ts: self.user.append((channel, ts)) + def note_file(self, file_id: str | None) -> None: + if file_id: + self.files.append(file_id) + class RecordingRouter: """A ChannelRouter stand-in that records what the channel hands it. @@ -326,14 +400,38 @@ async def wait_for_message( ) async def expect_no_message( - self, marker: str, settle: float = 6.0, + self, marker: str, channel, timeout: float = EVENT_TIMEOUT, ) -> None: - """Assert nothing carrying *marker* is routed within *settle* seconds. + """Assert the event met the policy and was refused by it. - The guardrail tests turn on this: a real Slack event reaches the - socket and must go no further. + Waiting and finding an empty router proves little on its own. An + event Slack never delivered, one it handed to another connection, + and one the harness dropped as stale all look the same from here. So + this first waits for the envelope to reach the channel, and only + then asserts nothing came out the far side. """ - await asyncio.sleep(settle) + diagnostics = getattr(channel._client, "_live_diagnostics", None) + assert diagnostics is not None, ( + "the live channel has no diagnostics, so a refusal cannot be " + "told apart from an event that never arrived" + ) + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while loop.time() < deadline: + if diagnostics.forwarded(marker): + break + await asyncio.sleep(0.1) + else: + dropped = diagnostics.dropped(marker) + raise AssertionError( + f"no envelope carrying {marker!r} reached the channel within " + f"{timeout}s, so the guardrail was never exercised " + f"({'the harness dropped it as stale' if dropped else 'Slack never delivered it'})", + ) + + # The envelope is in. Dispatch runs off the ack path, so give it room + # before concluding the policy stopped it. + await asyncio.sleep(REFUSAL_SETTLE_SECONDS) found = self._matching(marker) assert not found, ( f"expected the message to be refused, but the router received " @@ -489,10 +587,17 @@ async def watch(sock, req): channel=TEST_CHANNEL, text="nerve socket readiness probe", ) ts = probe["ts"] + # A message ts is Slack's own clock reading, which is what the + # staleness cutoff has to measure against. + note_slack_timestamp(ts) probes.append(ts) probe_posted[ts] = post_started if diagnostics: - diagnostics.emit("probe_posted", probe=len(probes)) + diagnostics.emit( + "probe_posted", + probe=len(probes), + clock_offset_seconds=round(_clock_offset or 0.0, 3), + ) settle = min(deadline, loop.time() + probe_interval) while loop.time() < settle: if ts in seen_fresh: @@ -612,12 +717,10 @@ def ignore_stale_events(channel) -> None: async def drop_stale(sock, req): payload = req.payload or {} - event_time = payload.get("event_time") - try: - too_old = time.time() - float(event_time) > EVENT_TIMEOUT - except (TypeError, ValueError): - too_old = False + age = event_age_seconds(payload.get("event_time")) + too_old = age is not None and age > EVENT_TIMEOUT + diagnostics = getattr(sock, "_live_diagnostics", None) if req.retry_attempt or too_old: await sock.send_socket_mode_response( SocketModeResponse(envelope_id=req.envelope_id), @@ -627,7 +730,11 @@ async def drop_stale(sock, req): ) if observer: await observer(sock, req) + if diagnostics: + diagnostics.note_dropped(req) return + if diagnostics: + diagnostics.note_forwarded(req) for fn in installed: await fn(sock, req) diff --git a/tests/test_slack_integration.py b/tests/test_slack_integration.py index 3fe6f140..523a229f 100644 --- a/tests/test_slack_integration.py +++ b/tests/test_slack_integration.py @@ -78,22 +78,26 @@ async def route(_client, req): ignore_stale_events(channel) try: class ProbeClient: - posts = 0 + # Slack stamps a message ts with the epoch second it was + # sent, and the harness calibrates its staleness clock from + # one, so a counter here would leave the cutoff blind. + def __init__(self): + self.sent: list[str] = [] async def chat_postMessage(self, **kwargs): - self.posts += 1 - ts = f"1.{self.posts}" + ts = f"{time.time():.6f}" + self.sent.append(ts) event = { "type": "message", "channel": kwargs["channel"], "channel_type": "channel", "user": "U0BOT", "ts": ts, "text": kwargs["text"], } - if self.posts == 2: + if len(self.sent) == 2: # The first probe missed the immediate attempts and # arrives only as a retry. Readiness must ack it before # returning, or its next retry poisons a later run. await slack.push_event( - {**event, "ts": "1.1"}, + {**event, "ts": self.sent[0]}, retry_attempt=2, retry_reason="timeout", ) @@ -108,10 +112,11 @@ async def chat_delete(self, **kwargs): }) return {"ok": True} + probe = ProbeClient() await wait_until_receiving( sink, timeout=2.0, - web_client=ProbeClient(), + web_client=probe, probe_interval=0.05, ) await slack.push("events_api", { @@ -120,7 +125,8 @@ async def chat_delete(self, **kwargs): }) await slack.settle() assert len(slack.acks) == 5 - assert "1.1" not in routed, "a marked retry reached the router" + assert probe.sent[0] not in routed, "a marked retry reached the router" + assert probe.sent[1] in routed, "the fresh probe never arrived" assert "3.1" not in routed, "an unmarked stale event reached the router" finally: await sink.close() diff --git a/tests/test_slack_live.py b/tests/test_slack_live.py index 2469b72e..082dda02 100644 --- a/tests/test_slack_live.py +++ b/tests/test_slack_live.py @@ -17,6 +17,7 @@ from __future__ import annotations import asyncio +import time from types import SimpleNamespace import pytest @@ -82,6 +83,113 @@ def build_socket(_channel, *, app_token=None, web_client=None): assert calls == [("xapp-test", web_client)] +@pytest.fixture(autouse=True) +def _keep_the_measured_clock_offset(): + """Leave the harness calibration as the live fixtures found it. + + It is module state measured from a real Slack timestamp, so a test that + fakes it must not hand a fabricated offset to the live tests. + """ + import tests.slack_live as harness + + saved = harness._clock_offset + yield + harness._clock_offset = saved + + +class TestStalenessCutoff: + """The cutoff has to survive a runner clock that disagrees with Slack.""" + + async def test_an_uncalibrated_harness_calls_nothing_stale(self, monkeypatch): + import tests.slack_live as harness + + monkeypatch.setattr(harness, "_clock_offset", None) + assert harness.event_age_seconds(time.time()) is None + + async def test_a_fresh_event_is_fresh_despite_a_skewed_runner(self, monkeypatch): + # The runner sits ten minutes ahead of Slack. Comparing the two + # clocks directly made every fresh event look stale, and a dropped + # event reads as a refusal on the guardrail tests. + import tests.slack_live as harness + + slack_now = 1_700_000_000.0 + monkeypatch.setattr(harness.time, "time", lambda: slack_now + 600.0) + harness.note_slack_timestamp(str(slack_now)) + age = harness.event_age_seconds(slack_now) + assert age is not None + assert abs(age) < 1.0 + assert age <= harness.EVENT_TIMEOUT + + async def test_an_event_from_an_earlier_run_is_still_stale(self, monkeypatch): + import tests.slack_live as harness + + slack_now = 1_700_000_000.0 + monkeypatch.setattr(harness.time, "time", lambda: slack_now + 600.0) + harness.note_slack_timestamp(str(slack_now)) + age = harness.event_age_seconds(slack_now - 3600.0) + assert age is not None + assert age > harness.EVENT_TIMEOUT + + async def test_a_missing_stamp_is_not_stale(self, monkeypatch): + import tests.slack_live as harness + + harness.note_slack_timestamp("1700000000.000100") + assert harness.event_age_seconds(None) is None + assert harness.event_age_seconds("not-a-number") is None + + +class TestRefusalControl: + """A quiet router only means a refusal if the event actually arrived.""" + + @staticmethod + def _req(text: str): + return SimpleNamespace( + payload={"event": {"text": text}}, + retry_attempt=0, + retry_reason=None, + type="events_api", + envelope_id="e1", + ) + + async def test_diagnostics_separate_forwarded_from_dropped(self): + from tests.slack_live import SocketDiagnostics + + diagnostics = SocketDiagnostics("control-test") + diagnostics.note_forwarded(self._req("hello nvz-aaa")) + diagnostics.note_dropped(self._req("stale nvz-bbb")) + + assert diagnostics.forwarded("nvz-aaa") + assert not diagnostics.dropped("nvz-aaa") + assert diagnostics.dropped("nvz-bbb") + # The distinction is the point: a dropped envelope never met the + # policy, so it cannot stand in for a refusal. + assert not diagnostics.forwarded("nvz-bbb") + assert not diagnostics.forwarded("nvz-never-sent") + + async def test_a_refusal_needs_the_envelope_to_have_arrived(self): + from tests.slack_live import SocketDiagnostics + + router = RecordingRouter() + diagnostics = SocketDiagnostics("control-test") + channel = SimpleNamespace( + _client=SimpleNamespace(_live_diagnostics=diagnostics), + ) + with pytest.raises(AssertionError, match="never delivered it"): + await router.expect_no_message("nvz-absent", channel, timeout=0.3) + + async def test_a_dropped_envelope_does_not_count_as_a_refusal(self): + from tests.slack_live import SocketDiagnostics + + router = RecordingRouter() + diagnostics = SocketDiagnostics("control-test") + diagnostics.note_dropped(self._req("stale nvz-old")) + channel = SimpleNamespace( + _client=SimpleNamespace(_live_diagnostics=diagnostics), + ) + with pytest.raises(AssertionError, match="dropped it as stale"): + await router.expect_no_message("nvz-old", channel, timeout=0.3) + + async def test_live_direct_messages_use_the_explicit_guardrail(): assert direct_message_guardrails("U123") == { "allow_users": ["U123"], @@ -153,6 +261,13 @@ async def posted(bot, _ack_outbound_events): await client.chat_delete(channel=channel, ts=ts) except Exception: pass # A test that already deleted it, or a stale ts. + if BOT_TOKEN and tracker.files: + client = make_client(BOT_TOKEN) + for file_id in tracker.files: + try: + await client.files_delete(file=file_id) + except Exception: + pass # Already gone, or removed with its message. # ---------------------------------------------------------------------- # @@ -302,13 +417,25 @@ async def test_a_file_upload_lands_in_the_conversation( cfg.slack = SlackConfig(bot_token=BOT_TOKEN, app_token="unused") channel = SlackChannel(cfg, router=None) # type: ignore[arg-type] channel._web = bot + channel._state = "running" assert await channel.send_file(format_target(TEST_CHANNEL), str(path)) + # chat.delete cannot remove an upload, so the file id has to be + # recorded for files.delete or the scratch channel keeps every copy. + listed = await bot.files_list(channel=TEST_CHANNEL, count=20) + for entry in listed.get("files") or []: + if entry.get("name") == "report.txt": + posted.note_file(entry.get("id")) + async def test_streaming_edits_then_removes_the_placeholder( self, bot, posted, ): placeholder = await bot.chat_postMessage(channel=TEST_CHANNEL, text="⏳") ts = placeholder["ts"] + # Tracked before the edits, so a rate limit part way through the loop + # does not leave the placeholder behind. Teardown tolerates a ts this + # test has already deleted itself. + posted.note_bot(TEST_CHANNEL, ts) for fragment in ("partial one", "partial one and two"): await bot.chat_update( channel=TEST_CHANNEL, ts=ts, text=_md_to_slack(fragment), diff --git a/tests/test_slack_live_inbound.py b/tests/test_slack_live_inbound.py index 1308769c..62578921 100644 --- a/tests/test_slack_live_inbound.py +++ b/tests/test_slack_live_inbound.py @@ -214,8 +214,10 @@ async def test_a_mention_reaches_the_router_and_is_answered_in_thread( m for m in replies["messages"] if m.get("user") == channel._bot_user_id ] assert bot_replies, "the bot did not reply in the thread" - assert bot_replies[0]["text"] == "*live reply*" + # Recorded before the text is checked, so a mismatch does not leave + # the bot's reply behind in the channel. posted.note_bot(TEST_CHANNEL, bot_replies[0]["ts"]) + assert bot_replies[0]["text"] == "*live reply*" async def test_channel_chatter_without_a_mention_is_left_alone( self, live_channel, human, posted, @@ -229,7 +231,7 @@ async def test_channel_chatter_without_a_mention_is_left_alone( channel=TEST_CHANNEL, text=f"just talking to my colleagues {marker}", ) posted.note_user(TEST_CHANNEL, sent["ts"]) - await router.expect_no_message(marker) + await router.expect_no_message(marker, channel) async def test_a_thread_reply_continues_without_another_mention( self, live_channel, human, posted, @@ -300,7 +302,7 @@ async def test_a_denied_user_never_reaches_the_router( channel=TEST_CHANNEL, text=f"<@{channel._bot_user_id}> let me in {marker}", ) posted.note_user(TEST_CHANNEL, sent["ts"]) - await router.expect_no_message(marker) + await router.expect_no_message(marker, channel) async def test_an_unconfigured_policy_refuses_a_real_message( self, live_channel, human, posted, @@ -313,7 +315,7 @@ async def test_an_unconfigured_policy_refuses_a_real_message( channel=TEST_CHANNEL, text=f"<@{channel._bot_user_id}> anyone home {marker}", ) posted.note_user(TEST_CHANNEL, sent["ts"]) - await router.expect_no_message(marker) + await router.expect_no_message(marker, channel) async def test_allowing_by_handle_resolves_through_users_info( self, live_channel, human, bot, posted, @@ -366,7 +368,7 @@ async def test_a_channel_outside_the_glob_is_refused( channel=TEST_CHANNEL, text=f"<@{channel._bot_user_id}> should not pass {marker}", ) posted.note_user(TEST_CHANNEL, sent["ts"]) - await router.expect_no_message(marker) + await router.expect_no_message(marker, channel) async def test_a_reaction_from_a_human_is_forwarded( self, live_channel, human, bot, posted, From 2fbf1fad1c3bbb75a452b8b5bd0dc1b551bf6e01 Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Tue, 25 Aug 2026 15:53:46 +0200 Subject: [PATCH 5/5] Cache the live reaction anchor the way the channel does The reaction test cached a bare conversation target. Nothing in the channel writes one for a shared channel: a top-level message is the root of its own thread, so every entry carries a thread ts. Only a notification card sits at conversation level, and a reaction there now has no session to join. Cache the anchor against its own thread, which is the target the channel would have stored for it. --- tests/test_slack_live_inbound.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/tests/test_slack_live_inbound.py b/tests/test_slack_live_inbound.py index 62578921..5169e40a 100644 --- a/tests/test_slack_live_inbound.py +++ b/tests/test_slack_live_inbound.py @@ -384,9 +384,14 @@ async def test_a_reaction_from_a_human_is_forwarded( ) posted.note_bot(TEST_CHANNEL, anchor["ts"]) # The channel only forwards reactions on messages it still has - # context for, which is what the outbound cache is for. + # context for, which is what the outbound cache is for. Every + # shared-channel entry the channel writes carries a thread, because a + # top-level message is the root of its own, so the target here is the + # one production would have cached. channel._cache_message( - anchor["ts"], format_target(TEST_CHANNEL), f"react to me {marker}", + anchor["ts"], + format_target(TEST_CHANNEL, anchor["ts"]), + f"react to me {marker}", ) await human.reactions_add( channel=TEST_CHANNEL, timestamp=anchor["ts"], name="tada",