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()