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()