patrickandClaude Sonnet 5
cfa707d8f9
Gleiches Muster wie hours_payouts.status (7166bbb): Model hatte nur
Python-default, kein server_default -> Migration 0044 zieht DB-Default
nach. Retention-Tests nutzten fehlenden Wert bei Raw-SQL-Insert und
fehlenden RLS-Bypass zwischen den Transaktionsgrenzen.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Gis16MnuwkYcivLrSxK1pD
118 lines
4.9 KiB
Python
118 lines
4.9 KiB
Python
"""Tests für DSGVO-Löschkonzept (retention_service.py)."""
|
|
from datetime import date, datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from httpx import AsyncClient
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.services.retention_service import run_retention_purge
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="session", loop_scope="session")
|
|
async def retention_headers(client: AsyncClient):
|
|
resp = await client.post("/api/v1/auth/register", json={
|
|
"company_name": "Retention GmbH",
|
|
"first_name": "Rita",
|
|
"last_name": "Tention",
|
|
"email": "admin@retentiongmbh.de",
|
|
"password": "Secret123",
|
|
})
|
|
assert resp.status_code == 201, resp.text
|
|
data = resp.json()
|
|
headers = {"Authorization": f"Bearer {data['access_token']}"}
|
|
me = await client.get("/api/v1/users/me", headers=headers)
|
|
return headers, me.json()["company_id"]
|
|
|
|
|
|
@pytest.mark.asyncio(loop_scope="session")
|
|
async def test_purge_deletes_old_time_entries_respects_retention(
|
|
client: AsyncClient, db_session: AsyncSession, retention_headers
|
|
):
|
|
headers, company_id = retention_headers
|
|
me = await client.get("/api/v1/users/me", headers=headers)
|
|
user_id = me.json()["id"]
|
|
|
|
old_date = date.today() - timedelta(days=11 * 365) # älter als Default 10 Jahre
|
|
recent_date = date.today() - timedelta(days=30)
|
|
|
|
await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))
|
|
await db_session.execute(text(
|
|
"INSERT INTO time_entries (id, user_id, date, start_time, end_time, break_minutes, status, source) "
|
|
"VALUES (gen_random_uuid(), :uid, :d, '08:00', '16:00', 0, 'approved', 'web')"
|
|
), {"uid": user_id, "d": old_date})
|
|
await db_session.execute(text(
|
|
"INSERT INTO time_entries (id, user_id, date, start_time, end_time, break_minutes, status, source) "
|
|
"VALUES (gen_random_uuid(), :uid, :d, '08:00', '16:00', 0, 'approved', 'web')"
|
|
), {"uid": user_id, "d": recent_date})
|
|
await db_session.commit()
|
|
|
|
await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))
|
|
result = await run_retention_purge(db_session, company_id=company_id)
|
|
await db_session.commit()
|
|
assert result["time_entries"] == 1
|
|
|
|
await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))
|
|
remaining = await db_session.execute(text(
|
|
"SELECT date FROM time_entries WHERE user_id = :uid"
|
|
), {"uid": user_id})
|
|
dates = [r[0] for r in remaining]
|
|
assert old_date not in dates
|
|
assert recent_date in dates
|
|
|
|
|
|
@pytest.mark.asyncio(loop_scope="session")
|
|
async def test_purge_configurable_retention_years(
|
|
client: AsyncClient, db_session: AsyncSession, retention_headers
|
|
):
|
|
headers, company_id = retention_headers
|
|
up = await client.patch("/api/v1/companies/me",
|
|
json={"settings": {"retention_lohn_years": 2}}, headers=headers)
|
|
assert up.status_code == 200, up.text
|
|
|
|
me = await client.get("/api/v1/users/me", headers=headers)
|
|
user_id = me.json()["id"]
|
|
three_years_ago = date.today() - timedelta(days=3 * 365)
|
|
|
|
await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))
|
|
await db_session.execute(text(
|
|
"INSERT INTO time_entries (id, user_id, date, start_time, end_time, break_minutes, status, source) "
|
|
"VALUES (gen_random_uuid(), :uid, :d, '08:00', '16:00', 0, 'approved', 'web')"
|
|
), {"uid": user_id, "d": three_years_ago})
|
|
await db_session.commit()
|
|
|
|
await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))
|
|
result = await run_retention_purge(db_session, company_id=company_id)
|
|
await db_session.commit()
|
|
assert result["time_entries"] >= 1
|
|
|
|
|
|
@pytest.mark.asyncio(loop_scope="session")
|
|
async def test_purge_expired_sessions_and_audit_logs_global(
|
|
client: AsyncClient, db_session: AsyncSession, retention_headers
|
|
):
|
|
headers, company_id = retention_headers
|
|
me = await client.get("/api/v1/users/me", headers=headers)
|
|
user_id = me.json()["id"]
|
|
|
|
expired = datetime.now(timezone.utc) - timedelta(days=1)
|
|
old_audit = datetime.now(timezone.utc) - timedelta(days=4 * 365)
|
|
|
|
await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))
|
|
await db_session.execute(text(
|
|
"INSERT INTO sessions (id, user_id, refresh_token_hash, expires_at) "
|
|
"VALUES (gen_random_uuid(), :uid, :h, :exp)"
|
|
), {"uid": user_id, "h": "expired-hash-retention-test", "exp": expired})
|
|
await db_session.execute(text(
|
|
"INSERT INTO audit_logs (id, company_id, user_id, action, created_at) "
|
|
"VALUES (gen_random_uuid(), :cid, :uid, 'test_old_action', :ts)"
|
|
), {"cid": company_id, "uid": user_id, "ts": old_audit})
|
|
await db_session.commit()
|
|
|
|
await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))
|
|
result = await run_retention_purge(db_session, company_id=None)
|
|
await db_session.commit()
|
|
assert result["sessions"] >= 1
|
|
assert result["audit_logs"] >= 1
|