283 lines
10 KiB
Python
283 lines
10 KiB
Python
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_dist_lists.backend.db.models import (
|
|
DistributionList,
|
|
DistributionListEntry,
|
|
DistributionListRevision,
|
|
DistributionListSnapshot,
|
|
)
|
|
from govoplan_dist_lists.backend.dsar_provider import (
|
|
DIST_LISTS_DSAR_CAPABILITY,
|
|
DistributionListsDsarProvider,
|
|
)
|
|
from govoplan_dist_lists.backend.manifest import manifest
|
|
|
|
|
|
NOW = datetime(2026, 8, 22, 16, 0, tzinfo=UTC)
|
|
|
|
|
|
def _recipient(
|
|
*,
|
|
key: str,
|
|
email: str,
|
|
account_id: str | None = None,
|
|
identity_id: str | None = None,
|
|
) -> dict[str, object]:
|
|
return {
|
|
"recipient_key": key,
|
|
"display_name": f"Recipient {email}",
|
|
"status": "usable",
|
|
"channels": [
|
|
{
|
|
"channel": "email",
|
|
"target": email,
|
|
"target_key": f"email:{email.casefold()}",
|
|
"status": "usable",
|
|
"contact_point_id": f"contact-point-{key}",
|
|
"decision_provenance": {"secret": "do-not-export-provenance"},
|
|
}
|
|
],
|
|
"account_id": account_id,
|
|
"identity_id": identity_id,
|
|
"contact_id": None,
|
|
"source_entry_ids": ["entry-1"],
|
|
"attributes": {"secret": "do-not-export-attributes"},
|
|
"provenance": {"secret": "do-not-export-recipient-provenance"},
|
|
}
|
|
|
|
|
|
class DistributionListsDsarProviderTests(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 = DistributionListsDsarProvider()
|
|
self._seed()
|
|
self.session.commit()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def _seed(self) -> None:
|
|
distribution_list = DistributionList(
|
|
id="list-1",
|
|
tenant_id="tenant-1",
|
|
scope_type="user",
|
|
scope_id="account-1",
|
|
name="Sensitive list name do not export",
|
|
description="Sensitive description do not export",
|
|
status="active",
|
|
current_revision_id="revision-1",
|
|
current_revision=1,
|
|
resource_revision=1,
|
|
created_by_account_id="account-1",
|
|
updated_by_account_id="account-1",
|
|
metadata_={"secret": "list-metadata-do-not-export"},
|
|
created_at=NOW,
|
|
updated_at=NOW,
|
|
)
|
|
revision = DistributionListRevision(
|
|
id="revision-1",
|
|
tenant_id="tenant-1",
|
|
distribution_list_id="list-1",
|
|
revision=1,
|
|
definition_kind="static",
|
|
definition_hash="definition-hash-do-not-export",
|
|
parameter_schema=[{"secret": "parameter-do-not-export"}],
|
|
constraints={"secret": "constraint-do-not-export"},
|
|
source_fingerprints=[{"secret": "fingerprint-do-not-export"}],
|
|
created_by_account_id="account-1",
|
|
created_at=NOW,
|
|
updated_at=NOW,
|
|
)
|
|
entry = DistributionListEntry(
|
|
id="entry-1",
|
|
tenant_id="tenant-1",
|
|
revision_id="revision-1",
|
|
entry_key="subject-email",
|
|
kind="raw_email",
|
|
mode="include",
|
|
source_provider="local",
|
|
source_resource_type="email",
|
|
source_resource_id="Subject@Example.test",
|
|
source_revision="source-revision-do-not-export",
|
|
source_fingerprint="source-fingerprint-do-not-export",
|
|
source_label="source-label-do-not-export",
|
|
source_metadata={"secret": "source-metadata-do-not-export"},
|
|
label="entry-label-do-not-export",
|
|
purpose="service_notice",
|
|
requested_channels=["email"],
|
|
order_index=0,
|
|
configuration={"secret": "configuration-do-not-export"},
|
|
created_at=NOW,
|
|
updated_at=NOW,
|
|
)
|
|
snapshot = DistributionListSnapshot(
|
|
id="snapshot-1",
|
|
tenant_id="tenant-1",
|
|
distribution_list_id="list-1",
|
|
revision_id="revision-1",
|
|
revision_number=1,
|
|
idempotency_key="idempotency-key-do-not-export",
|
|
request_={"secret": "request-do-not-export"},
|
|
expansion_hash="expansion-hash-do-not-export",
|
|
effective_at=NOW,
|
|
recipient_count=2,
|
|
excluded_count=0,
|
|
recipients=[
|
|
_recipient(
|
|
key="identity-1",
|
|
email="subject@example.test",
|
|
account_id="account-1",
|
|
identity_id="identity-1",
|
|
),
|
|
_recipient(
|
|
key="identity-other",
|
|
email="other-person-do-not-export@example.test",
|
|
account_id="account-other-do-not-export",
|
|
identity_id="identity-other-do-not-export",
|
|
),
|
|
],
|
|
excluded=[],
|
|
diagnostics=[{"secret": "diagnostics-do-not-export"}],
|
|
provider_evidence=[{"secret": "provider-evidence-do-not-export"}],
|
|
stale=False,
|
|
truncated=False,
|
|
created_by_account_id="account-1",
|
|
provenance={"secret": "snapshot-provenance-do-not-export"},
|
|
created_at=NOW,
|
|
updated_at=NOW,
|
|
)
|
|
other_tenant = DistributionList(
|
|
id="list-other-tenant",
|
|
tenant_id="tenant-2",
|
|
scope_type="user",
|
|
scope_id="account-1",
|
|
name="Other tenant do not export",
|
|
status="active",
|
|
current_revision_id="revision-other",
|
|
current_revision=1,
|
|
resource_revision=1,
|
|
)
|
|
self.session.add_all(
|
|
(distribution_list, revision, entry, snapshot, other_tenant)
|
|
)
|
|
|
|
def test_search_exports_only_exact_subject_projection(self) -> None:
|
|
self.assertIsInstance(self.provider, DsarProvider)
|
|
records = self.provider.search_subject(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=DsarSubjectRef(
|
|
account_id="account-1",
|
|
identity_id="identity-1",
|
|
email=" SUBJECT@example.test ",
|
|
),
|
|
)
|
|
self.assertEqual(
|
|
{
|
|
"distribution_list_subject_scope",
|
|
"distribution_list_definition_entry",
|
|
"distribution_list_snapshot_recipient",
|
|
"distribution_list_actor_attribution",
|
|
"distribution_list_revision_actor_attribution",
|
|
"distribution_list_snapshot_actor_attribution",
|
|
},
|
|
{record.resource_type for record in records},
|
|
)
|
|
exported = json.dumps([record.to_dict() for record in records])
|
|
self.assertIn("subject@example.test", exported.casefold())
|
|
for excluded in (
|
|
"other-person-do-not-export@example.test",
|
|
"account-other-do-not-export",
|
|
"identity-other-do-not-export",
|
|
"list-metadata-do-not-export",
|
|
"definition-hash-do-not-export",
|
|
"parameter-do-not-export",
|
|
"constraint-do-not-export",
|
|
"fingerprint-do-not-export",
|
|
"source-revision-do-not-export",
|
|
"source-fingerprint-do-not-export",
|
|
"source-label-do-not-export",
|
|
"source-metadata-do-not-export",
|
|
"configuration-do-not-export",
|
|
"idempotency-key-do-not-export",
|
|
"request-do-not-export",
|
|
"expansion-hash-do-not-export",
|
|
"diagnostics-do-not-export",
|
|
"provider-evidence-do-not-export",
|
|
"snapshot-provenance-do-not-export",
|
|
"do-not-export-provenance",
|
|
"do-not-export-attributes",
|
|
):
|
|
self.assertNotIn(excluded, exported)
|
|
|
|
def test_resource_references_only_narrow_verified_subjects(self) -> None:
|
|
self.assertEqual(
|
|
(),
|
|
self.provider.search_subject(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=DsarSubjectRef(
|
|
external_references={"dist_lists.snapshot": "snapshot-1"}
|
|
),
|
|
),
|
|
)
|
|
narrowed = self.provider.search_subject(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=DsarSubjectRef(
|
|
email="subject@example.test",
|
|
external_references={"dist_lists.snapshot": "snapshot-1"},
|
|
),
|
|
)
|
|
self.assertEqual(
|
|
["distribution_list_snapshot_recipient"],
|
|
[record.resource_type for record in narrowed],
|
|
)
|
|
conflict = self.provider.search_subject(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=DsarSubjectRef(
|
|
email="subject@example.test",
|
|
external_references={"dist_lists.email": "other@example.test"},
|
|
),
|
|
)
|
|
self.assertEqual((), conflict)
|
|
|
|
def test_definition_change_requires_review_and_snapshot_is_retained(self) -> None:
|
|
subject = DsarSubjectRef(email="subject@example.test")
|
|
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["distribution_list_definition_entry"])
|
|
self.assertEqual("retain", by_type["distribution_list_snapshot_recipient"])
|
|
|
|
def test_manifest_registers_provider_and_documentation(self) -> None:
|
|
self.assertIn(DIST_LISTS_DSAR_CAPABILITY, manifest.capability_factories)
|
|
self.assertIn(
|
|
"dist_lists.data-subject-requests",
|
|
{topic.id for topic in manifest.documentation},
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|