feat: add Playwright configuration and initial e2e test for authentication

- Created Playwright configuration file to set up testing environment.
- Added a new e2e test for user authentication in login.spec.ts.
- Updated tsconfig.node.json to include playwright.config.ts.
- Enhanced vite.config.ts to include API proxying for backend integration.
- Added a placeholder for last run test results in .last-run.json.
This commit is contained in:
2025-10-11 17:25:38 +02:00
parent 0c405ee6ca
commit 1099a738a3
17 changed files with 558 additions and 14 deletions

View File

@@ -0,0 +1,62 @@
from __future__ import annotations
from collections.abc import Callable
from typing import Optional
from sqlalchemy.orm import Session
from backend.app.db.session import SessionLocal
from backend.app.repositories import (
StationRepository,
TrainRepository,
TrainScheduleRepository,
TrackRepository,
UserRepository,
)
class SqlAlchemyUnitOfWork:
"""Coordinate transactional work across repositories."""
def __init__(self, session_factory: Callable[[], Session] = SessionLocal) -> None:
self._session_factory = session_factory
self.session: Optional[Session] = None
self._committed = False
self.users: UserRepository
self.stations: StationRepository
self.tracks: TrackRepository
self.trains: TrainRepository
self.train_schedules: TrainScheduleRepository
def __enter__(self) -> "SqlAlchemyUnitOfWork":
self.session = self._session_factory()
self.users = UserRepository(self.session)
self.stations = StationRepository(self.session)
self.tracks = TrackRepository(self.session)
self.trains = TrainRepository(self.session)
self.train_schedules = TrainScheduleRepository(self.session)
return self
def __exit__(self, exc_type, exc, _tb) -> None:
try:
if exc:
self.rollback()
elif not self._committed:
self.commit()
finally:
if self.session is not None:
self.session.close()
self.session = None
self._committed = False
def commit(self) -> None:
if self.session is None:
raise RuntimeError("Unit of work is not active")
self.session.commit()
self._committed = True
def rollback(self) -> None:
if self.session is None:
return
self.session.rollback()
self._committed = False

View File

@@ -12,8 +12,10 @@ from .base import (
StationModel,
TrackCreate,
TrackModel,
TrainScheduleCreate,
TrainCreate,
TrainModel,
UserCreate,
to_camel,
)
@@ -29,7 +31,9 @@ __all__ = [
"StationModel",
"TrackCreate",
"TrackModel",
"TrainScheduleCreate",
"TrainCreate",
"TrainModel",
"UserCreate",
"to_camel",
]

View File

@@ -85,3 +85,21 @@ class TrainCreate(CamelModel):
operator_id: str | None = None
home_station_id: str | None = None
consist: str | None = None
class TrainScheduleCreate(CamelModel):
train_id: str
station_id: str
sequence_index: int
scheduled_arrival: datetime | None = None
scheduled_departure: datetime | None = None
dwell_seconds: int | None = None
class UserCreate(CamelModel):
username: str
password_hash: str
email: str | None = None
full_name: str | None = None
role: str = "player"
preferences: str | None = None

View File

@@ -1,11 +1,15 @@
"""Repository abstractions for database access."""
from backend.app.repositories.stations import StationRepository
from backend.app.repositories.train_schedules import TrainScheduleRepository
from backend.app.repositories.tracks import TrackRepository
from backend.app.repositories.trains import TrainRepository
from backend.app.repositories.users import UserRepository
__all__ = [
"StationRepository",
"TrainScheduleRepository",
"TrackRepository",
"TrainRepository",
"UserRepository",
]

View File

@@ -0,0 +1,45 @@
from __future__ import annotations
import sqlalchemy as sa
from uuid import UUID
from sqlalchemy.orm import Session
from backend.app.db.models import TrainSchedule
from backend.app.models import TrainScheduleCreate
from backend.app.repositories.base import BaseRepository
class TrainScheduleRepository(BaseRepository[TrainSchedule]):
"""Persistence operations for train timetables."""
model = TrainSchedule
def __init__(self, session: Session) -> None:
super().__init__(session)
@staticmethod
def _ensure_uuid(value: UUID | str) -> UUID:
if isinstance(value, UUID):
return value
return UUID(str(value))
def list_for_train(self, train_id: UUID | str) -> list[TrainSchedule]:
identifier = self._ensure_uuid(train_id)
statement = (
sa.select(self.model)
.where(self.model.train_id == identifier)
.order_by(self.model.sequence_index.asc())
)
return list(self.session.scalars(statement))
def create(self, data: TrainScheduleCreate) -> TrainSchedule:
schedule = TrainSchedule(
train_id=self._ensure_uuid(data.train_id),
station_id=self._ensure_uuid(data.station_id),
sequence_index=data.sequence_index,
scheduled_arrival=data.scheduled_arrival,
scheduled_departure=data.scheduled_departure,
dwell_seconds=data.dwell_seconds,
)
self.session.add(schedule)
return schedule

View File

@@ -0,0 +1,37 @@
from __future__ import annotations
import sqlalchemy as sa
from sqlalchemy.orm import Session
from backend.app.db.models import User
from backend.app.models import UserCreate
from backend.app.repositories.base import BaseRepository
class UserRepository(BaseRepository[User]):
"""Data access helpers for user accounts."""
model = User
def __init__(self, session: Session) -> None:
super().__init__(session)
def get_by_username(self, username: str) -> User | None:
statement = sa.select(self.model).where(sa.func.lower(self.model.username) == username.lower())
return self.session.scalar(statement)
def list_recent(self, limit: int = 50) -> list[User]:
statement = sa.select(self.model).order_by(self.model.created_at.desc()).limit(limit)
return list(self.session.scalars(statement))
def create(self, data: UserCreate) -> User:
user = User(
username=data.username,
email=data.email,
full_name=data.full_name,
password_hash=data.password_hash,
role=data.role,
preferences=data.preferences,
)
self.session.add(user)
return user

