Adriano
e8cd1f05aa
Implements create/update/delete station, assign/unassign recipe, list_station_recipes (active only, ordered by code), and get_station_by_code. Explicit ORM-level assignment deletion in delete_station ensures cascade works engine-agnostically (SQLite tests + MySQL production). 10 TDD tests. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
115 lines
5.2 KiB
Python
115 lines
5.2 KiB
Python
"""Tests for station_service business logic."""
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from models.station import Station
|
|
from schemas.station import StationCreate, StationUpdate
|
|
from services.station_service import (
|
|
create_station, update_station, delete_station,
|
|
assign_recipe, unassign_recipe, list_station_recipes,
|
|
get_station_by_code,
|
|
)
|
|
from tests.conftest import _create_user, create_test_recipe
|
|
|
|
|
|
async def test_create_station_ok(db_session: AsyncSession):
|
|
admin = await _create_user(db_session, username="a1", is_admin=True)
|
|
station = await create_station(
|
|
db_session, StationCreate(code="ST-100", name="Pilot"), admin
|
|
)
|
|
assert station.id is not None
|
|
assert station.code == "ST-100"
|
|
|
|
|
|
async def test_create_station_duplicate_code(db_session: AsyncSession):
|
|
admin = await _create_user(db_session, username="a2", is_admin=True)
|
|
await create_station(db_session, StationCreate(code="ST-DUP", name="A"), admin)
|
|
with pytest.raises(HTTPException) as exc:
|
|
await create_station(db_session, StationCreate(code="ST-DUP", name="B"), admin)
|
|
assert exc.value.status_code == 409
|
|
|
|
|
|
async def test_update_station(db_session: AsyncSession):
|
|
admin = await _create_user(db_session, username="a3", is_admin=True)
|
|
station = await create_station(db_session, StationCreate(code="ST-U", name="Old"), admin)
|
|
updated = await update_station(
|
|
db_session, station.id, StationUpdate(name="New name"),
|
|
)
|
|
assert updated.name == "New name"
|
|
assert updated.code == "ST-U"
|
|
|
|
|
|
async def test_update_missing_station(db_session: AsyncSession):
|
|
with pytest.raises(HTTPException) as exc:
|
|
await update_station(db_session, 9999, StationUpdate(name="x"))
|
|
assert exc.value.status_code == 404
|
|
|
|
|
|
async def test_assign_and_list_recipes(db_session: AsyncSession):
|
|
admin = await _create_user(db_session, username="a4", is_admin=True)
|
|
station = await create_station(db_session, StationCreate(code="ST-R", name="R"), admin)
|
|
r1 = await create_test_recipe(db_session, user_id=admin.id, code="REC-R1")
|
|
r2 = await create_test_recipe(db_session, user_id=admin.id, code="REC-R2")
|
|
await assign_recipe(db_session, station.id, r1.id, admin)
|
|
await assign_recipe(db_session, station.id, r2.id, admin)
|
|
recipes = await list_station_recipes(db_session, station.id)
|
|
assert {r.code for r in recipes} == {"REC-R1", "REC-R2"}
|
|
|
|
|
|
async def test_assign_same_recipe_twice_is_409(db_session: AsyncSession):
|
|
admin = await _create_user(db_session, username="a5", is_admin=True)
|
|
station = await create_station(db_session, StationCreate(code="ST-D", name="D"), admin)
|
|
r = await create_test_recipe(db_session, user_id=admin.id, code="REC-D")
|
|
await assign_recipe(db_session, station.id, r.id, admin)
|
|
with pytest.raises(HTTPException) as exc:
|
|
await assign_recipe(db_session, station.id, r.id, admin)
|
|
assert exc.value.status_code == 409
|
|
|
|
|
|
async def test_unassign_recipe(db_session: AsyncSession):
|
|
admin = await _create_user(db_session, username="a6", is_admin=True)
|
|
station = await create_station(db_session, StationCreate(code="ST-UN", name="UN"), admin)
|
|
r = await create_test_recipe(db_session, user_id=admin.id, code="REC-UN")
|
|
await assign_recipe(db_session, station.id, r.id, admin)
|
|
await unassign_recipe(db_session, station.id, r.id)
|
|
recipes = await list_station_recipes(db_session, station.id)
|
|
assert recipes == []
|
|
|
|
|
|
async def test_get_station_by_code(db_session: AsyncSession):
|
|
admin = await _create_user(db_session, username="a7", is_admin=True)
|
|
await create_station(db_session, StationCreate(code="ST-FIND", name="F"), admin)
|
|
found = await get_station_by_code(db_session, "ST-FIND")
|
|
assert found is not None
|
|
assert found.name == "F"
|
|
missing = await get_station_by_code(db_session, "ST-NOPE")
|
|
assert missing is None
|
|
|
|
|
|
async def test_list_recipes_only_returns_active(db_session: AsyncSession):
|
|
admin = await _create_user(db_session, username="a8", is_admin=True)
|
|
station = await create_station(db_session, StationCreate(code="ST-A", name="A"), admin)
|
|
active = await create_test_recipe(db_session, user_id=admin.id, code="REC-AC")
|
|
inactive = await create_test_recipe(db_session, user_id=admin.id, code="REC-IN")
|
|
inactive.active = False
|
|
await db_session.flush()
|
|
await assign_recipe(db_session, station.id, active.id, admin)
|
|
await assign_recipe(db_session, station.id, inactive.id, admin)
|
|
recipes = await list_station_recipes(db_session, station.id)
|
|
assert [r.code for r in recipes] == ["REC-AC"]
|
|
|
|
|
|
async def test_delete_station_cascades_assignments(db_session: AsyncSession):
|
|
from sqlalchemy import select
|
|
from models.station import StationRecipeAssignment
|
|
admin = await _create_user(db_session, username="a9", is_admin=True)
|
|
station = await create_station(db_session, StationCreate(code="ST-DEL", name="D"), admin)
|
|
r = await create_test_recipe(db_session, user_id=admin.id, code="REC-DEL")
|
|
await assign_recipe(db_session, station.id, r.id, admin)
|
|
await delete_station(db_session, station.id)
|
|
remaining = await db_session.execute(
|
|
select(StationRecipeAssignment).where(StationRecipeAssignment.station_id == station.id)
|
|
)
|
|
assert remaining.scalars().all() == []
|