# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence. """The scenario injector CLI, including its exit codes.""" from __future__ import annotations import json import runpy import sys from pathlib import Path import httpx import pytest from fake_signal import cli from fake_signal.scenario import Expectation, GroupArrival, Injection, Scenario GROUPS = {"watched": "W==", "reply": "R==", "other": "O=="} class Recorder: """Stands in for the fake server's control plane.""" def __init__(self, sent: list[dict]) -> None: self.sent = sent self.injected: list[str] = [] self.added: list[list[dict]] = [] def __call__(self, request: httpx.Request) -> httpx.Response: path = request.url.path if path == "/_control/inject": self.injected.append(request.content.decode()) return httpx.Response(200, json={"delivered": 1}) if path == "/_control/groups/add": self.added.append(json.loads(request.content)) return httpx.Response(200, json={"groups": 1}) if path == "/_control/sent": return httpx.Response(200, json=self.sent) return httpx.Response(200, json={"status": "ok"}) def _patch_client(monkeypatch: pytest.MonkeyPatch, recorder: Recorder) -> None: real = httpx.AsyncClient def factory(**kwargs: object) -> httpx.AsyncClient: # base_url must survive: run_scenario uses relative paths. return real( transport=httpx.MockTransport(recorder), base_url=str(kwargs.get("base_url", "http://x")), ) monkeypatch.setattr(cli.httpx, "AsyncClient", factory) async def test_run_scenario_injects_and_passes(monkeypatch: pytest.MonkeyPatch) -> None: recorder = Recorder([{"base64_attachments": ["A"], "recipients": ["R=="]}]) _patch_client(monkeypatch, recorder) scenario = Scenario("s", inject=[Injection(text="!cat")], expect=[Expectation(kind="image")]) failures = await cli.run_scenario("http://x", scenario, GROUPS, settle=0) assert failures == [] assert "!cat" in recorder.injected[0] async def test_run_scenario_announces_arriving_groups(monkeypatch: pytest.MonkeyPatch) -> None: """A scenario can add a group to the account before it injects anything.""" recorder = Recorder([{"base64_attachments": ["A"], "recipients": ["R=="]}]) _patch_client(monkeypatch, recorder) scenario = Scenario( "s", arrive=[GroupArrival(id="reply", internal_id="watched")], inject=[Injection(text="!cat")], expect=[Expectation(kind="image")], ) assert await cli.run_scenario("http://x", scenario, GROUPS, settle=0) == [] assert recorder.added == [ [ { "id": "R==", "internal_id": "W==", "name": "added-later", "member": True, "joinable": True, } ] ] async def test_run_scenario_reports_a_mismatch(monkeypatch: pytest.MonkeyPatch) -> None: _patch_client(monkeypatch, Recorder([])) scenario = Scenario("s", expect=[Expectation(kind="image")]) assert await cli.run_scenario("http://x", scenario, GROUPS, settle=0, patience=0.05) def test_parse_args_collects_named_groups() -> None: args = cli.parse_args(["a.yaml", "--group", "watched=W==", "--group", "second=S=="]) assert args.scenarios == [Path("a.yaml")] assert cli.resolve_groups(args) == {"watched": "W==", "second": "S=="} @pytest.mark.parametrize("pair", ["watched", "=W==", "watched="]) def test_parse_args_rejects_a_malformed_group_pair(pair: str) -> None: """A typo here would silently rename a group instead of failing.""" args = cli.parse_args(["a.yaml", "--group", pair]) with pytest.raises(SystemExit, match="NAME=ID"): cli.resolve_groups(args) def _scenario_file(tmp_path: Path, expect: str) -> Path: path = tmp_path / "s.yaml" path.write_text(f"name: demo\ninject:\n - {{text: '!cat'}}\n{expect}", encoding="utf-8") return path def test_main_exits_zero_when_every_scenario_passes( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] ) -> None: _patch_client(monkeypatch, Recorder([{"base64_attachments": ["A"], "recipients": ["R=="]}])) path = _scenario_file(tmp_path, "expect:\n - {kind: image}\n") code = cli.main( [ str(path), "--group", "watched=W==", "--group", "reply=R==", "--settle", "0", "--patience", "0.05", ] ) assert code == 0 assert "ok demo" in capsys.readouterr().out def test_main_exits_non_zero_on_a_mismatch( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] ) -> None: """The e2e script is a gate only because this returns 1.""" _patch_client(monkeypatch, Recorder([])) path = _scenario_file(tmp_path, "expect:\n - {kind: image}\n") code = cli.main( [ str(path), "--group", "watched=W==", "--group", "reply=R==", "--settle", "0", "--patience", "0.05", ] ) assert code == 1 out = capsys.readouterr().out assert "FAIL demo" in out def test_module_runs_as_a_script(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: """`python -m fake_signal.cli` must exit with the adjudication code. runpy re-imports into a fresh namespace, so the patch has to land on the shared httpx module rather than on fake_signal.cli. """ _patch_client(monkeypatch, Recorder([])) path = _scenario_file(tmp_path, "expect:\n - {kind: image}\n") monkeypatch.setattr( sys, "argv", [ "cli", str(path), "--group", "watched=W==", "--group", "reply=R==", "--settle", "0", "--patience", "0.05", ], ) with pytest.raises(SystemExit) as exit_info: runpy.run_module("fake_signal.cli", run_name="__main__", alter_sys=True) assert exit_info.value.code == 1 class Slow(Recorder): """A server whose send only shows up after a few polls.""" def __init__(self, sent: list[dict], *, after: int) -> None: super().__init__(sent) self.after = after self.reads = 0 def __call__(self, request: httpx.Request) -> httpx.Response: if request.url.path == "/_control/sent": self.reads += 1 return httpx.Response(200, json=self.sent if self.reads > self.after else []) return super().__call__(request) async def test_a_reply_that_is_slow_is_waited_for_rather_than_missed( monkeypatch: pytest.MonkeyPatch, ) -> None: # The bug this replaces: a fixed settle expired before a reply that had # to fetch an image and shell out to exiftool, so the send landed after # the NEXT scenario's reset and was counted against that one instead. recorder = Slow([{"base64_attachments": ["A"], "recipients": ["R=="]}], after=3) _patch_client(monkeypatch, recorder) scenario = Scenario("s", inject=[Injection(text="!cat")], expect=[Expectation(kind="image")]) assert await cli.run_scenario("http://x", scenario, GROUPS, settle=0) == [] assert recorder.reads > 3 async def test_waiting_gives_up_and_reports_what_did_arrive( monkeypatch: pytest.MonkeyPatch, ) -> None: # Patience is a deadline, not a promise: a reply that never comes has to # fail the scenario rather than hang the whole run. recorder = Recorder([]) _patch_client(monkeypatch, recorder) scenario = Scenario("s", inject=[Injection(text="!cat")], expect=[Expectation(kind="image")]) failures = await cli.run_scenario("http://x", scenario, GROUPS, settle=0, patience=0.05) assert "captured 0" in failures[0] async def test_a_scenario_expecting_nothing_does_not_poll_at_all( monkeypatch: pytest.MonkeyPatch, ) -> None: # "nothing must happen" cannot be waited for -- there is no count to # reach -- so it waits out the settle and reads once. recorder = Recorder([]) _patch_client(monkeypatch, recorder) scenario = Scenario("s", inject=[Injection(text="chatter")]) assert await cli.run_scenario("http://x", scenario, GROUPS, settle=0) == []