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_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)]