Blame for tests/test_config.py

"""Settings from the environment, the deployment files, and that nothing the owner needs is left undocumented."""
import asyncio
import py_compile
import re
import socket
import tempfile
from pathlib import Path

import pytest

from app.config import DEFAULT_ERROR_LOG_CHAT_ID, Config, default_session_path, load_config
from app.health import start_health_server
from bot import COMMANDS, secrets_of

from .harness import App

ROOT = Path(__file__).resolve().parent.parent
REQUIRED = {"API_ID": "12345", "API_HASH": "abcdef0123456789", "BOT_TOKEN": "123456:ABCdefGHIjklMNOpqrSTUvwxYZ", "OWNER_IDS": "1001484109"}


def env_names() -> list:
    """Every environment variable app/config.py reads."""
    src = (ROOT / "app" / "config.py").read_text()
    names = set(re.findall(r'(?:getenv|_ints|_bool|_chat_id|_float|_whole)\(\s*"([A-Z][A-Z0-9_]+)"', src))
    names |= set(re.findall(r'environ\[\s*"([A-Z][A-Z0-9_]+)"', src))
    return sorted(names)


@pytest.fixture
def env(monkeypatch):
    for k in env_names():
        monkeypatch.delenv(k, raising=False)
    for k, v in REQUIRED.items():
        monkeypatch.setenv(k, v)
    return monkeypatch


# ----------------------------------------------------------------------------------------- defaults
def test_the_names_the_test_looks_for_are_all_found():
    names = env_names()
    for expected in ("API_ID", "BOT_TOKEN", "OWNER_IDS", "ERROR_LOG_CHAT_ID", "USERBOT_SESSION", "USERBOT_KEEP_ADMIN", "SESSION_PATH", "PORT"):
        assert expected in names, names


def test_defaults(env):
    cfg = load_config()
    assert cfg.error_log_chat_id == DEFAULT_ERROR_LOG_CHAT_ID == -1002525172451
    assert cfg.userbot_session == "" and cfg.userbot_keep_admin is False
    assert cfg.owners == frozenset({1001484109}) and cfg.admins == frozenset()
    assert cfg.edit_delay == 1.2 and cfg.replace_typed_links is True and cfg.port == 0
    assert cfg.database_url.startswith("sqlite")


@pytest.mark.parametrize(
    "raw,expected",
    [
        (None, -1002525172451),  # not set at all -> the default chat
        ("", -1002525172451),
        ("   ", -1002525172451),
        ("-1001234567890", -1001234567890),
        (" -1001234567890 ", -1001234567890),
        ("123456789", 123456789),  # a private chat with a person
        ("0", None),
        ("off", None),
        ("OFF", None),
        ("None", None),
        ("false", None),
        ("no", None),
        ("disabled", None),
    ],
)
def test_error_log_chat_id(env, raw, expected):
    if raw is not None:
        env.setenv("ERROR_LOG_CHAT_ID", raw)
    assert load_config().error_log_chat_id == expected


@pytest.mark.parametrize("raw", ["abc", "-100 123", "@mychannel", "1.5"])
def test_a_bad_error_log_chat_id_is_refused_with_its_name(env, raw):
    env.setenv("ERROR_LOG_CHAT_ID", raw)
    with pytest.raises(SystemExit) as e:
        load_config()
    assert "ERROR_LOG_CHAT_ID" in str(e.value) and repr(raw) in str(e.value)


def test_userbot_settings(env):
    env.setenv("USERBOT_SESSION", "  1BVtsOKwBu7abc  ")
    env.setenv("USERBOT_KEEP_ADMIN", "true")
    cfg = load_config()
    assert cfg.userbot_session == "1BVtsOKwBu7abc" and cfg.userbot_keep_admin is True
    for v, want in [("1", True), ("yes", True), ("ON", True), ("false", False), ("0", False), ("", False), ("nonsense", False)]:
        env.setenv("USERBOT_KEEP_ADMIN", v)
        assert load_config().userbot_keep_admin is want, v


def test_missing_settings_are_listed(env):
    env.delenv("API_HASH")
    env.delenv("OWNER_IDS")
    with pytest.raises(SystemExit) as e:
        load_config()
    assert "API_HASH" in str(e.value) and "OWNER_IDS" in str(e.value) and "API_ID" not in str(e.value)
    env.setenv("OWNER_IDS", "not a number")
    env.setenv("API_HASH", "x")
    with pytest.raises(SystemExit) as e:
        load_config()
    assert "OWNER_IDS" in str(e.value)


