patrickandClaude Sonnet 5
c60c0d6c90
Neuer retention_service.py: Lohn-/zeitrelevante Daten (time_entries, hours_payouts) werden nach konfigurierbarer Frist gelöscht (company.settings.retention_lohn_years, Default 10 Jahre). Technische Tabellen mit fester Frist: audit_logs (3 Jahre), abgelaufene sessions/password_resets (sofort). Täglicher Scheduler-Job (03:00 Uhr, Redis-Tageslock analog Reminder-Jobs) plus manuelle Trigger: POST /companies/me/run-retention-purge (COMPANY_ADMIN/HR, nur eigene Firma) und POST /admin/run-retention-purge (SUPER_ADMIN, global inkl. technischer Tabellen). Letzter offener Punkt aus dem DSGVO-Löschkonzept (Art. 15/17 bereits erledigt). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Gis16MnuwkYcivLrSxK1pD
109 lines
4.3 KiB
Python
109 lines
4.3 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()
|
|
return {"Authorization": f"Bearer {data['access_token']}"}, data["user"]["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(
|
|
"INSERT INTO time_entries (id, user_id, date, start_time, end_time, status, source) "
|
|
"VALUES (gen_random_uuid(), :uid, :d, '08:00', '16:00', '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, status, source) "
|
|
"VALUES (gen_random_uuid(), :uid, :d, '08:00', '16:00', 'approved', 'web')"
|
|
), {"uid": user_id, "d": recent_date})
|
|
await db_session.commit()
|
|
|
|
result = await run_retention_purge(db_session, company_id=company_id)
|
|
await db_session.commit()
|
|
assert result["time_entries"] == 1
|
|
|
|
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(
|
|
"INSERT INTO time_entries (id, user_id, date, start_time, end_time, status, source) "
|
|
"VALUES (gen_random_uuid(), :uid, :d, '08:00', '16:00', 'approved', 'web')"
|
|
), {"uid": user_id, "d": three_years_ago})
|
|
await db_session.commit()
|
|
|
|
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(
|
|
"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()
|
|
|
|
result = await run_retention_purge(db_session, company_id=None)
|
|
await db_session.commit()
|
|
assert result["sessions"] >= 1
|
|
assert result["audit_logs"] >= 1
|