From eaca78803d96b684c020625d6ee056c82b2196e6 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Fri, 21 Aug 2026 03:54:39 +0200 Subject: [PATCH] feat(quick-access): add governed DSAR coverage --- .../backend/dsar_provider.py | 417 ++++++++++++++++++ src/govoplan_quick_access/backend/manifest.py | 69 +++ tests/test_dsar_provider.py | 377 ++++++++++++++++ 3 files changed, 863 insertions(+) create mode 100644 src/govoplan_quick_access/backend/dsar_provider.py create mode 100644 tests/test_dsar_provider.py diff --git a/src/govoplan_quick_access/backend/dsar_provider.py b/src/govoplan_quick_access/backend/dsar_provider.py new file mode 100644 index 0000000..0f7c939 --- /dev/null +++ b/src/govoplan_quick_access/backend/dsar_provider.py @@ -0,0 +1,417 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from datetime import datetime, timezone + +from sqlalchemy import or_ +from sqlalchemy.orm import Session + +from govoplan_core.core.dsar import ( + DsarErasureActionRef, + DsarExecutionResultRef, + DsarRecordRef, + DsarSubjectRef, + dsar_capability_name, +) +from govoplan_quick_access.backend.db.models import QuickAccessProfile + + +QUICK_ACCESS_DSAR_CAPABILITY = dsar_capability_name("quick_access") +_MAX_PREFERENCES = 1_000 +_CONFLICT = object() + + +@dataclass(frozen=True, slots=True) +class _SubjectSelectors: + account_id: str + profile_id: str | None + + +class QuickAccessDsarProvider: + provider_id = "quick_access" + module_id = "quick_access" + + def search_subject( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + ) -> Sequence[DsarRecordRef]: + db = _session(session) + selectors = _subject_selectors(subject) + if selectors is None: + return () + + personal_query = db.query(QuickAccessProfile).filter( + QuickAccessProfile.scope_type == "user", + QuickAccessProfile.tenant_id == tenant_id, + QuickAccessProfile.scope_id == selectors.account_id, + ) + attribution_query = db.query(QuickAccessProfile).filter( + QuickAccessProfile.scope_type == "tenant", + QuickAccessProfile.tenant_id == tenant_id, + or_( + QuickAccessProfile.created_by == selectors.account_id, + QuickAccessProfile.updated_by == selectors.account_id, + ), + ) + if selectors.profile_id: + personal_query = personal_query.filter( + QuickAccessProfile.id == selectors.profile_id + ) + attribution_query = attribution_query.filter( + QuickAccessProfile.id == selectors.profile_id + ) + + records = [_personal_record(row) for row in personal_query.all()] + records.extend( + _attribution_record(row, selectors.account_id) + for row in attribution_query.all() + ) + return tuple( + sorted(records, key=lambda item: (item.resource_type, item.resource_id)) + ) + + def plan_erasure( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + records: Sequence[DsarRecordRef], + ) -> Sequence[DsarErasureActionRef]: + db = _session(session) + selectors = _subject_selectors(subject) + if selectors is None: + raise ValueError("Quick Access DSAR requires one corroborated account.") + + actions: list[DsarErasureActionRef] = [] + for record in records: + _validate_record(record) + if record.resource_type == "quick_access_tenant_attribution": + actions.append( + DsarErasureActionRef( + action_id=( + "quick_access:retain:quick_access_tenant_attribution:" + f"{record.resource_id}" + ), + provider_id=self.provider_id, + module_id=self.module_id, + kind="retain", + resource_type=record.resource_type, + resource_id=record.resource_id, + title=f"Retain {record.title}", + rationale=record.retention_reason + or "Tenant Quick Access policy attribution is accountability evidence.", + executable=False, + ) + ) + continue + + profile = ( + db.query(QuickAccessProfile) + .filter( + QuickAccessProfile.id == record.resource_id, + QuickAccessProfile.scope_type == "user", + QuickAccessProfile.tenant_id == tenant_id, + ) + .one_or_none() + ) + if ( + profile is None + or profile.scope_id != selectors.account_id + or (selectors.profile_id and profile.id != selectors.profile_id) + ): + actions.append(_manual_action(record)) + continue + actions.append( + DsarErasureActionRef( + action_id=( + "quick_access:delete:quick_access_personal_profile:" + f"{profile.id}:r{profile.revision}" + ), + provider_id=self.provider_id, + module_id=self.module_id, + kind="delete", + resource_type="quick_access_personal_profile", + resource_id=profile.id, + title="Delete personal Quick Access profile", + rationale=( + "The profile contains only account-owned presentation " + "preferences and can be recreated from governed defaults." + ), + executable=True, + irreversible=True, + metadata={ + "account_id": profile.scope_id, + "revision": profile.revision, + }, + ) + ) + return tuple(actions) + + def execute_erasure( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + actions: Sequence[DsarErasureActionRef], + request_id: str, + ) -> Sequence[DsarExecutionResultRef]: + db = _session(session) + selectors = _subject_selectors(subject) + if selectors is None: + raise ValueError("Quick Access DSAR requires one corroborated account.") + + results: list[DsarExecutionResultRef] = [] + for action in actions: + _validate_action(action) + if not action.executable or action.kind != "delete": + results.append( + DsarExecutionResultRef( + action_id=action.action_id, + status="blocked", + summary=( + "Tenant Quick Access policy attribution remains " + "institutional accountability evidence." + ), + ) + ) + continue + if action.resource_type != "quick_access_personal_profile": + raise ValueError("Unsupported executable Quick Access DSAR action.") + + profile = ( + db.query(QuickAccessProfile) + .filter( + QuickAccessProfile.id == action.resource_id, + QuickAccessProfile.scope_type == "user", + QuickAccessProfile.tenant_id == tenant_id, + ) + .one_or_none() + ) + if profile is None: + results.append( + DsarExecutionResultRef( + action_id=action.action_id, + status="unchanged", + summary="The personal Quick Access profile was already absent.", + evidence={"request_id": request_id}, + ) + ) + continue + + expected_account = str(action.metadata.get("account_id") or "") + expected_revision = action.metadata.get("revision") + if ( + profile.scope_id != selectors.account_id + or expected_account != selectors.account_id + or (selectors.profile_id and profile.id != selectors.profile_id) + ): + results.append( + DsarExecutionResultRef( + action_id=action.action_id, + status="blocked", + summary=( + "The profile is not owned by the corroborated subject " + "account and selector context." + ), + ) + ) + continue + if expected_revision != profile.revision: + results.append( + DsarExecutionResultRef( + action_id=action.action_id, + status="blocked", + summary=( + "The profile changed after the erasure plan; create a " + "new plan before deleting it." + ), + ) + ) + continue + + profile_id = profile.id + db.delete(profile) + db.flush() + results.append( + DsarExecutionResultRef( + action_id=action.action_id, + status="executed", + summary=( + "Deleted the personal Quick Access profile without " + "changing tenant policy, tools, or domain data." + ), + evidence={"request_id": request_id, "profile_id": profile_id}, + ) + ) + return tuple(results) + + +def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None: + references = subject.external_references + account_id = _coalesce( + subject.account_id, + references.get("quick_access.account"), + references.get("access.account"), + ) + profile_id = _coalesce( + references.get("quick_access.profile"), + references.get("quick_access.profile_id"), + ) + if _CONFLICT in (account_id, profile_id): + return None + normalized_account = _optional_string(account_id) + if normalized_account is None: + return None + return _SubjectSelectors( + account_id=normalized_account, + profile_id=_optional_string(profile_id), + ) + + +def _coalesce(*values: str | None) -> str | None | object: + normalized = {str(value).strip() for value in values if str(value or "").strip()} + if len(normalized) > 1: + return _CONFLICT + return next(iter(normalized), None) + + +def _optional_string(value: object) -> str | None: + return value if isinstance(value, str) and value else None + + +def _personal_record(profile: QuickAccessProfile) -> DsarRecordRef: + return DsarRecordRef( + provider_id="quick_access", + module_id="quick_access", + resource_type="quick_access_personal_profile", + resource_id=profile.id, + category="personal_interface_preference", + title="Personal Quick Access profile", + data={ + "revision": profile.revision, + "category_preferences": _preference_projection( + profile.category_preferences + ), + "tool_preferences": _preference_projection(profile.tool_preferences), + }, + observed_at=_aware(profile.updated_at), + ) + + +def _attribution_record( + profile: QuickAccessProfile, + account_id: str, +) -> DsarRecordRef: + activities = [] + if profile.created_by == account_id: + activities.append("created_tenant_quick_access_policy") + if profile.updated_by == account_id: + activities.append("updated_tenant_quick_access_policy") + return DsarRecordRef( + provider_id="quick_access", + module_id="quick_access", + resource_type="quick_access_tenant_attribution", + resource_id=profile.id, + category="operator_accountability_evidence", + title="Tenant Quick Access policy attribution", + data={ + "activities": activities, + "revision": profile.revision, + "created_at": _iso(profile.created_at), + "updated_at": _iso(profile.updated_at), + }, + observed_at=_aware(profile.updated_at), + immutable_evidence=True, + retention_reason=( + "Tenant policy authorship is retained as institutional " + "accountability evidence; policy preferences are excluded." + ), + ) + + +def _preference_projection(value: object) -> dict[str, dict[str, object]]: + if not isinstance(value, Mapping): + raise ValueError("Quick Access preference payload must be an object.") + if len(value) > _MAX_PREFERENCES: + raise ValueError( + "Quick Access preference limit exceeded; review the stored profile." + ) + projected: dict[str, dict[str, object]] = {} + for raw_key, raw_entry in value.items(): + key = str(raw_key) + if not key or len(key) > 255 or not isinstance(raw_entry, Mapping): + raise ValueError("Quick Access preference entry is invalid.") + entry: dict[str, object] = {} + if isinstance(raw_entry.get("enabled"), bool): + entry["enabled"] = raw_entry["enabled"] + if isinstance(raw_entry.get("forced"), bool): + entry["forced"] = raw_entry["forced"] + order = raw_entry.get("order") + if isinstance(order, int) and not isinstance(order, bool): + entry["order"] = order + projected[key] = entry + return projected + + +def _manual_action(record: DsarRecordRef) -> DsarErasureActionRef: + return DsarErasureActionRef( + action_id=f"quick_access:manual_review:{record.resource_type}:{record.resource_id}", + provider_id="quick_access", + module_id="quick_access", + kind="manual_review", + resource_type=record.resource_type, + resource_id=record.resource_id, + title=f"Review {record.title}", + rationale=( + "The selected profile is absent or no longer belongs to the exact " + "tenant, account, and narrowing selectors." + ), + executable=False, + ) + + +def _iso(value: datetime | None) -> str | None: + aware = _aware(value) + return aware.isoformat() if aware else None + + +def _aware(value: datetime | None) -> datetime | None: + if value is None or value.tzinfo is not None: + return value + return value.replace(tzinfo=timezone.utc) + + +def _session(value: object) -> Session: + if not isinstance(value, Session): + raise TypeError("Quick Access DSAR requires a SQLAlchemy Session.") + return value + + +def _validate_record(record: DsarRecordRef) -> None: + if record.provider_id != "quick_access" or record.module_id != "quick_access": + raise ValueError("Quick Access DSAR cannot plan a foreign provider record.") + if ( + record.resource_type + not in { + "quick_access_personal_profile", + "quick_access_tenant_attribution", + } + or not record.resource_id + ): + raise ValueError("Quick Access DSAR record identity is invalid.") + + +def _validate_action(action: DsarErasureActionRef) -> None: + if action.provider_id != "quick_access" or action.module_id != "quick_access": + raise ValueError("Quick Access DSAR cannot execute a foreign provider action.") + if not action.action_id.startswith("quick_access:"): + raise ValueError("Quick Access DSAR action identity is invalid.") + + +__all__ = ["QUICK_ACCESS_DSAR_CAPABILITY", "QuickAccessDsarProvider"] diff --git a/src/govoplan_quick_access/backend/manifest.py b/src/govoplan_quick_access/backend/manifest.py index 1414876..2b7d6de 100644 --- a/src/govoplan_quick_access/backend/manifest.py +++ b/src/govoplan_quick_access/backend/manifest.py @@ -11,6 +11,7 @@ from govoplan_core.core.module_guards import ( persistent_table_uninstall_guard, ) from govoplan_core.core.modules import ( + CapabilityDocumentation, DocumentationLink, DocumentationTopic, FrontendModule, @@ -25,6 +26,10 @@ from govoplan_core.core.provider_governance import declared_module_architecture from govoplan_core.core.views import ViewSurface from govoplan_core.db.base import Base from govoplan_quick_access.backend.db import models as quick_access_models +from govoplan_quick_access.backend.dsar_provider import ( + QUICK_ACCESS_DSAR_CAPABILITY, + QuickAccessDsarProvider, +) MODULE_ID = "quick_access" @@ -111,7 +116,56 @@ def _router(context: ModuleContext): return create_router(context.registry) +def _dsar_provider(_context: ModuleContext) -> QuickAccessDsarProvider: + return QuickAccessDsarProvider() + + DOCUMENTATION = ( + DocumentationTopic( + id="quick-access.data-subject-requests", + title="Quick Access data-subject requests", + summary=( + "Export or delete personal rail preferences while retaining " + "institutional Quick Access policy." + ), + body=( + "Quick Access contributes the user profile owned by the exact account " + "in the active tenant. The access package contains bounded category and " + "tool availability and ordering preferences but never follows a tool " + "into Mail, Postbox, Tasks, Files, or another owner module. Tenant policy " + "records are included only as minimized creation or update attribution " + "when the subject account performed that action; their preference payload " + "is excluded and their attribution is retained as institutional evidence. " + "System-wide policy is outside tenant-scoped requests. Erasure deletes " + "only the selected personal profile after owner and revision checks. The " + "effective rail then falls back to current module, system, and tenant " + "defaults, including all locked or forced items. A repeated execution is " + "unchanged and no tool or domain data is modified." + ), + layer="configured", + documentation_types=("admin", "user"), + audience=("user", "tenant_admin", "operator", "auditor"), + related_modules=("core", "access", "views"), + metadata={ + "help_contexts": [ + "quick_access.rail", + "quick_access.settings.personal", + "quick_access.admin.tenant", + "privacy.data-subject-requests", + ], + "consequence_classes": { + "export_personal_profile": ( + "Returns bounded rail preferences, never tool-owned data." + ), + "delete_personal_profile": ( + "Removes personal overrides; governed defaults apply again." + ), + "retain_tenant_attribution": ( + "Preserves minimized tenant-policy accountability evidence." + ), + }, + }, + ), DocumentationTopic( id="quick-access.user", title="Quick Access rail", @@ -229,10 +283,25 @@ manifest = ModuleManifest( provides_interfaces=( ModuleInterfaceProvider(name="quick_access.runtime", version="1.0.0"), ModuleInterfaceProvider(name="quick_access.preferences", version="1.0.0"), + ModuleInterfaceProvider( + name=QUICK_ACCESS_DSAR_CAPABILITY, + version="0.1.0", + ), ), permissions=PERMISSIONS, role_templates=ROLE_TEMPLATES, route_factory=_router, + capability_factories={QUICK_ACCESS_DSAR_CAPABILITY: _dsar_provider}, + capability_documentation={ + QUICK_ACCESS_DSAR_CAPABILITY: CapabilityDocumentation( + label="Quick Access data-subject request provider", + summary=( + "Exports and deletes personal Quick Access preferences while " + "retaining minimized tenant-policy attribution." + ), + contract_version="0.1.0", + ), + }, frontend=FrontendModule( module_id=MODULE_ID, package_name="@govoplan/quick-access-webui", diff --git a/tests/test_dsar_provider.py b/tests/test_dsar_provider.py new file mode 100644 index 0000000..6128478 --- /dev/null +++ b/tests/test_dsar_provider.py @@ -0,0 +1,377 @@ +from __future__ import annotations + +import json +import unittest + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + +from govoplan_core.core.dsar import ( + DsarErasureActionRef, + DsarProvider, + DsarRecordRef, + 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_quick_access.backend.db.models import QuickAccessProfile +from govoplan_quick_access.backend.dsar_provider import ( + QUICK_ACCESS_DSAR_CAPABILITY, + QuickAccessDsarProvider, +) +from govoplan_quick_access.backend.manifest import manifest + + +class _Registry: + def __init__( + self, + provider: QuickAccessDsarProvider, + *, + active: bool = True, + ) -> None: + self.provider = provider + self.active = active + + def capability_names(self): + return (QUICK_ACCESS_DSAR_CAPABILITY,) + + def capability_owner(self, name): + self._assert_capability(name) + return "quick_access" + + def tenant_entitlement_resolver(self): + active = self.active + + class _Resolver: + @staticmethod + def resolve(session, tenant_id): + del session, tenant_id + return type( + "State", + (), + {"effective_modules": ("quick_access",) if active else ()}, + )() + + return _Resolver() + + def require_tenant_capability(self, name, session, **kwargs): + del session, kwargs + self._assert_capability(name) + return self.provider + + def manifests(self): + return (type("Manifest", (), {"id": "quick_access"})(),) + + @staticmethod + def _assert_capability(name: str) -> None: + if name != QUICK_ACCESS_DSAR_CAPABILITY: + raise KeyError(name) + + +class QuickAccessDsarProviderTests(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 = QuickAccessDsarProvider() + self.assertIsInstance(self.provider, DsarProvider) + self._seed() + self.session.commit() + + def tearDown(self) -> None: + self.session.close() + self.engine.dispose() + + def _seed(self) -> None: + self.session.add_all( + ( + QuickAccessProfile( + id="profile-personal", + scope_type="user", + tenant_id="tenant-1", + scope_id="account-1", + scope_key="user:tenant-1:account-1", + category_preferences={"work": {"enabled": False}}, + tool_preferences={"tasks.mine": {"order": 5}}, + revision=3, + created_by="account-1", + updated_by="account-1", + ), + QuickAccessProfile( + id="profile-tenant", + scope_type="tenant", + tenant_id="tenant-1", + scope_id="tenant-1", + scope_key="tenant:tenant-1", + category_preferences={ + "private-policy-do-not-export": {"enabled": False} + }, + tool_preferences={"tasks.mine": {"forced": True}}, + revision=4, + created_by="account-1", + updated_by="account-other", + ), + QuickAccessProfile( + id="profile-other-account", + scope_type="user", + tenant_id="tenant-1", + scope_id="account-other", + scope_key="user:tenant-1:account-other", + category_preferences={"calendar": {"enabled": False}}, + tool_preferences={}, + revision=2, + created_by="account-other", + updated_by="account-other", + ), + QuickAccessProfile( + id="profile-other-tenant", + scope_type="user", + tenant_id="tenant-2", + scope_id="account-1", + scope_key="user:tenant-2:account-1", + category_preferences={"files": {"enabled": False}}, + tool_preferences={}, + revision=2, + created_by="account-1", + updated_by="account-1", + ), + QuickAccessProfile( + id="profile-system", + scope_type="system", + tenant_id=None, + scope_id=None, + scope_key="system:*", + category_preferences={"messages": {"enabled": False}}, + tool_preferences={}, + revision=2, + created_by="account-1", + updated_by="account-1", + ), + ) + ) + + def test_search_exports_personal_preferences_and_minimized_attribution( + self, + ) -> None: + records = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef(account_id="account-1"), + ) + + self.assertEqual( + [ + "quick_access_personal_profile", + "quick_access_tenant_attribution", + ], + [record.resource_type for record in records], + ) + exported = json.dumps([record.to_dict() for record in records]) + self.assertIn("tasks.mine", exported) + self.assertIn("created_tenant_quick_access_policy", exported) + self.assertNotIn("private-policy-do-not-export", exported) + self.assertNotIn("profile-other-account", exported) + self.assertNotIn("profile-other-tenant", exported) + self.assertNotIn("profile-system", exported) + + def test_profile_reference_narrows_and_alias_conflicts_fail_closed(self) -> None: + narrowed = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + account_id="account-1", + external_references={"quick_access.profile": "profile-personal"}, + ), + ) + conflict = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + account_id="account-1", + external_references={"quick_access.account": "account-other"}, + ), + ) + no_account = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + external_references={"quick_access.profile": "profile-personal"} + ), + ) + + self.assertEqual(["profile-personal"], [item.resource_id for item in narrowed]) + self.assertEqual((), conflict) + self.assertEqual((), no_account) + + def test_erasure_deletes_only_personal_profile_and_is_idempotent(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, + ) + + self.assertEqual(["delete", "retain"], [action.kind for action in actions]) + first = self.provider.execute_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + actions=actions, + request_id="dsar-1", + ) + second = self.provider.execute_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + actions=actions, + request_id="dsar-1", + ) + + self.assertEqual(["executed", "blocked"], [item.status for item in first]) + self.assertEqual(["unchanged", "blocked"], [item.status for item in second]) + self.assertIsNone(self.session.get(QuickAccessProfile, "profile-personal")) + self.assertIsNotNone(self.session.get(QuickAccessProfile, "profile-tenant")) + self.assertIsNotNone(self.session.get(QuickAccessProfile, "profile-system")) + + def test_changed_and_foreign_resources_are_blocked(self) -> None: + subject = DsarSubjectRef(account_id="account-1") + record = next( + item + for item in self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=subject, + ) + if item.resource_type == "quick_access_personal_profile" + ) + action = self.provider.plan_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + records=(record,), + )[0] + self.session.get(QuickAccessProfile, "profile-personal").revision += 1 + changed = self.provider.execute_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + actions=(action,), + request_id="dsar-1", + ) + self.assertEqual("blocked", changed[0].status) + + with self.assertRaisesRegex(ValueError, "foreign provider record"): + self.provider.plan_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + records=( + DsarRecordRef( + provider_id="dashboard", + module_id="dashboard", + resource_type="quick_access_personal_profile", + resource_id="profile-personal", + category="preference", + title="Foreign profile", + ), + ), + ) + with self.assertRaisesRegex(ValueError, "foreign provider action"): + self.provider.execute_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + actions=( + DsarErasureActionRef( + action_id="dashboard:delete:profile:profile-personal", + provider_id="dashboard", + module_id="dashboard", + kind="delete", + resource_type="quick_access_personal_profile", + resource_id="profile-personal", + title="Delete profile", + rationale="Foreign action", + executable=True, + ), + ), + request_id="dsar-1", + ) + + def test_core_workflow_reports_active_and_inactive_provider(self) -> None: + row = create_data_subject_request( + self.session, + tenant_id="tenant-1", + reference="DSAR-QUICK-ACCESS-1", + request_kind="access", + subject=DsarSubjectRef(account_id="account-1"), + purpose="Respond to a verified request.", + legal_basis="Article 15 GDPR", + due_at=None, + requested_by_account_id="privacy-officer", + ) + self.session.commit() + search_data_subject_request( + self.session, + registry=_Registry(self.provider), + row=row, + expected_revision=1, + ) + self.assertEqual( + [QUICK_ACCESS_DSAR_CAPABILITY], row.coverage["provider_capabilities"] + ) + self.assertEqual(2, row.search_result["record_count"]) + + inactive = create_data_subject_request( + self.session, + tenant_id="tenant-1", + reference="DSAR-QUICK-ACCESS-2", + request_kind="access", + subject=DsarSubjectRef(account_id="account-1"), + purpose="Respond to a verified request.", + legal_basis="Article 15 GDPR", + due_at=None, + requested_by_account_id="privacy-officer", + ) + self.session.commit() + search_data_subject_request( + self.session, + registry=_Registry(self.provider, active=False), + row=inactive, + expected_revision=1, + ) + self.assertEqual([], inactive.coverage["provider_capabilities"]) + self.assertEqual( + [QUICK_ACCESS_DSAR_CAPABILITY], + inactive.coverage["inactive_provider_capabilities"], + ) + + def test_manifest_registers_and_documents_the_capability(self) -> None: + self.assertIn(QUICK_ACCESS_DSAR_CAPABILITY, manifest.capability_factories) + self.assertIn( + QUICK_ACCESS_DSAR_CAPABILITY, + manifest.capability_documentation, + ) + self.assertIn( + QUICK_ACCESS_DSAR_CAPABILITY, + {item.name for item in manifest.provides_interfaces}, + ) + self.assertTrue( + any( + topic.id == "quick-access.data-subject-requests" + and {"admin", "user"}.issubset(topic.documentation_types) + for topic in manifest.documentation + ) + ) + + +if __name__ == "__main__": + unittest.main()