From 5959c9c92a5398057d93938c9ebf12775dfb6d5a Mon Sep 17 00:00:00 2001 From: AdrianoDev Date: Fri, 17 Apr 2026 21:56:22 +0200 Subject: [PATCH] feat(models): add Station and StationRecipeAssignment models TDD: test written first, confirmed failing with ModuleNotFoundError, then model implemented; all 3 new tests pass. conftest updated to import new models so Base.metadata.create_all picks up the tables. Co-Authored-By: Claude Sonnet 4.6 --- server/models/station.py | 69 ++++++++++++++++++++++++++++++ server/tests/conftest.py | 1 + server/tests/test_station_model.py | 55 ++++++++++++++++++++++++ 3 files changed, 125 insertions(+) create mode 100644 server/models/station.py create mode 100644 server/tests/test_station_model.py diff --git a/server/models/station.py b/server/models/station.py new file mode 100644 index 0000000..5488d99 --- /dev/null +++ b/server/models/station.py @@ -0,0 +1,69 @@ +"""Station and StationRecipeAssignment models. + +A Station represents a physical control point (typically one per tablet/PC). +Recipes are assigned to stations so that each station only sees the products +it is supposed to inspect. +""" +from datetime import datetime +from typing import TYPE_CHECKING, Optional + +from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint, func +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from database import Base + +if TYPE_CHECKING: + from models.recipe import Recipe + + +class Station(Base): + __tablename__ = "stations" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + code: Mapped[str] = mapped_column(String(100), unique=True, nullable=False, index=True) + name: Mapped[str] = mapped_column(String(255), nullable=False) + location: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) + notes: Mapped[Optional[str]] = mapped_column(Text, nullable=True) + active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, index=True) + created_by: Mapped[int] = mapped_column(Integer, ForeignKey("users.id"), nullable=False) + created_at: Mapped[datetime] = mapped_column( + DateTime, nullable=False, server_default=func.now() + ) + + assignments: Mapped[list["StationRecipeAssignment"]] = relationship( + back_populates="station", cascade="all, delete-orphan", lazy="selectin" + ) + + __table_args__ = ( + {"mysql_engine": "InnoDB", "mysql_charset": "utf8mb4"}, + ) + + def __repr__(self) -> str: + return f"" + + +class StationRecipeAssignment(Base): + __tablename__ = "station_recipe_assignments" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + station_id: Mapped[int] = mapped_column( + Integer, ForeignKey("stations.id", ondelete="CASCADE"), nullable=False, index=True + ) + recipe_id: Mapped[int] = mapped_column( + Integer, ForeignKey("recipes.id", ondelete="CASCADE"), nullable=False, index=True + ) + assigned_by: Mapped[int] = mapped_column(Integer, ForeignKey("users.id"), nullable=False) + assigned_at: Mapped[datetime] = mapped_column( + DateTime, nullable=False, server_default=func.now() + ) + + station: Mapped["Station"] = relationship(back_populates="assignments") + recipe: Mapped["Recipe"] = relationship(lazy="selectin") + + __table_args__ = ( + UniqueConstraint("station_id", "recipe_id", name="uq_station_recipe"), + {"mysql_engine": "InnoDB", "mysql_charset": "utf8mb4"}, + ) + + def __repr__(self) -> str: + return f"" diff --git a/server/tests/conftest.py b/server/tests/conftest.py index b424c9f..abe4d79 100644 --- a/server/tests/conftest.py +++ b/server/tests/conftest.py @@ -39,6 +39,7 @@ from models.task import RecipeTask, RecipeSubtask from models.measurement import Measurement from models.access_log import AccessLog from models.setting import SystemSetting, RecipeVersionAudit +from models.station import Station, StationRecipeAssignment from services.auth_service import hash_password, generate_api_key # --------------------------------------------------------------------------- diff --git a/server/tests/test_station_model.py b/server/tests/test_station_model.py new file mode 100644 index 0000000..cb5daf6 --- /dev/null +++ b/server/tests/test_station_model.py @@ -0,0 +1,55 @@ +"""Test the Station and StationRecipeAssignment ORM models.""" +import pytest +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from models.station import Station, StationRecipeAssignment +from tests.conftest import _create_user, create_test_recipe + + +async def test_create_station(db_session: AsyncSession): + admin = await _create_user(db_session, username="admin1", is_admin=True) + station = Station( + code="ST-001", + name="Linea 1", + location="Reparto Nord", + created_by=admin.id, + ) + db_session.add(station) + await db_session.flush() + await db_session.refresh(station) + assert station.id is not None + assert station.active is True + assert station.created_at is not None + + +async def test_station_code_is_unique(db_session: AsyncSession): + from sqlalchemy.exc import IntegrityError + admin = await _create_user(db_session, username="admin2", is_admin=True) + db_session.add(Station(code="ST-DUP", name="A", created_by=admin.id)) + await db_session.flush() + db_session.add(Station(code="ST-DUP", name="B", created_by=admin.id)) + with pytest.raises(IntegrityError): + await db_session.flush() + await db_session.rollback() + + +async def test_assign_recipe_to_station(db_session: AsyncSession): + admin = await _create_user(db_session, username="admin3", is_admin=True) + station = Station(code="ST-002", name="Linea 2", created_by=admin.id) + db_session.add(station) + await db_session.flush() + recipe = await create_test_recipe(db_session, user_id=admin.id, code="REC-X") + assignment = StationRecipeAssignment( + station_id=station.id, recipe_id=recipe.id, assigned_by=admin.id, + ) + db_session.add(assignment) + await db_session.flush() + result = await db_session.execute( + select(StationRecipeAssignment).where( + StationRecipeAssignment.station_id == station.id + ) + ) + assignments = result.scalars().all() + assert len(assignments) == 1 + assert assignments[0].recipe_id == recipe.id