2902164eff
Neue Tests in dieser Migration: - test_database.py (Merkliste-CRUD, Subscriptions, abgeordnetenwatch-Joins) - test_clustering.py (82% Coverage) - test_drucksache_typen.py (100%) - test_mail.py (86%) - test_monitoring.py (23 Tests) - test_abgeordnetenwatch.py (23 Tests, inkl. Drucksache-Extraction) - test_redline_parser.py (20 Tests fuer §INS§/§DEL§-Marker) - test_bug_regressions.py (PRAGMA, JWT-azp, CDU-PDF, PFLICHT-FRAKTIONEN, NRW-Titel) - test_embeddings_v3_v4.py (WRITE/READ-Pattern) - test_wahlprogramm_check.py (#128) - test_wahlprogramm_fetch.py (#138) - test_antrag/bewertung/abonnement_repository.py + test_llm_bewerter.py (DDD) - test_domain_behavior.py (5 Domain-Methoden boundary tests) - tests/e2e/test_ui.py (Playwright) Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
96 lines
3.2 KiB
Python
96 lines
3.2 KiB
Python
"""Tests für AbonnementRepository (#136, ADR 0008)."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
from app.repositories import (
|
|
AbonnementRepository,
|
|
InMemoryAbonnementRepository,
|
|
SqliteAbonnementRepository,
|
|
)
|
|
|
|
|
|
def _run(coro):
|
|
return asyncio.get_event_loop().run_until_complete(coro)
|
|
|
|
|
|
class TestProtocolConformance:
|
|
def test_in_memory_implements_protocol(self):
|
|
repo = InMemoryAbonnementRepository()
|
|
assert isinstance(repo, AbonnementRepository)
|
|
|
|
def test_sqlite_implements_protocol(self):
|
|
repo = SqliteAbonnementRepository()
|
|
assert isinstance(repo, AbonnementRepository)
|
|
|
|
|
|
class TestCreateAndList:
|
|
def test_create_returns_id(self):
|
|
repo = InMemoryAbonnementRepository()
|
|
sid = _run(repo.create("u1", "a@b.de"))
|
|
assert sid == 1
|
|
|
|
def test_create_increments_id(self):
|
|
repo = InMemoryAbonnementRepository()
|
|
s1 = _run(repo.create("u1", "a@b.de"))
|
|
s2 = _run(repo.create("u1", "c@d.de"))
|
|
assert s2 == s1 + 1
|
|
|
|
def test_list_by_user_filters(self):
|
|
repo = InMemoryAbonnementRepository()
|
|
_run(repo.create("u1", "a@b.de"))
|
|
_run(repo.create("u2", "c@d.de"))
|
|
rows = _run(repo.list_by_user("u1"))
|
|
assert len(rows) == 1
|
|
assert rows[0]["email"] == "a@b.de"
|
|
|
|
def test_list_all_returns_every_sub(self):
|
|
repo = InMemoryAbonnementRepository()
|
|
_run(repo.create("u1", "a@b.de"))
|
|
_run(repo.create("u2", "c@d.de"))
|
|
assert len(_run(repo.list_all())) == 2
|
|
|
|
|
|
class TestListDue:
|
|
def test_list_due_returns_unsent(self):
|
|
repo = InMemoryAbonnementRepository()
|
|
_run(repo.create("u1", "a@b.de"))
|
|
due = _run(repo.list_due())
|
|
assert len(due) == 1
|
|
|
|
def test_mark_sent_removes_from_due(self):
|
|
repo = InMemoryAbonnementRepository()
|
|
sid = _run(repo.create("u1", "a@b.de"))
|
|
_run(repo.mark_sent(sid))
|
|
assert _run(repo.list_due()) == []
|
|
|
|
def test_list_due_filters_by_frequency(self):
|
|
repo = InMemoryAbonnementRepository()
|
|
_run(repo.create("u1", "a@b.de", frequency="daily"))
|
|
_run(repo.create("u2", "c@d.de", frequency="weekly"))
|
|
daily = _run(repo.list_due("daily"))
|
|
weekly = _run(repo.list_due("weekly"))
|
|
assert len(daily) == 1 and daily[0]["email"] == "a@b.de"
|
|
assert len(weekly) == 1 and weekly[0]["email"] == "c@d.de"
|
|
|
|
|
|
class TestDelete:
|
|
def test_delete_checks_ownership(self):
|
|
repo = InMemoryAbonnementRepository()
|
|
sid = _run(repo.create("u1", "a@b.de"))
|
|
# Fremder User kann nicht löschen
|
|
assert _run(repo.delete("u2", sid)) is False
|
|
# Eigentümer kann löschen
|
|
assert _run(repo.delete("u1", sid)) is True
|
|
|
|
def test_delete_by_id_skips_ownership_check(self):
|
|
repo = InMemoryAbonnementRepository()
|
|
sid = _run(repo.create("u1", "a@b.de"))
|
|
# delete_by_id ist für Unsubscribe-Links (Token-gesichert), nicht für
|
|
# den Self-Service; kein User-Check
|
|
assert _run(repo.delete_by_id(sid)) is True
|
|
|
|
def test_delete_by_id_missing_returns_false(self):
|
|
repo = InMemoryAbonnementRepository()
|
|
assert _run(repo.delete_by_id(999)) is False
|