govoplan-projects/tests/test_dsar_provider.py

from __future__ import annotations

import json
import unittest
from datetime import UTC, datetime

from sqlalchemy import create_engine
from sqlalchemy.orm import Session

from govoplan_core.core.dsar import DsarProvider, DsarSubjectRef
from govoplan_core.db.base import Base
from govoplan_projects.backend.db.models import (
    ProjectMembershipGrant,
    ProjectObjectEvent,
    ProjectObjectIdentity,
    ProjectObjectRevision,
)
from govoplan_projects.backend.dsar_provider import (
    PROJECTS_DSAR_CAPABILITY,
    ProjectsDsarProvider,
)
from govoplan_projects.backend.manifest import manifest


NOW = datetime(2026, 8, 22, 15, 0, tzinfo=UTC)


class ProjectsDsarProviderTests(unittest.TestCase):
    def setUp(self) -> None:
        self.engine = create_engine("sqlite+pysqlite:///:memory:")
        Base.metadata.create_all(self.engine)
        self.session = Session(self.engine)
        self.provider = ProjectsDsarProvider()
        self._seed()
        self.session.commit()

    def tearDown(self) -> None:
        self.session.close()
        self.engine.dispose()

    def _seed(self) -> None:
        self.session.add_all(
            (
                ProjectObjectIdentity(
                    id="identity-row-1",
                    tenant_id="tenant-1",
                    object_kind="project",
                    object_id="project-1",
                    object_key="secret-object-key-do-not-export",
                    created_by="account-1",
                    created_at=NOW,
                    updated_at=NOW,
                ),
                ProjectObjectRevision(
                    id="revision-1",
                    tenant_id="tenant-1",
                    identity_id="identity-row-1",
                    object_kind="project",
                    object_id="project-1",
                    revision=2,
                    state="active",
                    title="Sensitive project title do not export",
                    visibility="restricted",
                    recorded_at=NOW,
                    search_text="project-search-content-do-not-export",
                    payload={"secret": "project-payload-do-not-export"},
                    changed_by="account-1",
                    created_at=NOW,
                    updated_at=NOW,
                ),
                ProjectMembershipGrant(
                    id="membership-1",
                    tenant_id="tenant-1",
                    object_kind="project",
                    object_id="project-1",
                    subject_kind="account",
                    subject_id="account-1",
                    role="member",
                    permissions=["read", "write"],
                    active=True,
                    source_revision=2,
                    created_at=NOW,
                    updated_at=NOW,
                ),
                ProjectMembershipGrant(
                    id="membership-other-person",
                    tenant_id="tenant-1",
                    object_kind="project",
                    object_id="project-1",
                    subject_kind="account",
                    subject_id="account-other-do-not-export",
                    role="owner",
                    permissions=["admin"],
                    active=True,
                    source_revision=2,
                ),
                ProjectObjectEvent(
                    id="event-row-1",
                    tenant_id="tenant-1",
                    object_kind="project",
                    object_id="project-1",
                    object_revision=2,
                    event_id="event-1",
                    event_type="projects.project.updated",
                    occurred_at=NOW,
                    actor_id="account-1",
                    idempotency_key="event-idempotency-do-not-export",
                    request_sha256="event-request-hash-do-not-export",
                    payload={"secret": "event-payload-do-not-export"},
                    created_at=NOW,
                    updated_at=NOW,
                ),
                ProjectMembershipGrant(
                    id="membership-other-tenant",
                    tenant_id="tenant-2",
                    object_kind="project",
                    object_id="project-other",
                    subject_kind="account",
                    subject_id="account-1",
                    role="member",
                    permissions=["read"],
                    active=True,
                    source_revision=1,
                ),
            )
        )

    def test_search_exports_membership_and_minimized_attribution(self) -> None:
        self.assertIsInstance(self.provider, DsarProvider)
        records = self.provider.search_subject(
            self.session,
            tenant_id="tenant-1",
            subject=DsarSubjectRef(account_id="account-1"),
        )
        self.assertEqual(
            {
                "project_membership",
                "project_identity_actor_attribution",
                "project_revision_actor_attribution",
                "project_event_actor_attribution",
            },
            {record.resource_type for record in records},
        )
        exported = json.dumps([record.to_dict() for record in records])
        self.assertIn("membership-1", exported)
        for excluded in (
            "account-other-do-not-export",
            "secret-object-key-do-not-export",
            "Sensitive project title do not export",
            "project-search-content-do-not-export",
            "project-payload-do-not-export",
            "event-idempotency-do-not-export",
            "event-request-hash-do-not-export",
            "event-payload-do-not-export",
        ):
            self.assertNotIn(excluded, exported)

    def test_requires_subject_identity_and_enforces_narrowing(self) -> None:
        self.assertEqual(
            (),
            self.provider.search_subject(
                self.session,
                tenant_id="tenant-1",
                subject=DsarSubjectRef(email="member@example.test"),
            ),
        )
        narrowed = self.provider.search_subject(
            self.session,
            tenant_id="tenant-1",
            subject=DsarSubjectRef(
                account_id="account-1",
                external_references={"projects.project": "project-1"},
            ),
        )
        self.assertEqual(4, len(narrowed))
        conflict = self.provider.search_subject(
            self.session,
            tenant_id="tenant-1",
            subject=DsarSubjectRef(
                account_id="account-1",
                external_references={
                    "projects.project": "project-1",
                    "projects.portfolio": "portfolio-1",
                },
            ),
        )
        self.assertEqual((), conflict)

    def test_membership_requires_review_and_attribution_is_retained(self) -> None:
        subject = DsarSubjectRef(account_id="account-1")
        records = self.provider.search_subject(
            self.session, tenant_id="tenant-1", subject=subject
        )
        actions = self.provider.plan_erasure(
            self.session,
            tenant_id="tenant-1",
            subject=subject,
            records=records,
        )
        by_type = {action.resource_type: action.kind for action in actions}
        self.assertEqual("manual_review", by_type["project_membership"])
        self.assertEqual("retain", by_type["project_event_actor_attribution"])

    def test_manifest_registers_provider_and_documentation(self) -> None:
        self.assertIn(PROJECTS_DSAR_CAPABILITY, manifest.capability_factories)
        self.assertIn(
            "projects.data-subject-requests",
            {topic.id for topic in manifest.documentation},
        )


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