feat(services): add station_service with CRUD and assignment logic
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>
This commit is contained in:
@@ -0,0 +1,114 @@
|
||||
"""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() == []
|
||||
Reference in New Issue
Block a user