"""Test di scripts/seed_fiera.py (issue #85): il seed del PC della fiera. Lo script non e' un modulo del pacchetto: si carica da file con importlib, come test_seed_collaudo. La parte "dato" gira sul DB in memoria dei test; l'ultimo test percorre invece la strada vera (file SQLite con `alembic upgrade head` + --db). """ import importlib.util import sqlite3 from argparse import Namespace from datetime import date from pathlib import Path import pytest from alembic import command from alembic.config import Config from sqlalchemy import func, select from src.backend.config import Settings, settings from src.backend.models.orm.measurement import Measurement from src.backend.models.orm.recipe import Recipe from src.backend.models.orm.station import Station, StationRecipeAssignment from src.backend.models.orm.user import User from src.backend.services.auth_service import authenticate_user from src.backend.tests.conftest import TestSessionFactory, test_engine _SCRIPT_PATH = Path(__file__).resolve().parents[3] / "scripts" / "seed_fiera.py" _ALEMBIC_INI = Path(__file__).resolve().parents[1] / "migrations" / "alembic.ini" FINE = date(2026, 10, 22) # data fissa: il test non dipende dal giorno in cui gira @pytest.fixture def seed_fiera(monkeypatch): spec = importlib.util.spec_from_file_location("seed_fiera", _SCRIPT_PATH) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) monkeypatch.setattr(module, "async_session_factory", TestSessionFactory) return module async def _utente(db_session, username) -> User: return (await db_session.execute( select(User).where(User.username == username) )).scalar_one() class TestSeedFiera: async def test_utenti_ruoli_e_login(self, seed_fiera, db_session): assert await seed_fiera.seed(False, False, FINE) == 0 attesi = { "adriano": ("123", True), "maker": ("123", False), "capoturno": ("123", False), "operatore": ("123", False), } for username, (password, admin) in attesi.items(): utente = await authenticate_user(db_session, username, password) assert utente is not None, username assert utente.is_admin is admin assert (await _utente(db_session, "maker")).roles == ["Maker"] assert "Supervisor" in (await _utente(db_session, "capoturno")).roles assert (await _utente(db_session, "operatore")).roles == ["MeasurementTec"] assert {"Maker", "Metrologist"} <= set((await _utente(db_session, "adriano")).roles) async def test_setup_risulta_completato(self, seed_fiera, client, monkeypatch): import src.backend.api.routers.setup as setup_mod monkeypatch.setattr(settings, "setup_password", "test-setup-password") monkeypatch.setattr(setup_mod, "engine", test_engine) monkeypatch.setattr(setup_mod, "async_session_factory", TestSessionFactory) # prima del seed il frontend vede il setup da fare, dopo no assert (await client.get("/api/setup/status")).json() == {"initialised": False} await seed_fiera.seed(False, False, FINE) assert (await client.get("/api/setup/status")).json() == {"initialised": True} async def test_stazione_e_ricette(self, seed_fiera, db_session): await seed_fiera.seed(False, False, FINE) stazione = (await db_session.execute( select(Station).where(Station.code == "ST-DEFAULT") )).scalar_one() assert stazione.camera_url == "http://127.0.0.1:8765" assert stazione.camera_source == "camera:601020" assegnate = (await db_session.execute( select(Recipe.code) .join(StationRecipeAssignment, StationRecipeAssignment.recipe_id == Recipe.id) .where(StationRecipeAssignment.station_id == stazione.id) )).scalars().all() assert sorted(assegnate) == ["FIERA-ALBERO", "FIERA-DXF", "FIERA-FLANGIA"] # la ricetta DXF ha un task dxf_compare e nessun grafo di visione from src.backend.models.orm.task import RecipeTask dxf = (await db_session.execute( select(RecipeTask).where(RecipeTask.task_type == "dxf_compare") )).scalars().all() assert len(dxf) == 1 and dxf[0].vision_json is None async def test_misure_nel_numero_atteso(self, seed_fiera, db_session): await seed_fiera.seed(False, False, FINE) # ogni pezzo ha una misura per quota: pezzi 150-300 per ricetta for codice, quote in (("FIERA-FLANGIA", 5), ("FIERA-ALBERO", 4)): misure = (await db_session.execute( select(func.count(Measurement.id)) .where(Measurement.version_id.in_( select(seed_fiera.RecipeVersion.id) .join(Recipe, Recipe.id == seed_fiera.RecipeVersion.recipe_id) .where(Recipe.code == codice) )) )).scalar_one() assert misure % quote == 0 assert 150 <= misure // quote <= 300, (codice, misure // quote) # lun-ven soltanto, e le misure compaiono nelle produzioni giorni = (await db_session.execute( select(seed_fiera.ProductionRun.started_at) )).scalars().all() assert giorni and all(g.weekday() < 5 for g in giorni) senza_run = (await db_session.execute( select(func.count(Measurement.id)).where(Measurement.production_run_id.is_(None)) )).scalar_one() assert senza_run == 0 async def test_statistiche_numeriche(self, seed_fiera, client, db_session): await seed_fiera.seed(False, False, FINE) adriano = await _utente(db_session, "adriano") from src.backend.services.auth_service import issue_api_key chiave = await issue_api_key(db_session, adriano) intestazioni = {"X-API-Key": chiave} flangia = (await db_session.execute( select(Recipe).where(Recipe.code == "FIERA-FLANGIA") )).scalar_one() quote = (await client.get( "/api/statistics/subtasks", headers=intestazioni, params={"recipe_id": flangia.id}, )).json() assert len(quote) == 5 cpk = [] for quota in quote: risposta = await client.get( "/api/statistics/capability", headers=intestazioni, params={"recipe_id": flangia.id, "subtask_id": quota["id"]}, ) assert risposta.status_code == 200 d = risposta.json() for campo in ("mean", "std_dev", "cp", "cpk"): assert isinstance(d[campo], (int, float)), (quota["id"], campo) cpk.append(d["cpk"]) # Cpk diversi per quota: una capace, una al limite o sotto assert max(cpk) > 1.4 assert min(cpk) < 1.1 async def test_secondo_giro_non_duplica_e_replace_rifa(self, seed_fiera, db_session): async def conta(): return [ (await db_session.execute(select(func.count(t.id)))).scalar_one() for t in (Recipe, Measurement, User) ] await seed_fiera.seed(False, False, FINE) primo = await conta() await seed_fiera.seed(False, False, FINE) assert await conta() == primo # utenti aggiornati, ricette e misure non duplicate await seed_fiera.seed(True, False, FINE) assert await conta() == primo # stessi numeri: il seme e' fisso async def test_dry_run_non_salva(self, seed_fiera, db_session): await seed_fiera.seed(False, True, FINE) assert (await db_session.execute(select(func.count(User.id)))).scalar_one() == 0 assert (await db_session.execute(select(func.count(Measurement.id)))).scalar_one() == 0 async def test_file_sqlite_con_alembic(seed_fiera, tmp_path, monkeypatch): """La strada vera: schema da `alembic upgrade head`, poi il seed con --db.""" db = tmp_path / "tmf.db" url = f"sqlite+aiosqlite:///{db.as_posix()}" monkeypatch.setattr(Settings, "database_url", property(lambda self: url)) # alembic fa asyncio.run(): fuori dal loop di pytest-asyncio, in un thread import asyncio await asyncio.to_thread(command.upgrade, Config(str(_ALEMBIC_INI)), "head") argomenti = Namespace(db=db, replace=False, dry_run=False, fine=FINE.isoformat()) assert await seed_fiera._main_async(argomenti) == 0 con = sqlite3.connect(db) try: assert con.execute("SELECT COUNT(*) FROM users WHERE is_admin=1").fetchone()[0] == 1 assert con.execute("SELECT COUNT(*) FROM recipes").fetchone()[0] == 3 assert con.execute("SELECT COUNT(*) FROM measurements").fetchone()[0] > 1000 finally: con.close()