fix(scripts): guardie di seed_collaudo.py per produzione e cancellazioni (#17)

--supervisor-password non ha piu' un default committato: se il
capoturno va creato e l'opzione manca, ne genera una casuale
(secrets.token_urlsafe(12)) e la stampa una sola volta.

--conferma-produzione richiesto quando APP_ENV (src.backend.config,
stessa regola di #33) non e' dev/test.

--replace elenca quante Measurement/ProductionRun/ProductionEvent
cancellerebbe e si ferma senza cancellare finche' non si aggiunge --si.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-21 22:45:14 +02:00
parent 54bc384650
commit 8864b2021b
2 changed files with 280 additions and 11 deletions
+90 -11
View File
@@ -21,16 +21,28 @@ are already in place:
Re-running finds the recipes already there and stops. --replace deletes them Re-running finds the recipes already there and stops. --replace deletes them
first, along with anything measured against them - which is what you want between first, along with anything measured against them - which is what you want between
one acceptance session and the next, and never what you want by accident. one acceptance session and the next, and never what you want by accident: it
prints how many Measurement/ProductionRun/ProductionEvent rows that is and
refuses to touch them without --si.
Refuses to run against a real deployment (APP_ENV outside dev/test, same rule
as src.backend.config) unless --conferma-produzione is also given: this script
deletes data, and a shop floor is not where to find that out by mistake.
If --supervisor-password is not given and the capoturno account does not exist
yet, a random one is generated and printed once - never committed, never a
default anyone could look up.
""" """
import argparse import argparse
import asyncio import asyncio
import secrets
import sys import sys
from sqlalchemy import delete, select from sqlalchemy import delete, func, select
sys.path.insert(0, "/app") sys.path.insert(0, "/app")
from src.backend.config import settings # noqa: E402
from src.backend.database import async_session_factory # noqa: E402 from src.backend.database import async_session_factory # noqa: E402
from src.backend.models.orm.measurement import Measurement # noqa: E402 from src.backend.models.orm.measurement import Measurement # noqa: E402
from src.backend.models.orm.production import ( # noqa: E402 from src.backend.models.orm.production import ( # noqa: E402
@@ -86,6 +98,43 @@ async def _find_recipes(session, codes):
return list(result.scalars().all()) return list(result.scalars().all())
async def _count_replace_impact(session, recipes) -> dict:
"""How many Measurement/ProductionRun/ProductionEvent --replace would delete.
Read-only mirror of what _delete_recipes actually removes, so the count
printed before the confirmation matches the count deleted after it.
"""
version_ids = []
run_ids = []
for recipe in recipes:
version_ids += (await session.execute(
select(RecipeVersion.id).where(RecipeVersion.recipe_id == recipe.id)
)).scalars().all()
run_ids += (await session.execute(
select(ProductionRun.id).where(ProductionRun.recipe_id == recipe.id)
)).scalars().all()
measurements = 0
if version_ids:
measurements = (await session.execute(
select(func.count()).select_from(Measurement)
.where(Measurement.version_id.in_(version_ids))
)).scalar_one()
events = 0
if run_ids:
events = (await session.execute(
select(func.count()).select_from(ProductionEvent)
.where(ProductionEvent.run_id.in_(run_ids))
)).scalar_one()
return {
"measurements": measurements,
"production_runs": len(run_ids),
"production_events": events,
}
async def _delete_recipes(session, recipes): async def _delete_recipes(session, recipes):
"""Remove the collaudo recipes and everything measured against them.""" """Remove the collaudo recipes and everything measured against them."""
for recipe in recipes: for recipe in recipes:
@@ -176,25 +225,33 @@ async def _ensure_supervisor(session, username, password):
An admin is accepted by the authorisation check, but the acceptance session An admin is accepted by the authorisation check, but the acceptance session
should exercise the role the shop floor will actually use. should exercise the role the shop floor will actually use.
`password` may be None: it is only needed when the account does not exist
yet, in which case a random one is generated here rather than falling
back to a committed default. Returns (user, created, password_used) so
the caller can print the generated one - the only place it appears.
""" """
existing = (await session.execute( existing = (await session.execute(
select(User).where(User.username == username) select(User).where(User.username == username)
)).scalar_one_or_none() )).scalar_one_or_none()
if existing is not None: if existing is not None:
return existing, False return existing, False, password
password_used = password or secrets.token_urlsafe(12)
user = await auth_service.create_user( user = await auth_service.create_user(
session, session,
username=username, username=username,
password=password, password=password_used,
display_name="Capo Turno (collaudo)", display_name="Capo Turno (collaudo)",
roles=["Supervisor", "MeasurementTec"], roles=["Supervisor", "MeasurementTec"],
) )
await session.flush() await session.flush()
return user, True return user, True, password_used
async def seed(station_code: str, replace: bool, supervisor_password: str) -> int: async def seed(
station_code: str, replace: bool, supervisor_password: str | None, si: bool,
) -> int:
async with async_session_factory() as session: async with async_session_factory() as session:
owner = (await session.execute( owner = (await session.execute(
select(User).where(User.is_admin == True) # noqa: E712 select(User).where(User.is_admin == True) # noqa: E712
@@ -217,6 +274,14 @@ async def seed(station_code: str, replace: bool, supervisor_password: str) -> in
print("Usa --replace per rifarle da zero (cancella anche le misure).") print("Usa --replace per rifarle da zero (cancella anche le misure).")
return 1 return 1
if existing: if existing:
impact = await _count_replace_impact(session, existing)
print(f"--replace cancellera' ({', '.join(r.code for r in existing)}):")
print(f" {impact['measurements']} misure")
print(f" {impact['production_runs']} produzioni")
print(f" {impact['production_events']} eventi di produzione")
if not si:
print("Nessuna cancellazione eseguita. Aggiungi --si per confermare.")
return 1
await _delete_recipes(session, existing) await _delete_recipes(session, existing)
print(f"Rimosse: {', '.join(r.code for r in existing)}") print(f"Rimosse: {', '.join(r.code for r in existing)}")
@@ -286,7 +351,7 @@ async def seed(station_code: str, replace: bool, supervisor_password: str) -> in
) )
await _assign(session, station, recipe_b, owner.id) await _assign(session, station, recipe_b, owner.id)
supervisor, created = await _ensure_supervisor( supervisor, created, password_used = await _ensure_supervisor(
session, "capoturno", supervisor_password, session, "capoturno", supervisor_password,
) )
@@ -296,7 +361,7 @@ async def seed(station_code: str, replace: bool, supervisor_password: str) -> in
print(" COLLAUDO-A ciclo di misura, lotto obbligatorio, intervallo 2 min") print(" COLLAUDO-A ciclo di misura, lotto obbligatorio, intervallo 2 min")
print(" COLLAUDO-B solo calibro, lotto e seriale obbligatori") print(" COLLAUDO-B solo calibro, lotto e seriale obbligatori")
if created: if created:
print(f"Capoturno creato: {supervisor.username} / {supervisor_password}") print(f"Capoturno creato: {supervisor.username} / {password_used}")
print(" cambiare la password prima di qualunque uso reale.") print(" cambiare la password prima di qualunque uso reale.")
else: else:
print(f"Capoturno gia' presente: {supervisor.username} (password invariata)") print(f"Capoturno gia' presente: {supervisor.username} (password invariata)")
@@ -309,11 +374,25 @@ def main() -> int:
help="codice della stazione a cui assegnare le ricette") help="codice della stazione a cui assegnare le ricette")
parser.add_argument("--replace", action="store_true", parser.add_argument("--replace", action="store_true",
help="cancella le ricette di collaudo esistenti e le misure fatte su di esse") help="cancella le ricette di collaudo esistenti e le misure fatte su di esse")
parser.add_argument("--supervisor-password", default="Collaudo2026!", parser.add_argument("--si", action="store_true",
help="password del capoturno, se va creato") help="conferma la cancellazione di --replace (senza, si limita a mostrarla)")
parser.add_argument("--supervisor-password", default=None,
help="password del capoturno, se va creato; "
"se omessa ne viene generata una casuale e stampata una volta")
parser.add_argument("--conferma-produzione", action="store_true",
help="richiesto per lanciare lo script con APP_ENV di produzione")
args = parser.parse_args() args = parser.parse_args()
if settings.app_env not in ("dev", "test") and not args.conferma_produzione:
print(
f"APP_ENV={settings.app_env!r} indica una produzione: "
"questo script scrive ricette e utenti di collaudo. "
"Rilancia con --conferma-produzione se e' davvero quello che vuoi."
)
return 2
return asyncio.run( return asyncio.run(
seed(args.station, args.replace, args.supervisor_password) seed(args.station, args.replace, args.supervisor_password, args.si)
) )
+190
View File
@@ -0,0 +1,190 @@
"""Tests for scripts/seed_collaudo.py (issue #17).
The script is not a package module (it is copied into the server container
and run directly), so it is loaded here via importlib from its file path.
"""
import asyncio
import importlib.util
from pathlib import Path
import pytest
from sqlalchemy import select
from src.backend.config import settings
from src.backend.models.orm.station import Station
from src.backend.models.orm.user import User
from src.backend.services.auth_service import create_user
from src.backend.tests.conftest import TestSessionFactory, test_engine
_SCRIPT_PATH = Path(__file__).resolve().parents[3] / "scripts" / "seed_collaudo.py"
def _load_script():
spec = importlib.util.spec_from_file_location("seed_collaudo", _SCRIPT_PATH)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
@pytest.fixture
def seed_collaudo(monkeypatch):
module = _load_script()
monkeypatch.setattr(module, "async_session_factory", TestSessionFactory)
return module
@pytest.fixture(autouse=True)
def reset_app_env(monkeypatch):
"""Every test in this file assumes the default (dev)."""
monkeypatch.setattr(settings, "app_env", "dev")
async def _make_admin_and_station(db_session) -> None:
admin = await create_user(
db_session, username="owner", password="OwnerPass1",
display_name="Owner", is_admin=True,
)
db_session.add(Station(
code="ST-DEFAULT", name="Default", created_by=admin.id,
))
await db_session.flush()
class TestReplaceRequiresSi:
"""--replace without --si only reports the impact, never deletes (#17c)."""
async def test_replace_without_si_does_not_delete(self, seed_collaudo, db_session):
await _make_admin_and_station(db_session)
# First run creates COLLAUDO-A/B.
rc = await seed_collaudo.seed("ST-DEFAULT", False, "FirstPass123", True)
assert rc == 0
# Second run with --replace but no --si: must not touch anything.
rc = await seed_collaudo.seed("ST-DEFAULT", True, "SecondPass123", False)
assert rc == 1
recipes = (await db_session.execute(
select(seed_collaudo.Recipe).where(
seed_collaudo.Recipe.code.in_(seed_collaudo.CODES)
)
)).scalars().all()
assert len(recipes) == 2
async def test_replace_with_si_deletes_and_recreates(self, seed_collaudo, db_session):
await _make_admin_and_station(db_session)
rc = await seed_collaudo.seed("ST-DEFAULT", False, "FirstPass123", True)
assert rc == 0
rc = await seed_collaudo.seed("ST-DEFAULT", True, "SecondPass123", True)
assert rc == 0
recipes = (await db_session.execute(
select(seed_collaudo.Recipe).where(
seed_collaudo.Recipe.code.in_(seed_collaudo.CODES)
)
)).scalars().all()
assert len(recipes) == 2
def test_main_with_fake_argv_replace_without_si_does_not_delete(
self, seed_collaudo, monkeypatch
):
"""The exact scenario from the analysis: main() with a fake argv,
--replace but no --si, must leave the recipes untouched.
Plain `def`, not `async def`: main() manages its own event loop via
asyncio.run(), which cannot be called from inside one already running
(as an `async def` test under pytest-asyncio would have).
"""
async def _setup():
async with TestSessionFactory() as session:
await _make_admin_and_station(session)
await session.commit()
asyncio.run(_setup())
assert asyncio.run(
seed_collaudo.seed("ST-DEFAULT", False, "FirstPass123", True)
) == 0
monkeypatch.setattr(
"sys.argv",
["seed_collaudo.py", "--station", "ST-DEFAULT", "--replace"],
)
rc = seed_collaudo.main()
assert rc == 1
async def _count():
async with TestSessionFactory() as session:
recipes = (await session.execute(
select(seed_collaudo.Recipe).where(
seed_collaudo.Recipe.code.in_(seed_collaudo.CODES)
)
)).scalars().all()
return len(recipes)
assert asyncio.run(_count()) == 2
class TestSupervisorPassword:
"""No default --supervisor-password: generated and printed once (#17a)."""
async def test_generates_random_password_when_not_given(
self, seed_collaudo, db_session
):
await _make_admin_and_station(db_session)
rc = await seed_collaudo.seed("ST-DEFAULT", False, None, True)
assert rc == 0
supervisor = (await db_session.execute(
select(User).where(User.username == "capoturno")
)).scalar_one_or_none()
assert supervisor is not None
# A real password was set (create_user always hashes something).
assert supervisor.password_hash
async def test_uses_given_password_when_provided(self, seed_collaudo, db_session):
await _make_admin_and_station(db_session)
rc = await seed_collaudo.seed("ST-DEFAULT", False, "ExplicitPass1", True)
assert rc == 0
from src.backend.services.auth_service import authenticate_user
user = await authenticate_user(db_session, "capoturno", "ExplicitPass1")
assert user is not None
async def test_existing_supervisor_password_untouched(
self, seed_collaudo, db_session
):
"""A second run (no --replace) must not need a password at all."""
await _make_admin_and_station(db_session)
assert await seed_collaudo.seed("ST-DEFAULT", False, "FirstPass123", True) == 0
# Recipes already exist, so seed() stops at the "already present"
# check before ever looking at supervisor_password.
rc = await seed_collaudo.seed("ST-DEFAULT", False, None, True)
assert rc == 1
class TestProductionGuard:
"""Refuses to run against APP_ENV outside dev/test without --conferma-produzione (#17b)."""
def test_main_refuses_in_production_without_confirmation(
self, seed_collaudo, monkeypatch
):
monkeypatch.setattr(settings, "app_env", "production")
monkeypatch.setattr(
"sys.argv", ["seed_collaudo.py", "--station", "ST-DEFAULT"]
)
rc = seed_collaudo.main()
assert rc == 2
def test_main_allows_dev_without_confirmation(self, seed_collaudo, monkeypatch):
monkeypatch.setattr(settings, "app_env", "dev")
# No admin/station set up: seed() itself will fail with rc=2, but that
# proves the production guard did not block it first.
monkeypatch.setattr(
"sys.argv", ["seed_collaudo.py", "--station", "NOPE"]
)
rc = seed_collaudo.main()
assert rc == 2 # "Nessun utente amministratore" / stazione inesistente