feat(campaign): enforce attachment delivery policies
This commit is contained in:
206
tests/test_attachment_building.py
Normal file
206
tests/test_attachment_building.py
Normal file
@@ -0,0 +1,206 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import tempfile
|
||||
import unittest
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
|
||||
from govoplan_campaign.backend.campaign.models import CampaignConfig
|
||||
from govoplan_campaign.backend.campaign.validation import validate_campaign_config
|
||||
from govoplan_campaign.backend.messages.builder import build_campaign_messages
|
||||
|
||||
|
||||
class CampaignAttachmentBuildTests(unittest.TestCase):
|
||||
def _no_attachment_config(
|
||||
self,
|
||||
behavior: str | None = None,
|
||||
legacy_allow: bool | None = None,
|
||||
*,
|
||||
configure_missing_rule: bool = False,
|
||||
) -> CampaignConfig:
|
||||
attachments: dict[str, object] = {
|
||||
"base_path": ".",
|
||||
"global": [],
|
||||
"zip": {"enabled": False, "archives": []},
|
||||
}
|
||||
if configure_missing_rule:
|
||||
attachments["global"] = [
|
||||
{
|
||||
"id": "missing",
|
||||
"label": "Missing attachment",
|
||||
"base_dir": ".",
|
||||
"file_filter": "missing.pdf",
|
||||
"required": False,
|
||||
"missing_behavior": "continue",
|
||||
}
|
||||
]
|
||||
if behavior is not None:
|
||||
attachments["send_without_attachments_behavior"] = behavior
|
||||
if legacy_allow is not None:
|
||||
attachments["send_without_attachments"] = legacy_allow
|
||||
return CampaignConfig.model_validate({
|
||||
"version": "1.0",
|
||||
"campaign": {"id": f"no-attachment-{behavior or legacy_allow}", "name": "No attachment policy", "mode": "test"},
|
||||
"fields": [],
|
||||
"global_values": {},
|
||||
"server": {"smtp": {"host": "smtp.example.invalid", "port": 587, "security": "starttls"}},
|
||||
"recipients": {"from": {"email": "sender@example.org", "type": "to"}, "allow_individual_to": True},
|
||||
"template": {"subject": "Subject", "text": "Body"},
|
||||
"attachments": attachments,
|
||||
"entries": {"inline": [{"id": "recipient-1", "to": [{"email": "recipient@example.org", "type": "to"}]}]},
|
||||
"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)
|
||||
campaign_file = root / "campaign.json"
|
||||
campaign_file.write_text("{}", encoding="utf-8")
|
||||
config = self._no_attachment_config(legacy_allow=False)
|
||||
|
||||
validation = validate_campaign_config(config, campaign_file=campaign_file, check_files=True)
|
||||
self.assertTrue(validation.ok)
|
||||
self.assertNotIn("missing_attachment_coverage", {issue.code for issue in validation.issues})
|
||||
|
||||
result = build_campaign_messages(config, campaign_file=campaign_file, output_dir=root / "out", write_eml=True)
|
||||
self.assertEqual(result.report.build_failed_count, 0)
|
||||
self.assertEqual(result.report.queueable_count, 1)
|
||||
message = result.report.messages[0]
|
||||
self.assertEqual(message.attachment_count, 0)
|
||||
self.assertIsNotNone(message.eml_path)
|
||||
self.assertNotIn("missing_attachment_coverage", {issue.code for issue in message.issues})
|
||||
self.assertIsNotNone(result.built_messages[0].mime)
|
||||
|
||||
def test_configured_attachment_rule_blocks_generation_when_no_files_resolve(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
campaign_file = root / "campaign.json"
|
||||
campaign_file.write_text("{}", encoding="utf-8")
|
||||
config = self._no_attachment_config(behavior="block", configure_missing_rule=True)
|
||||
|
||||
validation = validate_campaign_config(config, campaign_file=campaign_file, check_files=True)
|
||||
self.assertFalse(validation.ok)
|
||||
self.assertIn("missing_attachment_coverage", {issue.code for issue in validation.issues})
|
||||
|
||||
result = build_campaign_messages(config, campaign_file=campaign_file, output_dir=root / "out", write_eml=True)
|
||||
self.assertEqual(result.report.build_failed_count, 1)
|
||||
self.assertEqual(result.report.queueable_count, 0)
|
||||
message = result.report.messages[0]
|
||||
self.assertEqual(message.attachment_count, 0)
|
||||
self.assertIsNone(message.eml_path)
|
||||
self.assertIn("missing_attachment_coverage", {issue.code for issue in message.issues})
|
||||
self.assertIsNone(result.built_messages[0].mime)
|
||||
|
||||
def test_send_without_attachments_behavior_modes(self) -> None:
|
||||
cases = {
|
||||
"block": ("build_failed", "blocked", 0, "block", False),
|
||||
"ask": ("built", "needs_review", 0, "ask", True),
|
||||
"drop": ("built", "excluded", 0, "drop", True),
|
||||
"warn": ("built", "warning", 1, "warn", True),
|
||||
"continue": ("built", "ready", 1, None, True),
|
||||
}
|
||||
for behavior, (build_status, validation_status, queueable_count, issue_behavior, has_mime) in cases.items():
|
||||
with self.subTest(behavior=behavior):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
campaign_file = root / "campaign.json"
|
||||
campaign_file.write_text("{}", encoding="utf-8")
|
||||
config = self._no_attachment_config(behavior=behavior, configure_missing_rule=True)
|
||||
|
||||
validation = validate_campaign_config(config, campaign_file=campaign_file, check_files=True)
|
||||
if behavior == "block":
|
||||
self.assertFalse(validation.ok)
|
||||
else:
|
||||
self.assertTrue(validation.ok)
|
||||
|
||||
result = build_campaign_messages(config, campaign_file=campaign_file, output_dir=root / "out", write_eml=True)
|
||||
self.assertEqual(result.report.queueable_count, queueable_count)
|
||||
message = result.report.messages[0]
|
||||
self.assertEqual(message.build_status.value, build_status)
|
||||
self.assertEqual(message.validation_status.value, validation_status)
|
||||
coverage_issues = [issue for issue in message.issues if issue.code == "missing_attachment_coverage"]
|
||||
if issue_behavior is None:
|
||||
self.assertEqual(coverage_issues, [])
|
||||
else:
|
||||
self.assertEqual(coverage_issues[0].behavior, issue_behavior)
|
||||
self.assertEqual(result.built_messages[0].mime is not None, has_mime)
|
||||
|
||||
def test_missing_pattern_does_not_create_zip_member_or_count_as_attachment(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
documents = root / "documents"
|
||||
documents.mkdir()
|
||||
(documents / "matched.xlsx").write_bytes(b"matched workbook")
|
||||
campaign_file = root / "campaign.json"
|
||||
campaign_file.write_text("{}", encoding="utf-8")
|
||||
config = CampaignConfig.model_validate({
|
||||
"version": "1.0",
|
||||
"campaign": {"id": "zip-missing-pattern", "name": "ZIP missing pattern", "mode": "test"},
|
||||
"fields": [],
|
||||
"global_values": {},
|
||||
"server": {"smtp": {"host": "smtp.example.invalid", "port": 587, "security": "starttls"}},
|
||||
"recipients": {"from": {"email": "sender@example.org", "type": "to"}, "allow_individual_to": True},
|
||||
"template": {"subject": "Subject", "text": "Body"},
|
||||
"attachments": {
|
||||
"base_path": ".",
|
||||
"allow_individual": True,
|
||||
"send_without_attachments": False,
|
||||
"zip": {
|
||||
"enabled": True,
|
||||
"archives": [{"id": "bundle", "name": "recipient-bundle.zip", "standard": True}],
|
||||
},
|
||||
"global": [],
|
||||
},
|
||||
"entries": {
|
||||
"inline": [{
|
||||
"id": "recipient-1",
|
||||
"to": [{"email": "recipient@example.org", "type": "to"}],
|
||||
"attachments": [
|
||||
{"id": "matched", "base_dir": "documents", "file_filter": "matched.xlsx", "required": True},
|
||||
{
|
||||
"id": "missing",
|
||||
"base_dir": "documents",
|
||||
"file_filter": "missing.xlsx",
|
||||
"required": False,
|
||||
"missing_behavior": "warn",
|
||||
},
|
||||
],
|
||||
}]
|
||||
},
|
||||
"validation_policy": {
|
||||
"missing_email": "block",
|
||||
"template_error": "block",
|
||||
"missing_optional_attachment": "warn",
|
||||
},
|
||||
"delivery": {"imap_append_sent": {"enabled": False}},
|
||||
})
|
||||
|
||||
validation = validate_campaign_config(config, campaign_file=campaign_file, check_files=True)
|
||||
self.assertTrue(validation.ok)
|
||||
self.assertIn("missing_optional_attachment", {issue.code for issue in validation.issues})
|
||||
|
||||
result = build_campaign_messages(config, campaign_file=campaign_file, output_dir=root / "out", write_eml=False)
|
||||
self.assertEqual(result.report.built_count, 1)
|
||||
self.assertEqual(result.report.queueable_count, 1)
|
||||
message = result.report.messages[0]
|
||||
self.assertEqual(message.attachment_count, 1)
|
||||
summaries = {attachment.attachment_id: attachment for attachment in message.attachments}
|
||||
self.assertEqual(summaries["matched"].zip_filename, "recipient-bundle.zip")
|
||||
self.assertEqual(summaries["matched"].zip_entry_names, ["matched.xlsx"])
|
||||
self.assertIsNone(summaries["missing"].zip_filename)
|
||||
self.assertEqual(summaries["missing"].zip_entry_names, [])
|
||||
self.assertEqual(summaries["missing"].matches, [])
|
||||
|
||||
mime = result.built_messages[0].mime
|
||||
self.assertIsNotNone(mime)
|
||||
attachments = {part.get_filename(): part.get_payload(decode=True) for part in mime.iter_attachments()}
|
||||
self.assertEqual(set(attachments), {"recipient-bundle.zip"})
|
||||
with zipfile.ZipFile(io.BytesIO(attachments["recipient-bundle.zip"])) as archive:
|
||||
self.assertEqual(archive.namelist(), ["matched.xlsx"])
|
||||
self.assertEqual(archive.read("matched.xlsx"), b"matched workbook")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
68
tests/test_zip_service.py
Normal file
68
tests/test_zip_service.py
Normal file
@@ -0,0 +1,68 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import unittest
|
||||
import zipfile
|
||||
|
||||
try:
|
||||
import pyzipper
|
||||
except ImportError: # pragma: no cover
|
||||
pyzipper = None
|
||||
|
||||
from govoplan_campaign.backend.services.zip_service import create_zip_archive
|
||||
|
||||
|
||||
class ZipServiceTests(unittest.TestCase):
|
||||
def test_standard_password_zip_uses_zipcrypto_readable_by_stdlib(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
root = Path(temp_dir)
|
||||
source = root / "source.txt"
|
||||
source.write_text("Hello Windows ZIP", encoding="utf-8")
|
||||
output = root / "standard.zip"
|
||||
|
||||
create_zip_archive(output, [(source, "message.txt")], "secret", "zip_standard")
|
||||
|
||||
with zipfile.ZipFile(output) as archive:
|
||||
info = archive.getinfo("message.txt")
|
||||
self.assertEqual(info.compress_type, zipfile.ZIP_DEFLATED)
|
||||
self.assertTrue(info.flag_bits & 0x1)
|
||||
self.assertEqual(archive.read("message.txt", pwd=b"secret"), b"Hello Windows ZIP")
|
||||
|
||||
@unittest.skipIf(pyzipper is None, "pyzipper is not installed")
|
||||
def test_aes_password_zip_keeps_aes_encryption(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
root = Path(temp_dir)
|
||||
source = root / "source.txt"
|
||||
source.write_text("Hello AES ZIP", encoding="utf-8")
|
||||
output = root / "aes.zip"
|
||||
|
||||
create_zip_archive(output, [(source, "message.txt")], "secret", "aes")
|
||||
|
||||
with zipfile.ZipFile(output) as archive:
|
||||
info = archive.getinfo("message.txt")
|
||||
self.assertEqual(info.compress_type, 99)
|
||||
self.assertTrue(info.flag_bits & 0x1)
|
||||
|
||||
with pyzipper.AESZipFile(output) as archive:
|
||||
archive.setpassword(b"secret")
|
||||
self.assertEqual(archive.read("message.txt"), b"Hello AES ZIP")
|
||||
|
||||
def test_unprotected_zip_ignores_encryption_mode(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
root = Path(temp_dir)
|
||||
source = root / "source.txt"
|
||||
source.write_text("Plain ZIP", encoding="utf-8")
|
||||
output = root / "plain.zip"
|
||||
|
||||
create_zip_archive(output, [(source, "message.txt")], "", "zip_standard")
|
||||
|
||||
with zipfile.ZipFile(output) as archive:
|
||||
info = archive.getinfo("message.txt")
|
||||
self.assertEqual(info.compress_type, zipfile.ZIP_DEFLATED)
|
||||
self.assertFalse(info.flag_bits & 0x1)
|
||||
self.assertEqual(archive.read("message.txt"), b"Plain ZIP")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user