78 lines
2.6 KiB
Python
78 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine, event
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_access.backend.db.models import Account, Group, User
|
|
from govoplan_campaign.backend.db.models import Campaign
|
|
from govoplan_campaign.backend.manifest import _tenant_summary_batch
|
|
from govoplan_core.core.change_sequence import ChangeSequenceEntry
|
|
from govoplan_core.db.base import Base
|
|
|
|
|
|
class CampaignTenantSummaryBatchTests(unittest.TestCase):
|
|
def test_batch_summary_groups_all_tenants_in_one_query(self) -> None:
|
|
engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
Base.metadata.create_all(
|
|
engine,
|
|
tables=[
|
|
Account.__table__,
|
|
User.__table__,
|
|
Group.__table__,
|
|
Campaign.__table__,
|
|
ChangeSequenceEntry.__table__,
|
|
],
|
|
)
|
|
try:
|
|
with Session(engine) as session:
|
|
session.add_all(
|
|
[
|
|
Campaign(
|
|
id="campaign-1",
|
|
tenant_id="tenant-1",
|
|
external_id="one",
|
|
name="One",
|
|
),
|
|
Campaign(
|
|
id="campaign-2",
|
|
tenant_id="tenant-1",
|
|
external_id="two",
|
|
name="Two",
|
|
),
|
|
Campaign(
|
|
id="campaign-3",
|
|
tenant_id="tenant-2",
|
|
external_id="three",
|
|
name="Three",
|
|
),
|
|
]
|
|
)
|
|
session.commit()
|
|
query_count = 0
|
|
|
|
def count_query(*_args: object) -> None:
|
|
nonlocal query_count
|
|
query_count += 1
|
|
|
|
event.listen(engine, "before_cursor_execute", count_query)
|
|
try:
|
|
counts = _tenant_summary_batch(
|
|
session,
|
|
["tenant-1", "tenant-2", "tenant-empty"],
|
|
)
|
|
finally:
|
|
event.remove(engine, "before_cursor_execute", count_query)
|
|
|
|
self.assertEqual(1, query_count)
|
|
self.assertEqual({"campaigns": 2}, counts["tenant-1"])
|
|
self.assertEqual({"campaigns": 1}, counts["tenant-2"])
|
|
self.assertNotIn("tenant-empty", counts)
|
|
finally:
|
|
engine.dispose()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|