Lab056: compare 20 and 500 ksymbol/s in controlled RF session

tests/test_hackrf_process_check.py

@@ -0,0 +1,159 @@
+"""Достоверность перечисления процессов и запрет аппаратных действий при FAIL."""
+
+from __future__ import annotations
+
+import json
+import subprocess
+from types import SimpleNamespace
+
+import pytest
+
+from experiments import lab054_hardware_control as lab
+from tools import hackrf_process_check as check
+
+
+def snapshot(names=("python.exe",), errors=()):
+    return check.classify_snapshot("fixture", [check.ProcessIdentity(index + 1, name)
+                                               for index, name in enumerate(names)], errors)
+
+
+def completed(stdout="", stderr="", code=0):
+    return lambda command: subprocess.CompletedProcess(command, code, stdout, stderr)
+
+
+def test_clear_full_snapshot():
+    result = check.check_processes([lambda: snapshot()])
+    assert result.state is check.ProcessState.CLEAR
+    check.require_clear(result)
+
+
+@pytest.mark.parametrize("name", ["hackrf_transfer.exe", "hackrf_clock.exe", "HACKRF_DEBUG.EXE",
+                                  "hackrf_info.exe", "sdrangel.exe", "SDRSharp", "gqrx"])
+def test_busy_known_sdr_clients(name):
+    result = check.check_processes([lambda: snapshot((name,), errors=("partial snapshot",)),
+                                   lambda: pytest.fail("BUSY запрещает fallback")])
+    assert result.state is check.ProcessState.BUSY
+    with pytest.raises(RuntimeError):
+        check.require_clear(result)
+
+
+@pytest.mark.parametrize("names,errors", [((), ()), (("python",), ("Access denied",)), (("",), ())])
+def test_unknown_never_becomes_clear(names, errors):
+    result = check.check_processes([lambda: snapshot(names, errors)])
+    assert result.state is check.ProcessState.UNKNOWN
+    with pytest.raises(RuntimeError):
+        check.require_clear(result)
+
+
+def test_tasklist_access_denied_is_unknown():
+    result = check.scan_tasklist(completed(stderr="ERROR: Access denied", code=1))
+    assert result.state is check.ProcessState.UNKNOWN
+    assert "Access denied" in str(result.errors)
+
+
+def test_fallback_to_powershell_after_psutil_unavailable(monkeypatch):
+    powershell = check.scan_powershell
+    monkeypatch.setattr(check, "scan_psutil", lambda: snapshot((), ("ImportError: psutil",)))
+    monkeypatch.setattr(check, "scan_powershell", lambda: powershell(
+        completed('[{"Id":12,"ProcessName":"python"}]')))
+    monkeypatch.setattr(check, "scan_tasklist", lambda: pytest.fail("Полный CLEAR уже получен"))
+    result = check.check_processes()
+    assert result.state is check.ProcessState.CLEAR
+    assert [item.state for item in result.attempts] == [check.ProcessState.UNKNOWN, check.ProcessState.CLEAR]
+
+
+def test_tasklist_is_last_fallback(monkeypatch):
+    tasklist = check.scan_tasklist
+    monkeypatch.setattr(check, "scan_psutil", lambda: snapshot((), ("unavailable",)))
+    monkeypatch.setattr(check, "scan_powershell", lambda: snapshot((), ("denied",)))
+    monkeypatch.setattr(check, "scan_tasklist", lambda: tasklist(
+        completed('"python.exe","123","Console","1","123 K"\n')))
+    result = check.check_processes()
+    assert result.state is check.ProcessState.CLEAR
+    assert len(result.attempts) == 3
+
+
+@pytest.mark.parametrize("output", ["", "[]", "null", "not json", '[{"Id":4,"ProcessName":null}]'])
+def test_powershell_malformed_not_clear(output):
+    assert check.scan_powershell(completed(output)).state is check.ProcessState.UNKNOWN
+
+
+@pytest.mark.parametrize("output", ["", "INFO: No tasks are running", '"python.exe","bad pid","a","b","c"'])
+def test_tasklist_malformed_not_clear(output):
+    assert check.scan_tasklist(completed(output)).state is check.ProcessState.UNKNOWN
+
+
+def test_provider_exception_and_empty_provider_list_unknown():
+    def failed():
+        raise PermissionError("denied")
+    assert check.check_processes([failed]).state is check.ProcessState.UNKNOWN
+    assert check.check_processes([]).state is check.ProcessState.UNKNOWN
+
+
+@pytest.mark.parametrize("mechanism", ["powershell", "tasklist"])
+@pytest.mark.parametrize("malformed_first", [False, True])
+def test_busy_survives_malformed_neighbor(mechanism, malformed_first):
+    if mechanism == "powershell":
+        rows = [{"Id":42,"ProcessName":"hackrf_transfer"}, {}]
+        if malformed_first:
+            rows.reverse()
+        scan = check.scan_powershell(completed(json.dumps(rows)))
+    else:
+        rows = ['"hackrf_transfer.exe","42","Console","1","100 K"', "BROKEN"]
+        if malformed_first:
+            rows.reverse()
+        scan = check.scan_tasklist(completed("\n".join(rows)))
+    assert scan.errors
+    result = check.check_processes([lambda: scan, lambda: pytest.fail("Найденный BUSY нельзя потерять")])
+    assert result.state is check.ProcessState.BUSY
+
+
+@pytest.mark.parametrize("disappeared", [False, True])
+def test_psutil_access_denied_or_exited_process(monkeypatch, disappeared):
+    class NoSuchProcess(Exception):
+        pass
+
+    def unreadable():
+        raise NoSuchProcess() if disappeared else PermissionError("Access denied")
+
+    module = SimpleNamespace(NoSuchProcess=NoSuchProcess, process_iter=lambda: iter([
+        SimpleNamespace(pid=1, name=lambda: "python.exe"),
+        SimpleNamespace(pid=2, name=unreadable)]))
+    monkeypatch.setattr(check.importlib, "import_module", lambda name: module)
+    result = check.scan_psutil()
+    assert result.state is (check.ProcessState.CLEAR if disappeared else check.ProcessState.UNKNOWN)
+
+
+@pytest.mark.parametrize("state", list(check.ProcessState))
+def test_preflight_only_clear_reaches_hardware(tmp_path, monkeypatch, state):
+    actions = []
+    scan = snapshot(("hackrf_transfer.exe",)) if state is check.ProcessState.BUSY else snapshot()
+    if state is check.ProcessState.UNKNOWN:
+        scan = snapshot((), ("tasklist Access denied",))
+    result = check.ProcessCheck(state, (scan,))
+    monkeypatch.setattr(lab.phy.lab045, "_verify_device_inventory", lambda runner: actions.append("inventory") or {})
+    monkeypatch.setattr(lab, "clock_preflight", lambda runner, guard: actions.append("clock") or {"passed": True})
+    monkeypatch.setattr(lab, "disable_clock_outputs", lambda runner: actions.append("cleanup") or [])
+    if state is check.ProcessState.CLEAR:
+        assert lab.preflight_only(tmp_path, runner=lambda _: None, process_checker=lambda: result)["passed"]
+        assert actions == ["inventory", "clock", "cleanup"]
+    else:
+        with pytest.raises(RuntimeError):
+            lab.preflight_only(tmp_path, runner=lambda _: None, process_checker=lambda: result)
+        assert actions == []
+    report = json.loads((tmp_path / "common_clock_preflight.json").read_text(encoding="utf-8"))
+    assert report["process_check"]["state"] == state.value
+    assert report["rf_tx_count"] == 0
+
+
+def test_unknown_after_cleanup_rejects_preflight(tmp_path, monkeypatch):
+    states = iter([check.ProcessCheck(check.ProcessState.CLEAR, (snapshot(),)),
+                   check.ProcessCheck(check.ProcessState.UNKNOWN, (snapshot((), ("denied",)),))])
+    monkeypatch.setattr(lab.phy.lab045, "_verify_device_inventory", lambda runner: {})
+    monkeypatch.setattr(lab, "clock_preflight", lambda runner, guard: {"passed": True})
+    monkeypatch.setattr(lab, "disable_clock_outputs", lambda runner: [])
+    with pytest.raises(RuntimeError):
+        lab.preflight_only(tmp_path, runner=lambda _: None, process_checker=lambda: next(states))
+    report = json.loads((tmp_path / "common_clock_preflight.json").read_text(encoding="utf-8"))
+    assert not report["passed"]
+    assert report["process_check_after_cleanup"]["state"] == "UNKNOWN"