GovOPlaN/govoplan-core
1
0
Fork 0
Code Issues 43 Pull Requests Actions Packages Projects Releases Wiki Activity

test: align module contracts with Mail-owned delivery

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-21 17:15:19 +02:00
parent 57fe6c6006
commit 2eae5c4df6
3 changed files with 158 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/module-package-catalog.example.json
View File

@@ -84,7 +84,7 @@
"provides_interfaces": [ "provides_interfaces": [
{ {
"name": "mail.campaign_delivery", "name": "mail.campaign_delivery",
"version": "1.4.0" "version": "0.2.0"
} }
], ],
"artifact_integrity": { "artifact_integrity": {

264
tests/test_api_smoke.py
View File

@@ -100,6 +100,37 @@ class ApiSmokeTests(unittest.TestCase):
payload = response.json() payload = response.json()
return {"Authorization": f"Bearer {payload['access_token']}"}, payload return {"Authorization": f"Bearer {payload['access_token']}"}, payload
def _create_test_mail_profile(
self,
headers: dict[str, str],
*,
name: str,
include_imap: bool = False,
) -> str:
credentials = {
"smtp": {"username": "sender@example.org", "password": "test-secret"},
}
payload: dict[str, object] = {
"name": name,
"smtp": {
"host": "mock.smtp",
"port": 2525,
"security": "starttls",
},
"credentials": credentials,
}
if include_imap:
payload["imap"] = {
"host": "mock.imap",
"port": 993,
"security": "tls",
"sent_folder": "Sent",
}
credentials["imap"] = {"username": "sender@example.org", "password": "test-secret"}
response = self.client.post("/api/v1/mail/profiles", headers=headers, json=payload)
self.assertEqual(response.status_code, 201, response.text)
return str(response.json()["id"])
def test_recipient_import_mapping_profiles_are_db_backed(self) -> None: def test_recipient_import_mapping_profiles_are_db_backed(self) -> None:
headers, _ = self._login() headers, _ = self._login()
payload = { payload = {
@@ -224,6 +255,7 @@ class ApiSmokeTests(unittest.TestCase):
external_id: str, external_id: str,
recipient_count: int = 1, recipient_count: int = 1,
) -> tuple[str, str]: ) -> tuple[str, str]:
mail_profile_id = self._create_test_mail_profile(headers, name=f"{external_id} delivery")
entries = [ entries = [
{ {
"id": f"recipient-{index + 1}", "id": f"recipient-{index + 1}",
@@ -237,15 +269,7 @@ class ApiSmokeTests(unittest.TestCase):
"campaign": {"id": external_id, "name": external_id, "mode": "test"}, "campaign": {"id": external_id, "name": external_id, "mode": "test"},
"fields": [{"name": "first_name", "type": "string", "required": True}], "fields": [{"name": "first_name", "type": "string", "required": True}],
"global_values": {}, "global_values": {},
"server": { "server": {"mail_profile_id": mail_profile_id},
"smtp": {
"host": "mock.smtp",
"port": 2525,
"username": "sender@example.org",
"password": "test-secret",
"security": "starttls",
}
},
"recipients": { "recipients": {
"from": {"email": "sender@example.org", "name": "Sender", "type": "to"}, "from": {"email": "sender@example.org", "name": "Sender", "type": "to"},
"allow_individual_to": True, "allow_individual_to": True,
@@ -740,10 +764,40 @@ class ApiSmokeTests(unittest.TestCase):
assert version is not None assert version is not None
self.assertEqual(version.raw_json["server"], {"mail_profile_id": profile_id}) self.assertEqual(version.raw_json["server"], {"mail_profile_id": profile_id})
snapshot = version.execution_snapshot or {} snapshot = version.execution_snapshot or {}
self.assertIsNone(snapshot.get("smtp", {}).get("password")) self.assertEqual(snapshot["snapshot_version"], "5")
self.assertEqual(snapshot.get("smtp", {}).get("host"), "mock.smtp") self.assertEqual(snapshot["mail_profile_id"], profile_id)
self.assertNotIn("smtp", snapshot)
self.assertNotIn("imap", snapshot)
self.assertTrue(snapshot["smtp_transport_revision"])
self.assertTrue(snapshot["imap_transport_revision"])
def test_mail_profile_can_inherit_server_without_credentials(self) -> None: deleted = self.client.delete(f"/api/v1/mail/profiles/{profile_id}", headers=headers)
self.assertEqual(deleted.status_code, 200, deleted.text)
self.assertFalse(deleted.json()["is_active"])
self.assertFalse(deleted.json()["smtp_password_configured"])
self.assertFalse(deleted.json()["imap_password_configured"])
from govoplan_audit.backend.db.models import AuditLog
with SessionLocal() as session:
profile = session.get(MailServerProfile, profile_id)
self.assertIsNotNone(profile)
assert profile is not None
self.assertIsNone(profile.smtp_password_encrypted)
self.assertIsNone(profile.imap_password_encrypted)
audit = (
session.query(AuditLog)
.filter(
AuditLog.action == "mail.profile_credentials_deleted",
AuditLog.object_id == profile_id,
)
.one()
)
self.assertEqual(audit.details["deleted_protocols"], ["smtp", "imap"])
self.assertEqual(audit.details["deletion_reason"], "profile_deactivated")
self.assertNotIn("secret", repr(audit.details))
def test_campaign_runtime_does_not_receive_mail_owned_profile_credentials(self) -> None:
headers, _ = self._login() headers, _ = self._login()
created_profile = self.client.post( created_profile = self.client.post(
"/api/v1/mail/profiles", "/api/v1/mail/profiles",
@@ -775,15 +829,7 @@ class ApiSmokeTests(unittest.TestCase):
"campaign": {"id": "profile-local-creds", "name": "Profile local creds", "mode": "test"}, "campaign": {"id": "profile-local-creds", "name": "Profile local creds", "mode": "test"},
"fields": [{"name": "first_name", "type": "string", "required": True}], "fields": [{"name": "first_name", "type": "string", "required": True}],
"global_values": {}, "global_values": {},
"server": { "server": {"mail_profile_id": profile_id},
"mail_profile_id": profile_id,
"inherit_smtp_credentials": False,
"inherit_imap_credentials": False,
"credentials": {
"smtp": {"username": "campaign-smtp@example.org", "password": "campaign-smtp-secret"},
"imap": {"username": "campaign-imap@example.org", "password": "campaign-imap-secret"},
},
},
"recipients": { "recipients": {
"from": {"email": "sender@example.org", "name": "Sender", "type": "to"}, "from": {"email": "sender@example.org", "name": "Sender", "type": "to"},
"to": [{"email": "recipient@example.org", "type": "to"}], "to": [{"email": "recipient@example.org", "type": "to"}],
@@ -810,31 +856,43 @@ class ApiSmokeTests(unittest.TestCase):
self.assertEqual(built.status_code, 200, built.text) self.assertEqual(built.status_code, 200, built.text)
from govoplan_campaign.backend.db.models import CampaignVersion from govoplan_campaign.backend.db.models import CampaignVersion
from govoplan_campaign.backend.sending.execution import ExecutionSnapshot, runtime_imap_config, runtime_smtp_config from govoplan_campaign.backend.integrations import mail_integration
from govoplan_mail.backend.db.models import MailServerProfile
with SessionLocal() as session: with SessionLocal() as session:
version = session.get(CampaignVersion, version_id) version = session.get(CampaignVersion, version_id)
self.assertIsNotNone(version) self.assertIsNotNone(version)
assert version is not None assert version is not None
snapshot_payload = version.execution_snapshot or {} snapshot_payload = version.execution_snapshot or {}
self.assertEqual(snapshot_payload.get("smtp", {}).get("host"), "mock.smtp") self.assertEqual(snapshot_payload["snapshot_version"], "5")
self.assertEqual(snapshot_payload.get("smtp", {}).get("username"), "campaign-smtp@example.org") self.assertEqual(snapshot_payload["mail_profile_id"], profile_id)
self.assertIsNone(snapshot_payload.get("smtp", {}).get("password")) self.assertNotIn("smtp", snapshot_payload)
self.assertEqual(snapshot_payload.get("imap", {}).get("host"), "mock.imap") self.assertNotIn("imap", snapshot_payload)
self.assertEqual(snapshot_payload.get("imap", {}).get("username"), "campaign-imap@example.org") self.assertTrue(snapshot_payload["smtp_transport_revision"])
self.assertIsNone(snapshot_payload.get("imap", {}).get("password")) self.assertTrue(snapshot_payload["imap_transport_revision"])
snapshot = ExecutionSnapshot.model_validate(snapshot_payload) profile = session.get(MailServerProfile, profile_id)
smtp_runtime = runtime_smtp_config(session, version, snapshot) self.assertIsNotNone(profile)
imap_runtime = runtime_imap_config(session, version, snapshot) assert profile is not None
self.assertEqual(smtp_runtime.host, "mock.smtp") integration = mail_integration()
self.assertEqual(smtp_runtime.username, "campaign-smtp@example.org") summary = integration.campaign_profile_delivery_summary(
self.assertEqual(smtp_runtime.password, "campaign-smtp-secret") session,
self.assertIsNotNone(imap_runtime) tenant_id=profile.tenant_id,
assert imap_runtime is not None campaign_id=version.campaign_id,
self.assertEqual(imap_runtime.host, "mock.imap") profile_id=profile_id,
self.assertEqual(imap_runtime.username, "campaign-imap@example.org") )
self.assertEqual(imap_runtime.password, "campaign-imap-secret") self.assertTrue(summary["smtp_available"])
self.assertTrue(summary["imap_available"])
self.assertNotIn("mock.smtp", repr(summary))
self.assertNotIn("profile-smtp", repr(summary))
self.assertNotIn("secret", repr(summary))
for name in (
"smtp_config_from_profile",
"imap_config_from_profile",
"send_email_bytes",
"send_email_message",
):
self.assertFalse(hasattr(integration, name), name)
def test_health_schema_and_dev_mailbox_gates(self) -> None: def test_health_schema_and_dev_mailbox_gates(self) -> None:
public_health = self.client.get("/health") public_health = self.client.get("/health")
@@ -2885,7 +2943,10 @@ class ApiSmokeTests(unittest.TestCase):
self.assertFalse(delta_payload["full"]) self.assertFalse(delta_payload["full"])
self.assertEqual([item["id"] for item in delta_payload["jobs"]], [job_id]) self.assertEqual([item["id"] for item in delta_payload["jobs"]], [job_id])
self.assertEqual(delta_payload["jobs"][0]["send_status"], "outcome_unknown") self.assertEqual(delta_payload["jobs"][0]["send_status"], "outcome_unknown")
self.assertEqual(delta_payload["jobs"][0]["last_error"], "SMTP outcome needs reconciliation") self.assertEqual(
delta_payload["jobs"][0]["last_error"],
"SMTP delivery outcome requires operator reconciliation.",
)
self.assertEqual(delta_payload["counts"]["send"]["outcome_unknown"], 1) self.assertEqual(delta_payload["counts"]["send"]["outcome_unknown"], 1)
self.assertTrue(str(delta_payload["watermark"]).startswith("seq:")) self.assertTrue(str(delta_payload["watermark"]).startswith("seq:"))
@@ -3095,28 +3156,13 @@ class ApiSmokeTests(unittest.TestCase):
def test_campaign_create_validate_build_and_mock_send(self) -> None: def test_campaign_create_validate_build_and_mock_send(self) -> None:
headers, _ = self._login() headers, _ = self._login()
mail_profile_id = self._create_test_mail_profile(headers, name="API smoke delivery", include_imap=True)
campaign_json = { campaign_json = {
"version": "1.0", "version": "1.0",
"campaign": {"id": "api-smoke", "name": "API smoke campaign", "mode": "test"}, "campaign": {"id": "api-smoke", "name": "API smoke campaign", "mode": "test"},
"fields": [{"name": "first_name", "type": "string", "required": True}], "fields": [{"name": "first_name", "type": "string", "required": True}],
"global_values": {}, "global_values": {},
"server": { "server": {"mail_profile_id": mail_profile_id},
"smtp": {
"host": "smtp.example.invalid",
"port": 587,
"username": "sender@example.org",
"password": "test-secret",
"security": "starttls",
},
"imap": {
"enabled": True,
"host": "imap.example.invalid",
"port": 993,
"username": "sender@example.org",
"password": "test-secret",
"security": "tls",
},
},
"recipients": { "recipients": {
"from": {"email": "sender@example.org", "name": "Sender", "type": "to"}, "from": {"email": "sender@example.org", "name": "Sender", "type": "to"},
"allow_individual_to": True, "allow_individual_to": True,
@@ -3188,20 +3234,13 @@ class ApiSmokeTests(unittest.TestCase):
def test_managed_attachment_patterns_preview_build_and_mock_send(self) -> None: def test_managed_attachment_patterns_preview_build_and_mock_send(self) -> None:
headers, login = self._login() headers, login = self._login()
user_id = login["user"]["id"] user_id = login["user"]["id"]
mail_profile_id = self._create_test_mail_profile(headers, name="Managed attachments delivery")
campaign_json = { campaign_json = {
"version": "1.0", "version": "1.0",
"campaign": {"id": "managed-attachments", "name": "Managed attachments", "mode": "test"}, "campaign": {"id": "managed-attachments", "name": "Managed attachments", "mode": "test"},
"fields": [{"name": "invoice_number", "type": "string", "required": True}], "fields": [{"name": "invoice_number", "type": "string", "required": True}],
"global_values": {}, "global_values": {},
"server": { "server": {"mail_profile_id": mail_profile_id},
"smtp": {
"host": "smtp.example.invalid",
"port": 587,
"username": "sender@example.org",
"password": "test-secret",
"security": "starttls",
}
},
"recipients": { "recipients": {
"from": {"email": "sender@example.org", "name": "Sender", "type": "to"}, "from": {"email": "sender@example.org", "name": "Sender", "type": "to"},
"allow_individual_to": True, "allow_individual_to": True,
@@ -3401,20 +3440,13 @@ class ApiSmokeTests(unittest.TestCase):
def test_managed_attachment_unlinked_candidates_can_be_linked_on_validation(self) -> None: def test_managed_attachment_unlinked_candidates_can_be_linked_on_validation(self) -> None:
headers, login = self._login() headers, login = self._login()
user_id = login["user"]["id"] user_id = login["user"]["id"]
mail_profile_id = self._create_test_mail_profile(headers, name="Unlinked attachments delivery")
campaign_json = { campaign_json = {
"version": "1.0", "version": "1.0",
"campaign": {"id": "managed-unlinked-attachments", "name": "Managed unlinked attachments", "mode": "test"}, "campaign": {"id": "managed-unlinked-attachments", "name": "Managed unlinked attachments", "mode": "test"},
"fields": [{"name": "invoice_number", "type": "string", "required": True}], "fields": [{"name": "invoice_number", "type": "string", "required": True}],
"global_values": {}, "global_values": {},
"server": { "server": {"mail_profile_id": mail_profile_id},
"smtp": {
"host": "smtp.example.invalid",
"port": 587,
"username": "sender@example.org",
"password": "test-secret",
"security": "starttls",
}
},
"recipients": { "recipients": {
"from": {"email": "sender@example.org", "name": "Sender", "type": "to"}, "from": {"email": "sender@example.org", "name": "Sender", "type": "to"},
"allow_individual_to": True, "allow_individual_to": True,
@@ -3522,20 +3554,13 @@ class ApiSmokeTests(unittest.TestCase):
def test_managed_attachment_send_uses_frozen_build_artifact_after_file_changes(self) -> None: def test_managed_attachment_send_uses_frozen_build_artifact_after_file_changes(self) -> None:
headers, login = self._login() headers, login = self._login()
user_id = login["user"]["id"] user_id = login["user"]["id"]
mail_profile_id = self._create_test_mail_profile(headers, name="Frozen attachment delivery")
campaign_json = { campaign_json = {
"version": "1.0", "version": "1.0",
"campaign": {"id": "managed-send-freeze", "name": "Managed send freeze", "mode": "test"}, "campaign": {"id": "managed-send-freeze", "name": "Managed send freeze", "mode": "test"},
"fields": [], "fields": [],
"global_values": {}, "global_values": {},
"server": { "server": {"mail_profile_id": mail_profile_id},
"smtp": {
"host": "mock.smtp",
"port": 2525,
"username": "sender@example.org",
"password": "test-secret",
"security": "starttls",
}
},
"recipients": { "recipients": {
"from": {"email": "sender@example.org", "name": "Sender", "type": "to"}, "from": {"email": "sender@example.org", "name": "Sender", "type": "to"},
"allow_individual_to": True, "allow_individual_to": True,
@@ -3689,11 +3714,12 @@ class ApiSmokeTests(unittest.TestCase):
def test_non_blocking_review_conditions_can_be_accepted_in_bulk(self) -> None: def test_non_blocking_review_conditions_can_be_accepted_in_bulk(self) -> None:
headers, login = self._login() headers, login = self._login()
user_id = login["user"]["id"] user_id = login["user"]["id"]
mail_profile_id = self._create_test_mail_profile(headers, name="Bulk review delivery")
campaign_json = { campaign_json = {
"version": "1.0", "version": "1.0",
"campaign": {"id": "bulk-review", "name": "Bulk review", "mode": "test"}, "campaign": {"id": "bulk-review", "name": "Bulk review", "mode": "test"},
"fields": [], "global_values": {}, "fields": [], "global_values": {},
"server": {"smtp": {"host": "smtp.example.invalid", "port": 587, "security": "starttls"}}, "server": {"mail_profile_id": mail_profile_id},
"recipients": {"from": {"email": "sender@example.org", "type": "to"}, "allow_individual_to": True}, "recipients": {"from": {"email": "sender@example.org", "type": "to"}, "allow_individual_to": True},
"template": {"subject": "Warning test", "text": "Body"}, "template": {"subject": "Warning test", "text": "Body"},
"attachments": { "attachments": {
@@ -3741,12 +3767,13 @@ class ApiSmokeTests(unittest.TestCase):
def test_inactive_recipients_are_aggregated_but_not_built_or_reviewed(self) -> None: def test_inactive_recipients_are_aggregated_but_not_built_or_reviewed(self) -> None:
headers, login = self._login() headers, login = self._login()
user_id = login["user"]["id"] user_id = login["user"]["id"]
mail_profile_id = self._create_test_mail_profile(headers, name="Inactive recipients delivery")
campaign_json = { campaign_json = {
"version": "1.0", "version": "1.0",
"campaign": {"id": "inactive-recipients", "name": "Inactive recipients", "mode": "test"}, "campaign": {"id": "inactive-recipients", "name": "Inactive recipients", "mode": "test"},
"fields": [], "fields": [],
"global_values": {}, "global_values": {},
"server": {"smtp": {"host": "smtp.example.invalid", "port": 587, "security": "starttls"}}, "server": {"mail_profile_id": mail_profile_id},
"recipients": {"from": {"email": "sender@example.org", "name": "Sender", "type": "to"}, "allow_individual_to": True}, "recipients": {"from": {"email": "sender@example.org", "name": "Sender", "type": "to"}, "allow_individual_to": True},
"template": {"subject": "Hello", "text": "Active recipient only"}, "template": {"subject": "Hello", "text": "Active recipient only"},
"attachments": { "attachments": {
@@ -3801,12 +3828,13 @@ class ApiSmokeTests(unittest.TestCase):
def test_recipient_address_fields_and_merge_modes(self) -> None: def test_recipient_address_fields_and_merge_modes(self) -> None:
headers, _ = self._login() headers, _ = self._login()
mail_profile_id = self._create_test_mail_profile(headers, name="Recipient merge delivery")
campaign_json = { campaign_json = {
"version": "1.0", "version": "1.0",
"campaign": {"id": "recipient-address-fields", "name": "Recipient address fields", "mode": "test"}, "campaign": {"id": "recipient-address-fields", "name": "Recipient address fields", "mode": "test"},
"fields": [], "fields": [],
"global_values": {}, "global_values": {},
"server": {"smtp": {"host": "smtp.example.invalid", "port": 587, "security": "starttls"}}, "server": {"mail_profile_id": mail_profile_id},
"recipients": { "recipients": {
"from": [{"email": "global-from@example.org", "type": "to"}], "from": [{"email": "global-from@example.org", "type": "to"}],
"allow_individual_from": True, "allow_individual_from": True,
@@ -3867,12 +3895,13 @@ class ApiSmokeTests(unittest.TestCase):
def test_multiple_from_addresses_are_rejected(self) -> None: def test_multiple_from_addresses_are_rejected(self) -> None:
headers, _ = self._login() headers, _ = self._login()
mail_profile_id = self._create_test_mail_profile(headers, name="Multiple sender validation delivery")
campaign_json = { campaign_json = {
"version": "1.0", "version": "1.0",
"campaign": {"id": "multiple-from", "name": "Multiple From", "mode": "test"}, "campaign": {"id": "multiple-from", "name": "Multiple From", "mode": "test"},
"fields": [], "fields": [],
"global_values": {}, "global_values": {},
"server": {"smtp": {"host": "smtp.example.invalid", "port": 587, "security": "starttls"}}, "server": {"mail_profile_id": mail_profile_id},
"recipients": { "recipients": {
"from": [ "from": [
{"email": "first@example.org", "type": "to"}, {"email": "first@example.org", "type": "to"},
@@ -3892,12 +3921,13 @@ class ApiSmokeTests(unittest.TestCase):
def test_duplicate_zip_archive_filenames_block_validation(self) -> None: def test_duplicate_zip_archive_filenames_block_validation(self) -> None:
headers, _ = self._login() headers, _ = self._login()
mail_profile_id = self._create_test_mail_profile(headers, name="Duplicate ZIP validation delivery")
campaign_json = { campaign_json = {
"version": "1.0", "version": "1.0",
"campaign": {"id": "duplicate-zip-names", "name": "Duplicate ZIP names", "mode": "test"}, "campaign": {"id": "duplicate-zip-names", "name": "Duplicate ZIP names", "mode": "test"},
"fields": [], "fields": [],
"global_values": {}, "global_values": {},
"server": {"smtp": {"host": "smtp.example.invalid", "port": 587, "security": "starttls"}}, "server": {"mail_profile_id": mail_profile_id},
"recipients": {"from": {"email": "sender@example.org", "type": "to"}, "allow_individual_to": True}, "recipients": {"from": {"email": "sender@example.org", "type": "to"}, "allow_individual_to": True},
"template": {"subject": "ZIP validation", "text": "Body"}, "template": {"subject": "ZIP validation", "text": "Body"},
"attachments": { "attachments": {
@@ -3937,6 +3967,7 @@ class ApiSmokeTests(unittest.TestCase):
def test_recipient_zip_archive_with_field_password_and_rule_exclusions(self) -> None: def test_recipient_zip_archive_with_field_password_and_rule_exclusions(self) -> None:
headers, login = self._login() headers, login = self._login()
user_id = login["user"]["id"] user_id = login["user"]["id"]
mail_profile_id = self._create_test_mail_profile(headers, name="ZIP attachment delivery")
campaign_json = { campaign_json = {
"version": "1.0", "version": "1.0",
"campaign": {"id": "zip-attachments", "name": "ZIP attachments", "mode": "test"}, "campaign": {"id": "zip-attachments", "name": "ZIP attachments", "mode": "test"},
@@ -3946,7 +3977,7 @@ class ApiSmokeTests(unittest.TestCase):
{"name": "global_zip_password", "type": "password", "required": True}, {"name": "global_zip_password", "type": "password", "required": True},
], ],
"global_values": {"global_zip_password": "campaign-secret"}, "global_values": {"global_zip_password": "campaign-secret"},
"server": {"smtp": {"host": "smtp.example.invalid", "port": 587, "security": "starttls"}}, "server": {"mail_profile_id": mail_profile_id},
"recipients": { "recipients": {
"from": {"email": "sender@example.org", "name": "Sender", "type": "to"}, "from": {"email": "sender@example.org", "name": "Sender", "type": "to"},
"allow_individual_to": True, "allow_individual_to": True,
@@ -4133,14 +4164,15 @@ class ApiSmokeTests(unittest.TestCase):
def test_execution_snapshot_freezes_delivery_configuration_and_job_manifest(self) -> None: def test_execution_snapshot_detects_profile_drift_and_freezes_job_manifest(self) -> None:
headers, _ = self._login() headers, _ = self._login()
campaign_id, version_id = self._create_built_delivery_campaign( campaign_id, version_id = self._create_built_delivery_campaign(
headers, headers,
external_id="snapshot-freeze", external_id="snapshot-freeze",
) )
from govoplan_campaign.backend.db.models import CampaignJob, CampaignVersion from govoplan_campaign.backend.db.models import CampaignJob, CampaignVersion, SendAttempt
from govoplan_mail.backend.db.models import MailServerProfile
with SessionLocal() as session: with SessionLocal() as session:
version = session.get(CampaignVersion, version_id) version = session.get(CampaignVersion, version_id)
@@ -4148,23 +4180,24 @@ class ApiSmokeTests(unittest.TestCase):
assert version is not None assert version is not None
snapshot = version.execution_snapshot snapshot = version.execution_snapshot
self.assertIsInstance(snapshot, dict) self.assertIsInstance(snapshot, dict)
self.assertEqual(snapshot["snapshot_version"], "3") self.assertEqual(snapshot["snapshot_version"], "5")
self.assertEqual(snapshot["job_count"], 1) self.assertEqual(snapshot["job_count"], 1)
self.assertEqual(snapshot["queueable_job_count"], 1) self.assertEqual(snapshot["queueable_job_count"], 1)
self.assertTrue(snapshot["job_manifest_sha256"]) self.assertTrue(snapshot["job_manifest_sha256"])
self.assertTrue(snapshot["smtp_config_fingerprint"]) self.assertTrue(snapshot["smtp_transport_revision"])
self.assertIsNone(snapshot["smtp"].get("password")) self.assertNotIn("smtp", snapshot)
self.assertNotIn("imap", snapshot)
self.assertTrue(version.execution_snapshot_hash) self.assertTrue(version.execution_snapshot_hash)
job = session.query(CampaignJob).filter(CampaignJob.campaign_version_id == version_id).one() job = session.query(CampaignJob).filter(CampaignJob.campaign_version_id == version_id).one()
self.assertTrue(job.eml_sha256) self.assertTrue(job.eml_sha256)
self.assertTrue(job.message_id_header) self.assertTrue(job.message_id_header)
# Simulate accidental/config-drift mutation after the build. Sending profile = session.get(MailServerProfile, snapshot["mail_profile_id"])
# must still use the immutable mock SMTP snapshot, not raw_json. self.assertIsNotNone(profile)
mutated = json.loads(json.dumps(version.raw_json)) assert profile is not None
mutated["server"]["smtp"]["host"] = "smtp.example.invalid" profile.smtp_config = {**profile.smtp_config, "host": "smtp.example.invalid"}
version.raw_json = mutated profile.smtp_transport_revision = "manually-rotated-transport-revision"
session.add(version) session.add(profile)
session.commit() session.commit()
sent = self.client.post( sent = self.client.post(
@@ -4179,12 +4212,17 @@ class ApiSmokeTests(unittest.TestCase):
}, },
) )
self.assertEqual(sent.status_code, 200, sent.text) self.assertEqual(sent.status_code, 200, sent.text)
self.assertEqual(sent.json()["result"]["sent_count"], 1, sent.text) self.assertEqual(sent.json()["result"]["sent_count"], 0)
self.assertEqual(sent.json()["result"]["outcome_unknown_count"], 0, sent.text) self.assertEqual(sent.json()["result"]["failed_count"], 1)
self.assertIn("changed after this campaign was built", sent.json()["result"]["results"][0]["message"])
with SessionLocal() as session: with SessionLocal() as session:
job = session.query(CampaignJob).filter(CampaignJob.campaign_version_id == version_id).one() job = session.query(CampaignJob).filter(CampaignJob.campaign_version_id == version_id).one()
self.assertEqual(job.send_status, "smtp_accepted") self.assertEqual(job.send_status, "failed_permanent")
attempt = session.query(SendAttempt).filter(SendAttempt.job_id == job.id).one()
self.assertIn("changed after this campaign was built", attempt.error_message)
self.assertNotIn("smtp.example.invalid", attempt.error_message)
self.assertIsNone(attempt.smtp_response)
def test_send_now_sends_exact_generated_eml_bytes(self) -> None: def test_send_now_sends_exact_generated_eml_bytes(self) -> None:
headers, _ = self._login() headers, _ = self._login()
@@ -4256,21 +4294,14 @@ class ApiSmokeTests(unittest.TestCase):
def test_partial_smtp_recipient_refusal_is_recorded_without_retrying_accepted_delivery(self) -> None: def test_partial_smtp_recipient_refusal_is_recorded_without_retrying_accepted_delivery(self) -> None:
headers, _ = self._login() headers, _ = self._login()
from govoplan_mail.backend.dev.mock_mailbox import set_failures from govoplan_mail.backend.dev.mock_mailbox import set_failures
mail_profile_id = self._create_test_mail_profile(headers, name="Partial refusal delivery")
campaign_json = { campaign_json = {
"version": "1.0", "version": "1.0",
"campaign": {"id": "partial-refusal", "name": "Partial refusal", "mode": "test"}, "campaign": {"id": "partial-refusal", "name": "Partial refusal", "mode": "test"},
"fields": [], "fields": [],
"global_values": {}, "global_values": {},
"server": { "server": {"mail_profile_id": mail_profile_id},
"smtp": {
"host": "mock.smtp",
"port": 2525,
"username": "sender@example.org",
"password": "test-secret",
"security": "starttls",
}
},
"recipients": { "recipients": {
"from": {"email": "sender@example.org", "name": "Sender", "type": "to"}, "from": {"email": "sender@example.org", "name": "Sender", "type": "to"},
"allow_individual_to": True, "allow_individual_to": True,
@@ -5761,6 +5792,7 @@ class ApiSmokeTests(unittest.TestCase):
def test_campaign_acl_separates_capability_from_object_access(self) -> None: def test_campaign_acl_separates_capability_from_object_access(self) -> None:
owner_headers, _ = self._login() owner_headers, _ = self._login()
mail_profile_id = self._create_test_mail_profile(owner_headers, name="ACL campaign delivery")
access_role = self._create_role( access_role = self._create_role(
owner_headers, owner_headers,
slug="campaign-collaborator", slug="campaign-collaborator",
@@ -5786,7 +5818,7 @@ class ApiSmokeTests(unittest.TestCase):
"campaign": {"id": "acl-campaign", "name": "ACL campaign", "mode": "test"}, "campaign": {"id": "acl-campaign", "name": "ACL campaign", "mode": "test"},
"fields": [], "fields": [],
"global_values": {}, "global_values": {},
"server": {"smtp": {"host": "mock.smtp", "port": 2525, "security": "starttls"}}, "server": {"mail_profile_id": mail_profile_id},
"recipients": {"from": {"email": "sender@example.org", "type": "to"}}, "recipients": {"from": {"email": "sender@example.org", "type": "to"}},
"template": {"subject": "ACL", "text": "ACL"}, "template": {"subject": "ACL", "text": "ACL"},
"attachments": {"base_path": ".", "global": [], "allow_individual": False}, "attachments": {"base_path": ".", "global": [], "allow_individual": False},
@@ -5984,7 +6016,7 @@ class ApiSmokeTests(unittest.TestCase):
json={"campaign_json": inline_json}, json={"campaign_json": inline_json},
) )
self.assertEqual(blocked_inline.status_code, 422, blocked_inline.text) self.assertEqual(blocked_inline.status_code, 422, blocked_inline.text)
self.assertIn("Campaign-local inline mail settings", blocked_inline.json()["detail"]) self.assertIn("may only reference", blocked_inline.json()["detail"])
reusable_json = detail.json()["raw_json"] reusable_json = detail.json()["raw_json"]
reusable_json["server"] = {"mail_profile_id": tenant_profile_id} reusable_json["server"] = {"mail_profile_id": tenant_profile_id}
@@ -6105,7 +6137,7 @@ class ApiSmokeTests(unittest.TestCase):
} }
blocked = self.client.post("/api/v1/campaigns", headers=headers, json={"config": campaign_json}) blocked = self.client.post("/api/v1/campaigns", headers=headers, json={"config": campaign_json})
self.assertEqual(blocked.status_code, 422, blocked.text) self.assertEqual(blocked.status_code, 422, blocked.text)
self.assertIn("locked", blocked.json()["detail"]) self.assertIn("may only reference", blocked.json()["detail"])
campaign_json["server"] = {"mail_profile_id": profile_id} campaign_json["server"] = {"mail_profile_id": profile_id}
allowed = self.client.post("/api/v1/campaigns", headers=headers, json={"config": campaign_json}) allowed = self.client.post("/api/v1/campaigns", headers=headers, json={"config": campaign_json})

12
tests/test_module_system.py
View File

@@ -3130,18 +3130,24 @@ finally:
"version_max_exclusive": "0.2.0", "version_max_exclusive": "0.2.0",
}, modules["files"]["requires_interfaces"]) }, modules["files"]["requires_interfaces"])
self.assertEqual(["campaigns"], modules["files"]["optional_dependencies"]) self.assertEqual(["campaigns"], modules["files"]["optional_dependencies"])
self.assertIn({"name": "mail.campaign_delivery", "version": "0.1.6"}, modules["mail"]["provides_interfaces"]) self.assertIn({"name": "mail.campaign_delivery", "version": "0.2.0"}, modules["mail"]["provides_interfaces"])
self.assertIn({ self.assertIn({
"name": "files.campaign_attachments", "name": "files.campaign_attachments",
"optional": True, "optional": True,
"version_min": "0.1.0", "version_min": "0.1.0",
"version_max_exclusive": "0.2.0", "version_max_exclusive": "0.2.0",
}, modules["campaigns"]["requires_interfaces"]) }, modules["campaigns"]["requires_interfaces"])
self.assertIn({
"name": "mail.campaign_delivery",
"optional": True,
"version_min": "0.2.0",
"version_max_exclusive": "0.3.0",
}, modules["campaigns"]["requires_interfaces"])
self.assertEqual(["files", "mail", "notifications", "addresses"], modules["campaigns"]["optional_dependencies"]) self.assertEqual(["files", "mail", "notifications", "addresses"], modules["campaigns"]["optional_dependencies"])
self.assertEqual("requires_review", modules["files"]["migration_safety"]) self.assertEqual("requires_review", modules["files"]["migration_safety"])
self.assertIn("migration", modules["files"]["migration_notes"].lower()) self.assertIn("migration", modules["files"]["migration_notes"].lower())
self.assertEqual("0.1.8", modules["files"]["version"]) self.assertEqual("0.1.9", modules["files"]["version"])
self.assertIn("@v0.1.8", modules["files"]["python_ref"]) self.assertIn("@v0.1.9", modules["files"]["python_ref"])
def test_module_package_catalog_validates_remote_url_and_cache_fallback(self) -> None: def test_module_package_catalog_validates_remote_url_and_cache_fallback(self) -> None:
root = Path(tempfile.mkdtemp(prefix="govoplan-module-package-catalog-remote-", dir=_TEST_ROOT)) root = Path(tempfile.mkdtemp(prefix="govoplan-module-package-catalog-remote-", dir=_TEST_ROOT))