97 lines
3.9 KiB
Python
97 lines
3.9 KiB
Python
from sqlalchemy import delete, select
|
|
|
|
from app.core.db import SessionLocal
|
|
from app.models.audit import AuditEvent
|
|
from app.models.customer import Customer
|
|
|
|
|
|
def test_privacy_retention_uses_persisted_counts(ops_client):
|
|
response = ops_client.get("/api/v1/privacy/retention")
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["customers_total"] >= 180
|
|
assert body["minimum_booking_retention_days"] == 30
|
|
assert body["audit_retention_days"] == 2555
|
|
|
|
|
|
def test_privacy_retention_requires_manager(employee_client):
|
|
assert employee_client.get("/api/v1/privacy/retention").status_code == 403
|
|
|
|
|
|
def test_customer_export_is_downloadable_and_audited_without_mutation(ops_client):
|
|
response = ops_client.get("/api/v1/privacy/customers/CUS-0001/export")
|
|
assert response.status_code == 200
|
|
assert response.headers["content-disposition"] == 'attachment; filename="CUS-0001-privacy.json"'
|
|
assert response.json()["customer"]["public_ref"] == "CUS-0001"
|
|
assert isinstance(response.json()["bookings"], list)
|
|
events = ops_client.get("/api/v1/audit", params={"action": "privacy_customer_exported"}).json()
|
|
assert any(event["metadata"]["customer_ref"] == "CUS-0001" for event in events)
|
|
|
|
|
|
def test_active_customer_cannot_be_anonymized(ops_client):
|
|
response = ops_client.post(
|
|
"/api/v1/privacy/customers/CUS-0042/anonymize",
|
|
json={"confirmation": "CUS-0042", "reason": "Verified erasure request"},
|
|
)
|
|
assert response.status_code == 409
|
|
|
|
|
|
def test_eligible_customer_is_irreversibly_anonymized_without_pii_in_audit(ops_client):
|
|
with SessionLocal() as db:
|
|
customer = Customer(
|
|
public_ref="CUS-PRIVACY",
|
|
first_name="Private",
|
|
last_name="Person",
|
|
email="private.person@example.test",
|
|
phone="+32000000000",
|
|
postal_code="1000",
|
|
city="Brussels",
|
|
date_of_birth=None,
|
|
)
|
|
db.add(customer)
|
|
db.commit()
|
|
customer_id = customer.id
|
|
try:
|
|
mismatch = ops_client.post(
|
|
"/api/v1/privacy/customers/CUS-PRIVACY/anonymize",
|
|
json={"confirmation": "CUS-WRONG", "reason": "Verified erasure request"},
|
|
)
|
|
assert mismatch.status_code == 422
|
|
response = ops_client.post(
|
|
"/api/v1/privacy/customers/CUS-PRIVACY/anonymize",
|
|
json={"confirmation": "CUS-PRIVACY", "reason": "Verified erasure request"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "anonymized"
|
|
with SessionLocal() as db:
|
|
customer = db.scalar(select(Customer).where(Customer.id == customer_id))
|
|
assert customer is not None
|
|
assert customer.email is None
|
|
assert customer.phone is None
|
|
assert customer.first_name == "Anoniem"
|
|
event = db.scalar(
|
|
select(AuditEvent)
|
|
.where(AuditEvent.action == "privacy_customer_anonymized")
|
|
.order_by(AuditEvent.occurred_at.desc())
|
|
)
|
|
assert event is not None
|
|
serialized = f"{event.before_json}{event.after_json}{event.metadata_json}"
|
|
assert "private.person@example.test" not in serialized
|
|
finally:
|
|
with SessionLocal() as db:
|
|
db.execute(delete(AuditEvent).where(AuditEvent.entity_id == customer_id))
|
|
db.execute(delete(Customer).where(Customer.id == customer_id))
|
|
db.commit()
|
|
|
|
|
|
def test_audit_csv_export_is_bounded_and_audited(ops_client):
|
|
response = ops_client.get("/api/v1/audit/export.csv")
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"].startswith("text/csv")
|
|
assert response.text.startswith("id,occurred_at,actor_type")
|
|
invalid = ops_client.get(
|
|
"/api/v1/audit/export.csv",
|
|
params={"occurred_from": "2025-01-01T00:00:00Z", "occurred_to": "2026-01-01T00:00:00Z"},
|
|
)
|
|
assert invalid.status_code == 422
|