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

test(campaign): align synchronous send security contract

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-22 20:30:00 +02:00
parent f876345656
commit bc8afeb139
1 changed files with 17 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
tests/test_api_smoke.py
View File

@@ -4211,18 +4211,17 @@ class ApiSmokeTests(unittest.TestCase):
"enqueue_imap_task": False,
},
)
self.assertEqual(sent.status_code, 200, sent.text)
self.assertEqual(sent.json()["result"]["sent_count"], 0)
self.assertEqual(sent.json()["result"]["failed_count"], 1)
self.assertIn("changed after this campaign was built", sent.json()["result"]["results"][0]["message"])
self.assertEqual(sent.status_code, 422, sent.text)
self.assertIn("Synchronous preflight stopped before contacting SMTP", sent.json()["detail"])
with SessionLocal() as session:
job = session.query(CampaignJob).filter(CampaignJob.campaign_version_id == version_id).one()
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)
self.assertEqual(job.queue_status, "draft")
self.assertEqual(job.send_status, "not_queued")
self.assertEqual(
session.query(SendAttempt).filter(SendAttempt.job_id == job.id).count(),
0,
)
def test_send_now_sends_exact_generated_eml_bytes(self) -> None:
headers, _ = self._login()
@@ -4286,11 +4285,15 @@ class ApiSmokeTests(unittest.TestCase):
"enqueue_imap_task": False,
},
)
self.assertEqual(sent.status_code, 200, sent.text)
self.assertEqual(sent.json()["result"]["failed_count"], 1)
self.assertIn("Generated EML", sent.json()["result"]["results"][0]["message"])
self.assertEqual(sent.status_code, 422, sent.text)
self.assertIn("Synchronous preflight stopped before contacting SMTP", sent.json()["detail"])
self.assertEqual(list_records(kind="smtp"), [])
with SessionLocal() as session:
job = session.query(CampaignJob).filter(CampaignJob.campaign_version_id == version_id).one()
self.assertEqual(job.queue_status, "draft")
self.assertEqual(job.send_status, "not_queued")
def test_partial_smtp_recipient_refusal_is_recorded_without_retrying_accepted_delivery(self) -> None:
headers, _ = self._login()
from govoplan_mail.backend.dev.mock_mailbox import set_failures
@@ -4349,7 +4352,8 @@ class ApiSmokeTests(unittest.TestCase):
self.assertEqual(sent.status_code, 200, sent.text)
result = sent.json()["result"]
self.assertEqual(result["sent_count"], 1, sent.text)
self.assertIn("refused recipients", result["results"][0]["message"])
self.assertEqual(result["results"][0]["status"], "smtp_accepted")
self.assertNotIn("message", result["results"][0])
finally:
set_failures(smtp_reject_recipients_containing=None)