diff --git a/scripts/seed_collaudo.py b/scripts/seed_collaudo.py index 4ade59f..32b1c90 100644 --- a/scripts/seed_collaudo.py +++ b/scripts/seed_collaudo.py @@ -21,16 +21,28 @@ are already in place: 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 -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 asyncio +import secrets import sys -from sqlalchemy import delete, select +from sqlalchemy import delete, func, select 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.models.orm.measurement import Measurement # 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()) +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): """Remove the collaudo recipes and everything measured against them.""" 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 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( select(User).where(User.username == username) )).scalar_one_or_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( session, username=username, - password=password, + password=password_used, display_name="Capo Turno (collaudo)", roles=["Supervisor", "MeasurementTec"], ) 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: owner = (await session.execute( 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).") return 1 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) 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) - supervisor, created = await _ensure_supervisor( + supervisor, created, password_used = await _ensure_supervisor( 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-B solo calibro, lotto e seriale obbligatori") 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.") else: 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") parser.add_argument("--replace", action="store_true", help="cancella le ricette di collaudo esistenti e le misure fatte su di esse") - parser.add_argument("--supervisor-password", default="Collaudo2026!", - help="password del capoturno, se va creato") + parser.add_argument("--si", action="store_true", + 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() + + 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( - seed(args.station, args.replace, args.supervisor_password) + seed(args.station, args.replace, args.supervisor_password, args.si) ) diff --git a/src/backend/tests/test_seed_collaudo.py b/src/backend/tests/test_seed_collaudo.py new file mode 100644 index 0000000..5b1cd78 --- /dev/null +++ b/src/backend/tests/test_seed_collaudo.py @@ -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