124 lines
4.7 KiB
Python
124 lines
4.7 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
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.router import (
|
|
_campaign_jobs_grid_filter_expressions,
|
|
_campaign_jobs_ordering,
|
|
_campaign_jobs_page_response,
|
|
)
|
|
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),
|
|
]
|
|
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"<message-{entry_index}@example.test>",
|
|
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, 6)
|
|
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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|