GovOPlaN/govoplan-campaign
2
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add governed Campaign DSAR coverage

Browse Source
This commit is contained in:
zemion
2026-08-20 23:15:11 +02:00
parent 2a00d910df
commit 73cfad209a
5 changed files with 2135 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+3
View File
@@ -92,6 +92,9 @@ services can cooperate without importing campaign internals:
- `campaigns.policyContext` for retention/policy provenance
- `campaigns.deliveryTasks` for queued send and append-to-Sent workers
- `campaigns.retention` for campaign-owned retention cleanup
- `privacy.dsar.campaigns` for tenant-scoped recipient, version, delivery,
report-projection, and artifact-metadata discovery plus governed erasure
planning
Keep these capability payloads narrow: stable ids, policy payloads, and task
results only.
docs/CAMPAIGN_HANDBOOK.md
+24
View File
@@ -557,6 +557,30 @@ purpose, lawful basis, minimization, export control, and retention before the
campaign starts; do not use Campaign as a substitute consent or address-master
system.
The Core data-subject-request workflow discovers Campaign through the optional
`privacy.dsar.campaigns` capability. After the request's email, membership, and
namespaced Campaign references have been independently authorized and
corroborated, the provider searches only the effective tenant and isolates the
matching recipient entries and jobs. Its JSON result includes safe Campaign,
version, delivery-attempt, schedule, report-projection, share, import-mapping,
attachment, and generated-artifact metadata. Generated EML bytes and paths,
storage locators, delivery target snapshots, worker claims, idempotency
material, credentials, secret-like values, and unrelated recipients are never
embedded in that result. Authorized Campaign and Files review surfaces remain
the source for content that cannot safely be copied into the DSAR case.
Built, locked, published, terminal, delivered, or corrected records are
retained with an explicit reason and continue through Campaign's configured
retention and redaction process. Draft recipient content and user-owned
attachment content require coordinated manual review because copies may span
version JSON, jobs, generated messages, and managed files. The provider can
idempotently revoke an active share aimed at the subject and delete the
subject's personal recipient-import mapping profile. It does not rewrite
delivery evidence, delete generated artifacts, or report derived Campaign
counts as a separate store. Re-running an approved action is safe: already
revoked or absent data is reported as unchanged, and tenant, subject, and row
ownership are revalidated immediately before mutation.
### Audit and destructive actions
Material authoring, validation, locking, review, queueing, send, retry,
src/govoplan_campaign/backend/dsar_provider.py
+1327
View File
File diff suppressed because it is too large Load Diff
src/govoplan_campaign/backend/manifest.py
+76
View File
@@ -65,11 +65,19 @@ from govoplan_campaign.backend.documentation import (
CAMPAIGN_USER_DOCUMENTATION,
documentation_topics,
)
from govoplan_campaign.backend.dsar_provider import CAMPAIGN_DSAR_CAPABILITY
from govoplan_campaign.backend.search_source import create_campaign_search_source
register_campaign_change_tracking()
def _dsar_provider(context: ModuleContext) -> object:
del context
from govoplan_campaign.backend.dsar_provider import CampaignDsarProvider
return CampaignDsarProvider()
def _permission(
scope: str, label: str, description: str, category: str
) -> PermissionDefinition:
@@ -409,6 +417,7 @@ manifest = ModuleManifest(
name=REPORT_PROVIDER_CAPABILITY_PREFIX + "campaigns",
version="1.0.0",
),
ModuleInterfaceProvider(name=CAMPAIGN_DSAR_CAPABILITY, version="0.1.0"),
),
requires_interfaces=(
ModuleInterfaceRequirement(
@@ -659,6 +668,65 @@ manifest = ModuleManifest(
),
documentation=(
*CAMPAIGN_USER_DOCUMENTATION,
DocumentationTopic(
id="campaigns.privacy.data-subject-requests",
title="Review Campaign data in a data-subject request",
summary="Collect recipient, version, delivery, report, and artifact metadata without rewriting immutable evidence.",
body=(
"Campaign's DSAR provider searches the effective tenant by normalized recipient email, direct membership references, and namespaced Campaign job, entry, version, or Campaign references. "
"It isolates matching inline-recipient fields and job metadata, and reports built versions, delivery attempts, Postbox and print outcomes, message-action corrections, recipient-specific report projections, generated-message digests, and attachment metadata. It does not export EML bytes, object or local paths, provider target snapshots, worker claims, idempotency material, secrets, credentials, or unrelated recipient addresses. "
"Built, locked, published, terminal, delivered, or corrected records remain retained with a reason and continue through Campaign's configured retention/redaction process. Draft recipient content and user-owned attachment content require coordinated manual review because the same data may occur in version JSON, jobs, and generated artifacts. The provider can idempotently delete a personal recipient-import mapping profile and revoke an active Campaign share aimed at the subject. It never rewrites delivered evidence or deletes generated artifacts directly. Campaign reports are derived projections rather than a separate personal-data store."
),
layer="configured",
documentation_types=("admin",),
audience=("privacy_officer", "campaign_manager", "records_manager", "operator"),
order=42,
conditions=(
DocumentationCondition(
required_modules=("campaigns", "access"),
any_scopes=(
"access:privacy:read",
"access:privacy:manage",
"access:privacy:erase",
),
),
),
links=(
DocumentationLink(
label="Data-subject requests",
href="/admin?section=tenant-data-subject-requests",
kind="runtime",
),
DocumentationLink(
label="Campaign handbook",
href="govoplan-campaign/docs/CAMPAIGN_HANDBOOK.md",
kind="repository",
),
),
related_modules=("access", "audit", "files", "mail", "postbox", "reporting"),
metadata={
"kind": "workflow",
"route": "/admin?section=tenant-data-subject-requests",
"screen": "Data-subject requests",
"help_contexts": ["admin.privacy.data-subject-requests"],
"prerequisites": [
"The privacy request and recipient selectors have been independently authorized and corroborated.",
"The reviewer understands the effective Campaign retention policy and delivery-evidence obligations.",
],
"steps": [
"Run the Campaign provider search and review recipient, version, job, attempt, report-projection, and artifact dispositions.",
"Inspect matching draft content manually and keep every evidence retention reason with the case decision.",
"Execute only an approved user-owned mapping deletion or subject-targeted share revocation.",
"Use Campaign retention and artifact reconciliation for approved content redaction or expiry; do not mutate delivered evidence ad hoc.",
],
"limitations": [
"Generated EML bytes and attachment content are not embedded in the JSON export; authorized Campaign or Files review paths remain authoritative.",
"Draft recipient erasure is manual until a coordinated version/job/artifact rewrite contract can prove that no partial copy remains.",
],
"outcome": "Campaign personal data receives an explicit retained, review, revoke, or delete disposition without weakening delivery evidence.",
"verification": "Confirm matching recipients are isolated, no locator or credential material appears, report counts derive from the same matched jobs, and repeated reversible actions are unchanged.",
},
),
DocumentationTopic(
id="campaigns.access.child-evidence",
title="Explain access to Campaign child evidence",
@@ -1313,6 +1381,7 @@ manifest = ModuleManifest(
"govoplan_campaign.backend.reports.provider",
fromlist=["CampaignAggregateReportProvider"],
).CampaignAggregateReportProvider(),
CAMPAIGN_DSAR_CAPABILITY: _dsar_provider,
},
capability_documentation={
REPORT_PROVIDER_CAPABILITY_PREFIX + "campaigns": CapabilityDocumentation(
@@ -1325,6 +1394,13 @@ manifest = ModuleManifest(
documentation_types=("admin", "user"),
audience=("user", "reporting_analyst", "privacy_officer"),
),
CAMPAIGN_DSAR_CAPABILITY: CapabilityDocumentation(
label="Campaign data-subject request provider",
summary="Finds isolated recipient and Campaign evidence metadata and classifies governed erasure actions.",
contract_version="0.1.0",
documentation_types=("admin",),
audience=("privacy_officer", "campaign_manager", "records_manager"),
),
},
operational_check_providers=(
OperationalCheckProviderRegistration(
tests/test_dsar_provider.py
+705
View File
@@ -0,0 +1,705 @@
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, Group, User
from govoplan_campaign.backend.db.models import (
AttachmentBlob,
AttachmentInstance,
Campaign,
CampaignIssue,
CampaignJob,
CampaignMessageAction,
CampaignMessageActionAttempt,
CampaignSchedule,
CampaignShare,
CampaignVersion,
ImapAppendAttempt,
PostboxDeliveryAttempt,
PrintOutputAttempt,
RecipientImportMappingProfile,
SendAttempt,
)
from govoplan_campaign.backend.dsar_provider import (
CAMPAIGN_DSAR_CAPABILITY,
CampaignDsarProvider,
)
from govoplan_campaign.backend.manifest import manifest
from govoplan_core.core.change_sequence import ChangeSequenceEntry
from govoplan_core.core.dsar import DsarProvider, DsarSubjectRef
from govoplan_core.db.base import Base
from govoplan_core.privacy.dsar_workflow import (
DataSubjectRequest,
create_data_subject_request,
execute_data_subject_erasure,
plan_data_subject_erasure,
search_data_subject_request,
)
class _Registry:
def __init__(
self,
provider: CampaignDsarProvider,
*,
campaign_active: bool = True,
) -> None:
self.provider = provider
self.campaign_active = campaign_active
def capability_names(self):
return (CAMPAIGN_DSAR_CAPABILITY,)
def capability_owner(self, name):
self._assert_capability(name)
return "campaigns"
def tenant_entitlement_resolver(self):
campaign_active = self.campaign_active
class _Resolver:
@staticmethod
def resolve(session, tenant_id):
del session, tenant_id
return type(
"State",
(),
{"effective_modules": ("campaigns",) if campaign_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": "campaigns"})(),)
@staticmethod
def _assert_capability(name: str) -> None:
if name != CAMPAIGN_DSAR_CAPABILITY:
raise KeyError(name)
class CampaignDsarProviderTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine("sqlite:///:memory:", future=True)
Base.metadata.create_all(
bind=self.engine,
tables=[
Account.__table__,
User.__table__,
Group.__table__,
ChangeSequenceEntry.__table__,
DataSubjectRequest.__table__,
Campaign.__table__,
CampaignShare.__table__,
CampaignVersion.__table__,
CampaignJob.__table__,
CampaignIssue.__table__,
AttachmentBlob.__table__,
AttachmentInstance.__table__,
SendAttempt.__table__,
CampaignMessageAction.__table__,
CampaignMessageActionAttempt.__table__,
ImapAppendAttempt.__table__,
PostboxDeliveryAttempt.__table__,
PrintOutputAttempt.__table__,
RecipientImportMappingProfile.__table__,
CampaignSchedule.__table__,
],
)
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",
)
other_account = Account(
id="account-2",
email="other@example.test",
normalized_email="other@example.test",
display_name="Other",
)
self.user = User(
id="membership-1",
tenant_id="tenant-1",
account_id=self.account.id,
email="subject@example.test",
display_name="Subject",
)
self.other_user = User(
id="membership-2",
tenant_id="tenant-1",
account_id=other_account.id,
email="other@example.test",
display_name="Other",
)
self.campaign = Campaign(
id="campaign-1",
tenant_id="tenant-1",
created_by_user_id=self.other_user.id,
owner_user_id=self.other_user.id,
external_id="privacy-notice",
name="Privacy notice",
status="active",
)
self.version = CampaignVersion(
id="version-1",
campaign_id=self.campaign.id,
version_number=1,
workflow_state="built",
execution_snapshot_hash="a" * 64,
raw_json={
"entries": {
"inline": [
{
"id": "entry-subject",
"to": [
{
"email": "subject@example.test",
"name": "Subject Person",
}
],
"cc": [
{
"email": "other@example.test",
"name": "Unrelated person",
}
],
"body": "private-rendered-body-do-not-export",
"password": "inline-secret-do-not-export",
"case_reference": "CASE-SUBJECT-1",
},
{
"id": "entry-other",
"to": [{"email": "other@example.test"}],
"private_value": "other-recipient-data-do-not-export",
},
]
}
},
)
self.draft_version = CampaignVersion(
id="version-draft",
campaign_id=self.campaign.id,
version_number=2,
workflow_state="editing",
raw_json={
"entries": {
"inline": [
{
"id": "entry-draft-subject",
"to": [{"email": "subject@example.test"}],
"case_reference": "CASE-DRAFT-1",
}
]
}
},
)
self.job = CampaignJob(
id="job-subject",
tenant_id="tenant-1",
campaign_id=self.campaign.id,
campaign_version_id=self.version.id,
entry_index=0,
entry_id="entry-subject",
recipient_email="Subject@Example.Test",
subject="Your governed notice",
eml_storage_key="private/eml/key-do-not-export",
eml_local_path="/private/message-do-not-export.eml",
eml_size_bytes=512,
eml_sha256="b" * 64,
build_status="built",
validation_status="ready",
queue_status="completed",
send_status="smtp_accepted",
postbox_status="accepted",
print_status="accepted",
imap_status="appended",
attempt_count=1,
queued_at=now,
sent_at=now,
claim_token="job-claim-do-not-export",
resolved_recipients={
"from": {"email": "sender@example.test"},
"to": [{"email": "subject@example.test", "name": "Subject"}],
"cc": [{"email": "other@example.test", "name": "Other"}],
"legacy": ["other@example.test", "subject@example.test"],
},
resolved_attachments=[
{"storage_key": "resolved-attachment-key-do-not-export"}
],
)
other_job = CampaignJob(
id="job-other",
tenant_id="tenant-1",
campaign_id=self.campaign.id,
campaign_version_id=self.version.id,
entry_index=1,
entry_id="entry-other",
recipient_email="other@example.test",
subject="Other person's message",
build_status="built",
validation_status="ready",
)
tenant_two_campaign = Campaign(
id="campaign-tenant-2",
tenant_id="tenant-2",
external_id="other-tenant",
name="Other tenant data do not export",
)
tenant_two_version = CampaignVersion(
id="version-tenant-2",
campaign_id=tenant_two_campaign.id,
version_number=1,
raw_json={
"entries": {"inline": [{"to": [{"email": "subject@example.test"}]}]}
},
)
tenant_two_job = CampaignJob(
id="job-tenant-2",
tenant_id="tenant-2",
campaign_id=tenant_two_campaign.id,
campaign_version_id=tenant_two_version.id,
entry_index=0,
recipient_email="subject@example.test",
)
self.issue = CampaignIssue(
id="issue-1",
tenant_id="tenant-1",
campaign_id=self.campaign.id,
campaign_version_id=self.version.id,
job_id=self.job.id,
severity="warning",
code="delivery_warning",
message="issue-detail-do-not-export",
source="private-source-do-not-export",
behavior="review",
)
send_attempt = SendAttempt(
id="send-attempt-1",
job_id=self.job.id,
attempt_number=1,
status="accepted",
claim_token="attempt-claim-do-not-export",
smtp_status_code=250,
smtp_response="smtp-response-do-not-export",
error_message="transport-detail-do-not-export",
started_at=now,
finished_at=now,
)
postbox_attempt = PostboxDeliveryAttempt(
id="postbox-attempt-1",
tenant_id="tenant-1",
job_id=self.job.id,
target_key="target-key-do-not-export",
target_index=0,
attempt_number=1,
idempotency_key="postbox-idempotency-do-not-export",
status="accepted",
target_snapshot={"private": "snapshot-do-not-export"},
provider_delivery_id="delivery-1",
provider_message_id="message-1",
postbox_id="postbox-1",
address="subject@example.test",
evidence={"private": "postbox-evidence-do-not-export"},
started_at=now,
finished_at=now,
)
print_attempt = PrintOutputAttempt(
id="print-attempt-1",
tenant_id="tenant-1",
job_id=self.job.id,
attempt_number=1,
idempotency_key="print-idempotency-do-not-export",
status="accepted",
render_id="render-1",
artifact_sha256="c" * 64,
evidence={"private": "print-evidence-do-not-export"},
started_at=now,
finished_at=now,
)
self.share = CampaignShare(
id="share-1",
tenant_id="tenant-1",
campaign_id=self.campaign.id,
target_type="user",
target_id=self.user.id,
permission="read",
created_by_user_id=self.other_user.id,
)
self.profile = RecipientImportMappingProfile(
id="mapping-1",
tenant_id="tenant-1",
owner_user_id=self.user.id,
name="Subject mapping",
column_count=2,
headers=["email", "case_reference"],
normalized_headers=["email", "case_reference"],
ordered_header_fingerprint="d" * 64,
unordered_header_fingerprint="e" * 64,
delimiter=";",
header_rows=1,
quoted=True,
value_separators=",;|",
mappings=[
{"header": "email", "field": "to.0.email"},
{"secret": "profile-secret-do-not-export"},
],
)
blob = AttachmentBlob(
id="blob-1",
tenant_id="tenant-1",
sha256="f" * 64,
size_bytes=42,
mime_type="application/pdf",
storage_bucket="private-bucket-do-not-export",
storage_key="private-attachment-key-do-not-export",
)
attachment = AttachmentInstance(
id="attachment-1",
tenant_id="tenant-1",
owner_user_id=self.other_user.id,
campaign_id=self.campaign.id,
blob_id=blob.id,
logical_name="notice",
filename="notice.pdf",
tags=["notice"],
metadata_={"secret": "attachment-secret-do-not-export"},
)
schedule = CampaignSchedule(
id="schedule-1",
tenant_id="tenant-1",
campaign_id=self.campaign.id,
source_version_id=self.version.id,
created_by_user_id=self.other_user.id,
name="Recurring privacy notice",
delivery_mode="manual",
recurrence_kind="monthly",
starts_at=now,
next_fire_at=now,
max_occurrences=12,
source_snapshot={"private": "schedule-snapshot-do-not-export"},
source_snapshot_hash="1" * 64,
)
self.session.add_all(
[
self.account,
other_account,
self.user,
self.other_user,
self.campaign,
self.version,
self.draft_version,
self.job,
other_job,
tenant_two_campaign,
tenant_two_version,
tenant_two_job,
self.issue,
send_attempt,
postbox_attempt,
print_attempt,
self.share,
self.profile,
blob,
attachment,
schedule,
]
)
self.session.commit()
self.provider = CampaignDsarProvider()
self.subject = DsarSubjectRef(
membership_id=self.user.id,
email="subject@example.test",
)
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(CAMPAIGN_DSAR_CAPABILITY, provided_names)
provider = manifest.capability_factories[CAMPAIGN_DSAR_CAPABILITY](None)
self.assertIsInstance(provider, DsarProvider)
self.assertIn(
"campaigns.privacy.data-subject-requests",
{topic.id for topic in manifest.documentation},
)
def test_search_is_tenant_scoped_minimized_and_recipient_specific(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(
{
"campaign",
"campaign_version",
"campaign_recipient_job",
"campaign_message_artifact",
"campaign_delivery_issue",
"campaign_send_attempt",
"campaign_postbox_attempt",
"campaign_print_attempt",
"campaign_report_projection",
"campaign_share",
"recipient_import_mapping_profile",
"campaign_schedule",
"campaign_attachment",
}.issubset(resource_types)
)
report = next(
record
for record in records
if record.resource_type == "campaign_report_projection"
)
self.assertEqual(1, report.data["matched_job_count"])
artifact = next(
record
for record in records
if record.resource_type == "campaign_message_artifact"
)
self.assertEqual("b" * 64, artifact.data["sha256"])
self.assertEqual(512, artifact.data["size_bytes"])
serialized = repr([record.to_dict() for record in records])
for hidden in (
"job-tenant-2",
"Other tenant data do not export",
"job-other",
"other@example.test",
"Unrelated person",
"other-recipient-data-do-not-export",
"private-rendered-body-do-not-export",
"inline-secret-do-not-export",
"private/eml/key-do-not-export",
"/private/message-do-not-export.eml",
"resolved-attachment-key-do-not-export",
"job-claim-do-not-export",
"issue-detail-do-not-export",
"private-source-do-not-export",
"attempt-claim-do-not-export",
"smtp-response-do-not-export",
"transport-detail-do-not-export",
"target-key-do-not-export",
"postbox-idempotency-do-not-export",
"snapshot-do-not-export",
"postbox-evidence-do-not-export",
"print-idempotency-do-not-export",
"print-evidence-do-not-export",
"profile-secret-do-not-export",
"private-bucket-do-not-export",
"private-attachment-key-do-not-export",
"attachment-secret-do-not-export",
"schedule-snapshot-do-not-export",
):
self.assertNotIn(hidden, serialized)
def test_conflicting_email_references_fail_closed_for_recipient_data(self) -> None:
records = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(
email="subject@example.test",
external_references={"campaign.email": "other@example.test"},
),
)
self.assertEqual((), records)
def test_plan_retains_evidence_and_limits_execution_to_reversible_data(
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.assertTrue({"retain", "manual_review", "revoke", "delete"}.issubset(kinds))
self.assertTrue(
any(
action.action_id
== "campaigns:retain:campaign_recipient_job:job-subject"
for action in actions
)
)
self.assertTrue(
any(
action.action_id == "campaigns:review:campaign_version:version-draft"
for action in actions
)
)
executable_ids = {action.action_id for action in actions if action.executable}
self.assertEqual(
{
"campaigns:revoke:campaign_share:share-1",
"campaigns:delete:recipient_import_mapping_profile:mapping-1",
},
executable_ids,
)
def test_execution_is_revalidated_tenant_bound_and_idempotent(self) -> None:
actions = self._executable_actions()
wrong_tenant = self.provider.execute_erasure(
self.session,
tenant_id="tenant-2",
subject=self.subject,
actions=actions,
request_id="dsar-wrong-tenant",
)
self.assertEqual({"blocked"}, {result.status for result in wrong_tenant})
first = self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=self.subject,
actions=actions,
request_id="dsar-1",
)
self.assertEqual({"executed"}, {result.status for result in first})
self.session.flush()
self.assertIsNotNone(self.share.revoked_at)
self.assertIsNone(
self.session.get(RecipientImportMappingProfile, self.profile.id)
)
repeated = self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=self.subject,
actions=actions,
request_id="dsar-1",
)
self.assertEqual({"unchanged"}, {result.status for result in repeated})
def test_execution_blocks_when_mapping_owner_changed_after_planning(self) -> None:
delete_action = next(
action
for action in self._executable_actions()
if action.resource_type == "recipient_import_mapping_profile"
)
self.profile.owner_user_id = self.other_user.id
self.session.flush()
result = self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=self.subject,
actions=(delete_action,),
request_id="dsar-stale",
)
self.assertEqual("blocked", result[0].status)
self.assertIsNotNone(
self.session.get(RecipientImportMappingProfile, self.profile.id)
)
def test_core_workflow_discovers_active_provider_and_skips_it_when_disabled(
self,
) -> None:
request = create_data_subject_request(
self.session,
tenant_id="tenant-1",
reference="DSAR-CAMPAIGN-1",
request_kind="access_and_erasure",
subject=self.subject,
purpose="Respond to an authorized privacy request.",
legal_basis="Article 15 and 17 GDPR",
due_at=None,
requested_by_account_id="privacy-officer",
)
self.session.commit()
registry = _Registry(self.provider)
search_data_subject_request(
self.session,
registry=registry,
row=request,
expected_revision=1,
)
self.assertEqual("searched", request.status)
self.assertEqual(["campaigns"], request.coverage["covered_modules"])
self.assertEqual([], request.coverage["modules_without_provider"])
plan_data_subject_erasure(
self.session,
registry=registry,
row=request,
expected_revision=2,
)
executable_ids = [
action["action_id"]
for action in request.erasure_plan["actions"]
if action["executable"]
]
execute_data_subject_erasure(
self.session,
registry=registry,
row=request,
expected_revision=3,
action_ids=executable_ids,
)
self.assertEqual("completed", request.status)
disabled = create_data_subject_request(
self.session,
tenant_id="tenant-1",
reference="DSAR-CAMPAIGN-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, campaign_active=False),
row=disabled,
expected_revision=1,
)
self.assertEqual(0, disabled.search_result["record_count"])
self.assertEqual(
[CAMPAIGN_DSAR_CAPABILITY],
disabled.coverage["inactive_provider_capabilities"],
)
def _executable_actions(self):
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,
)
return tuple(action for action in actions if action.executable)
if __name__ == "__main__":
unittest.main()