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
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"}
    assert len(set(primo.values())) == 4  # segreti distinti
    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