72 lines
2.1 KiB
Python
72 lines
2.1 KiB
Python
from sqlalchemy import event, text
|
|
|
|
from app.auth.services import create_user
|
|
from app.extensions import db as database
|
|
from app.tasks.models import Task
|
|
from app.tasks.services import paginate_tasks
|
|
|
|
|
|
def seed_tasks(user_id: int, count: int) -> None:
|
|
database.session.add_all(
|
|
[
|
|
Task(
|
|
user_id=user_id,
|
|
title=f"Paginated task {number:03d}",
|
|
status="not_started",
|
|
priority="normal",
|
|
)
|
|
for number in range(count)
|
|
]
|
|
)
|
|
database.session.commit()
|
|
|
|
|
|
def login(client, email: str) -> None:
|
|
client.post(
|
|
"/auth/login",
|
|
data={"email": email, "password": "secure password"},
|
|
)
|
|
|
|
|
|
def test_task_list_paginates_at_fifty_items(client, db):
|
|
user = create_user("page_user", "page@example.com", "secure password")
|
|
seed_tasks(user.id, 55)
|
|
login(client, user.email)
|
|
|
|
first = client.get("/tasks/")
|
|
second = client.get("/tasks/?page=2")
|
|
|
|
assert first.data.count(b'class="task-card ') == 50
|
|
assert b"Page 1 of 2" in first.data
|
|
assert second.data.count(b'class="task-card ') == 5
|
|
assert b"Page 2 of 2" in second.data
|
|
assert client.get("/tasks/?page=0").status_code == 400
|
|
assert client.get("/tasks/?page=word").status_code == 400
|
|
assert client.get("/tasks/?page=3").status_code == 404
|
|
|
|
|
|
def test_paginated_service_uses_bounded_query_count(app, db):
|
|
user = create_user("query_user", "query@example.com", "secure password")
|
|
seed_tasks(user.id, 55)
|
|
user_id = user.id
|
|
query_count = 0
|
|
|
|
def count_query(*_args):
|
|
nonlocal query_count
|
|
query_count += 1
|
|
|
|
event.listen(database.engine, "before_cursor_execute", count_query)
|
|
try:
|
|
page = paginate_tasks(user_id, page=1)
|
|
finally:
|
|
event.remove(database.engine, "before_cursor_execute", count_query)
|
|
|
|
assert len(page.items) == 50
|
|
assert page.total == 55
|
|
assert query_count <= 3
|
|
|
|
|
|
def test_task_visibility_index_exists(db):
|
|
indexes = database.session.execute(text("PRAGMA index_list('task')")).all()
|
|
|
|
assert "ix_task_owner_visibility_status_due" in {row[1] for row in indexes}
|