fix(models): align Station unique constraint + extend tests

- Station.code: usa UniqueConstraint("code", name="uq_stations_code")
  esplicito in __table_args__ invece di unique=True sulla colonna,
  per allineamento con la migration 002 ed evitare drift Alembic.
- Aggiunge test test_duplicate_assignment_is_rejected per coprire
  il vincolo uq_station_recipe (regola business centrale del modello).
- Sposta import IntegrityError a module-level per consistenza.

Feedback da code-reviewer su Task 2.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-04-17 22:22:14 +02:00
parent 5959c9c92a
commit e36bbbb7d7
2 changed files with 21 additions and 2 deletions
+2 -1
View File
@@ -20,7 +20,7 @@ class Station(Base):
__tablename__ = "stations" __tablename__ = "stations"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
code: Mapped[str] = mapped_column(String(100), unique=True, nullable=False, index=True) code: Mapped[str] = mapped_column(String(100), nullable=False, index=True)
name: Mapped[str] = mapped_column(String(255), nullable=False) name: Mapped[str] = mapped_column(String(255), nullable=False)
location: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) location: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)
notes: Mapped[Optional[str]] = mapped_column(Text, nullable=True) notes: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
@@ -35,6 +35,7 @@ class Station(Base):
) )
__table_args__ = ( __table_args__ = (
UniqueConstraint("code", name="uq_stations_code"),
{"mysql_engine": "InnoDB", "mysql_charset": "utf8mb4"}, {"mysql_engine": "InnoDB", "mysql_charset": "utf8mb4"},
) )
+19 -1
View File
@@ -1,6 +1,7 @@
"""Test the Station and StationRecipeAssignment ORM models.""" """Test the Station and StationRecipeAssignment ORM models."""
import pytest import pytest
from sqlalchemy import select from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from models.station import Station, StationRecipeAssignment from models.station import Station, StationRecipeAssignment
@@ -24,7 +25,6 @@ async def test_create_station(db_session: AsyncSession):
async def test_station_code_is_unique(db_session: AsyncSession): 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) admin = await _create_user(db_session, username="admin2", is_admin=True)
db_session.add(Station(code="ST-DUP", name="A", created_by=admin.id)) db_session.add(Station(code="ST-DUP", name="A", created_by=admin.id))
await db_session.flush() await db_session.flush()
@@ -53,3 +53,21 @@ async def test_assign_recipe_to_station(db_session: AsyncSession):
assignments = result.scalars().all() assignments = result.scalars().all()
assert len(assignments) == 1 assert len(assignments) == 1
assert assignments[0].recipe_id == recipe.id assert assignments[0].recipe_id == recipe.id
async def test_duplicate_assignment_is_rejected(db_session: AsyncSession):
admin = await _create_user(db_session, username="admin4", is_admin=True)
station = Station(code="ST-003", name="Linea 3", 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-Y")
db_session.add(StationRecipeAssignment(
station_id=station.id, recipe_id=recipe.id, assigned_by=admin.id,
))
await db_session.flush()
db_session.add(StationRecipeAssignment(
station_id=station.id, recipe_id=recipe.id, assigned_by=admin.id,
))
with pytest.raises(IntegrityError):
await db_session.flush()
await db_session.rollback()