"""Достоверность перечисления процессов и запрет аппаратных действий при 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"