GovOPlaN/govoplan-campaign
2
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(campaign): govern attachment reuse

Browse Source
This commit is contained in:
zemion
2026-08-20 05:20:07 +02:00
parent f4fe534ee1
commit 112ef9dc31
16 changed files with 705 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_attachment_building.py
+160
View File
@@ -7,6 +7,7 @@ import zipfile
from pathlib import Path
from govoplan_campaign.backend.campaign.models import CampaignConfig
from govoplan_campaign.backend.campaign.loader import validate_against_schema
from govoplan_campaign.backend.campaign.validation import validate_campaign_config
from govoplan_campaign.backend.messages.builder import build_campaign_messages
@@ -56,6 +57,64 @@ class CampaignAttachmentBuildTests(unittest.TestCase):
"delivery": {"imap_append_sent": {"enabled": False}},
})
def _attachment_reuse_config(
self,
*,
action: str,
allow_within: str = "none",
recipient_emails: tuple[str, ...] = (
"first@example.org",
"second@example.org",
),
duplicate_rules: bool = False,
) -> CampaignConfig:
rules = [{
"id": "shared-file",
"base_dir": "documents",
"file_filter": "shared.pdf",
"required": True,
}]
if duplicate_rules:
rules.append({**rules[0], "id": "shared-file-again"})
return CampaignConfig.model_validate({
"version": "1.0",
"campaign": {
"id": f"reuse-{action}-{allow_within}",
"name": "Attachment reuse",
"mode": "test",
},
"server": {
"mail_profile_id": "profile-1",
"profile_capabilities": {"smtp_available": True},
},
"recipients": {
"from": {"email": "sender@example.org", "type": "to"},
"allow_individual_to": True,
},
"template": {"subject": "Subject", "text": "Body"},
"attachments": {
"global": rules,
"reuse_policy": {
"action": action,
"allow_within": allow_within,
},
},
"entries": {
"inline": [
{
"id": f"recipient-{index}",
"to": [{"email": email, "type": "to"}],
}
for index, email in enumerate(recipient_emails, start=1)
]
},
"validation_policy": {
"missing_email": "block",
"template_error": "block",
},
"delivery": {"imap_append_sent": {"enabled": False}},
})
def test_send_without_attachments_policy_does_not_block_when_no_rules_are_configured(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
@@ -275,6 +334,107 @@ class CampaignAttachmentBuildTests(unittest.TestCase):
self.assertEqual(archive.namelist(), ["matched.xlsx"])
self.assertEqual(archive.read("matched.xlsx"), b"matched workbook")
def test_attachment_reuse_action_controls_message_validation(self) -> None:
expected = {
"allow": ("ready", None, 0, 1),
"warn": ("warning", "warn", 1, 0),
"review": ("needs_review", "ask", 1, 0),
"block": ("blocked", "block", 1, 0),
}
for action, (status, behavior, violation_count, allowed_count) in expected.items():
with self.subTest(action=action), tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
documents = root / "documents"
documents.mkdir()
(documents / "shared.pdf").write_bytes(b"shared")
campaign_file = root / "campaign.json"
campaign_file.write_text("{}", encoding="utf-8")
config = self._attachment_reuse_config(action=action)
result = build_campaign_messages(
config,
campaign_file=campaign_file,
output_dir=root / "out",
write_eml=False,
)
self.assertEqual(
[status, status],
[message.validation_status.value for message in result.report.messages],
)
report = result.report.attachment_reuse
self.assertEqual(1, report["duplicate_file_count"])
self.assertEqual(violation_count, report["violation_file_count"])
self.assertEqual(allowed_count, report["allowed_file_count"])
finding = report["findings"][0]
self.assertEqual("shared.pdf", finding["file_name"])
self.assertNotIn(str(root), str(report))
issues = [
issue
for message in result.report.messages
for issue in message.issues
if issue.code == "duplicate_attachment_reuse"
]
if behavior is None:
self.assertEqual([], issues)
else:
self.assertEqual([behavior, behavior], [issue.behavior for issue in issues])
self.assertEqual(
{"action": action, "allow_within": "none"},
issues[0].details["policy"],
)
def test_attachment_reuse_can_be_allowed_within_recipient_or_message(self) -> None:
cases = (
("same_recipient", ("same@example.org", "same@example.org"), False, 2),
("same_message", ("same@example.org",), True, 2),
)
for allow_within, recipients, duplicate_rules, expected_use_count in cases:
with self.subTest(allow_within=allow_within), tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
documents = root / "documents"
documents.mkdir()
(documents / "shared.pdf").write_bytes(b"shared")
campaign_file = root / "campaign.json"
campaign_file.write_text("{}", encoding="utf-8")
result = build_campaign_messages(
self._attachment_reuse_config(
action="block",
allow_within=allow_within,
recipient_emails=recipients,
duplicate_rules=duplicate_rules,
),
campaign_file=campaign_file,
output_dir=root / "out",
write_eml=False,
)
self.assertEqual(
["ready"] * len(recipients),
[message.validation_status.value for message in result.report.messages],
)
report = result.report.attachment_reuse
self.assertEqual(1, report["allowed_file_count"])
self.assertEqual(0, report["violation_file_count"])
self.assertEqual(expected_use_count, report["findings"][0]["use_count"])
def test_attachment_reuse_policy_is_part_of_the_json_schema(self) -> None:
config = self._attachment_reuse_config(
action="review",
allow_within="same_recipient",
)
payload = config.model_dump(
mode="json",
by_alias=True,
exclude_none=True,
exclude_defaults=True,
)
payload["server"].pop("profile_capabilities", None)
validate_against_schema(payload)
def test_residual_files_become_a_separate_reviewed_report_or_attachment_message(self) -> None:
for mode, expected_attachment_count in (("report", 0), ("attach", 1)):
with self.subTest(mode=mode), tempfile.TemporaryDirectory() as tmp: