from __future__ import annotations import unittest from types import SimpleNamespace from unittest.mock import patch from fastapi import HTTPException 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, CampaignVersion from govoplan_campaign.backend.services.job_queries import ( _campaign_jobs_grid_filter_expressions, _campaign_jobs_ordering, _campaign_jobs_page_response, _campaign_jobs_query_context, _public_recipient_groups, ) from govoplan_core.core.change_sequence import ChangeSequenceEntry from govoplan_core.db.base import Base class CampaignJobListQueryTests(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__, Campaign.__table__, CampaignVersion.__table__, CampaignJob.__table__, ChangeSequenceEntry.__table__, ], ) self.session = Session(self.engine) rows = [ (0, "ordinary-0@example.test", "General notice", "ready", "draft", "not_queued", "not_requested", 0), (1, "ordinary-1@example.test", "General notice", "ready", "draft", "not_queued", "not_requested", 1), (2, "target-c@example.test", "Target notice C", "warning", "queued", "queued", "pending", 2), (3, "ordinary-3@example.test", "General notice", "blocked", "draft", "failed_permanent", "failed", 3), (4, "target-b@example.test", "Target notice B", "ready", "draft", "failed_temporary", "not_requested", 4), (5, "target-a@example.test", "Target notice A", "ready", "draft", "outcome_unknown", "outcome_unknown", 5), (6, "excluded@example.test", "Excluded notice", "excluded", "draft", "skipped", "skipped", 0), ] for entry_index, recipient, subject, validation, queue, send, imap, attempts in rows: self.session.add(CampaignJob( id=f"job-{entry_index}", tenant_id="tenant-1", campaign_id="campaign-1", campaign_version_id="version-1", entry_index=entry_index, entry_id=f"entry-{entry_index}", recipient_email=recipient, subject=subject, message_id_header=f"", eml_sha256=f"sha-{entry_index}", build_status="built", validation_status=validation, queue_status=queue, send_status=send, imap_status=imap, attempt_count=attempts, resolved_attachments=[], issues_snapshot=[], )) self.session.commit() def tearDown(self) -> None: self.session.close() self.engine.dispose() def test_grid_filters_apply_before_pagination_and_report_filtered_totals(self) -> None: base_filters = [CampaignJob.tenant_id == "tenant-1", CampaignJob.campaign_id == "campaign-1"] grid_filters = {"recipient": "target"} filtered = [*base_filters, *_campaign_jobs_grid_filter_expressions(grid_filters)] page = _campaign_jobs_page_response( self.session, campaign_id="campaign-1", version_id="version-1", base_filters=base_filters, filtered=filtered, reviewed_keys=set(), review_metadata={}, page=1, page_size=2, grid_filters=grid_filters, sort_by="recipient", sort_direction="asc", ) self.assertEqual(page.total, 3) self.assertEqual(page.total_unfiltered, 7) self.assertEqual(page.pages, 2) self.assertEqual( [row["recipient_email"] for row in page.jobs], ["target-a@example.test", "target-b@example.test"], ) self.assertIsNone(page.next_cursor) def test_list_and_integer_filters_share_the_full_backend_query(self) -> None: expressions = _campaign_jobs_grid_filter_expressions({ "send": 'list:["failed_temporary","outcome_unknown"]', "attempts": "gte:4", }) rows = ( self.session.query(CampaignJob) .filter(*expressions) .order_by(*_campaign_jobs_ordering("attempts", "desc")) .all() ) self.assertEqual([row.id for row in rows], ["job-5", "job-4"]) def test_invalid_list_filters_fail_closed(self) -> None: with self.assertRaises(HTTPException) as raised: _campaign_jobs_grid_filter_expressions({"send": 'list:["not-a-status"]'}) self.assertEqual(raised.exception.status_code, 422) def test_recipient_projection_preserves_each_group_order_without_extra_fields(self) -> None: recipients = {"to": [{"email": "second@example.test", "name": "Second", "secret": "never-project"}, {"email": "first@example.test"}], "cc": [{"email": "copy@example.test"}], "bcc": [{"email": "blind@example.test"}], "from": {"email": "sender@example.test"}, "private": "never-project"} result = _public_recipient_groups(recipients) self.assertEqual(list(result), ["to", "cc", "bcc"]) self.assertEqual([value["email"] for value in result["to"]], ["second@example.test", "first@example.test"]) self.assertNotIn("never-project", str(result)) def test_each_frozen_recipient_group_is_searchable_before_pagination(self) -> None: row = self.session.get(CampaignJob, "job-0") row.resolved_recipients = {"to": [{"email": "second-to@example.test"}], "cc": [{"email": "copy@example.test"}], "bcc": [{"email": "blind@example.test"}]} self.session.commit() for recipient in ("second-to", "copy@", "blind@"): with self.subTest(recipient=recipient): filters = _campaign_jobs_grid_filter_expressions({"recipient": recipient}) page = _campaign_jobs_page_response(self.session, campaign_id="campaign-1", version_id="version-1", base_filters=[CampaignJob.tenant_id == "tenant-1"], filtered=filters, reviewed_keys=set(), review_metadata={}, page=1, page_size=1, grid_filters={"recipient": recipient}) self.assertEqual(page.total, 1) self.assertEqual(page.jobs[0]["id"], "job-0") self.assertEqual(page.jobs[0]["resolved_recipients"]["bcc"], [{"email": "blind@example.test"}]) def test_free_search_also_matches_additional_recipients_and_keeps_tenant_scope(self) -> None: row = self.session.get(CampaignJob, "job-0") row.resolved_recipients = {"bcc": [{"email": "additional@example.test"}]} other = self.session.get(CampaignJob, "job-1") other.tenant_id = "other-tenant" other.resolved_recipients = row.resolved_recipients self.session.commit() principal = SimpleNamespace(tenant_id="tenant-1", has=lambda scope: scope == "campaigns:recipient:read") with patch("govoplan_campaign.backend.services.job_queries._get_campaign_for_principal"), \ patch("govoplan_campaign.backend.services.job_queries._get_campaign_for_tenant", return_value=SimpleNamespace(id="campaign-1")), \ patch("govoplan_campaign.backend.services.job_queries._review_metadata", return_value=({}, set())): _, _, filters, _, _ = _campaign_jobs_query_context(self.session, principal, campaign_id="campaign-1", version_id=None, send_status=None, validation_status=None, imap_status=None, query_text="additional@example.test") self.assertEqual([job.id for job in self.session.query(CampaignJob).filter(*filters)], ["job-0"]) def test_recipient_read_is_required_before_additional_addresses_are_queried(self) -> None: principal = SimpleNamespace(tenant_id="tenant-1", has=lambda _scope: False) with patch("govoplan_campaign.backend.services.job_queries._get_campaign_for_principal"), \ patch("govoplan_campaign.backend.services.job_queries._get_campaign_for_tenant") as lookup: with self.assertRaises(HTTPException) as raised: _campaign_jobs_query_context(self.session, principal, campaign_id="campaign-1", version_id=None, send_status=None, validation_status=None, imap_status=None, query_text="blind@example.test") self.assertEqual(raised.exception.status_code, 403) lookup.assert_not_called() def test_skipped_transport_filters_and_counts_remain_separate(self) -> None: base_filters = [CampaignJob.tenant_id == "tenant-1", CampaignJob.campaign_id == "campaign-1"] grid_filters = {"send": 'list:["skipped"]', "imap": 'list:["skipped"]'} filtered = [*base_filters, *_campaign_jobs_grid_filter_expressions(grid_filters)] page = _campaign_jobs_page_response( self.session, campaign_id="campaign-1", version_id="version-1", base_filters=base_filters, filtered=filtered, reviewed_keys=set(), review_metadata={}, page=1, page_size=20, grid_filters=grid_filters, ) self.assertEqual([row["id"] for row in page.jobs], ["job-6"]) self.assertEqual(page.counts["send"]["skipped"], 1) self.assertEqual(page.counts["send"]["not_queued"], 2) self.assertEqual(page.counts["imap"]["skipped"], 1) self.assertEqual(page.filtered_counts["send"], {"skipped": 1}) self.assertEqual(page.filtered_counts["imap"], {"skipped": 1}) if __name__ == "__main__": unittest.main()