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
|