signal-bot/tests/test_fake_account.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Registration and group endpoints on the fake server."""

from __future__ import annotations

import pytest
from fastapi.testclient import TestClient

from fake_signal.routes_account import CAPTCHA_REQUIRED, expected_code
from fake_signal.server import create_app
from fake_signal.state import State

NUMBER = "+15550001234"


@pytest.fixture
def state() -> State:
    return State(base_url="http://testserver")


@pytest.fixture
def client(state: State) -> TestClient:
    return TestClient(create_app(state))


def test_no_accounts_before_registration(client: TestClient) -> None:
    assert client.get("/v1/accounts").json() == []


def test_register_without_captcha_is_refused(client: TestClient) -> None:
    """The exact failure the real container returns, so the script can match it."""
    response = client.post(f"/v1/register/{NUMBER}", json={})
    assert response.status_code == 400
    assert "signalcaptchas.org" in response.json()["error"]
    assert response.json()["error"] == CAPTCHA_REQUIRED


def test_register_with_captcha_starts_verification(client: TestClient, state: State) -> None:
    response = client.post(f"/v1/register/{NUMBER}", json={"captcha": "signalcaptcha://x"})
    assert response.status_code == 201
    assert state.pending[NUMBER] == "001234"


def test_verify_completes_registration(client: TestClient) -> None:
    client.post(f"/v1/register/{NUMBER}", json={"captcha": "signalcaptcha://x"})
    code = expected_code(NUMBER)
    assert client.post(f"/v1/register/{NUMBER}/verify/{code}").status_code == 201
    assert client.get("/v1/accounts").json() == [NUMBER]


def test_verify_is_idempotent_on_the_account_list(client: TestClient) -> None:
    for _ in range(2):
        client.post(f"/v1/register/{NUMBER}", json={"captcha": "c"})
        client.post(f"/v1/register/{NUMBER}/verify/{expected_code(NUMBER)}")
    assert client.get("/v1/accounts").json() == [NUMBER]


def test_verify_rejects_a_wrong_code(client: TestClient) -> None:
    client.post(f"/v1/register/{NUMBER}", json={"captcha": "c"})
    response = client.post(f"/v1/register/{NUMBER}/verify/000000")
    assert response.status_code == 400
    assert "invalid verification code" in response.json()["error"]


def test_verify_without_a_pending_registration_is_refused(client: TestClient) -> None:
    response = client.post(f"/v1/register/{NUMBER}/verify/001234")
    assert response.status_code == 400
    assert "no registration in progress" in response.json()["error"]


def test_groups_require_a_registered_account(client: TestClient) -> None:
    response = client.get(f"/v1/groups/{NUMBER}")
    assert response.status_code == 400
    assert "not registered" in response.json()["error"]


def test_groups_are_listed_once_registered(client: TestClient, state: State) -> None:
    state.accounts.append(NUMBER)
    state.groups.extend([{"id": "gA==", "name": "Alpha"}, {"id": "gB==", "name": "Bravo"}])
    body = client.get(f"/v1/groups/{NUMBER}").json()
    assert [g["name"] for g in body] == ["Alpha", "Bravo"]


def test_expected_code_uses_the_last_six_digits() -> None:
    assert expected_code("+48 123 456 789") == "456789"


def test_control_can_seed_groups(client: TestClient, state: State) -> None:
    """The picker in scripts/lib_signal.sh is rehearsed against this."""
    state.accounts.append(NUMBER)
    body = [{"id": "gA==", "name": "Alpha"}, {"id": "gB==", "name": "Bravo"}]
    assert client.post("/_control/groups", json=body).json() == {"groups": 2}
    assert client.get(f"/v1/groups/{NUMBER}").json() == body


def test_seeding_groups_replaces_the_previous_list(client: TestClient, state: State) -> None:
    state.accounts.append(NUMBER)
    client.post("/_control/groups", json=[{"id": "old==", "name": "Old"}])
    client.post("/_control/groups", json=[{"id": "new==", "name": "New"}])
    assert [g["id"] for g in client.get(f"/v1/groups/{NUMBER}").json()] == ["new=="]


def test_control_seeds_registered_accounts(client: TestClient) -> None:
    """/v1/groups answers 400 for an unknown account, so the fake needs this."""
    assert client.get("/v1/groups/+15550000000").status_code == 400
    assert client.post("/_control/accounts", json=["+15550000000"]).json() == {"accounts": 1}
    assert client.get("/v1/groups/+15550000000").status_code == 200


def test_control_replaces_accounts_rather_than_appending(client: TestClient) -> None:
    client.post("/_control/accounts", json=["+15550000001"])
    client.post("/_control/accounts", json=["+15550000002"])
    assert client.get("/v1/accounts").json() == ["+15550000002"]


def test_devices_needs_a_registered_account(client: TestClient) -> None:
    response = client.get(f"/v1/devices/{NUMBER}")
    assert response.status_code == 400
    assert "not registered" in response.json()["error"]


def test_devices_answers_while_the_connection_is_alive(client: TestClient, state: State) -> None:
    state.accounts.append(NUMBER)
    response = client.get(f"/v1/devices/{NUMBER}")
    assert response.status_code == 200
    assert response.json()[0]["id"] == 1


def test_devices_reproduces_the_outage_when_the_connection_is_cut(
    client: TestClient, state: State
) -> None:
    """The 2026-08-29 shape: local state fine, anything upstream 400s."""
    state.accounts.append(NUMBER)
    assert client.post("/_control/connection", json={"alive": False}).json() == {"alive": False}

    response = client.get(f"/v1/devices/{NUMBER}")
    assert response.status_code == 400
    assert response.json()["error"] == "Failed to get linked devices: Closed unexpectedly"
    # Local state keeps answering, which is what made this invisible.
    assert client.get("/v1/accounts").status_code == 200

    assert client.post("/_control/connection", json={"alive": True}).json() == {"alive": True}
    assert client.get(f"/v1/devices/{NUMBER}").status_code == 200