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:
+90
-11
@@ -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)
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user