signal-bot/tests/test_routes_suggestions.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The management page's read of the wish list, and the export it writes."""

from __future__ import annotations

import dataclasses
import sqlite3
from pathlib import Path
from typing import TYPE_CHECKING

from signal_bot.manage.routes_suggestions import (
    EXPORT_BODY_LIMIT,
    EXPORT_NAME,
    default_export_path,
)
from signal_bot.store import Store
from tests.manage_fixture import client_for, deps_at

if TYPE_CHECKING:
    import pytest

    from signal_bot.manage.deps import ManageDeps


def seeded(tmp_path: Path) -> ManageDeps:
    """A management app pointed at a database with one of each row in it."""
    deps = deps_at(tmp_path)
    with Store(str(deps.database_path)) as store:
        store.suggestions.record_suggestion("g1", "!pogoda", "pokaz pogode", 101)
        store.suggestions.record_unknown("g1", "!cta", "!cta", 200)
        store.suggestions.record_unknown("g1", "!cta", "!cta teraz", 201)
    return deps


def test_the_fixture_never_points_at_the_real_database(tmp_path: Path) -> None:
    """Config.database_path is the *relative* "signal_bot.sqlite3".

    pytest runs from the repository root, where the operator's real database
    sits, and Store drops the ephemeral tables on a version mismatch. A
    defaulted field on ManageDeps would therefore have let the suite delete
    live cooldowns and announce state -- so the field has no default, and
    this asserts the fixture supplies a temporary one.
    """
    deps = deps_at(tmp_path)
    assert deps.database_path.is_relative_to(tmp_path)
    assert deps.export_path.is_relative_to(tmp_path)


def test_a_bot_that_has_never_run_is_not_an_error(tmp_path: Path) -> None:
    """No database yet means nothing collected, not a 500."""
    deps = deps_at(tmp_path)
    assert not deps.database_path.exists()
    body = client_for(deps).get("/api/suggestions").json()
    assert body == {"suggestions": [], "unknown": []}


def test_a_database_older_than_these_tables_is_not_an_error(tmp_path: Path) -> None:
    """Upgrading the bot must not make the page fail until it restarts."""
    deps = deps_at(tmp_path)
    sqlite3.connect(deps.database_path).close()
    body = client_for(deps).get("/api/suggestions").json()
    assert body == {"suggestions": [], "unknown": []}


def test_a_directory_where_the_database_should_be_is_not_an_error(tmp_path: Path) -> None:
    """is_file, not exists: a path that is not a file is still "nothing here"."""
    deps = deps_at(tmp_path)
    deps.database_path.mkdir()
    assert client_for(deps).get("/api/suggestions").json()["unknown"] == []


def test_an_unreadable_database_is_not_an_error(tmp_path: Path) -> None:
    """Whatever is at the path, the page still renders."""
    deps = deps_at(tmp_path)
    deps.database_path.write_text("not a database", encoding="utf-8")
    assert client_for(deps).get("/api/suggestions").json()["suggestions"] == []


def test_the_page_shows_ideas_and_tallies(tmp_path: Path) -> None:
    """Bodies included, because this response never leaves the loopback port."""
    body = client_for(seeded(tmp_path)).get("/api/suggestions").json()
    assert body["suggestions"][0]["command"] == "!pogoda"
    assert body["suggestions"][0]["body"] == "pokaz pogode"
    assert body["unknown"][0] == {
        "bodies": ["!cta", "!cta teraz"],
        "group_id": "g1",
        "hits": 2,
        "last_seen": 201,
        "word": "!cta",
    }


def test_the_export_keeps_notes_but_reduces_mistyped_words_to_counts(
    tmp_path: Path,
) -> None:
    """The asymmetry this whole feature is arranged around.

    A !zaproponuj body was written on purpose, to the maintainer. An
    unrecognised body is half-typed chat, and a half-typed name is one the
    personal-data gate cannot recognise -- so it becomes a number.
    """
    deps = seeded(tmp_path)
    answer = client_for(deps).post("/api/suggestions/export", json={}).json()
    assert answer == {"path": str(deps.export_path), "suggestions": 1, "unknown": 1}
    written = deps.export_path.read_text(encoding="utf-8")
    assert "pokaz pogode" in written
    assert "`!cta` x 2" in written
    assert "!cta teraz" not in written


def test_the_export_says_so_when_there_is_nothing_to_say(tmp_path: Path) -> None:
    """An empty file that reads as empty, rather than two bare headings."""
    deps = deps_at(tmp_path)
    client_for(deps).post("/api/suggestions/export", json={})
    assert deps.export_path.read_text(encoding="utf-8").count("_Nothing yet._") == 2


def test_a_very_long_body_is_truncated_in_the_export(tmp_path: Path) -> None:
    """One pasted paragraph must not bury the rest of the file."""
    deps = deps_at(tmp_path)
    with Store(str(deps.database_path)) as store:
        store.suggestions.record_suggestion("g1", "!x", "a" * 500, 1)
    client_for(deps).post("/api/suggestions/export", json={})
    written = deps.export_path.read_text(encoding="utf-8")
    assert "a" * EXPORT_BODY_LIMIT in written
    assert "a" * (EXPORT_BODY_LIMIT + 1) not in written


def test_the_default_export_path_is_never_inside_the_repository(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """A group-writable file in a public git tree is a leak and a wedge.

    The personal-data gate would refuse every future commit once a real name
    landed in it, and it cannot recognise the misspelled ones this file is
    full of. So the export goes where the album goes.
    """
    monkeypatch.setenv("STATE_DIRECTORY", str(tmp_path / "state"))
    assert default_export_path() == tmp_path / "state" / EXPORT_NAME
    monkeypatch.delenv("STATE_DIRECTORY")
    monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path / "share"))
    assert default_export_path() == tmp_path / "share" / "signal-bot" / EXPORT_NAME
    monkeypatch.delenv("XDG_DATA_HOME")
    assert default_export_path().is_relative_to(Path.home())


def test_the_export_does_not_need_its_directory_to_exist(tmp_path: Path) -> None:
    """First run on a fresh install, where the state directory is still empty."""
    deps = dataclasses.replace(deps_at(tmp_path), export_path=tmp_path / "a" / "b" / EXPORT_NAME)
    client_for(deps).post("/api/suggestions/export", json={})
    assert (tmp_path / "a" / "b" / EXPORT_NAME).is_file()