quality: 9 Tests fuer auto_rate_runs + ruff F401 in main.py

Phase 8 (Code-Pflege):

- Neue Test-Datei tests/test_auto_rate_runs.py (9 Cases) deckt
  record_auto_rate_run, list_auto_rate_runs, auto_rate_today_total
  und das Schema ab.
- list_auto_rate_runs sortiert jetzt by id DESC (statt started_at DESC),
  weil started_at nur sekundengenau ist und Sub-Sekunden-Inserts
  unstabilen Output produzierten.
- ruff --select F401 --fix auf main.py: 7 ungenutzte Imports entfernt
  (MAX_SEARCH_QUERY_LEN, import_json_assessments, KLEINE_ANFRAGE,
  BUNDESLAENDER, lokale sqlite3/json/timezone-Reimports). Tests
  weiterhin grün (74 passed).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

app/database.py

@@ -725,7 +725,11 @@ async def record_auto_rate_run(
 
 
 async def list_auto_rate_runs(limit: int = 20) -> list[dict]:
-    """Letzte N Runs (neueste zuerst)."""
+    """Letzte N Runs (neueste zuerst).
+
+    Ordnung: ``id DESC`` als stabiler Tiebreaker bei Sub-Sekunden-
+    Inserts (``started_at`` ist nur bis Sekunde genau).
+    """
     async with aiosqlite.connect(settings.db_path) as db:
         db.row_factory = aiosqlite.Row
         rows = await db.execute(
@@ -733,7 +737,7 @@ async def list_auto_rate_runs(limit: int = 20) -> list[dict]:
             SELECT id, started_at, source, bundesland, limit_requested,
                    n_attempted, n_succeeded, n_failed, n_skipped, error_summary
             FROM auto_rate_runs
-            ORDER BY started_at DESC LIMIT ?
+            ORDER BY id DESC LIMIT ?
             """,
             (limit,),
         )

app/main.py

@@ -16,7 +16,6 @@ from slowapi.util import get_remote_address
 from slowapi.errors import RateLimitExceeded
 
 from .validators import (
-    MAX_SEARCH_QUERY_LEN,
     validate_drucksache,
     validate_search_query,
 )
@@ -34,8 +33,7 @@ from .config import settings
 from .database import (
     init_db, get_job, create_job, update_job,
     get_all_assessments, get_assessment, delete_assessment,
-    upsert_assessment, import_json_assessments,
+    upsert_assessment, search_assessments,
-    search_assessments,
     toggle_bookmark, get_bookmarks, add_comment, get_comments, delete_comment,
     create_subscription, list_subscriptions, list_all_subscriptions, delete_subscription,
     delete_subscription_by_id,
@@ -1404,7 +1402,7 @@ async def search_landtag(
         external = adapter._filter_abstimmbar(await adapter.search(q, limit))
         # Zusätzliche Title-Heuristik: bei Adaptern die Typ='Drucksache' liefern
         # (NRW), Kleine-Anfrage-Frage-Pattern erkennen und ausfiltern.
-        from .drucksache_typen import likely_kleine_anfrage_titel, KLEINE_ANFRAGE
+        from .drucksache_typen import likely_kleine_anfrage_titel
         results = []
         for doc in external:
             if doc.typ_normiert == "sonstige" and likely_kleine_anfrage_titel(doc.title):
@@ -1798,7 +1796,7 @@ async def datenschutz_page(request: Request, current_user: Optional[dict] = Depe
 @app.get("/methodik", response_class=HTMLResponse)
 async def methodik_page(request: Request, current_user: Optional[dict] = Depends(get_current_user)):
     """Transparenz-/Methodik-Seite (#96)."""
-    from .bundeslaender import aktive_bundeslaender, BUNDESLAENDER
+    from .bundeslaender import aktive_bundeslaender
     from .embeddings import get_indexing_status
     from .analyzer import get_system_prompt, get_user_prompt_template
     from .protokoll_parsers import supported_bundeslaender as _plenum_supported
@@ -2356,7 +2354,6 @@ async def auswertungen_score_histogram(
     Liefert ein Bucket-Array fuer einen Histogramm-Chart. Filterbar
     ueber Bundesland + Wahlperiode (gleicher Pattern wie /matrix).
     """
-    import sqlite3
     from .auswertungen import _load_assessments
     rows = _load_assessments()
     from .wahlperioden import wahlperiode_for
@@ -2411,7 +2408,6 @@ async def auswertungen_themen_matrix(min_count: int = 3, bundesland: Optional[st
     mindestens `min_count` Assessments werden angezeigt. Optional auf
     ein Bundesland einschränken.
     """
-    import json as _json
     from collections import Counter, defaultdict
     from .parteien import normalize_partei
 
@@ -2548,7 +2544,7 @@ async def feed_xml(request: Request, bundesland: Optional[str] = None, partei: O
     - partei: optionaler Partei-Filter (CDU, SPD, GRÜNE, …) — matcht gegen fraktionen-Liste
     - limit: Anzahl Einträge (default 50, max 200)
     """
-    from datetime import datetime, timezone
+    from datetime import datetime
     import hashlib
     import html
 

tests/test_auto_rate_runs.py

@@ -0,0 +1,163 @@
+"""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