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