View File

@@ -12,6 +12,7 @@ except ImportError: # pragma: no cover - allow running without shapely at impor
Point = None # type: ignore[assignment]
from sqlalchemy.orm import Session
from sqlalchemy.exc import SQLAlchemyError
from backend.app.models import StationModel, TrackModel, TrainModel
from backend.app.repositories import StationRepository, TrackRepository, TrainRepository
@@ -94,9 +95,13 @@ def get_network_snapshot(session: Session) -> dict[str, list[dict[str, object]]]
track_repo = TrackRepository(session)
train_repo = TrainRepository(session)
stations_entities = station_repo.list_active()
tracks_entities = track_repo.list_all()
trains_entities = train_repo.list_all()
try:
stations_entities = station_repo.list_active()
tracks_entities = track_repo.list_all()
trains_entities = train_repo.list_all()
except SQLAlchemyError:
session.rollback()
return _fallback_snapshot()
if not stations_entities and not tracks_entities and not trains_entities:
return _fallback_snapshot()

View File

@@ -1,17 +1,40 @@
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, List
from datetime import datetime, timezone
from typing import Any, List, cast
from uuid import uuid4
import pytest
from backend.app.models import StationCreate, TrackCreate, TrainCreate
from backend.app.repositories import StationRepository, TrackRepository, TrainRepository
from backend.app.db.models import TrainSchedule, User
from backend.app.db.unit_of_work import SqlAlchemyUnitOfWork
from backend.app.models import (
StationCreate,
TrackCreate,
TrainCreate,
TrainScheduleCreate,
UserCreate,
)
from backend.app.repositories import (
StationRepository,
TrackRepository,
TrainRepository,
TrainScheduleRepository,
UserRepository,
)
from sqlalchemy.orm import Session
@dataclass
class DummySession:
added: List[Any] = field(default_factory=list)
scalars_result: List[Any] = field(default_factory=list)
scalar_result: Any = None
statements: List[Any] = field(default_factory=list)
committed: bool = False
rolled_back: bool = False
closed: bool = False
def add(self, instance: Any) -> None:
self.added.append(instance)
@@ -19,12 +42,26 @@ class DummySession:
def add_all(self, instances: list[Any]) -> None:
self.added.extend(instances)
def scalars(self, _statement: Any) -> list[Any]: # pragma: no cover - not used here
return []
def scalars(self, statement: Any) -> list[Any]:
self.statements.append(statement)
return list(self.scalars_result)
def scalar(self, statement: Any) -> Any:
self.statements.append(statement)
return self.scalar_result
def flush(self, _objects: list[Any] | None = None) -> None: # pragma: no cover - optional
return None
def commit(self) -> None: # pragma: no cover - optional
self.committed = True
def rollback(self) -> None: # pragma: no cover - optional
self.rolled_back = True
def close(self) -> None: # pragma: no cover - optional
self.closed = True
def test_station_repository_create_generates_geometry() -> None:
session = DummySession()
@@ -104,3 +141,96 @@ def test_train_repository_create_supports_optional_ids() -> None:
assert train.designation == "ICE 123"
assert str(train.home_station_id).endswith("1")
assert train.operator_id is None
def test_user_repository_create_persists_user() -> None:
session = DummySession()
repo = UserRepository(session) # type: ignore[arg-type]
user = repo.create(
UserCreate(
username="demo",
password_hash="hashed",
email="demo@example.com",
full_name="Demo Engineer",
role="admin",
)
)
assert session.added and session.added[0] is user
assert user.username == "demo"
assert user.role == "admin"
def test_user_repository_get_by_username_is_case_insensitive() -> None:
existing = User(username="Demo", password_hash="hashed", role="player")
session = DummySession(scalar_result=existing)
repo = UserRepository(session) # type: ignore[arg-type]
result = repo.get_by_username("demo")
assert result is existing
assert session.statements
def test_train_schedule_repository_create_converts_identifiers() -> None:
session = DummySession()
repo = TrainScheduleRepository(session) # type: ignore[arg-type]
train_id = uuid4()
station_id = uuid4()
schedule = repo.create(
TrainScheduleCreate(
train_id=str(train_id),
station_id=str(station_id),
sequence_index=1,
scheduled_arrival=datetime.now(timezone.utc),
dwell_seconds=90,
)
)
assert session.added and session.added[0] is schedule
assert schedule.train_id == train_id
assert schedule.station_id == station_id
def test_train_schedule_repository_list_for_train_orders_results() -> None:
train_id = uuid4()
schedules = [
TrainSchedule(train_id=train_id, station_id=uuid4(), sequence_index=2),
TrainSchedule(train_id=train_id, station_id=uuid4(), sequence_index=1),
]
session = DummySession(scalars_result=schedules)
repo = TrainScheduleRepository(session) # type: ignore[arg-type]
result = repo.list_for_train(train_id)
assert result == schedules
statement = session.statements[-1]
assert getattr(statement, "_order_by_clauses", ())
def test_unit_of_work_commits_and_closes_session() -> None:
session = DummySession()
uow = SqlAlchemyUnitOfWork(lambda: cast(Session, session))
with uow as active:
active.users.create(
UserCreate(username="demo", password_hash="hashed")
)
active.commit()
assert session.committed
assert session.closed
def test_unit_of_work_rolls_back_on_exception() -> None:
session = DummySession()
uow = SqlAlchemyUnitOfWork(lambda: cast(Session, session))
with pytest.raises(RuntimeError):
with uow:
raise RuntimeError("boom")
assert session.rolled_back
assert session.closed