feat: complete campaign wizards and retention reporting

This commit is contained in:
2026-07-31 04:21:34 +02:00
parent e689fdf495
commit fa4eb39e0b
18 changed files with 785 additions and 55 deletions
+110
View File
@@ -6,7 +6,10 @@ import unittest
from unittest.mock import patch
from govoplan_core.core.postbox import PostboxDeliveryReceiptSummaryRef
from govoplan_core.privacy.schemas import PrivacyRetentionPolicyItem
from govoplan_campaign.backend.reports.campaigns import (
_JobReportAggregate,
_campaign_retention_projection,
_campaign_postbox_receipt_summary,
_job_evidence_row,
_latest_by_job_id,
@@ -269,6 +272,113 @@ def test_postbox_receipt_report_aggregates_provider_state_without_identities() -
assert "identity_id" not in repr(report)
def test_retention_projection_explains_defaults_and_redacted_evidence() -> None:
aggregate = _JobReportAggregate(
eml_retained=2,
eml_expired=1,
eml_not_generated=3,
)
version = SimpleNamespace(
raw_json={
"_retention": {
"raw_json_redacted": True,
"redacted_at": "2026-07-08T12:30:00+00:00",
}
},
validation_summary={
"_retention": {
"report_detail_redacted": True,
"redacted_at": "2026-07-08T12:30:00+00:00",
}
},
build_summary={"ok": True},
)
with patch(
"govoplan_campaign.backend.reports.campaigns.capability",
return_value=None,
):
projection = _campaign_retention_projection(
object(), # type: ignore[arg-type]
campaign=SimpleNamespace(id="campaign-1"),
version=version, # type: ignore[arg-type]
aggregate=aggregate,
postbox_receipts={
"status": "available",
"currently_readable_delivery_count": 2,
"expired_message_count": 1,
},
)
assert projection["policy_status"] == "defaults"
assert projection["effective_policy"]["store_raw_campaign_json"] is True
assert projection["evidence"]["raw_campaign_json"]["state"] == "redacted"
assert projection["evidence"]["stored_report_detail"]["state"] == "partially_redacted"
assert projection["evidence"]["generated_eml"] == {
"state": "partially_expired",
"retained_count": 2,
"expired_count": 1,
"not_generated_count": 3,
}
assert projection["evidence"]["postbox_copies"]["state"] == "partially_expired"
assert "generated_eml" in projection["privacy_impact"]["minimized_categories"]
def test_retention_projection_uses_only_public_policy_contract_metadata() -> None:
class _RetentionService:
privacy_policy_from_settings = lambda self, *args, **kwargs: None
privacy_policy_from_session = lambda self, *args, **kwargs: None
set_privacy_policy = lambda self, *args, **kwargs: None
parent_privacy_policy = lambda self, *args, **kwargs: None
parent_privacy_policy_sources = lambda self, *args, **kwargs: []
get_privacy_policy_for_scope = lambda self, *args, **kwargs: {}
set_privacy_policy_for_scope = lambda self, *args, **kwargs: {}
sanitize_audit_details_for_policy = lambda self, *args, **kwargs: {}
apply_retention_policy = lambda self, *args, **kwargs: {}
def effective_privacy_policy(self, *args, **kwargs):
return PrivacyRetentionPolicyItem(
generated_eml_retention_days=30,
stored_report_detail_retention_days=90,
)
def effective_privacy_policy_sources(self, *args, **kwargs):
return [{
"scope_type": "tenant",
"path": "tenant:tenant-1",
"label": "Tenant",
"applied_fields": ["generated_eml_retention_days"],
"policy": {"generated_eml_retention_days": 30},
"provider_secret": "must-not-leak",
}]
with patch(
"govoplan_campaign.backend.reports.campaigns.capability",
return_value=_RetentionService(),
):
projection = _campaign_retention_projection(
object(), # type: ignore[arg-type]
campaign=SimpleNamespace(id="campaign-1"),
version=SimpleNamespace(
raw_json={},
validation_summary={},
build_summary={},
), # type: ignore[arg-type]
aggregate=_JobReportAggregate(),
postbox_receipts={"status": "not_applicable"},
)
assert projection["policy_status"] == "configured"
assert projection["effective_policy"]["generated_eml_retention_days"] == 30
assert projection["sources"] == [{
"scope_type": "tenant",
"path": "tenant:tenant-1",
"label": "Tenant",
"applied_fields": ["generated_eml_retention_days"],
}]
assert "must-not-leak" not in repr(projection)
class CampaignReportProjectionTests(unittest.TestCase):
def test_evidence_projection(self) -> None:
test_job_evidence_row_contains_transport_and_message_evidence()