def test_api_id_must_be_a_number(env):
    env.setenv("API_ID", "twelve")
    with pytest.raises(SystemExit) as e:
        load_config()
    assert "API_ID must be a number" in str(e.value)


def test_service_message_and_sync_settings(env):
    cfg = load_config()
    assert cfg.delete_service_messages is True and cfg.sync_interval_minutes == 60  # both are on unless somebody says otherwise
    env.setenv("DELETE_SERVICE_MESSAGES", "false")
    env.setenv("SYNC_INTERVAL_MINUTES", "15")
    cfg = load_config()
    assert cfg.delete_service_messages is False and cfg.sync_interval_minutes == 15
    env.setenv("DELETE_SERVICE_MESSAGES", "yes")
    assert load_config().delete_service_messages is True


def test_the_older_posts_of_a_channel_are_imported_unless_somebody_says_otherwise(env):
    assert load_config().import_old_posts is True
    env.setenv("IMPORT_OLD_POSTS", "false")
    assert load_config().import_old_posts is False
    env.setenv("IMPORT_OLD_POSTS", "")
    assert load_config().import_old_posts is True


@pytest.mark.parametrize(
    "raw,expected",
    [("0", 0), ("-5", 0), ("30", 30), ("2.9", 2), ("  45 ", 45), ("", 60), ("abc", 60), ("inf", 60), ("-inf", 60), ("nan", 60)],
)
def test_sync_interval_minutes_never_stops_the_bot_from_starting(env, raw, expected):
    env.setenv("SYNC_INTERVAL_MINUTES", raw)
    assert load_config().sync_interval_minutes == expected  # 0 = no automatic checks; nonsense = the default


def test_edit_delay_has_a_floor(env):
    env.setenv("EDIT_DELAY", "0.1")
    assert load_config().edit_delay == 0.3
    env.setenv("EDIT_DELAY", "banana")
    assert load_config().edit_delay == 1.2
    env.setenv("EDIT_DELAY", "2.5")
    assert load_config().edit_delay == 2.5


def test_the_session_is_not_kept_in_the_app_folder_by_default(env):
    path = Path(load_config().session_path).resolve()
    assert str(path).startswith(str(Path(tempfile.gettempdir()).resolve()))
    assert ROOT not in path.parents  # an app folder that a web server publishes must not contain the login
    assert Path(default_session_path()).name == "bot"
    env.setenv("SESSION_PATH", "data/bot")
    assert load_config().session_path == "data/bot"  # still possible when somebody wants it


# ------------------------------------------------------------------------------ secrets of the bot
def test_every_secret_is_collected_for_redaction():
    cfg = Config(
        api_id=1, api_hash="HASHHASHHASH", bot_token="123:TOKEN-TOKEN", owners=frozenset({1}), admins=frozenset(),
        database_url="postgresql://user:p%40ss%2Fword@host.example/db?sslmode=require", db_pool="null", session_path="x",
        session_string="SESSIONSTRINGTEXT", edit_delay=1.0, replace_typed_links=True, port=0, log_level="INFO",
        userbot_session="USERBOTSESSIONTEXT",
    )
    found = secrets_of(cfg)
    for s in ("HASHHASHHASH", "123:TOKEN-TOKEN", "SESSIONSTRINGTEXT", "USERBOTSESSIONTEXT", "p%40ss%2Fword", "p@ss/word"):
        assert s in found, (s, found)
    assert "" not in found
    plain = Config(**{**cfg.__dict__, "database_url": "sqlite+aiosqlite:///data/bot.db", "session_string": "", "userbot_session": ""})
    assert set(secrets_of(plain)) == {"123:TOKEN-TOKEN", "HASHHASHHASH"}  # nothing else to hide, and no crash on SQLite


# ------------------------------------------------------------------------------------- deployment files
def test_dockerfile_is_exactly_the_one_that_was_asked_for():
    lines = [ln.rstrip() for ln in (ROOT / "Dockerfile").read_text().splitlines() if ln.strip()]
    assert lines == [
        "FROM python:3.12-slim",
        "WORKDIR /app",
        "ENV PYTHONUNBUFFERED=1",
        "COPY requirements.txt .",
        "RUN pip install --no-cache-dir -r requirements.txt",
        "COPY . .",
        "CMD python -m http.server 7860 & python bot.py",
    ]


