gwoe-antragspruefer/tests/test_auto_rate_runs.py

"""Unit-Tests fuer auto_rate_runs DB-Helper (#173, Phase 8.2).

Nutzt das gleiche tmp-DB-Fixture-Pattern wie tests/test_database.py.
"""
from __future__ import annotations

import asyncio
import sys

import pytest

# Same aiosqlite-Cache-Schutz wie in test_database.py.
_aio = sys.modules.get("aiosqlite")
if _aio is not None and not hasattr(_aio, "connect"):
    del sys.modules["aiosqlite"]
import aiosqlite as _real_aiosqlite  # noqa: E402, F401
import importlib as _importlib  # noqa: E402

if "app.database" in sys.modules:
    _db_mod = sys.modules["app.database"]
    if not hasattr(getattr(_db_mod, "aiosqlite", None), "connect"):
        del sys.modules["app.database"]
        _importlib.import_module("app.database")
else:
    _importlib.import_module("app.database")


def run(coro):
    return asyncio.get_event_loop().run_until_complete(coro)


@pytest.fixture()
def db_path(tmp_path, monkeypatch):
    path = tmp_path / "test.db"
    from app.config import settings
    monkeypatch.setattr(settings, "db_path", str(path))
    return str(path)


@pytest.fixture()
def initialized_db(db_path):
    from app import database
    run(database.init_db())
    return db_path


class TestAutoRateRunsTable:
    """auto_rate_runs-Tabelle wird beim init_db angelegt."""

    def test_table_exists(self, initialized_db):
        import sqlite3
        conn = sqlite3.connect(initialized_db)
        try:
            row = conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table' AND name='auto_rate_runs'"
            ).fetchone()
            assert row is not None

            cols = {r[1] for r in conn.execute("PRAGMA table_info(auto_rate_runs)")}
            for required in (
                "id", "started_at", "source", "bundesland",
                "limit_requested", "n_attempted", "n_succeeded",
                "n_failed", "n_skipped", "error_summary",
            ):
                assert required in cols, f"Spalte {required!r} fehlt"
        finally:
            conn.close()


class TestRecordAutoRateRun:
    """record_auto_rate_run schreibt einen Run-Eintrag und liefert die ID."""

    def test_insert_returns_id(self, initialized_db):
        from app.database import record_auto_rate_run
        run_id = run(record_auto_rate_run(
            source="manual", limit_requested=10,
            bundesland="NRW",
            n_attempted=10, n_succeeded=8, n_failed=1, n_skipped=1,
        ))
        assert isinstance(run_id, int)
        assert run_id > 0

    def test_inserted_values_persisted(self, initialized_db):
        from app.database import record_auto_rate_run
        import sqlite3
        run_id = run(record_auto_rate_run(
            source="cron", limit_requested=30,
            bundesland=None,  # ALL
            n_attempted=25, n_succeeded=20, n_failed=2, n_skipped=3,
            error_summary="3 Drucksachen ohne Adapter",
        ))
        conn = sqlite3.connect(initialized_db)
        try:
            row = conn.execute(
                "SELECT source, bundesland, limit_requested, n_attempted, "
                "n_succeeded, n_failed, n_skipped, error_summary "
                "FROM auto_rate_runs WHERE id=?", (run_id,)
            ).fetchone()
        finally:
            conn.close()
        assert row == ("cron", None, 30, 25, 20, 2, 3, "3 Drucksachen ohne Adapter")

    def test_default_zero_counts(self, initialized_db):
        from app.database import record_auto_rate_run
        import sqlite3
        run_id = run(record_auto_rate_run(
            source="api", limit_requested=5,
        ))
        conn = sqlite3.connect(initialized_db)
        try:
            row = conn.execute(
                "SELECT n_attempted, n_succeeded, n_failed, n_skipped FROM auto_rate_runs WHERE id=?",
                (run_id,),
            ).fetchone()
        finally:
            conn.close()
        assert row == (0, 0, 0, 0)


class TestListAutoRateRuns:
    """list_auto_rate_runs liefert die letzten N Runs (neueste zuerst)."""

    def test_empty_db_returns_empty_list(self, initialized_db):
        from app.database import list_auto_rate_runs
        out = run(list_auto_rate_runs(limit=10))
        assert out == []

    def test_returns_newest_first(self, initialized_db):
        from app.database import record_auto_rate_run, list_auto_rate_runs
        for source in ("first", "second", "third"):
            run(record_auto_rate_run(source=source, limit_requested=10))
        out = run(list_auto_rate_runs(limit=10))
        # neueste zuerst — "third" sollte index 0 sein
        assert out[0]["source"] == "third"
        assert out[1]["source"] == "second"
        assert out[2]["source"] == "first"

    def test_respects_limit(self, initialized_db):
        from app.database import record_auto_rate_run, list_auto_rate_runs
        for i in range(5):
            run(record_auto_rate_run(source=f"r{i}", limit_requested=10))
        out = run(list_auto_rate_runs(limit=3))
        assert len(out) == 3


class TestAutoRateTodayTotal:
    """auto_rate_today_total summiert nur die heutigen Runs."""

    def test_empty_returns_zero(self, initialized_db):
        from app.database import auto_rate_today_total
        result = run(auto_rate_today_total())
        assert result == {"n_runs": 0, "total_attempted": 0, "total_succeeded": 0}

    def test_sums_today_runs(self, initialized_db):
        from app.database import record_auto_rate_run, auto_rate_today_total
        run(record_auto_rate_run(source="cron", limit_requested=30,
                                 n_attempted=10, n_succeeded=8))
        run(record_auto_rate_run(source="cron", limit_requested=30,
                                 n_attempted=15, n_succeeded=12))
        result = run(auto_rate_today_total())
        assert result["n_runs"] == 2
        assert result["total_attempted"] == 25
        assert result["total_succeeded"] == 20