141 lines
4.4 KiB
Python
141 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock
|
|
|
|
from govoplan_campaign.backend.retention import (
|
|
_apply_eml_retention,
|
|
_apply_raw_json_retention,
|
|
_apply_report_detail_retention,
|
|
)
|
|
|
|
|
|
def _querying(items: list[object]) -> MagicMock:
|
|
session = MagicMock()
|
|
session.query.return_value.order_by.return_value.all.return_value = items
|
|
session.query.return_value.filter.return_value.order_by.return_value.all.return_value = items
|
|
return session
|
|
|
|
|
|
def test_raw_json_retention_preserves_hash_for_final_version() -> None:
|
|
now = datetime.now(timezone.utc)
|
|
version = SimpleNamespace(
|
|
campaign_id="campaign-1",
|
|
updated_at=now - timedelta(days=10),
|
|
workflow_state="completed",
|
|
raw_json={"version": "1.0", "campaign": {"name": "Sensitive"}},
|
|
schema_version="1.0",
|
|
execution_snapshot={"campaign_json_sha256": "sealed-hash"},
|
|
)
|
|
session = _querying([version])
|
|
policy = SimpleNamespace(
|
|
store_raw_campaign_json=True,
|
|
raw_campaign_json_retention_days=1,
|
|
)
|
|
|
|
result = _apply_raw_json_retention(
|
|
session,
|
|
dry_run=False,
|
|
now=now,
|
|
policy_for_campaign_id=lambda _campaign_id: policy,
|
|
)
|
|
|
|
assert result["redacted"] == 1
|
|
assert version.raw_json == {
|
|
"version": "1.0",
|
|
"_retention": {
|
|
"raw_json_redacted": True,
|
|
"redacted_at": now.isoformat(),
|
|
"original_sha256": "sealed-hash",
|
|
},
|
|
}
|
|
session.add.assert_called_once_with(version)
|
|
|
|
|
|
def test_raw_json_retention_does_not_redact_unfinished_version() -> None:
|
|
now = datetime.now(timezone.utc)
|
|
original = {"version": "1.0", "campaign": {"name": "Editable"}}
|
|
version = SimpleNamespace(
|
|
campaign_id="campaign-1",
|
|
updated_at=now - timedelta(days=10),
|
|
workflow_state="editing",
|
|
raw_json=original,
|
|
schema_version="1.0",
|
|
execution_snapshot=None,
|
|
)
|
|
session = _querying([version])
|
|
policy = SimpleNamespace(
|
|
store_raw_campaign_json=True,
|
|
raw_campaign_json_retention_days=1,
|
|
)
|
|
|
|
result = _apply_raw_json_retention(
|
|
session,
|
|
dry_run=False,
|
|
now=now,
|
|
policy_for_campaign_id=lambda _campaign_id: policy,
|
|
)
|
|
|
|
assert result["skipped_not_final"] == 1
|
|
assert version.raw_json is original
|
|
session.add.assert_not_called()
|
|
|
|
|
|
def test_report_detail_retention_keeps_aggregate_evidence() -> None:
|
|
now = datetime.now(timezone.utc)
|
|
version = SimpleNamespace(
|
|
campaign_id="campaign-1",
|
|
updated_at=now - timedelta(days=10),
|
|
validation_summary={"ready": 4, "issues": [{"entry_id": "secret"}]},
|
|
build_summary={"built": 3, "jobs": [{"recipient": "person@example.test"}]},
|
|
)
|
|
session = _querying([version])
|
|
policy = SimpleNamespace(stored_report_detail_retention_days=1)
|
|
|
|
result = _apply_report_detail_retention(
|
|
session,
|
|
dry_run=False,
|
|
now=now,
|
|
policy_for_campaign_id=lambda _campaign_id: policy,
|
|
)
|
|
|
|
assert result["summaries_redacted"] == 2
|
|
assert version.validation_summary["ready"] == 4
|
|
assert "issues" not in version.validation_summary
|
|
assert version.build_summary["built"] == 3
|
|
assert "jobs" not in version.build_summary
|
|
assert version.validation_summary["_retention"]["report_detail_redacted"] is True
|
|
session.add.assert_called_once_with(version)
|
|
|
|
|
|
def test_eml_retention_removes_only_terminal_artifact(tmp_path) -> None:
|
|
now = datetime.now(timezone.utc)
|
|
eml_path = tmp_path / "message.eml"
|
|
eml_path.write_bytes(b"message")
|
|
job = SimpleNamespace(
|
|
campaign_id="campaign-1",
|
|
updated_at=now - timedelta(days=10),
|
|
queue_status="draft",
|
|
send_status="smtp_accepted",
|
|
imap_status="appended",
|
|
eml_local_path=str(eml_path),
|
|
eml_storage_key="campaigns/message.eml",
|
|
)
|
|
session = _querying([job])
|
|
policy = SimpleNamespace(generated_eml_retention_days=1)
|
|
|
|
result = _apply_eml_retention(
|
|
session,
|
|
dry_run=False,
|
|
now=now,
|
|
policy_for_campaign_id=lambda _campaign_id: policy,
|
|
)
|
|
|
|
assert result["files_deleted"] == 1
|
|
assert result["metadata_cleared"] == 1
|
|
assert not eml_path.exists()
|
|
assert job.eml_local_path is None
|
|
assert job.eml_storage_key is None
|
|
session.add.assert_called_once_with(job)
|