style: use SQLAlchemy 2.0 select() in sync test helpers

Match the select()/scalars() convention already used throughout
app/db/repositories.py instead of the legacy Query API.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Bastian Wagner
2026-08-15 15:54:52 +02:00
parent c4e986e3f8
commit 4aaf490bc5
2 changed files with 4 additions and 3 deletions

View File

@@ -3,7 +3,7 @@ from typing import Callable
import pytest import pytest
from cryptography.fernet import Fernet from cryptography.fernet import Fernet
from sqlalchemy import create_engine from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session, sessionmaker from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool from sqlalchemy.pool import StaticPool
@@ -68,7 +68,7 @@ def settings(tmp_path: Path) -> StubSettings:
def load_only_activity(session_factory) -> Callable[[int], Activity]: def load_only_activity(session_factory) -> Callable[[int], Activity]:
def _load(user_id: int) -> Activity: def _load(user_id: int) -> Activity:
with session_factory() as session: with session_factory() as session:
activities = ActivityRepository(session).session.query(Activity).filter_by(user_id=user_id).all() activities = list(session.scalars(select(Activity).where(Activity.user_id == user_id)))
assert len(activities) == 1, f"expected exactly one activity for user {user_id}, found {len(activities)}" assert len(activities) == 1, f"expected exactly one activity for user {user_id}, found {len(activities)}"
return activities[0] return activities[0]

View File

@@ -1,6 +1,7 @@
from pathlib import Path from pathlib import Path
import pytest import pytest
from sqlalchemy import select
from app.db.models import ActivityStatus, SyncRun, SyncRunStatus, SyncUser from app.db.models import ActivityStatus, SyncRun, SyncRunStatus, SyncUser
@@ -78,7 +79,7 @@ async def test_sync_run_repository_wiring_records_run(manager, seeded_user: Sync
await manager.sync_user(seeded_user.id) await manager.sync_user(seeded_user.id)
with session_factory() as session: with session_factory() as session:
runs = session.query(SyncRun).filter_by(user_id=seeded_user.id).all() runs = list(session.scalars(select(SyncRun).where(SyncRun.user_id == seeded_user.id)))
assert len(runs) == 1 assert len(runs) == 1
run = runs[0] run = runs[0]
assert run.status == SyncRunStatus.SUCCESS assert run.status == SyncRunStatus.SUCCESS