121 lines
3.9 KiB
Python
121 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
|
|
from govoplan_campaign.backend.db.models import (
|
|
JobBuildStatus,
|
|
JobQueueStatus,
|
|
JobSendStatus,
|
|
JobValidationStatus,
|
|
)
|
|
from govoplan_campaign.backend.sending.jobs import (
|
|
_queue_validation_statuses,
|
|
_select_campaign_jobs_for_queue,
|
|
)
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self) -> None:
|
|
self.added: list[object] = []
|
|
|
|
def add(self, value: object) -> None:
|
|
self.added.append(value)
|
|
|
|
|
|
def _job(entry_id: str, **overrides):
|
|
values = {
|
|
"entry_id": entry_id,
|
|
"entry_index": int(entry_id),
|
|
"send_status": JobSendStatus.NOT_QUEUED.value,
|
|
"queue_status": JobQueueStatus.DRAFT.value,
|
|
"validation_status": JobValidationStatus.READY.value,
|
|
"build_status": JobBuildStatus.BUILT.value,
|
|
"eml_local_path": f"{entry_id}.eml",
|
|
"eml_storage_key": None,
|
|
"last_error": "old error",
|
|
"queued_at": None,
|
|
"claimed_at": "claimed",
|
|
"claim_token": "token",
|
|
"smtp_started_at": "started",
|
|
"outcome_unknown_at": "unknown",
|
|
}
|
|
values.update(overrides)
|
|
return SimpleNamespace(**values)
|
|
|
|
|
|
class CampaignQueueSelectionTests(unittest.TestCase):
|
|
def test_selects_queueable_jobs_without_reclassifying_retry_states(self):
|
|
skipped_send = _job("1", send_status=JobSendStatus.FAILED_TEMPORARY.value)
|
|
skipped_queue = _job("2", queue_status=JobQueueStatus.PAUSED.value)
|
|
blocked_validation = _job(
|
|
"3",
|
|
validation_status=JobValidationStatus.WARNING.value,
|
|
)
|
|
blocked_missing_eml = _job(
|
|
"4",
|
|
eml_local_path=None,
|
|
eml_storage_key=None,
|
|
)
|
|
ready = _job("5")
|
|
reviewed = _job(
|
|
"6",
|
|
validation_status=JobValidationStatus.NEEDS_REVIEW.value,
|
|
)
|
|
session = FakeSession()
|
|
|
|
queued, skipped_count, blocked_count = _select_campaign_jobs_for_queue(
|
|
session,
|
|
jobs=[
|
|
skipped_send,
|
|
skipped_queue,
|
|
blocked_validation,
|
|
blocked_missing_eml,
|
|
ready,
|
|
reviewed,
|
|
],
|
|
allowed_validation=_queue_validation_statuses(include_warnings=False),
|
|
reviewed_needs_review_keys={"6"},
|
|
dry_run=False,
|
|
)
|
|
|
|
self.assertEqual(queued, [ready, reviewed])
|
|
self.assertEqual(skipped_count, 2)
|
|
self.assertEqual(blocked_count, 2)
|
|
self.assertIn("generated EML", blocked_missing_eml.last_error)
|
|
self.assertEqual(session.added, [ready, reviewed])
|
|
for job in queued:
|
|
self.assertEqual(job.queue_status, JobQueueStatus.QUEUED.value)
|
|
self.assertEqual(job.send_status, JobSendStatus.QUEUED.value)
|
|
self.assertIsNotNone(job.queued_at)
|
|
self.assertIsNone(job.claimed_at)
|
|
self.assertIsNone(job.claim_token)
|
|
self.assertIsNone(job.smtp_started_at)
|
|
self.assertIsNone(job.outcome_unknown_at)
|
|
self.assertIsNone(job.last_error)
|
|
|
|
def test_dry_run_does_not_mutate_queueable_job(self):
|
|
warning = _job(
|
|
"1",
|
|
validation_status=JobValidationStatus.WARNING.value,
|
|
)
|
|
session = FakeSession()
|
|
|
|
queued, skipped_count, blocked_count = _select_campaign_jobs_for_queue(
|
|
session,
|
|
jobs=[warning],
|
|
allowed_validation=_queue_validation_statuses(include_warnings=True),
|
|
reviewed_needs_review_keys=set(),
|
|
dry_run=True,
|
|
)
|
|
|
|
self.assertEqual(queued, [warning])
|
|
self.assertEqual((skipped_count, blocked_count), (0, 0))
|
|
self.assertEqual(warning.queue_status, JobQueueStatus.DRAFT.value)
|
|
self.assertEqual(warning.send_status, JobSendStatus.NOT_QUEUED.value)
|
|
self.assertEqual(session.added, [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|