feat(license): wafer license gate, runtime warning, and first-launch wizard
- Gate READ/ERASE/DEBUG behind wafer license check; log hint to load .bin - waferRuntimeExceeded() warns when onboard runtime > 350s - LicenseModel: remove-license (soft hide, .bin stays on disk) via removed_licenses.txt; re-loading un-hides - FirstLaunchWizard: 2-step popup (license + save dir) shown on first run - Per-tab file memory in HomePage (map/graph/data don't bleed state) - Merge test_device_controller_license into test_device_controller
This commit is contained in:
@@ -50,6 +50,11 @@ def test_detect_wafer_spawns_thread(controller):
|
||||
assert controller.operationInProgress
|
||||
|
||||
def test_read_memory_without_detect_return_error(controller):
|
||||
# License the wafer so this test exercises the "no port" branch rather
|
||||
# than short-circuiting on the license guard.
|
||||
controller._license_lookup = lambda s: "02"
|
||||
controller._last_wafer_info = {"serialNumber": "A00001"}
|
||||
|
||||
emitted_result = None
|
||||
def on_read_result(result):
|
||||
nonlocal emitted_result
|
||||
@@ -91,6 +96,9 @@ def test_clear_session(controller):
|
||||
assert controller.lastWaferInfo == ["", "", 0, "", 0, 0]
|
||||
|
||||
def test_erase_memory_spawns_thread(controller):
|
||||
# License the wafer so the new license guard doesn't block the erase.
|
||||
controller._license_lookup = lambda s: "02"
|
||||
controller._last_wafer_info = {"serialNumber": "A00001"}
|
||||
with patch("threading.Thread") as mock_thread:
|
||||
controller.eraseMemory("COM1")
|
||||
mock_thread.assert_called_once()
|
||||
@@ -109,11 +117,32 @@ def test_erase_memory_handler(controller):
|
||||
assert not controller.operationInProgress
|
||||
|
||||
def test_read_debug_spawns_thread(controller):
|
||||
# License the wafer so the license guard doesn't block the debug read.
|
||||
controller._license_lookup = lambda s: "02"
|
||||
controller._last_wafer_info = {"serialNumber": "A00001"}
|
||||
with patch("threading.Thread") as mock_thread:
|
||||
controller.readDebug("COM1")
|
||||
mock_thread.assert_called_once()
|
||||
assert controller.operationInProgress
|
||||
|
||||
def test_read_debug_denied_without_license(controller):
|
||||
"""readDebug() returns D1 sensor data, so it obeys the same license guard."""
|
||||
controller._license_lookup = lambda s: ""
|
||||
controller._last_wafer_info = {"serialNumber": "A00001"}
|
||||
|
||||
emitted_result = None
|
||||
def on_debug_result(result):
|
||||
nonlocal emitted_result
|
||||
emitted_result = result
|
||||
controller.debugResult.connect(on_debug_result)
|
||||
|
||||
with patch("threading.Thread") as mock_thread:
|
||||
controller.readDebug("COM1")
|
||||
mock_thread.assert_not_called()
|
||||
|
||||
assert emitted_result == {"success": False, "error": "Wafer not licensed"}
|
||||
assert not controller.operationInProgress
|
||||
|
||||
def test_read_debug_handler(controller):
|
||||
emitted_result = None
|
||||
def on_debug_result(result):
|
||||
@@ -300,3 +329,136 @@ def test_export_dir_created_under_save_data_dir(controller):
|
||||
|
||||
assert result == str(Path(controller.saveDataDir) / "Export")
|
||||
assert Path(result).is_dir()
|
||||
|
||||
|
||||
def test_wafer_licensed_default_no_license(controller):
|
||||
"""Default license_lookup (empty lambda) -> waferLicensed() is False."""
|
||||
controller._last_wafer_info = {"serialNumber": "P00001"}
|
||||
assert controller.waferLicensed() is False
|
||||
|
||||
|
||||
def test_wafer_licensed_with_valid_license(mock_settings):
|
||||
"""A license_lookup returning a level -> waferLicensed() is True."""
|
||||
import shutil
|
||||
import tempfile
|
||||
|
||||
temp_dir = tempfile.mkdtemp()
|
||||
try:
|
||||
controller = DeviceController(
|
||||
mock_settings, temp_dir, license_lookup=lambda s: "02"
|
||||
)
|
||||
controller._last_wafer_info = {"serialNumber": "A00001"}
|
||||
assert controller.waferLicensed() is True
|
||||
finally:
|
||||
shutil.rmtree(temp_dir)
|
||||
|
||||
|
||||
def test_read_memory_denied_without_license(controller):
|
||||
"""readMemoryAsync() must refuse to spawn a thread for an unlicensed wafer."""
|
||||
controller._license_lookup = lambda s: ""
|
||||
controller._last_wafer_info = {"serialNumber": "A00001"}
|
||||
controller._selected_port = "COM1"
|
||||
|
||||
emitted_result = None
|
||||
def on_read_result(result):
|
||||
nonlocal emitted_result
|
||||
emitted_result = result
|
||||
controller.readResult.connect(on_read_result)
|
||||
|
||||
with patch("threading.Thread") as mock_thread:
|
||||
controller.readMemoryAsync()
|
||||
mock_thread.assert_not_called()
|
||||
|
||||
assert emitted_result is not None
|
||||
assert "error" in emitted_result
|
||||
assert not controller.operationInProgress
|
||||
|
||||
|
||||
def test_erase_memory_denied_without_license(controller):
|
||||
"""eraseMemory() must refuse to spawn a thread for an unlicensed wafer."""
|
||||
controller._license_lookup = lambda s: ""
|
||||
controller._last_wafer_info = {"serialNumber": "A00001"}
|
||||
|
||||
emitted_result = None
|
||||
def on_erase_result(result):
|
||||
nonlocal emitted_result
|
||||
emitted_result = result
|
||||
controller.eraseResult.connect(on_erase_result)
|
||||
|
||||
with patch("threading.Thread") as mock_thread:
|
||||
controller.eraseMemory("COM1")
|
||||
mock_thread.assert_not_called()
|
||||
|
||||
assert emitted_result == {"success": False}
|
||||
assert not controller.operationInProgress
|
||||
|
||||
|
||||
def test_read_memory_allowed_with_license(controller):
|
||||
"""readMemoryAsync() proceeds and spawns a thread once the wafer is licensed."""
|
||||
controller._license_lookup = lambda s: "02"
|
||||
controller._last_wafer_info = {"serialNumber": "A00001"}
|
||||
controller._selected_port = "COM1"
|
||||
|
||||
with patch("threading.Thread") as mock_thread:
|
||||
controller.readMemoryAsync()
|
||||
mock_thread.assert_called_once()
|
||||
assert controller.operationInProgress
|
||||
|
||||
|
||||
def test_erase_memory_allowed_with_license(controller):
|
||||
"""eraseMemory() proceeds and spawns a thread once the wafer is licensed."""
|
||||
controller._license_lookup = lambda s: "02"
|
||||
controller._last_wafer_info = {"serialNumber": "A00001"}
|
||||
|
||||
with patch("threading.Thread") as mock_thread:
|
||||
controller.eraseMemory("COM1")
|
||||
mock_thread.assert_called_once()
|
||||
assert controller.operationInProgress
|
||||
|
||||
|
||||
def test_wafer_runtime_exceeded_false_under_threshold(controller):
|
||||
controller._last_wafer_info = {"runtime": 100}
|
||||
assert controller.waferRuntimeExceeded() is False
|
||||
|
||||
|
||||
def test_wafer_runtime_exceeded_true_over_threshold(controller):
|
||||
controller._last_wafer_info = {"runtime": 999}
|
||||
assert controller.waferRuntimeExceeded() is True
|
||||
|
||||
|
||||
def test_detect_finished_warns_on_runtime_exceeded(controller):
|
||||
info = WaferInfo(
|
||||
family_code="P",
|
||||
serial_number="P00001",
|
||||
sensor_count=244,
|
||||
mfg_date_hex="12345678",
|
||||
sensor_assigned_hex="01",
|
||||
locked_hex="00",
|
||||
runtime=999,
|
||||
cycle_count=10,
|
||||
)
|
||||
log_messages = []
|
||||
controller.activityLogUpdated.connect(lambda msg: log_messages.append(msg))
|
||||
|
||||
controller._handle_detect_finished("COM1", info)
|
||||
|
||||
assert any("exceeds recommended" in msg for msg in log_messages)
|
||||
|
||||
|
||||
def test_detect_finished_no_warning_under_threshold(controller):
|
||||
info = WaferInfo(
|
||||
family_code="P",
|
||||
serial_number="P00001",
|
||||
sensor_count=244,
|
||||
mfg_date_hex="12345678",
|
||||
sensor_assigned_hex="01",
|
||||
locked_hex="00",
|
||||
runtime=100,
|
||||
cycle_count=10,
|
||||
)
|
||||
log_messages = []
|
||||
controller.activityLogUpdated.connect(lambda msg: log_messages.append(msg))
|
||||
|
||||
controller._handle_detect_finished("COM1", info)
|
||||
|
||||
assert not any("exceeds recommended" in msg for msg in log_messages)
|
||||
|
||||
@@ -1,8 +0,0 @@
|
||||
# TODO P4.2: unit tests for the DeviceController license guard (plan §4.2,
|
||||
# docs/pending/license-gating-plan.md). Two cases:
|
||||
# 1. license_lookup=lambda s: "" + fake _last_wafer_info → readMemoryAsync()
|
||||
# emits readResult {"error": ...} and spawns no thread.
|
||||
# 2. lookup returning "02" → proceeds to _read_worker (mock DeviceService).
|
||||
# THINKING: guard is a trust boundary (plan §2.1); these fail if someone
|
||||
# reorders the busy/license checks or renames the "serialNumber" key.
|
||||
# Depends on P1.2 + P2.1 being implemented.
|
||||
@@ -135,6 +135,63 @@ def test_model_load_license_file_url(tmp_path, qapp):
|
||||
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()
|
||||
@@ -167,6 +224,23 @@ def test_replay_state_none_then_temporary(tmp_path, qapp):
|
||||
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")
|
||||
|
||||
Reference in New Issue
Block a user