govoplan-decisions/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_core.privacy.dsar_workflow import (
    create_data_subject_request,
    search_data_subject_request,
)
from govoplan_decisions.backend.db.models import FormalDecisionRevision
from govoplan_decisions.backend.dsar_provider import (
    DECISIONS_DSAR_CAPABILITY,
    DecisionsDsarProvider,
)
from govoplan_decisions.backend.manifest import manifest


NOW = datetime(2026, 8, 21, 14, 0, tzinfo=UTC)


class _Registry:
    def __init__(self, provider: DecisionsDsarProvider) -> None:
        self.provider = provider

    def capability_names(self):
        return (DECISIONS_DSAR_CAPABILITY,)

    def capability_owner(self, name):
        if name != DECISIONS_DSAR_CAPABILITY:
            raise KeyError(name)
        return "decisions"

    def tenant_entitlement_resolver(self):
        class _Resolver:
            @staticmethod
            def resolve(session, tenant_id):
                del session, tenant_id
                return type("State", (), {"effective_modules": ("decisions",)})()

        return _Resolver()

    def require_tenant_capability(self, name, session, **kwargs):
        del session, kwargs
        if name != DECISIONS_DSAR_CAPABILITY:
            raise KeyError(name)
        return self.provider

    def manifests(self):
        return (type("Manifest", (), {"id": "decisions"})(),)


class DecisionsDsarProviderTests(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 = DecisionsDsarProvider()
        self.assertIsInstance(self.provider, DsarProvider)
        self.session.add_all(
            (
                self._decision(
                    "revision-1", decision_id="decision-1", created_by="account-1"
                ),
                self._decision(
                    "revision-other",
                    decision_id="decision-other",
                    created_by="account-other",
                ),
                self._decision(
                    "revision-other-tenant",
                    tenant_id="tenant-2",
                    decision_id="decision-other-tenant",
                    created_by="account-1",
                ),
            )
        )
        self.session.commit()

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

    @staticmethod
    def _decision(
        row_id: str,
        *,
        decision_id: str,
        created_by: str,
        tenant_id: str = "tenant-1",
    ) -> FormalDecisionRevision:
        return FormalDecisionRevision(
            id=row_id,
            tenant_id=tenant_id,
            decision_id=decision_id,
            revision="1",
            decision_type="permit",
            state="effective",
            valid_from=NOW,
            recorded_at=NOW,
            payload={
                "reasoning": f"protected-reasoning-{row_id}-do-not-export",
                "operative_result": f"protected-result-{row_id}-do-not-export",
                "digest": f"payload-digest-{row_id}-do-not-export",
            },
            created_by=created_by,
        )

    @staticmethod
    def _subject() -> DsarSubjectRef:
        return DsarSubjectRef(account_id="account-1")

    def test_search_exports_minimized_creator_attribution(self) -> None:
        records = self.provider.search_subject(
            self.session, tenant_id="tenant-1", subject=self._subject()
        )
        self.assertEqual(["revision-1"], [record.resource_id for record in records])
        exported = json.dumps(records[0].to_dict())
        self.assertIn("decision-1", exported)
        self.assertIn("recorded_formal_decision_revision", exported)
        self.assertNotIn("protected-reasoning", exported)
        self.assertNotIn("protected-result", exported)
        self.assertNotIn("payload-digest", exported)
        self.assertNotIn("decision-other", exported)

    def test_decision_narrowing_and_actor_conflicts_fail_closed(self) -> None:
        narrowed = self.provider.search_subject(
            self.session,
            tenant_id="tenant-1",
            subject=DsarSubjectRef(
                account_id="account-1",
                external_references={"decisions.decision": "decision-1"},
            ),
        )
        conflict = self.provider.search_subject(
            self.session,
            tenant_id="tenant-1",
            subject=DsarSubjectRef(
                account_id="account-1",
                external_references={"decisions.actor": "account-other"},
            ),
        )
        decision_only = self.provider.search_subject(
            self.session,
            tenant_id="tenant-1",
            subject=DsarSubjectRef(
                external_references={"decisions.decision": "decision-1"}
            ),
        )
        self.assertEqual(["revision-1"], [record.resource_id for record in narrowed])
        self.assertEqual((), conflict)
        self.assertEqual((), decision_only)

    def test_erasure_is_retain_only(self) -> None:
        records = self.provider.search_subject(
            self.session, tenant_id="tenant-1", subject=self._subject()
        )
        actions = self.provider.plan_erasure(
            self.session,
            tenant_id="tenant-1",
            subject=self._subject(),
            records=records,
        )
        self.assertEqual(["retain"], [action.kind for action in actions])
        results = self.provider.execute_erasure(
            self.session,
            tenant_id="tenant-1",
            subject=self._subject(),
            actions=actions,
            request_id="dsar-decisions-1",
        )
        self.assertEqual(["blocked"], [result.status for result in results])
        self.assertEqual(3, self.session.query(FormalDecisionRevision).count())

    def test_manifest_and_core_workflow_discover_provider(self) -> None:
        self.assertIn(DECISIONS_DSAR_CAPABILITY, manifest.capability_factories)
        row = create_data_subject_request(
            self.session,
            tenant_id="tenant-1",
            reference="DSAR-DECISIONS-1",
            request_kind="access",
            subject=self._subject(),
            purpose="Decision attribution access request",
            legal_basis=None,
            due_at=None,
            requested_by_account_id="operator-1",
        )
        search_data_subject_request(
            self.session,
            registry=_Registry(self.provider),
            row=row,
            expected_revision=row.resource_revision,
        )
        self.assertEqual("searched", row.status)
        self.assertEqual(1, row.search_result["record_count"])


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