pyGUI/tests/test_license.py

"""Tests for src/pygui/backend/license/ (manager + model)."""

import os
import time

from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes

from pygui.backend.license.license_manager import (
    LICENSE_KEY,
    parse_license_blob,
    read_license_files,
    split_data_iv,
)
from pygui.backend.license.license_model import LicenseModel


# ===== Fixture helpers =====
def _encrypt_aes(plaintext: bytes, key: bytes, iv: bytes) -> bytes:
    """AES-128-CBC + PKCS7, mirroring what the C# license generator does."""
    pad_len = 16 - (len(plaintext) % 16)
    padded = plaintext + bytes([pad_len]) * pad_len
    cipher = Cipher(algorithms.AES(key), modes.CBC(iv), backend=default_backend())
    enc = cipher.encryptor()
    return enc.update(padded) + enc.finalize()


def make_license_blob(
    serial: str = "A00001",
    mfg_hex: str = "0001E240",
    level: str = "01",
    lic_hex: str = "0001E24F",
) -> bytes:
    """Build a C#-format license blob: ciphertext || iv, fixed-offset text."""
    text = f"{serial} {mfg_hex} {level} {lic_hex}"
    iv = bytes(range(16))
    return _encrypt_aes(text.encode("utf-8"), LICENSE_KEY, iv) + iv


def write_license(directory, serial="A00001", level="01", **kw) -> str:
    path = directory / f"{serial}-{level}.bin"
    path.write_bytes(make_license_blob(serial=serial, level=level, **kw))
    return str(path)


# ===== Manager: blob parsing =====
def test_parse_valid_blob():
    lic = parse_license_blob(make_license_blob())
    assert lic is not None
    assert lic.serial == "A00001"
    assert lic.level == "01"
    assert lic.level_int == 1
    assert lic.mfg_date_hex == "0001E240"
    assert lic.mfg_date_decimal == "123456"
    assert lic.lic_date_decimal == str(int("0001E24F", 16))


def test_parse_blob_too_short():
    assert parse_license_blob(b"short") is None


def test_parse_blob_garbage():
    assert parse_license_blob(os.urandom(64)) is None


def test_parse_blob_short_text():
    iv = bytes(16)
    blob = _encrypt_aes(b"A00001 tiny", LICENSE_KEY, iv) + iv
    assert parse_license_blob(blob) is None


def test_split_data_iv():
    data, iv = split_data_iv(bytes(48))
    assert len(data) == 32
    assert len(iv) == 16


# ===== Manager: directory scan =====
def test_read_license_files_valid(tmp_path):
    write_license(tmp_path)
    licenses = read_license_files(tmp_path)
    assert len(licenses) == 1
    assert licenses[0].serial == "A00001"


def test_read_license_files_missing_dir(tmp_path):
    assert read_license_files(tmp_path / "nope") == []


def test_read_license_files_bad_filename_skipped(tmp_path):
    (tmp_path / "notalicense.bin").write_bytes(make_license_blob())
    assert read_license_files(tmp_path) == []


def test_read_license_files_name_payload_mismatch(tmp_path):
    # payload says A00001-01 but file claims B00002-03 → tamper, reject
    (tmp_path / "B00002-03.bin").write_bytes(make_license_blob())
    assert read_license_files(tmp_path) == []


def test_read_license_files_corrupt_skipped(tmp_path):
    (tmp_path / "A00001-01.bin").write_bytes(os.urandom(64))
    assert read_license_files(tmp_path) == []


# ===== Model =====
def test_model_lists_licenses(tmp_path, qapp):
    (tmp_path / "licenses").mkdir()
    write_license(tmp_path / "licenses")
    model = LicenseModel(str(tmp_path))
    rows = model.licenses
    assert len(rows) == 1
    assert rows[0]["serial"] == "A00001"
    assert rows[0]["level"] == "01"
    assert rows[0]["mfgDate"] == "123456"


def test_model_load_license_file(tmp_path, qapp):
    src_dir = tmp_path / "downloads"
    src_dir.mkdir()
    src = write_license(src_dir, serial="B00002", level="00")
    model = LicenseModel(str(tmp_path / "appdata"))
    assert model.loadLicenseFile(src) is True
    assert len(model.licenses) == 1
    assert model.licenses[0]["serial"] == "B00002"
    # copy landed in <data_dir>/licenses
    assert (tmp_path / "appdata" / "licenses" / "B00002-00.bin").exists()


def test_model_load_license_file_url(tmp_path, qapp):
    src_dir = tmp_path / "dl"
    src_dir.mkdir()
    src = write_license(src_dir)
    model = LicenseModel(str(tmp_path / "appdata"))
    assert model.loadLicenseFile("file://" + src) is True


