589 lines
21 KiB
Python
589 lines
21 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from govoplan_access.backend.db.models import Account, User
|
|
from govoplan_addresses.backend.db.models import (
|
|
AddressBook,
|
|
AddressImportRun,
|
|
AddressList,
|
|
AddressListEntry,
|
|
AddressSyncConflict,
|
|
AddressSyncSource,
|
|
AddressSyncTombstone,
|
|
Contact,
|
|
ContactChannelRule,
|
|
ContactEmail,
|
|
ContactFieldProvenance,
|
|
ContactMergeRecord,
|
|
ContactPhone,
|
|
ContactPointQualityDecision,
|
|
ContactPostalAddress,
|
|
ContactRedirect,
|
|
)
|
|
from govoplan_addresses.backend.dsar_provider import (
|
|
ADDRESSES_DSAR_CAPABILITY,
|
|
AddressesDsarProvider,
|
|
)
|
|
from govoplan_addresses.backend.manifest import manifest
|
|
from govoplan_core.core.dsar import (
|
|
DsarErasureActionRef,
|
|
DsarProvider,
|
|
DsarSubjectRef,
|
|
)
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_core.privacy.dsar_workflow import (
|
|
create_data_subject_request,
|
|
search_data_subject_request,
|
|
)
|
|
|
|
|
|
class _Registry:
|
|
def __init__(
|
|
self,
|
|
provider: AddressesDsarProvider,
|
|
*,
|
|
addresses_active: bool = True,
|
|
) -> None:
|
|
self.provider = provider
|
|
self.addresses_active = addresses_active
|
|
|
|
def capability_names(self):
|
|
return (ADDRESSES_DSAR_CAPABILITY,)
|
|
|
|
def capability_owner(self, name):
|
|
self._assert_capability(name)
|
|
return "addresses"
|
|
|
|
def tenant_entitlement_resolver(self):
|
|
addresses_active = self.addresses_active
|
|
|
|
class _Resolver:
|
|
@staticmethod
|
|
def resolve(session, tenant_id):
|
|
del session, tenant_id
|
|
return type(
|
|
"State",
|
|
(),
|
|
{"effective_modules": (("addresses",) if addresses_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": "addresses"})(),)
|
|
|
|
@staticmethod
|
|
def _assert_capability(name: str) -> None:
|
|
if name != ADDRESSES_DSAR_CAPABILITY:
|
|
raise KeyError(name)
|
|
|
|
|
|
class AddressesDsarProviderTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite:///:memory:", future=True)
|
|
Base.metadata.create_all(bind=self.engine)
|
|
self.session = sessionmaker(bind=self.engine, future=True)()
|
|
now = datetime.now(timezone.utc)
|
|
|
|
self.account = Account(
|
|
id="account-1",
|
|
email="subject@example.test",
|
|
normalized_email="subject@example.test",
|
|
display_name="Subject",
|
|
password_hash="password-secret-do-not-export",
|
|
)
|
|
self.user = User(
|
|
id="membership-1",
|
|
tenant_id="tenant-1",
|
|
account_id=self.account.id,
|
|
email=self.account.email,
|
|
display_name="Subject",
|
|
)
|
|
self.book = AddressBook(
|
|
id="book-1",
|
|
tenant_id="tenant-1",
|
|
scope_type="tenant",
|
|
scope_id="tenant-1",
|
|
name="Residents",
|
|
created_by_account_id=self.account.id,
|
|
metadata_={"secret": "book-metadata-do-not-export"},
|
|
)
|
|
self.contact = Contact(
|
|
id="contact-1",
|
|
tenant_id="tenant-1",
|
|
address_book_id=self.book.id,
|
|
display_name="Subject Person",
|
|
given_name="Subject",
|
|
family_name="Person",
|
|
organization="Example household",
|
|
note="A bounded subject note",
|
|
tags=["resident"],
|
|
source_kind="carddav",
|
|
source_ref="https://source.invalid/private/contact.vcf",
|
|
source_payload_kind="vcard",
|
|
source_payload_raw="raw-source-payload-do-not-export",
|
|
source_revision="revision-7",
|
|
provenance={"secret": "contact-provenance-do-not-export"},
|
|
created_by_account_id=self.account.id,
|
|
metadata_={"secret": "contact-metadata-do-not-export"},
|
|
)
|
|
self.email = ContactEmail(
|
|
id="email-1",
|
|
contact_id=self.contact.id,
|
|
label="private",
|
|
email="Subject@Example.test",
|
|
original_email="Subject@Example.test",
|
|
normalized_email="subject@example.test",
|
|
provenance={"secret": "email-provenance-do-not-export"},
|
|
is_primary=True,
|
|
)
|
|
self.phone = ContactPhone(
|
|
id="phone-1",
|
|
contact_id=self.contact.id,
|
|
label="mobile",
|
|
phone="+49 30 123456",
|
|
original_phone="030 123456",
|
|
normalized_phone="+4930123456",
|
|
provenance={"secret": "phone-provenance-do-not-export"},
|
|
is_primary=True,
|
|
)
|
|
self.postal = ContactPostalAddress(
|
|
id="postal-1",
|
|
contact_id=self.contact.id,
|
|
label="home",
|
|
street="Example Street 1",
|
|
postal_code="10115",
|
|
locality="Berlin",
|
|
country="DE",
|
|
original_value={"secret": "postal-original-do-not-export"},
|
|
normalized_value={"secret": "postal-normalized-do-not-export"},
|
|
provenance={"secret": "postal-provenance-do-not-export"},
|
|
is_primary=True,
|
|
)
|
|
self.address_list = AddressList(
|
|
id="list-1",
|
|
tenant_id="tenant-1",
|
|
address_book_id=self.book.id,
|
|
name="District residents",
|
|
created_by_account_id="another-account",
|
|
)
|
|
self.list_entry = AddressListEntry(
|
|
id="entry-1",
|
|
address_list_id=self.address_list.id,
|
|
contact_id=self.contact.id,
|
|
contact_email_id=self.email.id,
|
|
target_kind="email",
|
|
metadata_={"secret": "list-entry-metadata-do-not-export"},
|
|
)
|
|
self.channel_rule = ContactChannelRule(
|
|
id="rule-1",
|
|
tenant_id="tenant-1",
|
|
contact_id=self.contact.id,
|
|
channel="email",
|
|
purpose="resident-notice",
|
|
contact_point_id=self.email.id,
|
|
decision="allow",
|
|
legal_basis="public task",
|
|
evidence_ref="records://consent/evidence-1",
|
|
reason="Current resident preference",
|
|
effective_from=now,
|
|
created_by_account_id="another-account",
|
|
metadata_={"secret": "rule-metadata-do-not-export"},
|
|
)
|
|
self.quality = ContactPointQualityDecision(
|
|
id="quality-1",
|
|
tenant_id="tenant-1",
|
|
contact_id=self.contact.id,
|
|
channel="email",
|
|
contact_point_id=self.email.id,
|
|
state="valid",
|
|
reason_code="verified",
|
|
reason="Verified by operator",
|
|
evidence_ref="files://private/evidence",
|
|
effective_from=now,
|
|
created_by_account_id="another-account",
|
|
metadata_={"secret": "quality-metadata-do-not-export"},
|
|
)
|
|
self.provenance = ContactFieldProvenance(
|
|
id="provenance-1",
|
|
tenant_id="tenant-1",
|
|
contact_id=self.contact.id,
|
|
field_path="emails[0].email",
|
|
value={"secret": "field-value-do-not-export"},
|
|
source_kind="carddav",
|
|
source_ref="https://source.invalid/private",
|
|
source_revision="revision-7",
|
|
precedence=10,
|
|
selected=True,
|
|
reason_code="source_authority",
|
|
explanation="Selected from the authoritative source",
|
|
visibility="operator",
|
|
created_by_account_id="another-account",
|
|
metadata_={"secret": "field-metadata-do-not-export"},
|
|
)
|
|
self.sync_source = AddressSyncSource(
|
|
id="source-1",
|
|
tenant_id="tenant-1",
|
|
address_book_id=self.book.id,
|
|
connector_type="carddav",
|
|
display_name="Residents CardDAV",
|
|
external_account_ref="private-account-ref-do-not-export",
|
|
external_address_book_ref="private-book-ref-do-not-export",
|
|
sync_token="sync-token-do-not-export",
|
|
etag="private-etag-do-not-export",
|
|
remote_revision="private-remote-revision-do-not-export",
|
|
last_diagnostic={"secret": "diagnostic-do-not-export"},
|
|
created_by_account_id=self.account.id,
|
|
metadata_={"secret": "source-metadata-do-not-export"},
|
|
)
|
|
self.tombstone = AddressSyncTombstone(
|
|
id="tombstone-1",
|
|
tenant_id="tenant-1",
|
|
sync_source_id=self.sync_source.id,
|
|
address_book_id=self.book.id,
|
|
contact_id=self.contact.id,
|
|
remote_uid="private-uid-do-not-export",
|
|
resource_href="private-href-do-not-export",
|
|
synced_at=now,
|
|
metadata_={"secret": "tombstone-metadata-do-not-export"},
|
|
)
|
|
self.conflict = AddressSyncConflict(
|
|
id="conflict-1",
|
|
tenant_id="tenant-1",
|
|
sync_source_id=self.sync_source.id,
|
|
address_book_id=self.book.id,
|
|
contact_id=self.contact.id,
|
|
remote_uid="private-conflict-uid-do-not-export",
|
|
resource_href="private-conflict-href-do-not-export",
|
|
field_path="family_name",
|
|
local_value={"secret": "local-value-do-not-export"},
|
|
remote_value={"secret": "remote-value-do-not-export"},
|
|
status="resolved",
|
|
resolution="local",
|
|
resolved_at=now,
|
|
resolved_by_account_id=self.account.id,
|
|
metadata_={"secret": "conflict-metadata-do-not-export"},
|
|
)
|
|
self.import_run = AddressImportRun(
|
|
id="import-1",
|
|
tenant_id="tenant-1",
|
|
address_book_id=self.book.id,
|
|
source_filename="contacts.csv",
|
|
source_format="csv",
|
|
input_hash="a" * 64,
|
|
plan_hash="b" * 64,
|
|
status="applied",
|
|
row_count=1,
|
|
statistics={"secret": "statistics-do-not-export"},
|
|
diagnostics=[{"secret": "import-diagnostic-do-not-export"}],
|
|
plan_data=[{"secret": "import-plan-do-not-export"}],
|
|
result_evidence={"secret": "import-result-do-not-export"},
|
|
created_by_account_id=self.account.id,
|
|
applied_at=now,
|
|
)
|
|
self.merge = ContactMergeRecord(
|
|
id="merge-1",
|
|
tenant_id="tenant-1",
|
|
address_book_id=self.book.id,
|
|
winner_contact_id=self.contact.id,
|
|
loser_contact_ids=["old-contact-1"],
|
|
status="active",
|
|
reason="Duplicate contact",
|
|
survivorship={"secret": "survivorship-do-not-export"},
|
|
decisions=[{"secret": "merge-decisions-do-not-export"}],
|
|
before_payload={"secret": "merge-before-do-not-export"},
|
|
after_payload={"secret": "merge-after-do-not-export"},
|
|
before_hash="c" * 64,
|
|
after_hash="d" * 64,
|
|
created_by_account_id="another-account",
|
|
provenance={"secret": "merge-provenance-do-not-export"},
|
|
)
|
|
self.redirect = ContactRedirect(
|
|
id="redirect-1",
|
|
tenant_id="tenant-1",
|
|
source_contact_id="old-contact-1",
|
|
target_contact_id=self.contact.id,
|
|
merge_record_id=self.merge.id,
|
|
)
|
|
self.unrelated = Contact(
|
|
id="contact-unrelated",
|
|
tenant_id="tenant-1",
|
|
address_book_id=self.book.id,
|
|
display_name="Unrelated Person",
|
|
note="unrelated-person-do-not-export",
|
|
)
|
|
unrelated_email = ContactEmail(
|
|
id="email-unrelated",
|
|
contact_id=self.unrelated.id,
|
|
email="unrelated@example.test",
|
|
original_email="unrelated@example.test",
|
|
normalized_email="unrelated@example.test",
|
|
)
|
|
tenant_two_book = AddressBook(
|
|
id="book-tenant-2",
|
|
tenant_id="tenant-2",
|
|
scope_type="tenant",
|
|
scope_id="tenant-2",
|
|
name="Other tenant",
|
|
)
|
|
tenant_two_contact = Contact(
|
|
id="contact-tenant-2",
|
|
tenant_id="tenant-2",
|
|
address_book_id=tenant_two_book.id,
|
|
display_name="Other Tenant Subject",
|
|
note="other-tenant-do-not-export",
|
|
)
|
|
tenant_two_email = ContactEmail(
|
|
id="email-tenant-2",
|
|
contact_id=tenant_two_contact.id,
|
|
email="subject@example.test",
|
|
original_email="subject@example.test",
|
|
normalized_email="subject@example.test",
|
|
)
|
|
self.session.add_all(
|
|
[
|
|
self.account,
|
|
self.user,
|
|
self.book,
|
|
self.contact,
|
|
self.email,
|
|
self.phone,
|
|
self.postal,
|
|
self.address_list,
|
|
self.list_entry,
|
|
self.channel_rule,
|
|
self.quality,
|
|
self.provenance,
|
|
self.sync_source,
|
|
self.tombstone,
|
|
self.conflict,
|
|
self.import_run,
|
|
self.merge,
|
|
self.redirect,
|
|
self.unrelated,
|
|
unrelated_email,
|
|
tenant_two_book,
|
|
tenant_two_contact,
|
|
tenant_two_email,
|
|
]
|
|
)
|
|
self.session.commit()
|
|
self.provider = AddressesDsarProvider()
|
|
self.subject = DsarSubjectRef(
|
|
account_id=self.account.id,
|
|
email=self.account.email,
|
|
)
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_manifest_publishes_protocol_conforming_provider(self) -> None:
|
|
provided_names = {item.name for item in manifest.provides_interfaces}
|
|
self.assertIn(ADDRESSES_DSAR_CAPABILITY, provided_names)
|
|
provider = manifest.capability_factories[ADDRESSES_DSAR_CAPABILITY](None)
|
|
self.assertIsInstance(provider, DsarProvider)
|
|
|
|
def test_search_is_tenant_scoped_related_and_minimized(self) -> None:
|
|
records = self.provider.search_subject(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=self.subject,
|
|
)
|
|
|
|
resource_types = {record.resource_type for record in records}
|
|
self.assertTrue(
|
|
{
|
|
"addresses_contact",
|
|
"addresses_contact_email",
|
|
"addresses_contact_phone",
|
|
"addresses_contact_postal_address",
|
|
"addresses_list_membership",
|
|
"addresses_channel_rule",
|
|
"addresses_quality_decision",
|
|
"addresses_field_provenance",
|
|
"addresses_merge_record",
|
|
"addresses_contact_redirect",
|
|
"addresses_sync_tombstone",
|
|
"addresses_sync_conflict",
|
|
"addresses_address_book_attribution",
|
|
"addresses_sync_source_attribution",
|
|
"addresses_import_run_attribution",
|
|
}.issubset(resource_types)
|
|
)
|
|
serialized = repr([record.to_dict() for record in records])
|
|
excluded_values = (
|
|
"password-secret-do-not-export",
|
|
"raw-source-payload-do-not-export",
|
|
"contact-provenance-do-not-export",
|
|
"book-metadata-do-not-export",
|
|
"field-value-do-not-export",
|
|
"sync-token-do-not-export",
|
|
"private-account-ref-do-not-export",
|
|
"private-remote-revision-do-not-export",
|
|
"local-value-do-not-export",
|
|
"remote-value-do-not-export",
|
|
"import-plan-do-not-export",
|
|
"merge-before-do-not-export",
|
|
"merge-after-do-not-export",
|
|
"unrelated-person-do-not-export",
|
|
"other-tenant-do-not-export",
|
|
)
|
|
for value in excluded_values:
|
|
self.assertNotIn(value, serialized)
|
|
|
|
def test_conflicting_selectors_and_uncorroborated_reference_fail_closed(
|
|
self,
|
|
) -> None:
|
|
conflict = self.provider.search_subject(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=DsarSubjectRef(
|
|
email="subject@example.test",
|
|
external_references={"addresses.email": "other@example.test"},
|
|
),
|
|
)
|
|
uncorroborated = self.provider.search_subject(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=DsarSubjectRef(
|
|
email="subject@example.test",
|
|
external_references={"addresses.contact": self.unrelated.id},
|
|
),
|
|
)
|
|
|
|
self.assertEqual((), conflict)
|
|
self.assertEqual((), uncorroborated)
|
|
|
|
def test_plan_retains_evidence_and_routes_contact_data_to_review(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,
|
|
)
|
|
|
|
kinds = {action.kind for action in actions}
|
|
self.assertEqual({"manual_review", "retain"}, kinds)
|
|
self.assertFalse(any(action.executable for action in actions))
|
|
retained = [action for action in actions if action.kind == "retain"]
|
|
self.assertTrue(retained)
|
|
self.assertTrue(all(action.rationale for action in retained))
|
|
|
|
results = self.provider.execute_erasure(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=self.subject,
|
|
actions=actions,
|
|
request_id="dsar-addresses-1",
|
|
)
|
|
self.assertEqual({"blocked"}, {result.status for result in results})
|
|
self.assertIsNotNone(self.session.get(Contact, self.contact.id))
|
|
|
|
def test_execution_rejects_foreign_or_forged_executable_actions(self) -> None:
|
|
foreign = DsarErasureActionRef(
|
|
action_id="mail:delete:contact:contact-1",
|
|
provider_id="mail",
|
|
module_id="mail",
|
|
kind="delete",
|
|
resource_type="addresses_contact",
|
|
resource_id=self.contact.id,
|
|
title="Foreign delete",
|
|
rationale="Must be rejected",
|
|
executable=True,
|
|
)
|
|
forged = DsarErasureActionRef(
|
|
action_id="addresses:delete:addresses_contact:contact-1",
|
|
provider_id="addresses",
|
|
module_id="addresses",
|
|
kind="delete",
|
|
resource_type="addresses_contact",
|
|
resource_id=self.contact.id,
|
|
title="Forged delete",
|
|
rationale="Must be rejected",
|
|
executable=True,
|
|
)
|
|
|
|
with self.assertRaises(ValueError):
|
|
self.provider.execute_erasure(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=self.subject,
|
|
actions=(foreign,),
|
|
request_id="dsar-addresses-2",
|
|
)
|
|
with self.assertRaises(ValueError):
|
|
self.provider.execute_erasure(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=self.subject,
|
|
actions=(forged,),
|
|
request_id="dsar-addresses-2",
|
|
)
|
|
|
|
def test_core_workflow_discovers_active_and_inactive_provider(self) -> None:
|
|
request = create_data_subject_request(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
reference="DSAR-ADDRESSES-1",
|
|
request_kind="access",
|
|
subject=self.subject,
|
|
purpose="Respond to an authorized privacy 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=request,
|
|
expected_revision=1,
|
|
)
|
|
self.assertEqual("searched", request.status)
|
|
self.assertEqual(["addresses"], request.coverage["covered_modules"])
|
|
self.assertEqual([], request.coverage["modules_without_provider"])
|
|
|
|
disabled = create_data_subject_request(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
reference="DSAR-ADDRESSES-DISABLED",
|
|
request_kind="access",
|
|
subject=self.subject,
|
|
purpose="Verify disabled-module coverage.",
|
|
legal_basis="Article 15 GDPR",
|
|
due_at=None,
|
|
requested_by_account_id="privacy-officer",
|
|
)
|
|
search_data_subject_request(
|
|
self.session,
|
|
registry=_Registry(self.provider, addresses_active=False),
|
|
row=disabled,
|
|
expected_revision=1,
|
|
)
|
|
|
|
self.assertEqual(0, disabled.search_result["record_count"])
|
|
self.assertEqual(
|
|
[ADDRESSES_DSAR_CAPABILITY],
|
|
disabled.coverage["inactive_provider_capabilities"],
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|