41 lines
1.5 KiB
Python
41 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from k1link.device_plugins.xgrids_k1.archive import discover_legacy_viewer_sessions
|
|
from k1link.sessions import ActiveSessionLease, recover_stale_active_session_marker
|
|
|
|
|
|
def test_active_session_lease_hides_live_evidence_until_release(tmp_path: Path) -> None:
|
|
sessions = tmp_path / "sessions"
|
|
session = sessions / "20260717T011050Z_viewer_live"
|
|
|
|
lease = ActiveSessionLease.acquire(sessions, session)
|
|
try:
|
|
session.mkdir()
|
|
capture = session / "captures" / "mqtt_live"
|
|
capture.mkdir(parents=True)
|
|
(capture / "mqtt.raw.k1mqtt").write_bytes(b"K1MQTT\x00unfinished")
|
|
candidates = discover_legacy_viewer_sessions(sessions)
|
|
assert candidates == ()
|
|
assert recover_stale_active_session_marker(sessions) is False
|
|
finally:
|
|
lease.release()
|
|
|
|
assert not (sessions / ".current_session").exists()
|
|
candidates = discover_legacy_viewer_sessions(sessions)
|
|
assert len(candidates) == 1
|
|
assert candidates[0].session_id == session.name
|
|
assert candidates[0].replayable is False
|
|
|
|
|
|
def test_startup_recovery_removes_only_an_unlocked_stale_marker(tmp_path: Path) -> None:
|
|
sessions = tmp_path / "sessions"
|
|
sessions.mkdir()
|
|
marker = sessions / ".current_session"
|
|
marker.write_text("20260717T011050Z_viewer_live\n", encoding="utf-8")
|
|
|
|
assert recover_stale_active_session_marker(sessions) is True
|
|
assert not marker.exists()
|
|
assert recover_stale_active_session_marker(sessions) is False
|