signal-bot/tests/test_config.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Configuration loading, validation and URL derivation."""

from __future__ import annotations

import pytest

from signal_bot.config import (
    Config,
    ConfigError,
    canonical_group_id,
    group_id_aliases,
    load_config,
)

# The real spellings of one group, as the API reports them.
PREFIXED = "group.WTI5dVptbG5MV2R5YjNWd0xXbGtMVE15TFdKNWRHVnpJU0VoSVNFaElRPT0="
INTERNAL = "Y29uZmlnLWdyb3VwLWlkLTMyLWJ5dGVzISEhISEhIQ=="

BASE_ENV = {
    "PHONE_NUMBER": "+15550000000",
    "GROUP_IDS": "group.watched==",
}


def test_loads_minimal_environment() -> None:
    config = load_config(dict(BASE_ENV))
    assert config.phone_number == "+15550000000"
    assert config.api_base_url == "http://localhost:9922"
    assert config.report_hour == 21
    assert config.report_minute == 37


@pytest.mark.parametrize("missing", ["PHONE_NUMBER", "GROUP_IDS"])
def test_refuses_to_start_without_required_value(missing: str) -> None:
    env = dict(BASE_ENV)
    del env[missing]
    with pytest.raises(ConfigError, match=missing):
        load_config(env)


@pytest.mark.parametrize("missing", ["PHONE_NUMBER", "GROUP_IDS"])
def test_refuses_to_start_on_blank_value(missing: str) -> None:
    """A linked device sees every chat, so an unscoped bot must not start."""
    env = dict(BASE_ENV) | {missing: "   "}
    with pytest.raises(ConfigError, match=missing):
        load_config(env)


def test_overrides_are_honoured() -> None:
    env = dict(BASE_ENV) | {
        "API_BASE_URL": "https://signal.example:8443",
        "DATABASE_PATH": "/var/lib/bot/bot.sqlite3",
        "REPORT_HOUR": "6",
        "REPORT_MINUTE": "5",
        "COOLDOWN_SECONDS": "0",
    }
    config = load_config(env)
    assert config.api_base_url == "https://signal.example:8443"
    assert config.database_path == "/var/lib/bot/bot.sqlite3"
    assert config.report_hour == 6
    assert config.cooldown_seconds == 0


def test_blank_optionals_fall_back_to_defaults() -> None:
    env = dict(BASE_ENV) | {"API_BASE_URL": "  ", "DATABASE_PATH": "", "REPORT_HOUR": ""}
    config = load_config(env)
    assert config.api_base_url == "http://localhost:9922"
    assert config.database_path == "signal_bot.sqlite3"
    assert config.report_hour == 21


def test_rejects_non_integer_option() -> None:
    with pytest.raises(ConfigError, match="must be an integer"):
        load_config(dict(BASE_ENV) | {"REPORT_HOUR": "quarter past"})


def test_rejects_negative_option() -> None:
    with pytest.raises(ConfigError, match="must not be negative"):
        load_config(dict(BASE_ENV) | {"COOLDOWN_SECONDS": "-1"})


@pytest.mark.parametrize(("hour", "minute"), [("24", "0"), ("0", "60")])
def test_rejects_out_of_range_clock(hour: str, minute: str) -> None:
    env = dict(BASE_ENV) | {"REPORT_HOUR": hour, "REPORT_MINUTE": minute}
    with pytest.raises(ConfigError, match="out of range"):
        load_config(env)


def test_reads_os_environ_when_no_mapping_given(monkeypatch: pytest.MonkeyPatch) -> None:
    for key, value in BASE_ENV.items():
        monkeypatch.setenv(key, value)
    config = load_config(use_dotenv=False)
    assert config.group_ids == ("group.watched==",)


def test_derived_urls() -> None:
    config = load_config(dict(BASE_ENV))
    assert config.receive_url == (
        "ws://localhost:9922/v1/receive/+15550000000?send_read_receipts=false"
    )
    assert config.send_url == "http://localhost:9922/v2/send"
    assert config.attachments_url == "http://localhost:9922/v1/attachments"
    assert config.about_url == "http://localhost:9922/v1/about"


def test_https_base_url_yields_wss() -> None:
    config = Config(
        phone_number="+1",
        group_ids=("group.g==",),
        api_base_url="https://signal.example",
    )
    assert config.receive_url.startswith("wss://signal.example/")


def test_group_id_aliases_adds_the_internal_form() -> None:
    """The API spells one group two ways; the bot must answer to both."""
    # Real value observed live: the base64 decodes to the internal_id the
    # same endpoint reports, and to what envelopes actually carry.
    configured = "group.WTI5dVptbG5MV2R5YjNWd0xXbGtMVE15TFdKNWRHVnpJU0VoSVNFaElRPT0="
    assert group_id_aliases(configured) == {
        configured,
        "Y29uZmlnLWdyb3VwLWlkLTMyLWJ5dGVzISEhISEhIQ==",
    }


def test_group_id_aliases_offers_the_prefixed_form_of_an_internal_id() -> None:
    """An internal id must still be addressable: /v2/send only takes group.*."""
    aliases = group_id_aliases(INTERNAL)
    assert aliases == {INTERNAL, PREFIXED}


def test_group_id_aliases_tolerates_undecodable_ids() -> None:
    """A hand-edited GROUP_ID must not crash startup before the real check."""
    assert group_id_aliases("group.not!valid!base64") == {"group.not!valid!base64"}


def test_watched_group_ids_exposes_both_spellings() -> None:
    config = load_config({"PHONE_NUMBER": "+15550000000", "GROUP_IDS": PREFIXED})
    assert INTERNAL in config.watched_group_ids


def test_groups_url_is_scoped_to_the_account() -> None:
    config = load_config(
        {
            "PHONE_NUMBER": "+15550000000",
            "GROUP_IDS": "group.g==",
            "API_BASE_URL": "http://localhost:9922",
        }
    )
    assert config.groups_url == "http://localhost:9922/v1/groups/+15550000000"


def test_an_unknown_log_level_is_rejected() -> None:
    with pytest.raises(ConfigError, match="LOG_LEVEL"):
        load_config(
            {
                "PHONE_NUMBER": "+15550000000",
                "GROUP_IDS": "group.g==",
                "LOG_LEVEL": "chatty",
            }
        )


@pytest.mark.parametrize("port", ["0", "65536"])
def test_a_port_outside_the_valid_range_is_rejected(port: str) -> None:
    with pytest.raises(ConfigError, match="BIND_PORT"):
        load_config(
            {
                "PHONE_NUMBER": "+15550000000",
                "GROUP_IDS": "group.g==",
                "BIND_PORT": port,
            }
        )


def test_several_groups_are_parsed_in_order() -> None:
    """The point of the change: one process, a test group and a real one."""
    config = load_config(dict(BASE_ENV) | {"GROUP_IDS": f"{PREFIXED}, group.other=="})
    assert config.group_ids == (PREFIXED, "group.other==")
    assert config.resolve_group(INTERNAL) == PREFIXED
    assert config.resolve_group("group.other==") == "group.other=="


def test_group_id_is_still_accepted_as_the_one_group_spelling() -> None:
    """An .env written before the list form must keep working untouched."""
    env = {"PHONE_NUMBER": "+15550000000", "GROUP_ID": PREFIXED}
    assert load_config(env).group_ids == (PREFIXED,)


def test_setting_both_spellings_is_refused() -> None:
    env = dict(BASE_ENV) | {"GROUP_ID": PREFIXED}
    with pytest.raises(ConfigError, match="not both"):
        load_config(env)


def test_the_removed_send_target_fails_startup() -> None:
    """Silently ignoring it would post the daily report into the void."""
    with pytest.raises(ConfigError, match="GROUP_ID_SEND has been removed"):
        load_config(dict(BASE_ENV) | {"GROUP_ID_SEND": PREFIXED})


def test_blank_entries_between_separators_are_skipped() -> None:
    config = load_config(dict(BASE_ENV) | {"GROUP_IDS": f" {PREFIXED} , , "})
    assert config.group_ids == (PREFIXED,)


def test_a_list_of_only_separators_is_refused() -> None:
    with pytest.raises(ConfigError, match="only separators"):
        load_config(dict(BASE_ENV) | {"GROUP_IDS": " , , "})


def test_one_group_listed_in_both_spellings_is_refused() -> None:
    """Two scopes for one group would split its tally and its cooldowns."""
    with pytest.raises(ConfigError, match="listed twice"):
        load_config(dict(BASE_ENV) | {"GROUP_IDS": f"{PREFIXED},{INTERNAL}"})


def test_an_internal_id_is_stored_in_its_addressable_form() -> None:
    """Replies are sent to the canonical id, and /v2/send rejects the other."""
    config = load_config(dict(BASE_ENV) | {"GROUP_IDS": INTERNAL})
    assert config.group_ids == (PREFIXED,)
    assert config.resolve_group(INTERNAL) == PREFIXED


def test_canonical_group_id_is_idempotent() -> None:
    assert canonical_group_id(PREFIXED) == PREFIXED
    assert canonical_group_id(canonical_group_id(INTERNAL)) == PREFIXED


def test_resolve_group_rejects_direct_messages_and_strangers() -> None:
    config = load_config(dict(BASE_ENV))
    assert config.resolve_group(None) is None
    assert config.resolve_group("group.stranger==") is None