TieMeasureFlow/packaging/windows/tests/test_avvio.py

"""Test di packaging/windows/avvio.py (TieMeasureFlow#76): licenza firmata con la chiave di prova
del contratto v1, nessun server vero (avvia_server e migrazioni sono sostituiti).

Si lanciano con l'ambiente 3.13 del pacchetto (serve visionsuite.license):
  UV_PROJECT_ENVIRONMENT=build/venv313 uv run --python 3.13 --frozen --extra server --extra client \
      --extra vision-worker --group windows --with pytest pytest packaging/windows/tests -q
"""
from __future__ import annotations

import base64
import functools
import importlib.util
import json
import os
import sqlite3
import sys
import types
from datetime import date, timedelta
from pathlib import Path

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

RADICE = Path(__file__).resolve().parents[3]
# vettore di prova del contratto licenza v1 (chiave finta, solo test)
SEED_TEST = base64.b64decode("AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=")
PUBBLICA_TEST = base64.b64decode("A6EHv/POEL4dcN0Y50vAmWfk1jCbpQ1fHdyGZBJVMbg=")
IMPRONTA = "ab" * 32


def _firma(dati: dict) -> str:
    # stessa canonicalizzazione del contratto (byte firmati)
    byte = json.dumps(dati, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
    return base64.b64encode(Ed25519PrivateKey.from_private_bytes(SEED_TEST).sign(byte)).decode()


@pytest.fixture
def avvio(monkeypatch, tmp_path):
    # avvio.py non e' in un pacchetto: lo si carica dal percorso
    spec = importlib.util.spec_from_file_location("avvio_tmf", RADICE / "packaging/windows/avvio.py")
    m = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(m)
    monkeypatch.setenv("PROGRAMDATA", str(tmp_path))  # ProgramData finta
    monkeypatch.delenv("TIELOGIC_LICENZA", raising=False)
    monkeypatch.setattr(os, "environ", os.environ.copy())  # le variabili impostate da avvio non escono dal test
    monkeypatch.setattr(m.lic, "impronta", lambda: IMPRONTA)
    # verifica reale (formato, firma, prodotto, macchina, scadenza) ma con la chiave di prova
    verifica_reale = m.lic.verifica
    monkeypatch.setattr(m.lic, "verifica", functools.partial(
        verifica_reale, chiave_pubblica=PUBBLICA_TEST, impronta_attesa=IMPRONTA))
    m.chiamate = []
    monkeypatch.setattr(m, "avvia_server", lambda host, dati: m.chiamate.append((host, dati)) or [])
    monkeypatch.setattr(m, "esegui_migrazioni", lambda: m.chiamate.append("migrazioni"))
    m.aperti = []
    monkeypatch.setattr(m.webbrowser, "open", lambda url: m.aperti.append(url))
    monkeypatch.setattr(m, "_prepara_log", lambda dati=None: None)
    monkeypatch.setattr(m, "_porta_aperta", lambda porta: False)
    return m


def _scrivi_licenza(tmp_path: Path, **modifiche) -> Path:
    dati = {"versione": 1, "prodotto": "tiemeasureflow", "cliente": "Prova Srl", "impronta": IMPRONTA,
            "emessa": date.today().isoformat(), "scadenza": (date.today() + timedelta(days=30)).isoformat()}
    dati.update(modifiche)
    p = tmp_path / "Tielogic" / "licenze" / "tiemeasureflow.json"
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_text(json.dumps({"dati": dati, "firma": _firma(dati)}), encoding="utf-8")
    return p


def test_licenza_mancante_non_avvia_e_scrive_impronta(avvio, capsys):
    assert avvio.main(["--senza-finestre"]) == 1
    assert avvio.chiamate == [] and avvio.aperti == []
    assert avvio.percorso_impronta().read_text(encoding="utf-8").strip() == IMPRONTA
    err = capsys.readouterr().err
    assert "Licenza non trovata" in err and IMPRONTA in err and "tiemeasureflow.json" in err


@pytest.mark.parametrize("modifica,testo", [
    ({"scadenza": "2020-01-01"}, "scaduta"),
    ({"impronta": "cd" * 32}, "altra macchina"),
    ({"prodotto": "tievision-studio"}, "altro prodotto"),
])
def test_licenza_non_valida_non_avvia(avvio, tmp_path, capsys, modifica, testo):
    _scrivi_licenza(tmp_path, **modifica)
    assert avvio.main(["--senza-finestre"]) == 1
    assert avvio.chiamate == []
    assert testo in capsys.readouterr().err
    assert avvio.percorso_impronta().read_text(encoding="utf-8").strip() == IMPRONTA


def test_licenza_manomessa_non_avvia(avvio, tmp_path, capsys):
    p = _scrivi_licenza(tmp_path)
    doc = json.loads(p.read_text(encoding="utf-8"))
    doc["dati"]["cliente"] = "Altro Cliente"  # dati cambiati dopo la firma
    p.write_text(json.dumps(doc), encoding="utf-8")
    assert avvio.main(["--senza-finestre"]) == 1
    assert "Firma" in capsys.readouterr().err


def test_licenza_valida_prepara_dati_e_avvia(avvio, tmp_path):
    _scrivi_licenza(tmp_path)
    assert avvio.main(["--senza-finestre", "--senza-browser", "--host", "0.0.0.0"]) == 0
    dati = tmp_path / "Tielogic" / "TieMeasureFlow"
    # prima le migrazioni, poi i server
    assert avvio.chiamate == ["migrazioni", ("0.0.0.0", dati)]
    assert (dati / "config.env").is_file() and (dati / "uploads").is_dir() and (dati / "camera").is_dir()
    assert not avvio.percorso_impronta().exists()  # con licenza valida non si scrive nulla
    assert avvio.aperti == []


def test_variabili_del_contratto(avvio, tmp_path):
    _scrivi_licenza(tmp_path)
    avvio.main(["--senza-finestre", "--senza-browser"])
    e = os.environ
    dati = tmp_path / "Tielogic" / "TieMeasureFlow"
    assert e["APP_ENV"] == "production"
    assert e["DB_URL"] == f"sqlite+aiosqlite:///{(dati / 'tmf.db').as_posix()}" and "\\" not in e["DB_URL"]
    assert e["UPLOAD_DIR"] == str(dati / "uploads") == e["VISION_RECIPES_ROOT"]
    assert e["VISION_WORKER_URL"] == "http://127.0.0.1:8100"
    assert e["API_SERVER_URL"] == "http://127.0.0.1:8000"
    assert e["SESSION_COOKIE_SECURE"] == "0"
    for k in ("SERVER_SECRET_KEY", "CLIENT_SECRET_KEY", "SETUP_PASSWORD", "DB_PASSWORD"):
        assert len(e[k]) >= 32


def test_versione_motore_visione_letta_dalla_radice_sorgente(avvio, tmp_path, monkeypatch):
    # #109: da sorgente il file vision-engine-version.txt sta nella radice del repo e arriva al worker
    prog = tmp_path / "prog"
    prog.mkdir()
    (prog / "vision-engine-version.txt").write_bytes(b"a" * 40)
    monkeypatch.setattr(avvio, "cartella_programma", lambda: prog)
    avvio.imposta_ambiente(tmp_path / "dati", "127.0.0.1", {k: "x" * 32 for k in avvio.SEGRETI})
    assert os.environ["VISION_ENGINE_VERSION"] == "a" * 40


def test_config_env_generato_una_volta_e_riletto(avvio, tmp_path):
    dati = tmp_path / "Tielogic" / "TieMeasureFlow"
    primo = avvio.carica_segreti(dati)
    testo = (dati / "config.env").read_text(encoding="utf-8")
    assert set(primo) == {"SERVER_SECRET_KEY", "CLIENT_SECRET_KEY", "SETUP_PASSWORD", "DB_PASSWORD", "STATION_CODE"}
    assert len({primo[k] for k in avvio.SEGRETI}) == 4  # segreti distinti
    assert primo["STATION_CODE"] == "ST-DEFAULT"
    secondo = avvio.carica_segreti(dati)
    assert secondo == primo and (dati / "config.env").read_text(encoding="utf-8") == testo


def test_config_env_completa_i_segreti_mancanti(avvio, tmp_path):
    dati = tmp_path / "Tielogic" / "TieMeasureFlow"
    dati.mkdir(parents=True)
    (dati / "config.env").write_text("# nota\nSETUP_PASSWORD=scelta-da-me\n", encoding="utf-8")
    v = avvio.carica_segreti(dati)
    assert v["SETUP_PASSWORD"] == "scelta-da-me" and len(v["SERVER_SECRET_KEY"]) >= 32
    assert avvio.carica_segreti(dati) == v


def test_seconda_istanza_apre_solo_il_browser(avvio, tmp_path, monkeypatch):
    _scrivi_licenza(tmp_path)
    monkeypatch.setattr(avvio, "_porta_aperta", lambda porta: True)
    monkeypatch.setattr(avvio, "_e_tmf", lambda: True)
    assert avvio.main(["--senza-finestre"]) == 0
    assert avvio.aperti == ["http://127.0.0.1:8780/"] and avvio.chiamate == []


def test_porta_occupata_da_altro_programma(avvio, tmp_path, monkeypatch, capsys):
    _scrivi_licenza(tmp_path)
    monkeypatch.setattr(avvio, "_porta_aperta", lambda porta: True)
    monkeypatch.setattr(avvio, "_e_tmf", lambda: False)
    assert avvio.main(["--senza-finestre"]) == 2
    assert "8780" in capsys.readouterr().err and avvio.chiamate == []


def test_migrazione_fallita_esce_con_3(avvio, tmp_path, monkeypatch, capsys):
    _scrivi_licenza(tmp_path)

    def rotta():
        raise RuntimeError("db bloccato")

    monkeypatch.setattr(avvio, "esegui_migrazioni", rotta)
    assert avvio.main(["--senza-finestre"]) == 3
    assert "db bloccato" in capsys.readouterr().err


def test_impronta_si_stampa_senza_licenza(avvio, capsys):
    assert avvio.main(["--impronta", "--senza-finestre"]) == 0
    assert IMPRONTA in capsys.readouterr().out


def test_nessun_utente_rileva_il_primo_avvio(avvio, tmp_path):
    db = tmp_path / "t.db"
    assert avvio._nessun_utente(db) is False  # file assente: nessuna conclusione
    con = sqlite3.connect(db)
    con.execute("CREATE TABLE users (id INTEGER)")
    con.commit()
    assert avvio._nessun_utente(db) is True
    con.execute("INSERT INTO users VALUES (1)")
    con.commit()
    con.close()
    assert avvio._nessun_utente(db) is False


def test_chiavi_extra_di_config_env_passano_ma_non_sovrascrivono_il_contratto(avvio, tmp_path):
    _scrivi_licenza(tmp_path)
    dati = tmp_path / "Tielogic" / "TieMeasureFlow"
    dati.mkdir(parents=True)
    (dati / "config.env").write_text("STATION_CODE=ST-FIERA\nSTATION_SWITCH_ENABLED=1\nAPP_ENV=dev\n", encoding="utf-8")
    avvio.main(["--senza-finestre", "--senza-browser"])
    assert os.environ["STATION_CODE"] == "ST-FIERA" and os.environ["STATION_SWITCH_ENABLED"] == "1"
    assert os.environ["APP_ENV"] == "production"  # le variabili fisse vincono


# --- TieMeasureFlow#82: strumenti nel pacchetto (nel binario non ci sono gli entry point) ---

def _modulo_strumenti(monkeypatch, moduli):
    # finto _strumenti.py generato da build.ps1
    finto = types.ModuleType("_strumenti")
    finto.MODULI = tuple(moduli)
    monkeypatch.setitem(sys.modules, "_strumenti", finto)


def test_registra_strumenti_da_sorgente_senza_modulo_generato(avvio, monkeypatch):
    # _strumenti.py esiste solo dopo build.ps1: se manca (sorgente) bastano gli entry point
    monkeypatch.setitem(sys.modules, "_strumenti", None)  # None in sys.modules = ImportError all'import
    from visionsuite.task.graph import registry
    assert avvio.registra_strumenti() == []
    assert len(registry()) > 0


def test_registra_strumenti_modulo_inesistente_non_solleva(avvio, monkeypatch):
    _modulo_strumenti(monkeypatch, ("visionsuite.tools.image", "visionsuite.tools.non_esiste"))
    assert avvio.registra_strumenti() == ["visionsuite.tools.non_esiste"]  # raccolto, non sollevato


def test_registra_strumenti_importa_i_moduli_elencati(avvio, monkeypatch):
    # senza entry point (come nell'exe) i moduli di MODULI sono l'unica via: li si importa per nome
    _modulo_strumenti(monkeypatch, ("visionsuite.tools.image",))
    visti = []
    monkeypatch.setattr(avvio.importlib, "import_module", lambda nome: visti.append(nome))
    assert avvio.registra_strumenti() == [] and visti == ["visionsuite.tools.image"]


def test_controlla_strumenti_senza_licenza_scrive_json(avvio, monkeypatch, capsys, tmp_path):
    monkeypatch.setitem(sys.modules, "_strumenti", None)

    # la licenza non deve essere nemmeno chiesta, e server e migrazioni non devono partire
    def mai(*a, **k):
        raise AssertionError("licenza, server o migrazioni chiamati da --controlla-strumenti")

    monkeypatch.setattr(avvio.lic, "verifica", mai)
    monkeypatch.setattr(avvio.lic, "cerca_licenza", mai)
    monkeypatch.setattr(avvio, "avvia_server", mai)
    monkeypatch.setattr(avvio, "esegui_migrazioni", mai)
    assert avvio.main(["--controlla-strumenti", "--senza-finestre"]) == 0
    dati = json.loads(capsys.readouterr().out)
    assert dati["strumenti"] > 0 and dati["strumenti"] == len(dati["kind"])
    assert dati["falliti"] == []
    scritto = tmp_path / "Tielogic" / "TieMeasureFlow" / "log" / "controllo-strumenti.json"
    assert json.loads(scritto.read_text(encoding="utf-8")) == dati
    assert avvio.aperti == [] and not (tmp_path / "Tielogic" / "TieMeasureFlow" / "tmf.db").exists()  # nessun DB


def test_controlla_strumenti_exit_3_se_un_modulo_fallisce(avvio, monkeypatch, capsys):
    _modulo_strumenti(monkeypatch, ("visionsuite.tools.non_esiste",))
    assert avvio.main(["--controlla-strumenti"]) == 3
    assert json.loads(capsys.readouterr().out)["falliti"] == ["visionsuite.tools.non_esiste"]


def test_controlla_strumenti_exit_3_se_il_catalogo_e_vuoto(avvio, monkeypatch, capsys):
    # niente strumenti registrati (es. entry point assenti e _strumenti vuoto): il controllo deve fallire
    import visionsuite.task.graph as graph
    monkeypatch.setattr(avvio, "registra_strumenti", lambda: [])
    monkeypatch.setattr(graph, "registry", lambda: {})
    assert avvio.main(["--controlla-strumenti"]) == 3
    assert json.loads(capsys.readouterr().out)["strumenti"] == 0


def test_avvio_normale_registra_gli_strumenti_prima_del_server(avvio, tmp_path, monkeypatch):
    # main() chiama registra_strumenti() dopo licenza e migrazioni e prima di avvia_server
    _scrivi_licenza(tmp_path)
    monkeypatch.setattr(avvio, "registra_strumenti", lambda: avvio.chiamate.append("strumenti") or [])
    assert avvio.main(["--senza-finestre", "--senza-browser"]) == 0
    dati = tmp_path / "Tielogic" / "TieMeasureFlow"
    assert avvio.chiamate == ["migrazioni", "strumenti", ("127.0.0.1", dati)]


# ---------------------------------------------------------------- scalda camere (#108)

def _db_camere(percorso):
    # #108: db minimo con le due tabelle lette da camere_da_scaldare
    con = sqlite3.connect(percorso)
    con.execute("CREATE TABLE stations (id INTEGER PRIMARY KEY, camera_url TEXT, camera_source TEXT)")
    con.execute("CREATE TABLE recipe_tasks (id INTEGER PRIMARY KEY, acquisition_json TEXT)")
    con.executemany("INSERT INTO stations (camera_url, camera_source) VALUES (?, ?)", [
        ("http://127.0.0.1:8765", "camera:601020"),   # valida
        (None, None),                                  # stazione senza camera
        ("http://127.0.0.1:8765", "camera:folder"),   # non apre hardware
    ])
    con.executemany("INSERT INTO recipe_tasks (acquisition_json) VALUES (?)", [
        (json.dumps({"camera_url": "http://127.0.0.1:8765", "camera_source": "camera:601020"}),),  # duplicato
        (json.dumps({"camera_url": "http://10.0.0.5:8765/", "camera_source": "camera:777"}),),     # valida
        ("{rotto",),                                                                               # JSON non valido
        (None,),
    ])
    con.commit()
    con.close()


def test_camere_da_scaldare_filtra_e_deduplica(avvio, tmp_path):
    db = tmp_path / "tmf.db"
    _db_camere(db)
    assert avvio.camere_da_scaldare(db) == [
        ("http://127.0.0.1:8765", "camera:601020"),
        ("http://10.0.0.5:8765/", "camera:777"),
    ]


def test_camere_da_scaldare_db_inesistente(avvio, tmp_path):
    assert avvio.camere_da_scaldare(tmp_path / "non_esiste.db") == []


def test_scalda_camere_una_chiamata_per_coppia(avvio, tmp_path, monkeypatch):
    db = tmp_path / "tmf.db"
    _db_camere(db)
    monkeypatch.setattr(avvio, "_porta_aperta", lambda porta: True)
    chiamate = []

    class Risposta:
        status = 200
        def __enter__(self): return self
        def __exit__(self, *a): return False

    def finto_urlopen(richiesta, timeout=None):
        chiamate.append((richiesta.full_url, json.loads(richiesta.data), richiesta.get_method(), timeout))
        return Risposta()

    monkeypatch.setattr(avvio.urllib.request, "urlopen", finto_urlopen)
    avvio.scalda_camere(db)
    corpo = {"trigger": "continuo", "timeout_ms": 5000}
    assert chiamate == [
        ("http://127.0.0.1:8765/camera/grab", {"source": "camera:601020", **corpo}, "POST", 90),
        ("http://10.0.0.5:8765/camera/grab", {"source": "camera:777", **corpo}, "POST", 90),
    ]


def test_scalda_camere_eccezione_urlopen_non_esce(avvio, tmp_path, monkeypatch):
    db = tmp_path / "tmf.db"
    _db_camere(db)
    monkeypatch.setattr(avvio, "_porta_aperta", lambda porta: True)
    tentativi = []

    def urlopen_rotto(richiesta, timeout=None):
        tentativi.append(richiesta.full_url)
        raise OSError("camera irraggiungibile")

    monkeypatch.setattr(avvio.urllib.request, "urlopen", urlopen_rotto)
    avvio.scalda_camere(db)  # non solleva e passa alla camera successiva
    assert len(tentativi) == 2