zemion
2be4a0c598
Module Package Release / publish-packages (push) Successful in 13s
Release v0.1.23. Coordinated integrity review: GovOPlaN/govoplan-core#298.
95 lines
4.2 KiB
Python
Executable File
95 lines
4.2 KiB
Python
Executable File
from __future__ import annotations
|
|
|
|
from dataclasses import replace
|
|
import unittest
|
|
|
|
from sqlalchemy import event, select
|
|
from sqlalchemy.dialects import postgresql
|
|
|
|
import test_ticket_service as fixture
|
|
from govoplan_tickets.backend import service
|
|
from govoplan_tickets.backend.db.models import Ticket
|
|
|
|
|
|
class TicketListQueryTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.fixture = fixture.TicketServiceTests()
|
|
self.fixture.setUp()
|
|
self.engine = self.fixture.Session.kw["bind"]
|
|
with self.fixture.Session() as session:
|
|
for index in range(50):
|
|
service.create_ticket(
|
|
session, self.fixture.manager,
|
|
record=replace(fixture._record(reporter="reporter-1"), ticket_id=f"ticket-{index:03}", ticket_number=f"TKT-{index:03}"),
|
|
idempotency_key=f"fixture-{index}",
|
|
)
|
|
session.commit()
|
|
|
|
def tearDown(self):
|
|
self.engine.dispose()
|
|
|
|
def test_one_row_page_loads_one_row_with_exact_total_and_stable_ties(self):
|
|
with self.fixture.Session() as session:
|
|
session.query(Ticket).update({Ticket.updated_at: fixture.NOW})
|
|
session.commit()
|
|
loaded = []
|
|
queries = []
|
|
|
|
def row_loaded(target, context):
|
|
loaded.append(target.id)
|
|
|
|
def executed(conn, cursor, statement, parameters, context, executemany):
|
|
if statement.lstrip().upper().startswith("SELECT"):
|
|
queries.append(statement)
|
|
|
|
event.listen(Ticket, "load", row_loaded)
|
|
event.listen(self.engine, "before_cursor_execute", executed)
|
|
try:
|
|
with self.fixture.Session() as session:
|
|
page, total = service.list_tickets(session, self.fixture.manager, offset=1, limit=1)
|
|
self.assertEqual(50, total)
|
|
self.assertEqual(["ticket-001"], [item.ticket_id for item in page])
|
|
self.assertEqual(["ticket-001"], loaded)
|
|
self.assertEqual(2, len(queries))
|
|
self.assertIn("LIMIT", queries[-1])
|
|
finally:
|
|
event.remove(Ticket, "load", row_loaded)
|
|
event.remove(self.engine, "before_cursor_execute", executed)
|
|
|
|
def test_count_and_page_match_current_object_authorization(self):
|
|
viewer = fixture._Principal("viewer-1", {service.READ_SCOPE})
|
|
viewer.group_ids = ("group-1",)
|
|
with self.fixture.Session() as session:
|
|
rows = session.query(Ticket).order_by(Ticket.id).all()
|
|
rows[0].visibility = "tenant"
|
|
rows[1].assignee = {"kind": "account", "id": viewer.account_id}
|
|
rows[2].reporter = {"kind": "account", "id": viewer.account_id}
|
|
rows[3].requester = {"kind": "account", "id": viewer.account_id}
|
|
rows[4].participants = [{"kind": "group", "id": "group-1"}]
|
|
rows[5].created_by = viewer.account_id
|
|
rows[6].participants = ['{"kind":"account","id":"viewer-1"}']
|
|
rows[7].tenant_id = "tenant-2"
|
|
rows[7].visibility = "tenant"
|
|
session.commit()
|
|
expected = {row.id for row in rows if service._can_read_row(viewer, row)}
|
|
actual = set()
|
|
for offset in range(len(expected)):
|
|
page, total = service.list_tickets(session, viewer, offset=offset, limit=1)
|
|
self.assertEqual(len(expected), total)
|
|
self.assertEqual(1, len(page))
|
|
actual.add(page[0].ticket_id)
|
|
self.assertEqual(expected, actual)
|
|
self.assertEqual(6, len(actual))
|
|
rows[4].participants = []
|
|
session.commit()
|
|
self.assertEqual(5, service.list_tickets(session, viewer, limit=1)[1])
|
|
viewer.scopes = frozenset()
|
|
self.assertEqual(((), 0), service.list_tickets(session, viewer, limit=1))
|
|
|
|
def test_read_predicate_compiles_for_postgresql_with_bound_subjects(self):
|
|
viewer = fixture._Principal("viewer' OR 1=1 --", {service.READ_SCOPE})
|
|
compiled = select(Ticket.id).where(service._ticket_read_predicate(viewer)).compile(dialect=postgresql.dialect())
|
|
self.assertIn("json_array_elements", str(compiled))
|
|
self.assertNotIn(viewer.account_id, str(compiled))
|
|
self.assertIn(viewer.account_id, compiled.params.values())
|