201 lines
7.8 KiB
Python
201 lines
7.8 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_access.backend.db.models import Account, Group, User
|
|
from govoplan_campaign.backend.db.models import (
|
|
Campaign,
|
|
CampaignJob,
|
|
CampaignStatus,
|
|
CampaignVersion,
|
|
JobBuildStatus,
|
|
JobQueueStatus,
|
|
JobSendStatus,
|
|
JobValidationStatus,
|
|
)
|
|
from govoplan_campaign.backend.sending.jobs import (
|
|
QueueingError,
|
|
cancel_campaign_jobs,
|
|
pause_campaign_jobs,
|
|
resume_campaign_jobs,
|
|
)
|
|
from govoplan_campaign.backend.reports.campaigns import _campaign_report_cards, _version_info
|
|
from govoplan_core.core.change_sequence import ChangeSequenceEntry, ChangeSequenceRetentionFloor
|
|
from govoplan_core.db.base import Base
|
|
|
|
|
|
class CampaignQueueControlTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
Base.metadata.create_all(
|
|
self.engine,
|
|
tables=[
|
|
Account.__table__,
|
|
User.__table__,
|
|
Group.__table__,
|
|
ChangeSequenceEntry.__table__,
|
|
ChangeSequenceRetentionFloor.__table__,
|
|
Campaign.__table__,
|
|
CampaignVersion.__table__,
|
|
CampaignJob.__table__,
|
|
],
|
|
)
|
|
self.session = Session(self.engine)
|
|
self._add_campaign("campaign-1", "tenant-1", "version-1")
|
|
self._add_campaign("campaign-2", "tenant-2", "version-2")
|
|
self._add_job("queued", queue="queued", send="queued")
|
|
self._add_job("paused", queue="paused", send="queued")
|
|
self._add_job("failed", queue="draft", send="failed_temporary")
|
|
self._add_job("unknown", queue="draft", send="outcome_unknown")
|
|
self._add_job("accepted", queue="draft", send="smtp_accepted")
|
|
self._add_job("claimed", queue="sending", send="claimed")
|
|
self._add_job(
|
|
"other-tenant",
|
|
tenant_id="tenant-2",
|
|
campaign_id="campaign-2",
|
|
version_id="version-2",
|
|
queue="queued",
|
|
send="queued",
|
|
)
|
|
self.session.commit()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_pause_and_resume_change_only_valid_jobs_in_the_selected_tenant(self) -> None:
|
|
with patch("govoplan_campaign.backend.sending.jobs._emit_campaign_status_notification"):
|
|
paused = pause_campaign_jobs(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
campaign_id="campaign-1",
|
|
)
|
|
resumed = resume_campaign_jobs(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
campaign_id="campaign-1",
|
|
enqueue_celery=False,
|
|
)
|
|
|
|
self.assertEqual(paused["paused_count"], 1)
|
|
self.assertEqual(resumed, {
|
|
"campaign_id": "campaign-1",
|
|
"resumed_count": 2,
|
|
"enqueued_count": 0,
|
|
})
|
|
self.session.expire_all()
|
|
self.assertEqual(self.session.get(Campaign, "campaign-1").status, CampaignStatus.QUEUED.value)
|
|
version = self.session.get(CampaignVersion, "version-1")
|
|
self.assertEqual(version.delivery_mode, "database_queue")
|
|
self.assertIsNotNone(version.delivery_mode_selected_at)
|
|
self.assertEqual(self.session.get(CampaignJob, "queued").queue_status, JobQueueStatus.QUEUED.value)
|
|
self.assertEqual(self.session.get(CampaignJob, "paused").queue_status, JobQueueStatus.QUEUED.value)
|
|
self.assertEqual(self.session.get(CampaignJob, "other-tenant").queue_status, JobQueueStatus.QUEUED.value)
|
|
|
|
def test_cancel_is_auditable_idempotent_state_change_and_protects_started_effects(self) -> None:
|
|
with patch("govoplan_campaign.backend.sending.jobs._emit_campaign_status_notification"):
|
|
first = cancel_campaign_jobs(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
campaign_id="campaign-1",
|
|
)
|
|
second = cancel_campaign_jobs(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
campaign_id="campaign-1",
|
|
)
|
|
|
|
self.assertEqual(first["cancelled_count"], 3)
|
|
self.assertEqual(first["protected_count"], 3)
|
|
self.assertEqual(second["cancelled_count"], 0)
|
|
self.assertEqual(second["protected_count"], 3)
|
|
self.session.expire_all()
|
|
for job_id in ("queued", "paused", "failed"):
|
|
job = self.session.get(CampaignJob, job_id)
|
|
self.assertEqual(job.queue_status, JobQueueStatus.CANCELLED.value)
|
|
self.assertEqual(job.send_status, JobSendStatus.CANCELLED.value)
|
|
self.assertEqual(self.session.get(CampaignJob, "unknown").send_status, JobSendStatus.OUTCOME_UNKNOWN.value)
|
|
self.assertEqual(self.session.get(CampaignJob, "accepted").send_status, JobSendStatus.SMTP_ACCEPTED.value)
|
|
self.assertEqual(self.session.get(CampaignJob, "claimed").send_status, JobSendStatus.CLAIMED.value)
|
|
self.assertEqual(self.session.get(CampaignJob, "other-tenant").send_status, JobSendStatus.QUEUED.value)
|
|
|
|
def test_controls_fail_closed_for_a_campaign_owned_by_another_tenant(self) -> None:
|
|
for control in (pause_campaign_jobs, resume_campaign_jobs, cancel_campaign_jobs):
|
|
with self.subTest(control=control.__name__), self.assertRaises(QueueingError):
|
|
control(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
campaign_id="campaign-2",
|
|
)
|
|
|
|
def test_queue_projection_exposes_exact_action_counts_and_persisted_mode_only(self) -> None:
|
|
version = self.session.get(CampaignVersion, "version-1")
|
|
version.delivery_mode = "worker_queue"
|
|
version.delivery_mode_selected_at = version.updated_at
|
|
jobs = (
|
|
self.session.query(CampaignJob)
|
|
.filter(CampaignJob.tenant_id == "tenant-1")
|
|
.all()
|
|
)
|
|
|
|
cards = _campaign_report_cards(version, jobs)
|
|
projected_version = _version_info(version)
|
|
|
|
self.assertEqual(cards["retryable"], 1)
|
|
self.assertEqual(cards["queueable_unattempted"], 0)
|
|
self.assertEqual(cards["cancellable"], 3)
|
|
self.assertEqual(projected_version["delivery_mode"], "worker_queue")
|
|
self.assertIn("delivery_mode_selected_at", projected_version)
|
|
self.assertNotIn("execution_snapshot", projected_version)
|
|
|
|
def _add_campaign(self, campaign_id: str, tenant_id: str, version_id: str) -> None:
|
|
self.session.add(Campaign(
|
|
id=campaign_id,
|
|
tenant_id=tenant_id,
|
|
external_id=f"external-{campaign_id}",
|
|
name=campaign_id,
|
|
status=CampaignStatus.QUEUED.value,
|
|
current_version_id=version_id,
|
|
settings={},
|
|
mail_profile_policy={},
|
|
))
|
|
self.session.add(CampaignVersion(
|
|
id=version_id,
|
|
campaign_id=campaign_id,
|
|
version_number=1,
|
|
raw_json={},
|
|
))
|
|
|
|
def _add_job(
|
|
self,
|
|
job_id: str,
|
|
*,
|
|
queue: str,
|
|
send: str,
|
|
tenant_id: str = "tenant-1",
|
|
campaign_id: str = "campaign-1",
|
|
version_id: str = "version-1",
|
|
) -> None:
|
|
self.session.add(CampaignJob(
|
|
id=job_id,
|
|
tenant_id=tenant_id,
|
|
campaign_id=campaign_id,
|
|
campaign_version_id=version_id,
|
|
entry_index=len(self.session.new),
|
|
entry_id=f"entry-{job_id}",
|
|
build_status=JobBuildStatus.BUILT.value,
|
|
validation_status=JobValidationStatus.READY.value,
|
|
queue_status=queue,
|
|
send_status=send,
|
|
resolved_attachments=[],
|
|
issues_snapshot=[],
|
|
))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|