from __future__ import annotations from datetime import UTC, datetime, timedelta from types import SimpleNamespace import pytest from govoplan_campaign.backend.report_privacy_policy import ( CampaignReportPrivacyPolicy, CampaignReportPrivacyPolicyError, DEFAULT_SMALL_CELL_THRESHOLD, effective_campaign_report_privacy_policy, ) from govoplan_campaign.backend.reports.aggregate import build_aggregate_campaign_report class _PolicySession: def __init__(self, settings: dict[str, object] | None = None) -> None: self.tenant = SimpleNamespace(settings=settings or {}) def get(self, _model, _id): return self.tenant def _policy(threshold: int = 5) -> CampaignReportPrivacyPolicy: return CampaignReportPrivacyPolicy( small_cell_threshold=threshold, source="test", deployment_small_cell_threshold=threshold, ) def _campaign() -> SimpleNamespace: now = datetime(2026, 7, 22, 8, 0, tzinfo=UTC) return SimpleNamespace( id="campaign-safe", tenant_id="tenant-safe", external_id="must-not-leak-external-id", name="Semester notification", description="Business-safe description", status="partially_completed", owner_user_id="must-not-leak-owner", current_version_id="must-not-leak-version-id", updated_at=now, ) def _version(*, inactive_count: int = 0) -> SimpleNamespace: return SimpleNamespace( id="must-not-leak-version-id", version_number=7, build_summary={"inactive_count": inactive_count, "build_token": "must-not-leak-token"}, execution_snapshot={"smtp": {"password": "must-not-leak-secret"}}, raw_json={"entries": [{"email": "must-not-leak@example.test"}]}, ) def _job(index: int, send_status: str, **overrides: object) -> SimpleNamespace: started = datetime(2026, 7, 22, 8, 0, tzinfo=UTC) + timedelta(minutes=index) values: dict[str, object] = { "id": f"job-{index}", "recipient_email": f"private-{index}@example.test", "subject": f"Personal subject {index}", "last_error": "smtp.internal.example provider-secret", "resolved_attachments": [{"storage_key": f"private/{index}"}], "send_status": send_status, "validation_status": "ready", "build_status": "built", "queued_at": started, "smtp_started_at": started, "sent_at": started if send_status in {"smtp_accepted", "sent"} else None, "outcome_unknown_at": started if send_status == "outcome_unknown" else None, } values.update(overrides) return SimpleNamespace(**values) def test_aggregate_projection_has_a_strict_recipient_free_shape() -> None: jobs = [ *[_job(index, "smtp_accepted") for index in range(10)], *[_job(index + 10, "failed_permanent") for index in range(5)], ] payload = build_aggregate_campaign_report( campaign=_campaign(), # type: ignore[arg-type] version=_version(inactive_count=5), # type: ignore[arg-type] jobs=jobs, # type: ignore[arg-type] policy=_policy(), generated_at=datetime(2026, 7, 22, 12, 0, tzinfo=UTC), ).model_dump(mode="json") assert set(payload) == { "generated_at", "campaign", "version_number", "completion_state", "population", "outcomes", "time_range", "privacy", } assert set(payload["campaign"]) == {"id", "name", "status"} assert payload["population"]["denominator"] == {"value": 15, "suppressed": False} assert payload["outcomes"]["smtp_accepted"] == {"value": 10, "suppressed": False} assert payload["outcomes"]["failed"] == {"value": 5, "suppressed": False} assert payload["completion_state"] == "partially_completed" serialized = repr(payload) for forbidden in ( "private-0@example.test", "Personal subject", "smtp.internal.example", "provider-secret", "storage_key", "must-not-leak", "Business-safe description", "job-0", ): assert forbidden not in serialized def test_small_cells_use_primary_and_complementary_suppression() -> None: jobs = [ *[_job(index, "smtp_accepted") for index in range(8)], _job(8, "failed_permanent"), ] report = build_aggregate_campaign_report( campaign=_campaign(), # type: ignore[arg-type] version=_version(inactive_count=1), # type: ignore[arg-type] jobs=jobs, # type: ignore[arg-type] policy=_policy(5), ) assert report.population.denominator.value == 9 assert report.outcomes.failed.suppressed is True assert report.outcomes.failed.value is None assert report.outcomes.smtp_accepted.suppressed is True assert report.outcomes.smtp_accepted.value is None assert report.population.inactive_source_entries.suppressed is True assert report.time_range.suppressed is True assert report.privacy.suppression_applied is True def test_all_small_cells_also_suppress_the_denominator_and_state() -> None: jobs = [_job(0, "smtp_accepted"), _job(1, "failed_permanent")] report = build_aggregate_campaign_report( campaign=_campaign(), # type: ignore[arg-type] version=_version(), # type: ignore[arg-type] jobs=jobs, # type: ignore[arg-type] policy=_policy(5), ) assert report.population.denominator.model_dump() == {"value": None, "suppressed": True} assert report.completion_state == "suppressed" assert report.outcomes.smtp_accepted.value is None assert report.outcomes.failed.value is None def test_explicitly_skipped_jobs_are_exclusions_not_unattempted_outcomes() -> None: jobs = [_job(index, "skipped") for index in range(5)] report = build_aggregate_campaign_report( campaign=_campaign(), # type: ignore[arg-type] version=_version(), # type: ignore[arg-type] jobs=jobs, # type: ignore[arg-type] policy=_policy(5), ) assert report.outcomes.excluded.value == 5 assert report.outcomes.not_attempted.value == 0 assert report.completion_state == "not_started" def test_report_privacy_policy_defaults_to_five_and_tenant_can_only_strengthen() -> None: default = effective_campaign_report_privacy_policy( _PolicySession(), # type: ignore[arg-type] tenant_id="tenant-1", environ={}, ) assert default.small_cell_threshold == DEFAULT_SMALL_CELL_THRESHOLD == 5 assert default.source == "deployment_default" strengthened = effective_campaign_report_privacy_policy( _PolicySession( {"campaign_report_privacy_policy": {"small_cell_threshold": 10}} ), # type: ignore[arg-type] tenant_id="tenant-1", environ={"GOVOPLAN_CAMPAIGN_REPORT_SMALL_CELL_THRESHOLD": "5"}, ) assert strengthened.small_cell_threshold == 10 assert strengthened.source == "tenant" floor = effective_campaign_report_privacy_policy( _PolicySession( {"campaign_report_privacy_policy": {"small_cell_threshold": 3}} ), # type: ignore[arg-type] tenant_id="tenant-1", environ={"GOVOPLAN_CAMPAIGN_REPORT_SMALL_CELL_THRESHOLD": "7"}, ) assert floor.small_cell_threshold == 7 assert floor.source == "deployment_floor" @pytest.mark.parametrize("value", [True, 0, 1, 101, "2.5", "disabled"]) def test_invalid_report_privacy_policy_fails_closed(value: object) -> None: with pytest.raises(CampaignReportPrivacyPolicyError): effective_campaign_report_privacy_policy( _PolicySession(), # type: ignore[arg-type] tenant_id="tenant-1", environ={"GOVOPLAN_CAMPAIGN_REPORT_SMALL_CELL_THRESHOLD": value}, # type: ignore[dict-item] )