def test_github_workflow_syncs_to_the_space_without_a_hard_coded_token():
    text = (ROOT / ".github" / "workflows" / "hf-sync.yml").read_text()
    assert "huggingface.co/spaces/Mega-Evolutions/ButtonBot" in text
    assert "${{ secrets.HF_TOKEN }}" in text and "branches: [main]" in text and "workflow_dispatch" in text
    assert "--exclude='README.md'" in text and "--exclude='.github'" in text  # the Space keeps its own README
    assert not re.search(r"hf_[A-Za-z0-9]{20,}", text)  # no real token pasted in
    yaml = pytest.importorskip("yaml")
    doc = yaml.safe_load(text)
    assert doc["jobs"]["sync-to-hub"]["steps"][1]["env"]["HF_TOKEN"] == "${{ secrets.HF_TOKEN }}"


def test_nothing_secret_can_end_up_in_the_image_or_the_repository():
    docker = (ROOT / ".dockerignore").read_text().split()
    for entry in (".env", "*.session", "data/*", "tests"):
        assert entry in docker, entry
    git = (ROOT / ".gitignore").read_text().split()
    for entry in (".env", "*.session"):
        assert entry in git, entry


def test_env_example_documents_every_setting():
    text = (ROOT / ".env.example").read_text()
    listed = set(re.findall(r"^#?\s*([A-Z][A-Z0-9_]+)=", text, re.M))
    missing = [n for n in env_names() if n not in listed]
    assert not missing, f".env.example does not mention: {missing}"
    unknown = sorted(listed - set(env_names()))
    assert not unknown, f".env.example mentions settings the bot does not read: {unknown}"
    assert re.search(r"^ERROR_LOG_CHAT_ID=-1002525172451$", text, re.M)


def test_helper_scripts_compile_and_the_userbot_one_stands_alone():
    for name in ("make_userbot_session.py", "make_session.py", "bot.py"):
        py_compile.compile(str(ROOT / name), doraise=True)
    src = (ROOT / "make_userbot_session.py").read_text()
    assert "from app" not in src and "import app" not in src  # runs on its own computer with only telethon installed
    assert "StringSession" in src and "USERBOT_SESSION=" in src


def test_the_testing_commands_are_gone_everywhere():
    names = [c for c, _ in COMMANDS]
    assert "testedit" not in names and "selftest" not in names
    assert {"new", "posts", "repost", "replace", "undo", "userbot", "help"} <= set(names)
    assert {"autopost", "broadcast", "handleswap", "shift", "sync", "export", "import"} <= set(names)
    assert "fixlinks" not in names
    assert "testerror" not in names
    for path in list((ROOT / "app").rglob("*.py")) + [ROOT / "bot.py"]:
        text = path.read_text().lower()
        assert "testedit" not in text and "selftest" not in text and "testerror" not in text, path


def test_every_menu_command_has_a_handler(tmp_path):
    async def go():
        app = App(tmp_path)
        await app.start()
        patterns = [b.pattern for b, _ in app.tg.handlers if getattr(b, "pattern", None)]
        for name, _ in COMMANDS:
            assert any(p(f"/{name}") for p in patterns), f"/{name} is in the menu but nothing answers it"
        await app.db.close()

    asyncio.run(go())


def test_the_help_text_lists_the_commands_that_exist():
    from app.handlers.basic import HELP

    for name in (
        "new", "posts", "repost", "replace", "undo", "userbot", "export", "import", "addchannel", "channels",
        "autopost", "broadcast", "handleswap", "shift", "sync",
    ):
        assert f"/{name}" in HELP, name
    assert "--channel" not in HELP and "/testerror" not in HELP and "/fixlinks" not in HELP and "--clean" not in HELP


# ------------------------------------------------------------------------------------------ health port
def free_port() -> int:
    with socket.socket() as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


def test_health_endpoint_answers_ok():
    async def go():
        port = free_port()
        server = await start_health_server(port)
        assert server is not None
        reader, writer = await asyncio.open_connection("127.0.0.1", port)
        writer.write(b"GET / HTTP/1.1\r\nHost: x\r\n\r\n")
        await writer.drain()
        data = await asyncio.wait_for(reader.read(), 5)
        assert data.startswith(b"HTTP/1.1 200 OK") and data.endswith(b"ok")
        writer.close()
        server.close()
        await server.wait_closed()

    asyncio.run(go())


def test_a_port_that_is_already_taken_does_not_stop_the_bot():
    """The Dockerfile starts `python -m http.server 7860` first; if PORT=7860 is set too, the bot must carry on."""

    async def go():
        busy = socket.socket()
        busy.bind(("0.0.0.0", 0))
        busy.listen()
        try:
            assert await start_health_server(busy.getsockname()[1]) is None
        finally:
            busy.close()

    asyncio.run(go())