213 lines
11 KiB
Python
213 lines
11 KiB
Python
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": {
|
|
"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": 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": {
|
|
"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": {
|
|
"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()
|