246 lines
8.7 KiB
Python
246 lines
8.7 KiB
Python
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_outcome_suppression_also_hides_overlapping_population_count() -> None:
|
|
jobs = [
|
|
*[
|
|
_job(
|
|
index,
|
|
"smtp_accepted",
|
|
validation_status="blocked" if index < 4 else "ready",
|
|
)
|
|
for index in range(8)
|
|
],
|
|
_job(8, "skipped", validation_status="excluded"),
|
|
]
|
|
|
|
report = 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(5),
|
|
)
|
|
|
|
assert report.outcomes.excluded.suppressed is True
|
|
assert report.population.excluded_or_blocked_jobs.model_dump() == {
|
|
"value": None,
|
|
"suppressed": True,
|
|
}
|
|
# Source entries without jobs are outside the outcome denominator and do
|
|
# not overlap the suppressed partition, so their threshold-safe count stays visible.
|
|
assert report.population.inactive_source_entries.model_dump() == {
|
|
"value": 5,
|
|
"suppressed": False,
|
|
}
|
|
|
|
|
|
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]
|
|
)
|