signal-bot/tests/test_cli.py

# 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) == []