diff --git a/src/govoplan_campaign/backend/report_privacy_policy.py b/src/govoplan_campaign/backend/report_privacy_policy.py new file mode 100644 index 0000000..c541a8c --- /dev/null +++ b/src/govoplan_campaign/backend/report_privacy_policy.py @@ -0,0 +1,109 @@ +from __future__ import annotations + +import os +from dataclasses import dataclass +from typing import Any, Mapping + +from sqlalchemy.orm import Session + +from govoplan_core.tenancy.scope import Tenant + + +DEFAULT_SMALL_CELL_THRESHOLD = 5 +MIN_SMALL_CELL_THRESHOLD = 2 +MAX_SMALL_CELL_THRESHOLD = 100 +SMALL_CELL_THRESHOLD_ENV = "GOVOPLAN_CAMPAIGN_REPORT_SMALL_CELL_THRESHOLD" +CAMPAIGN_REPORT_POLICY_SETTINGS_KEY = "campaign_report_privacy_policy" +SMALL_CELL_THRESHOLD_SETTINGS_KEY = "small_cell_threshold" + + +class CampaignReportPrivacyPolicyError(RuntimeError): + pass + + +@dataclass(frozen=True, slots=True) +class CampaignReportPrivacyPolicy: + small_cell_threshold: int + source: str + deployment_small_cell_threshold: int + tenant_small_cell_threshold: int | None = None + + def as_dict(self) -> dict[str, Any]: + return { + "small_cell_threshold": self.small_cell_threshold, + "source": self.source, + "deployment_small_cell_threshold": self.deployment_small_cell_threshold, + "tenant_small_cell_threshold": self.tenant_small_cell_threshold, + "deployment_setting": SMALL_CELL_THRESHOLD_ENV, + "tenant_setting": ( + f"tenant.settings.{CAMPAIGN_REPORT_POLICY_SETTINGS_KEY}." + f"{SMALL_CELL_THRESHOLD_SETTINGS_KEY}" + ), + } + + +def effective_campaign_report_privacy_policy( + session: Session, + *, + tenant_id: str, + environ: Mapping[str, str] | None = None, +) -> CampaignReportPrivacyPolicy: + env = os.environ if environ is None else environ + deployment_raw = env.get(SMALL_CELL_THRESHOLD_ENV) + deployment_value = _configured_threshold( + deployment_raw, + source=SMALL_CELL_THRESHOLD_ENV, + default=DEFAULT_SMALL_CELL_THRESHOLD, + ) + tenant = session.get(Tenant, tenant_id) + tenant_raw = _tenant_threshold_value(tenant.settings if tenant is not None else None) + if tenant_raw is None: + return CampaignReportPrivacyPolicy( + small_cell_threshold=deployment_value, + source="deployment" if deployment_raw not in (None, "") else "deployment_default", + deployment_small_cell_threshold=deployment_value, + ) + + tenant_value = _configured_threshold( + tenant_raw, + source=( + f"tenant.settings.{CAMPAIGN_REPORT_POLICY_SETTINGS_KEY}." + f"{SMALL_CELL_THRESHOLD_SETTINGS_KEY}" + ), + ) + effective_value = max(deployment_value, tenant_value) + return CampaignReportPrivacyPolicy( + small_cell_threshold=effective_value, + source="tenant" if tenant_value >= deployment_value else "deployment_floor", + deployment_small_cell_threshold=deployment_value, + tenant_small_cell_threshold=tenant_value, + ) + + +def _tenant_threshold_value(settings: Mapping[str, Any] | None) -> object | None: + if not isinstance(settings, Mapping): + return None + policy = settings.get(CAMPAIGN_REPORT_POLICY_SETTINGS_KEY) + if not isinstance(policy, Mapping): + return None + return policy.get(SMALL_CELL_THRESHOLD_SETTINGS_KEY) + + +def _configured_threshold(value: object, *, source: str, default: int | None = None) -> int: + if value is None or (isinstance(value, str) and not value.strip()): + if default is not None: + return default + raise CampaignReportPrivacyPolicyError(f"{source} must be configured as an integer") + if isinstance(value, bool): + raise CampaignReportPrivacyPolicyError(f"{source} must be an integer, not a boolean") + try: + parsed = int(value) + except (TypeError, ValueError) as exc: + raise CampaignReportPrivacyPolicyError(f"{source} must be an integer") from exc + if str(parsed) != str(value).strip() and not isinstance(value, int): + raise CampaignReportPrivacyPolicyError(f"{source} must be an integer") + if parsed < MIN_SMALL_CELL_THRESHOLD or parsed > MAX_SMALL_CELL_THRESHOLD: + raise CampaignReportPrivacyPolicyError( + f"{source} must be between {MIN_SMALL_CELL_THRESHOLD} and {MAX_SMALL_CELL_THRESHOLD}" + ) + return parsed diff --git a/src/govoplan_campaign/backend/reports/aggregate.py b/src/govoplan_campaign/backend/reports/aggregate.py new file mode 100644 index 0000000..4e82f7c --- /dev/null +++ b/src/govoplan_campaign/backend/reports/aggregate.py @@ -0,0 +1,479 @@ +from __future__ import annotations + +from collections import Counter +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Literal + +from pydantic import BaseModel, ConfigDict +from sqlalchemy import case, func, or_ +from sqlalchemy.orm import Session + +from govoplan_campaign.backend.db.models import Campaign, CampaignJob, CampaignVersion +from govoplan_campaign.backend.report_privacy_policy import ( + CampaignReportPrivacyPolicy, + effective_campaign_report_privacy_policy, +) + + +class AggregateCampaignReportError(RuntimeError): + pass + + +class AggregateReportCampaign(BaseModel): + model_config = ConfigDict(extra="forbid") + + id: str + name: str + status: str + + +class AggregateReportCampaignListItem(AggregateReportCampaign): + updated_at: datetime + + +class AggregateReportCampaignList(BaseModel): + model_config = ConfigDict(extra="forbid") + + campaigns: list[AggregateReportCampaignListItem] + + +class AggregateCount(BaseModel): + model_config = ConfigDict(extra="forbid") + + value: int | None + suppressed: bool = False + + +class AggregatePopulation(BaseModel): + model_config = ConfigDict(extra="forbid") + + denominator: AggregateCount + denominator_definition: str + inactive_source_entries: AggregateCount + excluded_or_blocked_jobs: AggregateCount + + +class AggregateOutcomeCounts(BaseModel): + model_config = ConfigDict(extra="forbid") + + smtp_accepted: AggregateCount + failed: AggregateCount + outcome_unknown: AggregateCount + queued_or_active: AggregateCount + cancelled: AggregateCount + excluded: AggregateCount + not_attempted: AggregateCount + + +class AggregateTimeRange(BaseModel): + model_config = ConfigDict(extra="forbid") + + first_activity_at: datetime | None + last_activity_at: datetime | None + suppressed: bool + + +class AggregatePrivacy(BaseModel): + model_config = ConfigDict(extra="forbid") + + small_cell_threshold: int + suppression_applied: bool + rule: str + + +class AggregateCampaignReport(BaseModel): + model_config = ConfigDict(extra="forbid") + + generated_at: datetime + campaign: AggregateReportCampaign + version_number: int | None + completion_state: Literal[ + "not_started", + "in_progress", + "completed", + "partially_completed", + "incomplete", + "outcome_unknown", + "suppressed", + ] + population: AggregatePopulation + outcomes: AggregateOutcomeCounts + time_range: AggregateTimeRange + privacy: AggregatePrivacy + + +_OUTCOME_KEYS = ( + "smtp_accepted", + "failed", + "outcome_unknown", + "queued_or_active", + "cancelled", + "excluded", + "not_attempted", +) + + +def generate_aggregate_campaign_report( + session: Session, + *, + tenant_id: str, + campaign_id: str, + version_id: str | None = None, +) -> AggregateCampaignReport: + """Build the deliberately small, recipient-free Campaign report projection.""" + + campaign = _get_campaign(session, tenant_id=tenant_id, campaign_id=campaign_id) + version = _selected_version(session, campaign, version_id) + facts = _query_aggregate_facts( + session, + tenant_id=tenant_id, + campaign_id=campaign.id, + version=version, + ) + policy = effective_campaign_report_privacy_policy(session, tenant_id=tenant_id) + return _build_aggregate_campaign_report( + campaign=campaign, + version=version, + facts=facts, + policy=policy, + ) + + +def _get_campaign(session: Session, *, tenant_id: str, campaign_id: str) -> Campaign: + campaign = ( + session.query(Campaign) + .filter(Campaign.tenant_id == tenant_id, Campaign.id == campaign_id) + .one_or_none() + ) + if campaign is None: + raise AggregateCampaignReportError("Campaign not found") + return campaign + + +def _selected_version( + session: Session, + campaign: Campaign, + version_id: str | None, +) -> CampaignVersion | None: + selected_id = version_id or campaign.current_version_id + if not selected_id: + return None + version = session.get(CampaignVersion, selected_id) + if version is None or version.campaign_id != campaign.id: + raise AggregateCampaignReportError("Campaign version not found") + return version + + +def _query_aggregate_facts( + session: Session, + *, + tenant_id: str, + campaign_id: str, + version: CampaignVersion | None, +) -> _AggregateFacts: + if version is None: + return _AggregateFacts.empty() + + accepted = CampaignJob.send_status.in_({"smtp_accepted", "sent"}) + failed = CampaignJob.send_status.in_({"failed_temporary", "failed_permanent"}) + unknown = CampaignJob.send_status == "outcome_unknown" + active = CampaignJob.send_status.in_({"queued", "claimed", "sending"}) + cancelled = CampaignJob.send_status == "cancelled" + excluded = CampaignJob.send_status == "skipped" + excluded_or_blocked = or_( + CampaignJob.validation_status.in_({"blocked", "excluded", "inactive"}), + CampaignJob.build_status != "built", + ) + row = ( + session.query( + func.count(CampaignJob.id).label("denominator"), + func.sum(case((accepted, 1), else_=0)).label("smtp_accepted"), + func.sum(case((failed, 1), else_=0)).label("failed"), + func.sum(case((unknown, 1), else_=0)).label("outcome_unknown"), + func.sum(case((active, 1), else_=0)).label("queued_or_active"), + func.sum(case((cancelled, 1), else_=0)).label("cancelled"), + func.sum(case((excluded, 1), else_=0)).label("excluded"), + func.sum( + case((or_(accepted, failed, unknown, active, cancelled, excluded), 0), else_=1) + ).label("not_attempted"), + func.sum(case((excluded_or_blocked, 1), else_=0)).label("excluded_or_blocked"), + func.min(CampaignJob.queued_at).label("queued_min"), + func.max(CampaignJob.queued_at).label("queued_max"), + func.min(CampaignJob.smtp_started_at).label("smtp_min"), + func.max(CampaignJob.smtp_started_at).label("smtp_max"), + func.min(CampaignJob.sent_at).label("sent_min"), + func.max(CampaignJob.sent_at).label("sent_max"), + func.min(CampaignJob.outcome_unknown_at).label("unknown_min"), + func.max(CampaignJob.outcome_unknown_at).label("unknown_max"), + ) + .filter( + CampaignJob.tenant_id == tenant_id, + CampaignJob.campaign_id == campaign_id, + CampaignJob.campaign_version_id == version.id, + ) + .one() + ) + values = row._mapping + activity = [ + values[key] + for key in ( + "queued_min", + "queued_max", + "smtp_min", + "smtp_max", + "sent_min", + "sent_max", + "unknown_min", + "unknown_max", + ) + if values[key] is not None + ] + return _AggregateFacts( + outcomes={key: int(values[key] or 0) for key in _OUTCOME_KEYS}, + denominator=int(values["denominator"] or 0), + excluded_or_blocked=int(values["excluded_or_blocked"] or 0), + first_activity_at=min(activity) if activity else None, + last_activity_at=max(activity) if activity else None, + ) + + +def build_aggregate_campaign_report( + *, + campaign: Campaign, + version: CampaignVersion | None, + jobs: list[CampaignJob], + policy: CampaignReportPrivacyPolicy, + generated_at: datetime | None = None, +) -> AggregateCampaignReport: + return _build_aggregate_campaign_report( + campaign=campaign, + version=version, + facts=_facts_from_jobs(jobs), + policy=policy, + generated_at=generated_at, + ) + + +def _build_aggregate_campaign_report( + *, + campaign: Campaign, + version: CampaignVersion | None, + facts: _AggregateFacts, + policy: CampaignReportPrivacyPolicy, + generated_at: datetime | None = None, +) -> AggregateCampaignReport: + outcome_values = facts.outcomes + outcomes, denominator = _suppress_partition( + outcome_values, + threshold=policy.small_cell_threshold, + ) + inactive_entries = _inactive_entry_count(version) + standalone_counts = { + "inactive_source_entries": _suppress_standalone_count( + inactive_entries, + threshold=policy.small_cell_threshold, + ), + "excluded_or_blocked_jobs": _suppress_standalone_count( + facts.excluded_or_blocked, + threshold=policy.small_cell_threshold, + ), + } + suppression_applied = denominator.suppressed or any( + item.suppressed for item in (*outcomes.values(), *standalone_counts.values()) + ) + first_activity = facts.first_activity_at + last_activity = facts.last_activity_at + suppress_time_range = suppression_applied or ( + 0 < facts.denominator < policy.small_cell_threshold + ) + + return AggregateCampaignReport( + generated_at=generated_at or datetime.now(timezone.utc), + campaign=AggregateReportCampaign( + id=campaign.id, + name=campaign.name, + status=campaign.status, + ), + version_number=version.version_number if version else None, + completion_state=( + "suppressed" + if denominator.suppressed + else _completion_state(outcome_values, facts.denominator) + ), + population=AggregatePopulation( + denominator=denominator, + denominator_definition=( + "All recipient delivery jobs built for the selected campaign version. " + "Inactive source entries are excluded and reported separately." + ), + inactive_source_entries=standalone_counts["inactive_source_entries"], + excluded_or_blocked_jobs=standalone_counts["excluded_or_blocked_jobs"], + ), + outcomes=AggregateOutcomeCounts(**outcomes), + time_range=AggregateTimeRange( + first_activity_at=None if suppress_time_range else first_activity, + last_activity_at=None if suppress_time_range else last_activity, + suppressed=suppress_time_range and first_activity is not None, + ), + privacy=AggregatePrivacy( + small_cell_threshold=policy.small_cell_threshold, + suppression_applied=suppression_applied, + rule=( + "Positive counts below the threshold are hidden. At least one additional " + "count or the denominator is hidden when needed to prevent subtraction." + ), + ), + ) + + +def aggregate_report_campaign_item(campaign: Campaign) -> AggregateReportCampaignListItem: + return AggregateReportCampaignListItem( + id=campaign.id, + name=campaign.name, + status=campaign.status, + updated_at=campaign.updated_at, + ) + + +def _outcome_counts(jobs: list[CampaignJob]) -> dict[str, int]: + counts: Counter[str] = Counter() + for job in jobs: + status = job.send_status + if status in {"smtp_accepted", "sent"}: + counts["smtp_accepted"] += 1 + elif status in {"failed_temporary", "failed_permanent"}: + counts["failed"] += 1 + elif status == "outcome_unknown": + counts["outcome_unknown"] += 1 + elif status in {"queued", "claimed", "sending"}: + counts["queued_or_active"] += 1 + elif status == "cancelled": + counts["cancelled"] += 1 + elif status == "skipped": + counts["excluded"] += 1 + else: + counts["not_attempted"] += 1 + return {key: counts[key] for key in _OUTCOME_KEYS} + + +@dataclass(frozen=True, slots=True) +class _AggregateFacts: + outcomes: dict[str, int] + denominator: int + excluded_or_blocked: int + first_activity_at: datetime | None + last_activity_at: datetime | None + + @classmethod + def empty(cls) -> _AggregateFacts: + return cls( + outcomes={key: 0 for key in _OUTCOME_KEYS}, + denominator=0, + excluded_or_blocked=0, + first_activity_at=None, + last_activity_at=None, + ) + + +def _facts_from_jobs(jobs: list[CampaignJob]) -> _AggregateFacts: + first_activity, last_activity = _activity_range(jobs) + return _AggregateFacts( + outcomes=_outcome_counts(jobs), + denominator=len(jobs), + excluded_or_blocked=sum( + 1 + for job in jobs + if job.validation_status in {"blocked", "excluded", "inactive"} + or job.build_status != "built" + ), + first_activity_at=first_activity, + last_activity_at=last_activity, + ) + + +def _suppress_partition( + counts: dict[str, int], + *, + threshold: int, +) -> tuple[dict[str, AggregateCount], AggregateCount]: + total = sum(counts.values()) + suppressed = {key for key, value in counts.items() if 0 < value < threshold} + suppress_denominator = False + if suppressed: + companions = [ + (value, key) + for key, value in counts.items() + if key not in suppressed and value > 0 + ] + if companions: + suppressed.add(max(companions)[1]) + else: + suppress_denominator = True + denominator = AggregateCount( + value=None if suppress_denominator else total, + suppressed=suppress_denominator, + ) + return ( + { + key: AggregateCount( + value=None if key in suppressed else value, + suppressed=key in suppressed, + ) + for key, value in counts.items() + }, + denominator, + ) + + +def _suppress_standalone_count(value: int, *, threshold: int) -> AggregateCount: + if 0 < value < threshold: + return AggregateCount(value=None, suppressed=True) + return AggregateCount(value=value, suppressed=False) + + +def _completion_state( + counts: dict[str, int], + total: int, +) -> Literal[ + "not_started", + "in_progress", + "completed", + "partially_completed", + "incomplete", + "outcome_unknown", +]: + if total == 0 or counts["not_attempted"] + counts["excluded"] == total: + return "not_started" + if counts["outcome_unknown"]: + return "outcome_unknown" + if counts["queued_or_active"]: + return "in_progress" + accepted = counts["smtp_accepted"] + if accepted == total: + return "completed" + if accepted: + return "partially_completed" + return "incomplete" + + +def _activity_range(jobs: list[CampaignJob]) -> tuple[datetime | None, datetime | None]: + activity = [ + value + for job in jobs + for value in ( + job.queued_at, + job.smtp_started_at, + job.sent_at, + job.outcome_unknown_at, + ) + if value is not None + ] + if not activity: + return None, None + return min(activity), max(activity) + + +def _inactive_entry_count(version: CampaignVersion | None) -> int: + build_summary = version.build_summary if version and isinstance(version.build_summary, dict) else {} + return int(build_summary.get("inactive_count") or build_summary.get("inactive_entries_count") or 0) diff --git a/src/govoplan_campaign/backend/router.py b/src/govoplan_campaign/backend/router.py index 3eb3abb..1def878 100644 --- a/src/govoplan_campaign/backend/router.py +++ b/src/govoplan_campaign/backend/router.py @@ -106,6 +106,14 @@ from govoplan_campaign.backend.db.models import ( ) from govoplan_core.db.session import get_session from govoplan_campaign.backend.reports.campaigns import CampaignReportError, generate_campaign_report, generate_jobs_csv +from govoplan_campaign.backend.report_privacy_policy import CampaignReportPrivacyPolicyError +from govoplan_campaign.backend.reports.aggregate import ( + AggregateCampaignReport, + AggregateCampaignReportError, + AggregateReportCampaignList, + aggregate_report_campaign_item, + generate_aggregate_campaign_report, +) from govoplan_campaign.backend.response_security import ( public_campaign_payload, public_delivery_result_message, @@ -962,6 +970,52 @@ def delete_recipient_import_mapping_profile( return Response(status_code=status.HTTP_204_NO_CONTENT) +@router.get("/aggregate-reports", response_model=AggregateReportCampaignList) +def list_aggregate_campaign_reports( + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(require_scope("campaigns:report:read")), +): + """List only the business metadata needed to select an aggregate report.""" + + campaigns = ( + _campaign_query_for_principal(session, principal) + .order_by(Campaign.updated_at.desc()) + .all() + ) + return AggregateReportCampaignList( + campaigns=[aggregate_report_campaign_item(campaign) for campaign in campaigns] + ) + + +@router.get("/aggregate-reports/{campaign_id}", response_model=AggregateCampaignReport) +def aggregate_campaign_report( + campaign_id: str, + version_id: str | None = None, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(require_scope("campaigns:report:read")), +): + """Return the privacy-safe aggregate projection without recipient detail.""" + + _get_campaign_for_principal(session, campaign_id, principal) + try: + return generate_aggregate_campaign_report( + session, + tenant_id=principal.tenant_id, + campaign_id=campaign_id, + version_id=version_id, + ) + except AggregateCampaignReportError as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Campaign report not found", + ) from exc + except CampaignReportPrivacyPolicyError as exc: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="Campaign report privacy policy is invalid", + ) from exc + + @router.get("/{campaign_id}/address-lookup", response_model=CampaignAddressLookupResponse) def lookup_campaign_addresses( campaign_id: str, @@ -2937,6 +2991,7 @@ def get_job_diagnostics( """Return infrastructure details only to campaign operators/admins.""" _get_campaign_for_principal(session, campaign_id, principal) + _require_permission(principal, "campaigns:recipient:read") campaign = _get_campaign_for_tenant(session, campaign_id, principal.tenant_id) job = session.get(CampaignJob, job_id) if not job or job.campaign_id != campaign.id or job.tenant_id != principal.tenant_id: @@ -2992,8 +3047,7 @@ def campaign_report( principal: ApiPrincipal = Depends(require_scope("campaigns:report:read")), ): _get_campaign_for_principal(session, campaign_id, principal) - if include_jobs: - _require_permission(principal, "campaigns:recipient:read") + _require_permission(principal, "campaigns:recipient:read") """Return the full JSON report for one campaign.""" try: @@ -3048,6 +3102,7 @@ def email_campaign_report( principal: ApiPrincipal = Depends(require_scope("campaigns:report:send")), ): campaign = _get_campaign_for_principal(session, campaign_id, principal) + _require_permission(principal, "campaigns:recipient:export") selected_version_id = payload.version_id or campaign.current_version_id selected_version = session.get(CampaignVersion, selected_version_id) if selected_version_id else None if selected_version is not None and selected_version.campaign_id == campaign.id: @@ -3055,8 +3110,6 @@ def email_campaign_report( principal, selected_version.raw_json if isinstance(selected_version.raw_json, dict) else {}, ) - if payload.include_jobs or payload.attach_jobs_csv: - _require_permission(principal, "campaigns:recipient:export") """Generate a campaign report and send it to one or more email addresses.""" try: diff --git a/tests/test_aggregate_report.py b/tests/test_aggregate_report.py new file mode 100644 index 0000000..ff04692 --- /dev/null +++ b/tests/test_aggregate_report.py @@ -0,0 +1,212 @@ +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] + ) diff --git a/tests/test_aggregate_report_routes.py b/tests/test_aggregate_report_routes.py new file mode 100644 index 0000000..e6cd3ac --- /dev/null +++ b/tests/test_aggregate_report_routes.py @@ -0,0 +1,186 @@ +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import Mock, patch + +import pytest +from fastapi import HTTPException +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + +from govoplan_access.backend.db.models import Account, Group, User +from govoplan_campaign.backend import router +from govoplan_campaign.backend.db.models import Campaign, CampaignJob, CampaignVersion +from govoplan_campaign.backend.reports.aggregate import ( + AggregateCampaignReportError, + generate_aggregate_campaign_report, +) +from govoplan_campaign.backend.schemas import ReportEmailRequest +from govoplan_core.core.change_sequence import ChangeSequenceEntry +from govoplan_core.db.base import Base +from govoplan_core.tenancy.scope import Tenant, create_scope_tables + + +class _Principal: + def __init__(self, *scopes: str, tenant_id: str = "tenant-1") -> None: + self.scopes = set(scopes) + self.tenant_id = tenant_id + self.user = SimpleNamespace(id="reader-1") + + def has(self, scope: str) -> bool: + return scope in self.scopes + + +def test_full_report_and_job_detail_reject_aggregate_only_principal() -> None: + principal = _Principal("campaigns:report:read") + session = Mock() + campaign = SimpleNamespace(id="campaign-1", tenant_id="tenant-1") + + with ( + patch.object(router, "_get_campaign_for_principal", return_value=campaign), + pytest.raises(HTTPException) as full_report_denied, + ): + router.campaign_report( + "campaign-1", + session=session, + principal=principal, # type: ignore[arg-type] + ) + assert full_report_denied.value.status_code == 403 + assert "campaigns:recipient:read" in full_report_denied.value.detail + + with ( + patch.object(router, "_get_campaign_for_principal", return_value=campaign), + pytest.raises(HTTPException) as job_detail_denied, + ): + router.get_job_detail( + "campaign-1", + "job-1", + session=session, + principal=principal, # type: ignore[arg-type] + ) + assert job_detail_denied.value.status_code == 403 + + with ( + patch.object(router, "_get_campaign_for_principal", return_value=campaign), + pytest.raises(HTTPException) as report_email_denied, + ): + router.email_campaign_report( + "campaign-1", + ReportEmailRequest(to=["auditor@example.test"]), + session=session, + principal=_Principal("campaigns:report:send"), # type: ignore[arg-type] + ) + assert report_email_denied.value.status_code == 403 + assert "campaigns:recipient:export" in report_email_denied.value.detail + + with ( + patch.object(router, "_get_campaign_for_principal", return_value=campaign), + pytest.raises(HTTPException) as diagnostics_denied, + ): + router.get_job_diagnostics( + "campaign-1", + "job-1", + session=session, + principal=_Principal("campaigns:diagnostic:read"), # type: ignore[arg-type] + ) + assert diagnostics_denied.value.status_code == 403 + assert "campaigns:recipient:read" in diagnostics_denied.value.detail + + +def test_aggregate_route_uses_only_the_safe_projection() -> None: + principal = _Principal("campaigns:report:read") + session = Mock() + safe_projection = Mock() + + with ( + patch.object(router, "_get_campaign_for_principal") as acl, + patch.object( + router, + "generate_aggregate_campaign_report", + return_value=safe_projection, + ) as generate, + ): + result = router.aggregate_campaign_report( + "campaign-1", + session=session, + principal=principal, # type: ignore[arg-type] + ) + + assert result is safe_projection + acl.assert_called_once_with(session, "campaign-1", principal) + generate.assert_called_once_with( + session, + tenant_id="tenant-1", + campaign_id="campaign-1", + version_id=None, + ) + + +def test_aggregate_projection_is_tenant_isolated_and_needs_no_optional_module() -> None: + engine = create_engine("sqlite+pysqlite:///:memory:") + create_scope_tables(engine) + Base.metadata.create_all( + engine, + tables=[ + Account.__table__, + User.__table__, + Group.__table__, + Campaign.__table__, + CampaignVersion.__table__, + CampaignJob.__table__, + ChangeSequenceEntry.__table__, + ], + ) + with Session(engine) as session: + session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant 1", settings={})) + campaign = Campaign( + id="campaign-1", + tenant_id="tenant-1", + external_id="external-1", + name="Safe aggregate", + description=None, + status="sent", + ) + version = CampaignVersion( + id="version-1", + campaign_id=campaign.id, + version_number=1, + raw_json={"mail": {"profile_id": "optional-module-not-loaded"}}, + schema_version="5", + build_summary={}, + ) + campaign.current_version_id = version.id + session.add_all([campaign, version]) + for index in range(5): + session.add(CampaignJob( + id=f"job-{index}", + tenant_id="tenant-1", + campaign_id=campaign.id, + campaign_version_id=version.id, + entry_index=index, + recipient_email=f"private-{index}@example.test", + subject="Private", + build_status="built", + validation_status="ready", + queue_status="queued", + send_status="smtp_accepted", + imap_status="not_requested", + )) + session.commit() + + report = generate_aggregate_campaign_report( + session, + tenant_id="tenant-1", + campaign_id="campaign-1", + ) + assert report.population.denominator.value == 5 + assert report.outcomes.smtp_accepted.value == 5 + + with pytest.raises(AggregateCampaignReportError): + generate_aggregate_campaign_report( + session, + tenant_id="tenant-2", + campaign_id="campaign-1", + ) + + engine.dispose()