diff --git a/.github/workflows/slack-live.yml b/.github/workflows/slack-live.yml new file mode 100644 index 00000000..c1997ed7 --- /dev/null +++ b/.github/workflows/slack-live.yml @@ -0,0 +1,82 @@ +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: + +# 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 + +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..8d838273 --- /dev/null +++ b/tests/slack_live.py @@ -0,0 +1,778 @@ +"""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 + +# 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.""" + 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 + # 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 + 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 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 ( + 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: + """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: + self.bot.append((channel, ts)) + + 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. + + 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, channel, timeout: float = EVENT_TIMEOUT, + ) -> None: + """Assert the event met the policy and was refused by it. + + 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. + """ + 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 " + 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"] + # 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), + 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: + 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 {} + 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), + ) + observer = getattr( + sock, "_live_diagnostics_request_listener", None, + ) + 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) + + 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(cfg, router) + 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..523a229f 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,98 @@ 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: + # 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): + 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 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": self.sent[0]}, + 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} + + probe = ProbeClient() + await wait_until_receiving( + sink, + timeout=2.0, + web_client=probe, + 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 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() + 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..082dda02 --- /dev/null +++ b/tests/test_slack_live.py @@ -0,0 +1,483 @@ +"""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 +import time +from types import SimpleNamespace + +import pytest +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, + split_message, +) +from tests.slack_live import ( + BOT_TOKEN, + HAVE_OUTBOUND, + NO_EMAIL_BOT_TOKEN, + TEST_CHANNEL, + USER_TOKEN, + Posted, + RecordingRouter, + build_channel, + direct_message_guardrails, + 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)] + + +@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"], + "allow_direct_messages": True, + } + + +# ---------------------------------------------------------------------- # +# 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. + 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. + + +# ---------------------------------------------------------------------- # +# 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(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), + ) + 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..5169e40a --- /dev/null +++ b/tests/test_slack_live_inbound.py @@ -0,0 +1,443 @@ +"""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 copy +import inspect +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, + direct_message_guardrails, + 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, _ = _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) + config = copy.deepcopy(channel.config) + for field, default in ( + ("allow_users", []), ("deny_users", []), + ("allow_direct_messages", False), + ("allow_channels", []), ("deny_channels", []), + ("commands", None), + ): + 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, config + + 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" + # 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, + ): + 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, channel) + + 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, **direct_message_guardrails(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}" + + +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( + 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, channel) + + 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, channel) + + 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, channel) + + 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. 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, anchor["ts"]), + 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)