govoplan-core/tests/test_dsar_api.py

from __future__ import annotations

from types import SimpleNamespace
import unittest
from unittest.mock import patch

from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from govoplan_core.auth import ApiPrincipal, get_api_principal
from govoplan_core.core.access import PrincipalRef
from govoplan_core.core.dsar import DsarRecordRef
from govoplan_core.db.base import Base
from govoplan_core.db.session import get_session
from govoplan_core.privacy.dsar_workflow import DataSubjectRequest
from govoplan_core.server.dsar import router


class _Provider:
    provider_id = "example"
    module_id = "example"

    def search_subject(self, session, *, tenant_id, subject):
        del session, tenant_id, subject
        return (
            DsarRecordRef(
                provider_id="example",
                module_id="example",
                resource_type="profile",
                resource_id="profile-1",
                category="profile",
                title="Example profile",
            ),
        )

    def plan_erasure(self, session, *, tenant_id, subject, records):
        del session, tenant_id, subject, records
        return ()

    def execute_erasure(self, session, *, tenant_id, subject, actions, request_id):
        del session, tenant_id, subject, actions, request_id
        return ()


class _Registry:
    provider = _Provider()

    def capability_names(self):
        return ("privacy.dsar.example",)

    def require_tenant_capability(self, name, session, **kwargs):
        del name, session, kwargs
        return self.provider

    def manifests(self):
        return (SimpleNamespace(id="example"),)


class DsarApiTests(unittest.TestCase):
    def setUp(self) -> None:
        self.engine = create_engine(
            "sqlite+pysqlite://",
            connect_args={"check_same_thread": False},
            poolclass=StaticPool,
        )
        DataSubjectRequest.__table__.create(self.engine)
        self.session = sessionmaker(bind=self.engine, expire_on_commit=False)()
        self.principal = _principal(
            "access:privacy:read",
            "access:privacy:manage",
            "access:privacy:export",
            "access:privacy:erase",
        )
        app = FastAPI()
        app.state.govoplan_registry = _Registry()
        app.include_router(router, prefix="/api/v1")
        app.dependency_overrides[get_session] = lambda: self.session
        app.dependency_overrides[get_api_principal] = lambda: self.principal
        self.audit_patch = patch("govoplan_core.server.dsar.audit_event")
        self.audit_patch.start()
        self.client = TestClient(app)

    def tearDown(self) -> None:
        self.client.close()
        self.audit_patch.stop()
        self.session.close()
        Base.metadata.drop_all(self.engine, tables=[DataSubjectRequest.__table__])
        self.engine.dispose()

    def test_mutation_requires_strong_revision_and_export_is_available(self) -> None:
        created = self.client.post(
            "/api/v1/admin/privacy/data-subject-requests",
            json={
                "reference": "DSAR-1",
                "request_kind": "access",
                "subject": {"email": "ada@example.test"},
                "purpose": "Verified access request",
            },
        )
        self.assertEqual(201, created.status_code, created.text)
        item = created.json()["request"]

        missing = self.client.post(
            f"/api/v1/admin/privacy/data-subject-requests/{item['id']}/search",
            json={"base_revision": item["resource_revision"]},
        )
        self.assertEqual(428, missing.status_code, missing.text)

        searched = self.client.post(
            f"/api/v1/admin/privacy/data-subject-requests/{item['id']}/search",
            headers={"If-Match": item["etag"]},
            json={"base_revision": item["resource_revision"]},
        )
        self.assertEqual(200, searched.status_code, searched.text)
        self.assertEqual(1, searched.json()["search"]["record_count"])

        stale = self.client.post(
            f"/api/v1/admin/privacy/data-subject-requests/{item['id']}/search",
            headers={"If-Match": item["etag"]},
            json={"base_revision": item["resource_revision"]},
        )
        self.assertEqual(409, stale.status_code, stale.text)

        exported = self.client.get(
            f"/api/v1/admin/privacy/data-subject-requests/{item['id']}/export"
        )
        self.assertEqual(200, exported.status_code, exported.text)
        self.assertIn("govoplan.dsars.export.v1", exported.text)

    def test_privacy_scopes_are_independent(self) -> None:
        self.principal = _principal("access:privacy:read")
        denied = self.client.post(
            "/api/v1/admin/privacy/data-subject-requests",
            json={
                "reference": "DSAR-2",
                "request_kind": "access",
                "subject": {"email": "ada@example.test"},
                "purpose": "Verified access request",
            },
        )
        self.assertEqual(403, denied.status_code, denied.text)
        self.assertEqual(200, self.client.get("/api/v1/admin/privacy/data-subject-requests").status_code)


def _principal(*scopes: str) -> ApiPrincipal:
    return ApiPrincipal(
        principal=PrincipalRef(
            account_id="account-operator",
            membership_id="membership-operator",
            tenant_id="tenant-1",
            scopes=frozenset(scopes),
        ),
        account=SimpleNamespace(id="account-operator"),
        user=SimpleNamespace(id="membership-operator"),
    )


if __name__ == "__main__":
    unittest.main()