def test_model_remove_license_keeps_bin_on_disk(tmp_path, qapp):
    (tmp_path / "licenses").mkdir()
    write_license(tmp_path / "licenses", serial="A00001", level="01")
    model = LicenseModel(str(tmp_path))
    assert len(model.licenses) == 1

    assert model.removeLicense("A00001", "01") is True
    assert model.licenses == []
    # app-side removal only — the vendor-issued .bin is never deleted
    assert (tmp_path / "licenses" / "A00001-01.bin").exists()

    # sticky across a fresh scan (not just until the next refresh)
    model.refresh()
    assert model.licenses == []

    # sticky across a real relaunch — brand new instance, no shared state
    relaunched = LicenseModel(str(tmp_path))
    assert relaunched.licenses == []


def test_model_remove_license_missing_returns_false(tmp_path, qapp):
    model = LicenseModel(str(tmp_path))
    assert model.removeLicense("Z99999", "00") is False


def test_model_reload_after_remove_unhides(tmp_path, qapp):
    src_dir = tmp_path / "dl"
    src_dir.mkdir()
    src = write_license(src_dir, serial="A00001", level="01")
    model = LicenseModel(str(tmp_path / "appdata"))
    assert model.loadLicenseFile(src) is True
    assert len(model.licenses) == 1

    model.removeLicense("A00001", "01")
    assert model.licenses == []

    # explicit reload of the same file brings it back
    assert model.loadLicenseFile(src) is True
    assert len(model.licenses) == 1


def test_model_reload_in_place_unhides(tmp_path, qapp):
    """Un-hiding by re-picking the .bin from its own <data_dir>/licenses/ dir —
    src and dst resolve to the same file, must not error on the copy step."""
    model = LicenseModel(str(tmp_path / "appdata"))
    write_license(tmp_path / "appdata" / "licenses", serial="A00001", level="01")
    model.refresh()
    assert len(model.licenses) == 1

    model.removeLicense("A00001", "01")
    assert model.licenses == []

    in_place = tmp_path / "appdata" / "licenses" / "A00001-01.bin"
    assert model.loadLicenseFile(str(in_place)) is True
    assert len(model.licenses) == 1


def test_model_load_rejects_invalid(tmp_path, qapp):
    src_dir = tmp_path / "dl"
    src_dir.mkdir()
    bad = src_dir / "A00001-01.bin"
    bad.write_bytes(os.urandom(64))
    model = LicenseModel(str(tmp_path / "appdata"))
    assert model.loadLicenseFile(str(bad)) is False
    assert model.licenses == []


def test_model_load_rejects_missing(tmp_path, qapp):
    model = LicenseModel(str(tmp_path))
    assert model.loadLicenseFile(str(tmp_path / "ghost.bin")) is False


def test_model_level_for_wafer(tmp_path, qapp):
    (tmp_path / "licenses").mkdir()
    write_license(tmp_path / "licenses", serial="A00001", level="02")
    model = LicenseModel(str(tmp_path))
    assert model.levelForWafer("A00001") == "02"
    assert model.levelForWafer("Z99999") == ""


# ===== Replay trial =====
def test_replay_state_none_then_temporary(tmp_path, qapp):
    model = LicenseModel(str(tmp_path))
    assert model.replayLicenseState() == "none"
    assert model.startReplayTrial() is True
    assert model.replayLicenseState() == "temporary"
    assert model.replayTrialExpired() is False


def test_start_replay_trial_does_not_reset_existing_marker(tmp_path, qapp):
    """Repeat calls must not push the clock forward — the original
    touch(exist_ok=True) bumped mtime on every call, which would let anyone
    re-trigger the trial start (e.g. an over-limit wafer re-detected) reset
    the 30-day window indefinitely."""
    model = LicenseModel(str(tmp_path))
    assert model.startReplayTrial() is True
    first_mtime = model._trial_marker.stat().st_mtime

    # Simulate the marker being old (as if most of the trial has elapsed).
    old_time = first_mtime - 25 * 86400
    os.utime(model._trial_marker, (old_time, old_time))

    assert model.startReplayTrial() is True
    assert model._trial_marker.stat().st_mtime == old_time


def test_replay_state_permanent_with_level1(tmp_path, qapp):
    (tmp_path / "licenses").mkdir()
    write_license(tmp_path / "licenses", level="01")
    model = LicenseModel(str(tmp_path))
    assert model.replayLicenseState() == "permanent"


def test_replay_level0_not_permanent(tmp_path, qapp):
    (tmp_path / "licenses").mkdir()
    write_license(tmp_path / "licenses", level="00")
    model = LicenseModel(str(tmp_path))
    assert model.replayLicenseState() == "none"


def test_replay_trial_expiry(tmp_path, qapp):
    model = LicenseModel(str(tmp_path))
    model.startReplayTrial()
    marker = tmp_path / "replay.lic"
    old = time.time() - 31 * 86400
    os.utime(marker, (old, old))
    assert model.replayTrialExpired() is True


def test_replay_trial_expired_without_marker(tmp_path, qapp):
    model = LicenseModel(str(tmp_path))
    assert model.replayTrialExpired() is True