diff --git a/README.md b/README.md index cd09233..9f70928 100644 --- a/README.md +++ b/README.md @@ -13,3 +13,14 @@ contact points. Cases and other procedures consume effective Party projections through `parties.resolver` instead of copying representation logic. See [docs/PARTIES_DOMAIN.md](docs/PARTIES_DOMAIN.md). + +## Data-subject requests + +The module contributes `privacy.dsar.parties`. Searches are tenant-scoped and +accept canonical identity/account selectors or explicit `parties.party` and +`parties.revision` references. Results expose a minimized typed projection; +the stored payload, free-text change reasons, evidence identifiers, and +unrelated counterpart details are not exported. Historical revisions and +operator attribution are retained, while a current active party fact requires +authorized manual review through the normal party lifecycle. The provider does +not automatically erase or rewrite procedure evidence. diff --git a/docs/PARTIES_DOMAIN.md b/docs/PARTIES_DOMAIN.md index 8bd8c5b..1fb3e87 100644 --- a/docs/PARTIES_DOMAIN.md +++ b/docs/PARTIES_DOMAIN.md @@ -28,3 +28,22 @@ Party writes are append-only, replay-safe, tenant-bound, and protected by optimistic concurrency. The procedure identity cannot change across revisions. Database restore is the recovery unit; downstream effects preserve exact Party, contact-snapshot, and representation evidence references. + +## Data-subject requests + +Parties publishes the `privacy.dsar.parties` capability. A search uses the exact +tenant plus a canonical identity, a supported account-backed external subject, +or an explicit `parties.party`/`parties.revision` reference. Combining a direct +reference with a conflicting canonical selector returns no data. The bounded +search exports typed party, procedure, channel, snapshot, and representation +facts. It never exports the raw JSON payload, free-text change reasons, +evidence identifiers, or unrelated counterpart party identifiers. An account +that only recorded a revision receives a separate, minimized attribution +record rather than the affected party's content. + +Every stored revision and operator attribution is immutable accountability +evidence and therefore receives a retention action. The current active party +fact receives a non-executable manual-review action: an authorized operator +must correct, expire, supersede, or explicitly revoke representation through +the normal versioned lifecycle after reviewing procedural and third-party +consequences. DSAR execution cannot mutate Parties data automatically. diff --git a/src/govoplan_parties/backend/dsar_provider.py b/src/govoplan_parties/backend/dsar_provider.py new file mode 100644 index 0000000..f067e21 --- /dev/null +++ b/src/govoplan_parties/backend/dsar_provider.py @@ -0,0 +1,520 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from datetime import datetime, timezone + +from sqlalchemy.orm import Session + +from govoplan_core.core.dsar import ( + DsarErasureActionRef, + DsarExecutionResultRef, + DsarRecordRef, + DsarSubjectRef, + dsar_capability_name, +) +from govoplan_core.core.institutional import ( + PartyRepresentation, + PartySubjectReference, + ProcedureParty, +) +from govoplan_parties.backend.db.models import ProcedurePartyRevision +from govoplan_parties.backend.service import party_from_mapping + + +PARTIES_DSAR_CAPABILITY = dsar_capability_name("parties") +_MAX_CANDIDATES = 5_000 +_MAX_RECORDS = 5_000 + + +@dataclass(frozen=True, slots=True) +class _SubjectSelectors: + account_id: str | None + identity_id: str | None + party_id: str | None + revision_id: str | None + + @property + def has_canonical_selector(self) -> bool: + return bool(self.account_id or self.identity_id) + + @property + def has_recognized_selector(self) -> bool: + return bool( + self.account_id or self.identity_id or self.party_id or self.revision_id + ) + + +class PartiesDsarProvider: + provider_id = "parties" + module_id = "parties" + + def search_subject( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + ) -> Sequence[DsarRecordRef]: + db = _session(session) + selectors = _subject_selectors(subject) + if selectors is None or not selectors.has_recognized_selector: + return () + + rows = _tenant_rows(db, tenant_id=tenant_id) + selected = [ + (row, party_from_mapping(row.payload)) + for row in rows + if _direct_match(row, selectors) + or _payload_may_match(row.payload, selectors) + or _actor_matches(row, selectors) + ] + if _direct_reference_conflicts(selected, selectors): + return () + + records: list[DsarRecordRef] = [] + seen: set[tuple[str, str]] = set() + + def append(record: DsarRecordRef) -> None: + key = (record.resource_type, record.resource_id) + if key in seen: + return + if len(records) >= _MAX_RECORDS: + raise ValueError( + "Parties DSAR result limit exceeded; narrow the subject selectors." + ) + seen.add(key) + records.append(record) + + for row, party in selected: + subject_fields = _subject_match_fields(party.subject, selectors) + direct = _direct_match(row, selectors) + if subject_fields or (direct and not selectors.has_canonical_selector): + match_fields = list(subject_fields) + if direct: + match_fields.append("reference") + append(_revision_record(row, party, match_fields=match_fields)) + if row.superseded_at is None and row.status == "active": + append(_active_fact_record(row, party, match_fields=match_fields)) + + if _actor_matches(row, selectors): + append(_operator_attribution_record(row)) + + return tuple(records) + + def plan_erasure( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + records: Sequence[DsarRecordRef], + ) -> Sequence[DsarErasureActionRef]: + del tenant_id + _session(session) + if _subject_selectors(subject) is None: + raise ValueError("Parties DSAR subject selectors conflict.") + actions: list[DsarErasureActionRef] = [] + for record in records: + _validate_record(record) + if record.immutable_evidence: + kind = "retain" + title = f"Retain {record.title}" + rationale = record.retention_reason or ( + "Procedure-party history is retained as institutional evidence." + ) + else: + kind = "manual_review" + title = f"Review {record.title}" + rationale = ( + "An authorized operator must correct, revoke, expire, or supersede " + "the current procedure-party fact through the governed lifecycle " + "after reviewing procedure, representation, and third-party effects." + ) + actions.append( + DsarErasureActionRef( + action_id=( + f"parties:{kind}:{record.resource_type}:{record.resource_id}" + ), + provider_id=self.provider_id, + module_id=self.module_id, + kind=kind, + resource_type=record.resource_type, + resource_id=record.resource_id, + title=title, + rationale=rationale, + executable=False, + ) + ) + return tuple(actions) + + def execute_erasure( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + actions: Sequence[DsarErasureActionRef], + request_id: str, + ) -> Sequence[DsarExecutionResultRef]: + del tenant_id + _session(session) + if _subject_selectors(subject) is None: + raise ValueError("Parties DSAR subject selectors conflict.") + results: list[DsarExecutionResultRef] = [] + for action in actions: + _validate_action(action) + if action.executable: + raise ValueError( + "Parties DSAR does not publish executable erasure actions." + ) + results.append( + DsarExecutionResultRef( + action_id=action.action_id, + status="blocked", + summary=( + "Use the governed procedure-party or representation lifecycle " + "after institutional-evidence and third-party review." + ), + evidence={"request_id": request_id}, + ) + ) + return tuple(results) + + +def _tenant_rows( + session: Session, + *, + tenant_id: str, +) -> list[ProcedurePartyRevision]: + rows = ( + session.query(ProcedurePartyRevision) + .filter(ProcedurePartyRevision.tenant_id == tenant_id) + .order_by( + ProcedurePartyRevision.party_id.asc(), + ProcedurePartyRevision.recorded_at.asc(), + ProcedurePartyRevision.id.asc(), + ) + .limit(_MAX_CANDIDATES + 1) + .all() + ) + if len(rows) > _MAX_CANDIDATES: + raise ValueError( + "Parties DSAR candidate limit exceeded; use an explicit Parties reference." + ) + return rows + + +def _revision_record( + row: ProcedurePartyRevision, + party: ProcedureParty, + *, + match_fields: Sequence[str], +) -> DsarRecordRef: + return _record( + "parties_procedure_party_revision", + row.id, + "procedure_party_history", + "Procedure-party revision", + { + "match_fields": list(match_fields), + "party_id": row.party_id, + "revision": row.revision, + "previous_revision_id": row.previous_revision_id, + "status": row.status, + "procedure": { + "kind": row.procedure_kind, + "owner_module": row.procedure_owner_module, + "id": row.procedure_id, + }, + "role": _bounded_text(party.role, 120), + "subject": _subject_data(party.subject), + "valid_from": _iso(row.valid_from), + "valid_to": _iso(row.valid_to), + "recorded_at": _iso(row.recorded_at), + "superseded_at": _iso(row.superseded_at), + "preferred_channels": _bounded_values(party.preferred_channels), + "permitted_channels": _bounded_values(party.permitted_channels), + "delivery_recipient": party.delivery_recipient, + "contact_snapshot_refs": _bounded_values( + party.contact_snapshot_refs, + limit=200, + ), + "representations": _representation_data( + party.representations, + party_id=row.party_id, + ), + "evidence_count": len(party.evidence), + }, + observed_at=row.recorded_at, + immutable=True, + retention_reason=( + "Procedure-party revisions, explicit representation changes, and their " + "sequence are retained as institutional and accountability evidence." + ), + ) + + +def _active_fact_record( + row: ProcedurePartyRevision, + party: ProcedureParty, + *, + match_fields: Sequence[str], +) -> DsarRecordRef: + return _record( + "parties_current_procedure_party", + row.party_id, + "current_procedure_party_fact", + "Current procedure-party fact", + { + "match_fields": list(match_fields), + "revision_record_id": row.id, + "revision": row.revision, + "procedure": { + "kind": row.procedure_kind, + "owner_module": row.procedure_owner_module, + "id": row.procedure_id, + }, + "role": _bounded_text(party.role, 120), + "delivery_recipient": party.delivery_recipient, + "representation_count": len(party.representations), + }, + observed_at=row.recorded_at, + ) + + +def _operator_attribution_record( + row: ProcedurePartyRevision, +) -> DsarRecordRef: + return _record( + "parties_operator_attribution", + row.id, + "operator_accountability_evidence", + "Procedure-party recording attribution", + { + "match_fields": ["created_by"], + "activity": "recorded_procedure_party_revision", + "procedure": { + "kind": row.procedure_kind, + "owner_module": row.procedure_owner_module, + "id": row.procedure_id, + }, + "revision": row.revision, + "recorded_at": _iso(row.recorded_at), + }, + observed_at=row.recorded_at, + immutable=True, + retention_reason=( + "Operator attribution is retained as accountability evidence; unrelated " + "party identity, contact, representation, and evidence data is excluded." + ), + ) + + +def _representation_data( + representations: Sequence[PartyRepresentation], + *, + party_id: str, +) -> list[dict[str, object]]: + values: list[dict[str, object]] = [] + for representation in representations[:200]: + positions = [] + if representation.representative_party_ref.object_id == party_id: + positions.append("representative") + if representation.represented_party_ref.object_id == party_id: + positions.append("represented") + if not positions: + continue + values.append( + { + "subject_positions": positions, + "power_ref": _bounded_text(representation.power_ref, 255), + "permitted_actions": _bounded_values( + representation.permitted_actions, + limit=100, + ), + "revision": representation.temporal.revision, + "valid_from": _iso(representation.temporal.valid_from), + "valid_to": _iso(representation.temporal.valid_to), + "recorded_at": _iso(representation.temporal.recorded_at), + "revoked_at": _iso(representation.revoked_at), + "evidence_count": len(representation.evidence), + } + ) + return values + + +def _subject_data(subject: PartySubjectReference) -> dict[str, object]: + return { + "kind": subject.kind, + "provider": _bounded_text(subject.provider, 120), + "subject_id": subject.subject_id, + "version": _bounded_text(subject.version, 120), + } + + +def _payload_may_match( + payload: Mapping[str, object], + selectors: _SubjectSelectors, +) -> bool: + raw_subject = payload.get("subject") + if not isinstance(raw_subject, Mapping): + return False + subject_id = _normalized_id(raw_subject.get("subject_id")) + if selectors.identity_id and subject_id == selectors.identity_id: + return True + return bool(selectors.account_id and subject_id == selectors.account_id) + + +def _subject_match_fields( + subject: PartySubjectReference, + selectors: _SubjectSelectors, +) -> list[str]: + fields: list[str] = [] + if ( + selectors.identity_id + and subject.kind == "identity" + and subject.subject_id == selectors.identity_id + ): + fields.append("identity_id") + if ( + selectors.account_id + and subject.kind == "external" + and subject.provider.casefold() in {"access", "account", "accounts"} + and subject.subject_id == selectors.account_id + ): + fields.append("account_id") + return fields + + +def _actor_matches( + row: ProcedurePartyRevision, + selectors: _SubjectSelectors, +) -> bool: + return bool(selectors.account_id and row.created_by == selectors.account_id) + + +def _direct_match( + row: ProcedurePartyRevision, + selectors: _SubjectSelectors, +) -> bool: + return bool( + (selectors.party_id and row.party_id == selectors.party_id) + or (selectors.revision_id and row.id == selectors.revision_id) + ) + + +def _direct_reference_conflicts( + rows: Sequence[tuple[ProcedurePartyRevision, ProcedureParty]], + selectors: _SubjectSelectors, +) -> bool: + direct_rows = [(row, party) for row, party in rows if _direct_match(row, selectors)] + if selectors.party_id and selectors.revision_id: + revision_row = next( + (row for row, _party in rows if row.id == selectors.revision_id), + None, + ) + if revision_row is None or revision_row.party_id != selectors.party_id: + return True + if not selectors.has_canonical_selector: + return bool((selectors.party_id or selectors.revision_id) and not direct_rows) + return any( + not _subject_match_fields(party.subject, selectors) + for _row, party in direct_rows + ) or bool((selectors.party_id or selectors.revision_id) and not direct_rows) + + +def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None: + groups = { + "account_id": ( + subject.account_id, + subject.external_references.get("parties.account"), + subject.external_references.get("access.account"), + ), + "identity_id": ( + subject.identity_id, + subject.external_references.get("parties.identity"), + subject.external_references.get("identity.id"), + ), + "party_id": (subject.external_references.get("parties.party"),), + "revision_id": (subject.external_references.get("parties.revision"),), + } + normalized: dict[str, str | None] = {} + for key, values in groups.items(): + distinct = {value for item in values if (value := _normalized_id(item))} + if len(distinct) > 1: + return None + normalized[key] = next(iter(distinct), None) + return _SubjectSelectors(**normalized) + + +def _validate_record(record: DsarRecordRef) -> None: + if record.provider_id != "parties" or record.module_id != "parties": + raise ValueError("Parties DSAR received a foreign provider record.") + + +def _validate_action(action: DsarErasureActionRef) -> None: + if action.provider_id != "parties" or action.module_id != "parties": + raise ValueError("Parties DSAR received a foreign provider action.") + + +def _record( + resource_type: str, + resource_id: str, + category: str, + title: str, + data: Mapping[str, object], + *, + observed_at: datetime | None, + immutable: bool = False, + retention_reason: str | None = None, +) -> DsarRecordRef: + return DsarRecordRef( + provider_id="parties", + module_id="parties", + resource_type=resource_type, + resource_id=resource_id, + category=category, + title=title, + data=data, + observed_at=observed_at, + immutable_evidence=immutable, + retention_reason=retention_reason, + source_path="/parties", + ) + + +def _session(value: object) -> Session: + if not isinstance(value, Session): + raise TypeError("Parties DSAR provider requires a SQLAlchemy session.") + return value + + +def _bounded_values( + values: Sequence[str], + *, + limit: int = 100, +) -> list[str]: + return [str(value)[:255] for value in values[:limit]] + + +def _bounded_text(value: str | None, limit: int) -> str | None: + return value[:limit] if value else None + + +def _normalized_id(value: object) -> str | None: + if value is None: + return None + normalized = str(value).strip() + return normalized or None + + +def _iso(value: datetime | None) -> str | None: + if value is None: + return None + if value.tzinfo is None: + value = value.replace(tzinfo=timezone.utc) + return value.isoformat() + + +__all__ = ["PARTIES_DSAR_CAPABILITY", "PartiesDsarProvider"] diff --git a/src/govoplan_parties/backend/manifest.py b/src/govoplan_parties/backend/manifest.py index 2479952..1b40059 100644 --- a/src/govoplan_parties/backend/manifest.py +++ b/src/govoplan_parties/backend/manifest.py @@ -8,6 +8,10 @@ from govoplan_core.core.modules import CapabilityDocumentation, DocumentationLin from govoplan_core.core.provider_governance import declared_module_architecture from govoplan_core.db.base import Base from govoplan_parties.backend.db import models as party_models +from govoplan_parties.backend.dsar_provider import ( + PARTIES_DSAR_CAPABILITY, + PartiesDsarProvider, +) from govoplan_parties.backend.service import SqlPartyResolver @@ -34,12 +38,16 @@ def _resolver(_context: ModuleContext) -> SqlPartyResolver: return SqlPartyResolver() +def _dsar_provider(_context: ModuleContext) -> PartiesDsarProvider: + return PartiesDsarProvider() + + manifest = ModuleManifest( id=MODULE_ID, name=MODULE_NAME, version=MODULE_VERSION, optional_dependencies=("identity", "organizations", "addresses", "cases", "workflow_engine", "decisions", "policy", "audit"), - provides_interfaces=(ModuleInterfaceProvider(name="parties.procedure", version="0.1.0"), ModuleInterfaceProvider(name="parties.representation", version="0.1.0")), + provides_interfaces=(ModuleInterfaceProvider(name="parties.procedure", version="0.1.0"), ModuleInterfaceProvider(name="parties.representation", version="0.1.0"), ModuleInterfaceProvider(name=PARTIES_DSAR_CAPABILITY, version="0.1.0")), permissions=( _permission(READ_SCOPE, "View procedure parties", "View procedure-local roles, contact snapshots, and representation evidence."), _permission(WRITE_SCOPE, "Manage procedure parties", "Create and revise procedure parties and revoke representation powers."), @@ -50,8 +58,8 @@ manifest = ModuleManifest( RoleTemplate(slug="party_reader", name="Party reader", description="Inspect procedure parties and authority evidence.", permissions=(READ_SCOPE,)), ), route_factory=_router, - capability_factories={CAPABILITY_PARTY_RESOLVER: _resolver}, - capability_documentation={CAPABILITY_PARTY_RESOLVER: CapabilityDocumentation(label="Procedure Party resolver", summary="Returns effective, tenant-bound procedure parties and representation powers.", contract_version="0.1.0")}, + capability_factories={CAPABILITY_PARTY_RESOLVER: _resolver, PARTIES_DSAR_CAPABILITY: _dsar_provider}, + capability_documentation={CAPABILITY_PARTY_RESOLVER: CapabilityDocumentation(label="Procedure Party resolver", summary="Returns effective, tenant-bound procedure parties and representation powers.", contract_version="0.1.0"), PARTIES_DSAR_CAPABILITY: CapabilityDocumentation(label="Parties data-subject request provider", summary="Finds minimized, tenant-scoped procedure-party data and preserves governed revision evidence.", contract_version="0.1.0")}, migration_spec=MigrationSpec( module_id=MODULE_ID, metadata=Base.metadata, @@ -62,6 +70,16 @@ manifest = ModuleManifest( ), uninstall_guard_providers=(persistent_table_uninstall_guard(party_models.ProcedurePartyRevision, label="Parties"),), documentation=( + DocumentationTopic( + id="parties.data-subject-requests", + title="Procedure-party data-subject requests", + summary="Export subject-linked party facts while preserving procedure and representation evidence.", + body="The Parties DSAR provider matches exact tenant-bound canonical identity or supported account selectors and explicit Parties references. It exports typed procedure, party, channel, snapshot, and representation fields, but not the stored opaque payload, free-text change reasons, evidence identifiers, or unrelated counterpart party identifiers. Immutable revisions and operator attribution are retained. A current active party fact is flagged for authorized manual review; correction, expiry, supersession, and representation revocation must use the governed lifecycle because Parties never performs automatic DSAR mutation. Explicit references that conflict with a canonical selector fail closed. Searches are bounded, and operator-attribution matches expose no unrelated party identity or contact content.", + layer="configured", + documentation_types=("admin", "user"), + audience=("user", "operator", "module_admin", "auditor"), + links=(DocumentationLink(label="Parties domain and recovery", href="govoplan-parties/docs/PARTIES_DOMAIN.md", kind="repository"),), + ), DocumentationTopic( id="parties.procedure-and-representation", title="Procedure parties and representation", diff --git a/tests/test_dsar_provider.py b/tests/test_dsar_provider.py new file mode 100644 index 0000000..7a6ce6f --- /dev/null +++ b/tests/test_dsar_provider.py @@ -0,0 +1,456 @@ +from __future__ import annotations + +import json +import unittest +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + +from govoplan_core.core.dsar import ( + DsarErasureActionRef, + DsarProvider, + DsarRecordRef, + DsarSubjectRef, +) +from govoplan_core.core.institutional import ( + EvidenceReference, + InstitutionalReference, + PartyRepresentation, + PartySubjectReference, + ProcedureParty, + TemporalRevision, +) +from govoplan_core.db.base import Base +from govoplan_core.privacy.dsar_workflow import ( + create_data_subject_request, + search_data_subject_request, +) +from govoplan_parties.backend.dsar_provider import ( + PARTIES_DSAR_CAPABILITY, + PartiesDsarProvider, +) +from govoplan_parties.backend.manifest import manifest +from govoplan_parties.backend.service import record_procedure_party + + +NOW = datetime(2026, 8, 21, 10, 0, tzinfo=UTC) + + +@dataclass +class _Principal: + tenant_id: str + account_id: str + + +class _Registry: + def __init__( + self, + provider: PartiesDsarProvider, + *, + parties_active: bool = True, + ) -> None: + self.provider = provider + self.parties_active = parties_active + + def capability_names(self): + return (PARTIES_DSAR_CAPABILITY,) + + def capability_owner(self, name): + self._assert_capability(name) + return "parties" + + def tenant_entitlement_resolver(self): + parties_active = self.parties_active + + class _Resolver: + @staticmethod + def resolve(session, tenant_id): + del session, tenant_id + return type( + "State", + (), + {"effective_modules": ("parties",) if parties_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": "parties"})(),) + + @staticmethod + def _assert_capability(name: str) -> None: + if name != PARTIES_DSAR_CAPABILITY: + raise KeyError(name) + + +def _ref( + kind: str, + object_id: str, + owner: str, + *, + tenant_id: str, + version: str | None = "1", +) -> InstitutionalReference: + return InstitutionalReference( + kind=kind, # type: ignore[arg-type] + owner_module=owner, + object_id=object_id, + tenant_id=tenant_id, + version=version, + valid_at=NOW, + ) + + +def _party( + party_id: str, + subject_id: str, + *, + tenant_id: str = "tenant-1", + revision: str = "1", + procedure_id: str = "case-1", + subject_kind: str = "identity", + subject_provider: str = "identity", +) -> ProcedureParty: + recorded_at = NOW + timedelta(minutes=int(revision) - 1) + representation = PartyRepresentation( + representative_party_ref=_ref( + "party", + party_id, + "parties", + tenant_id=tenant_id, + version="1", + ), + represented_party_ref=_ref( + "party", + "counterparty-private-id", + "parties", + tenant_id=tenant_id, + version="1", + ), + power_ref="power-1", + permitted_actions=("receive",), + temporal=TemporalRevision( + revision="1", + valid_from=NOW, + recorded_at=NOW, + change_reason="private-representation-reason-do-not-export", + ), + evidence=( + EvidenceReference( + kind="document", + owner_module="files", + evidence_id="private-power-document-do-not-export", + tenant_id=tenant_id, + version="1", + ), + ), + ) + return ProcedureParty( + reference=_ref( + "party", + party_id, + "parties", + tenant_id=tenant_id, + version=revision, + ), + procedure_ref=_ref( + "case", + procedure_id, + "cases", + tenant_id=tenant_id, + ), + role="representative", + subject=PartySubjectReference( + kind=subject_kind, # type: ignore[arg-type] + provider=subject_provider, + subject_id=subject_id, + tenant_id=tenant_id, + version="1", + ), + temporal=TemporalRevision( + revision=revision, + valid_from=NOW, + recorded_at=recorded_at, + change_reason="private-party-change-reason-do-not-export", + ), + preferred_channels=("postbox",), + permitted_channels=("postbox", "mail"), + delivery_recipient=True, + representations=(representation,), + contact_snapshot_refs=("addresses:snapshot-1",), + evidence=( + EvidenceReference( + kind="document", + owner_module="files", + evidence_id="private-party-document-do-not-export", + tenant_id=tenant_id, + version=revision, + ), + ), + ) + + +class PartiesDsarProviderTests(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 = PartiesDsarProvider() + self.assertIsInstance(self.provider, DsarProvider) + + operator = _Principal("tenant-1", "operator-1") + record_procedure_party( + self.session, + operator, + party=_party("party-1", "identity-1"), + ) + record_procedure_party( + self.session, + operator, + party=_party("party-1", "identity-1", revision="2"), + expected_revision="1", + ) + record_procedure_party( + self.session, + operator, + party=_party( + "party-unrelated", + "identity-unrelated", + procedure_id="case-private-unrelated", + ), + ) + record_procedure_party( + self.session, + _Principal("tenant-2", "operator-2"), + party=_party( + "party-other-tenant", + "identity-1", + tenant_id="tenant-2", + procedure_id="case-other-tenant", + ), + ) + self.session.commit() + + def tearDown(self) -> None: + self.session.close() + self.engine.dispose() + + def test_identity_search_is_tenant_scoped_minimized_and_versioned(self) -> None: + records = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef(identity_id="identity-1"), + ) + + revisions = [ + item + for item in records + if item.resource_type == "parties_procedure_party_revision" + ] + current = [ + item + for item in records + if item.resource_type == "parties_current_procedure_party" + ] + self.assertEqual(2, len(revisions)) + self.assertEqual(1, len(current)) + self.assertEqual(["1", "2"], [item.data["revision"] for item in revisions]) + self.assertTrue(all(item.immutable_evidence for item in revisions)) + self.assertFalse(current[0].immutable_evidence) + + exported = json.dumps([item.to_dict() for item in records], sort_keys=True) + self.assertIn("identity-1", exported) + self.assertIn("addresses:snapshot-1", exported) + self.assertNotIn("identity-unrelated", exported) + self.assertNotIn("case-private-unrelated", exported) + self.assertNotIn("case-other-tenant", exported) + self.assertNotIn("counterparty-private-id", exported) + self.assertNotIn("private-party-document-do-not-export", exported) + self.assertNotIn("private-power-document-do-not-export", exported) + self.assertNotIn("private-party-change-reason-do-not-export", exported) + self.assertNotIn("private-representation-reason-do-not-export", exported) + + def test_account_attribution_does_not_export_unrelated_party_content(self) -> None: + records = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef(account_id="operator-1"), + ) + + self.assertEqual(3, len(records)) + self.assertTrue( + all( + item.resource_type == "parties_operator_attribution" for item in records + ) + ) + exported = json.dumps([item.to_dict() for item in records], sort_keys=True) + self.assertNotIn("identity-1", exported) + self.assertNotIn("identity-unrelated", exported) + self.assertNotIn("counterparty-private-id", exported) + self.assertNotIn("addresses:snapshot-1", exported) + + def test_direct_references_and_canonical_conflicts_fail_closed(self) -> None: + direct = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef(external_references={"parties.party": "party-1"}), + ) + self.assertEqual(3, len(direct)) + + conflict = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + identity_id="identity-1", + external_references={"parties.party": "party-unrelated"}, + ), + ) + self.assertEqual((), conflict) + + alias_conflict = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + identity_id="identity-1", + external_references={"parties.identity": "identity-other"}, + ), + ) + self.assertEqual((), alias_conflict) + + def test_planning_retains_history_and_routes_current_fact_to_review(self) -> None: + subject = DsarSubjectRef(identity_id="identity-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(2, sum(item.kind == "retain" for item in actions)) + self.assertEqual(1, sum(item.kind == "manual_review" for item in actions)) + self.assertTrue(all(not item.executable for item in actions)) + results = self.provider.execute_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + actions=actions, + request_id="dsar-1", + ) + self.assertTrue(all(item.status == "blocked" for item in results)) + + def test_foreign_records_and_actions_are_rejected(self) -> None: + subject = DsarSubjectRef(identity_id="identity-1") + foreign_record = DsarRecordRef( + provider_id="foreign", + module_id="foreign", + resource_type="foreign", + resource_id="foreign-1", + category="foreign", + title="Foreign record", + ) + with self.assertRaisesRegex(ValueError, "foreign provider record"): + self.provider.plan_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + records=(foreign_record,), + ) + + foreign_action = DsarErasureActionRef( + action_id="foreign:delete:1", + provider_id="foreign", + module_id="foreign", + kind="delete", + resource_type="foreign", + resource_id="foreign-1", + title="Delete foreign", + rationale="No", + executable=True, + ) + with self.assertRaisesRegex(ValueError, "foreign provider action"): + self.provider.execute_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + actions=(foreign_action,), + request_id="dsar-1", + ) + + def test_workflow_discovers_only_the_active_tenant_capability(self) -> None: + active = create_data_subject_request( + self.session, + tenant_id="tenant-1", + reference="DSAR-PARTIES-1", + request_kind="access", + subject=DsarSubjectRef(identity_id="identity-1"), + purpose="Subject access request", + legal_basis=None, + due_at=None, + requested_by_account_id="privacy-operator", + ) + search_data_subject_request( + self.session, + registry=_Registry(self.provider), + row=active, + expected_revision=1, + ) + self.assertEqual("searched", active.status) + self.assertEqual( + [PARTIES_DSAR_CAPABILITY], active.coverage["provider_capabilities"] + ) + self.assertEqual(["parties"], active.coverage["covered_modules"]) + + inactive = create_data_subject_request( + self.session, + tenant_id="tenant-1", + reference="DSAR-PARTIES-2", + request_kind="access", + subject=DsarSubjectRef(identity_id="identity-1"), + purpose="Inactive module coverage", + legal_basis=None, + due_at=None, + requested_by_account_id="privacy-operator", + ) + search_data_subject_request( + self.session, + registry=_Registry(self.provider, parties_active=False), + row=inactive, + expected_revision=1, + ) + self.assertEqual([], inactive.coverage["provider_capabilities"]) + self.assertEqual( + [PARTIES_DSAR_CAPABILITY], + inactive.coverage["inactive_provider_capabilities"], + ) + self.assertEqual(0, inactive.search_result["record_count"]) + + def test_manifest_registers_and_documents_the_capability(self) -> None: + self.assertIn(PARTIES_DSAR_CAPABILITY, manifest.capability_factories) + self.assertIn(PARTIES_DSAR_CAPABILITY, manifest.capability_documentation) + self.assertIn( + PARTIES_DSAR_CAPABILITY, + {item.name for item in manifest.provides_interfaces}, + ) + self.assertTrue( + any( + topic.id == "parties.data-subject-requests" + and {"admin", "user"}.issubset(topic.documentation_types) + for topic in manifest.documentation + ) + ) + + +if __name__ == "__main__": + unittest.main()