152 lines
6.7 KiB
Python
152 lines
6.7 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from govoplan_campaign.backend import router
|
|
from govoplan_campaign.backend.campaign.models import CampaignConfig
|
|
from govoplan_campaign.backend.campaign.validation import validate_campaign_config
|
|
from govoplan_campaign.backend.persistence.versions import validate_campaign_partial
|
|
|
|
|
|
class CampaignPartialValidationTests(unittest.TestCase):
|
|
def test_partial_validation_reports_counts_across_sections(self) -> None:
|
|
result = validate_campaign_partial({
|
|
"campaign": {},
|
|
"recipients": {"from": {}},
|
|
"entries": {"source": {}, "mapping": {}},
|
|
"template": {"body_mode": "both"},
|
|
"attachments": {},
|
|
"delivery": {"rate_limit": {"messages_per_minute": 0}},
|
|
})
|
|
|
|
codes = {item["code"] for item in result["issues"]}
|
|
self.assertFalse(result["ok"])
|
|
self.assertEqual(3, result["error_count"])
|
|
self.assertIn("missing_campaign_id", codes)
|
|
self.assertIn("missing_email_mapping", codes)
|
|
self.assertIn("missing_template_text_body", codes)
|
|
self.assertIn("missing_attachment_base_path", codes)
|
|
self.assertIn("invalid_rate_limit", codes)
|
|
|
|
def test_partial_validation_filters_to_requested_section(self) -> None:
|
|
result = validate_campaign_partial(
|
|
{
|
|
"campaign": {},
|
|
"template": {"body_mode": "text"},
|
|
"delivery": {"rate_limit": {"messages_per_minute": "fast"}},
|
|
},
|
|
section="template",
|
|
)
|
|
|
|
self.assertTrue(result["ok"])
|
|
self.assertEqual("template", result["section"])
|
|
self.assertEqual(0, result["error_count"])
|
|
self.assertEqual({"missing_subject", "missing_template_text_body"}, {item["code"] for item in result["issues"]})
|
|
|
|
def test_address_lookup_returns_unavailable_when_addresses_capability_absent(self) -> None:
|
|
with patch.object(router, "_get_campaign_for_principal", return_value=None), patch.object(router, "_registry_capability", return_value=None):
|
|
response = router.lookup_campaign_addresses("campaign-1", query="", session=object(), principal=object())
|
|
|
|
self.assertFalse(response.available)
|
|
self.assertEqual(response.candidates, [])
|
|
|
|
def test_address_lookup_accepts_empty_query_for_suggestion_preload(self) -> None:
|
|
class LookupCapability:
|
|
def lookup(self, _session, _principal, *, query: str, limit: int):
|
|
self.query = query
|
|
self.limit = limit
|
|
return [{
|
|
"contact_id": "contact-1",
|
|
"address_book_id": "book-1",
|
|
"display_name": "Ada Lovelace",
|
|
"email": "ada@example.local",
|
|
"email_label": "work",
|
|
"tags": [],
|
|
"source_kind": "local",
|
|
"provenance": {"module": "addresses"},
|
|
}]
|
|
|
|
capability = LookupCapability()
|
|
with patch.object(router, "_get_campaign_for_principal", return_value=None), patch.object(router, "_registry_capability", return_value=capability):
|
|
response = router.lookup_campaign_addresses("campaign-1", query="", limit=100, session=object(), principal=object())
|
|
|
|
self.assertEqual(capability.query, "")
|
|
self.assertEqual(capability.limit, 100)
|
|
self.assertTrue(response.available)
|
|
self.assertEqual(response.candidates[0].email, "ada@example.local")
|
|
|
|
|
|
class CampaignSemanticValidationTests(unittest.TestCase):
|
|
def test_semantic_validation_reports_zip_delivery_and_external_mapping_issues(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / "recipients.csv").write_text("email\nada@example.local\n", encoding="utf-8")
|
|
campaign_file = root / "campaign.json"
|
|
config = CampaignConfig.model_validate({
|
|
"version": "1.0",
|
|
"campaign": {"id": "campaign-1", "name": "Campaign", "mode": "send"},
|
|
"fields": [
|
|
{"name": "pin", "type": "string", "can_override": False},
|
|
{"name": "secret", "type": "password"},
|
|
],
|
|
"global_values": {"unknown": "value"},
|
|
"recipients": {
|
|
"from": [{"email": "sender@example.local"}],
|
|
"to": [{"email": "recipient@example.local"}],
|
|
},
|
|
"template": {"subject": "Subject", "text": "Body", "body_mode": "text"},
|
|
"attachments": {
|
|
"base_path": "attachments",
|
|
"zip": {
|
|
"enabled": True,
|
|
"archives": [
|
|
{
|
|
"id": "main",
|
|
"name": "bundle.zip",
|
|
"standard": True,
|
|
"password_enabled": True,
|
|
"password_field": "secret",
|
|
"password_scope": "global",
|
|
}
|
|
],
|
|
},
|
|
},
|
|
"entries": {
|
|
"source": {"type": "csv", "path": "recipients.csv", "has_header": True},
|
|
"mapping": {"fields.pin": "missing_pin", "fields.unknown": "missing_unknown"},
|
|
"defaults": {
|
|
"attachments": [
|
|
{
|
|
"base_dir": "",
|
|
"file_filter": "*.pdf",
|
|
"zip": {"archive_id": "missing-archive"},
|
|
}
|
|
]
|
|
},
|
|
},
|
|
"delivery": {"imap_append_sent": {"enabled": True}},
|
|
})
|
|
|
|
report = validate_campaign_config(config, campaign_file=campaign_file, check_files=True)
|
|
|
|
codes = {issue.code for issue in report.issues}
|
|
self.assertFalse(report.ok)
|
|
self.assertEqual("external:csv", report.entries_mode)
|
|
self.assertIsNone(report.entries_count)
|
|
self.assertIn("unknown_global_value", codes)
|
|
self.assertIn("zip_global_password_value_missing", codes)
|
|
self.assertIn("zip_archive_unknown", codes)
|
|
self.assertIn("delivery_imap_enabled_without_server_imap", codes)
|
|
self.assertIn("missing_smtp_config", codes)
|
|
self.assertIn("unknown_mapping_target", codes)
|
|
self.assertIn("mapping_target_not_overridable", codes)
|
|
self.assertIn("mapping_columns_missing", codes)
|
|
self.assertIn("attachments_base_path_not_found", codes)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|