141 lines
5.4 KiB
Python
141 lines
5.4 KiB
Python
"""Lab054: фиксированные RF-команды и отказы runner без оборудования."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
|
|
import pytest
|
|
|
|
from experiments import lab054_hardware_control as lab
|
|
from tools.hackrf_process_check import ProcessCheck, ProcessState
|
|
|
|
|
|
@pytest.mark.parametrize("failure", [None, "master", "slave", "clkin", "receive"])
|
|
def test_clock_gates_never_transmit(failure):
|
|
commands = []
|
|
|
|
def runner(command):
|
|
commands.append(command)
|
|
output, code = "", 0
|
|
if "--si5351c" in command:
|
|
serial = command[command.index("-d") + 1]
|
|
bad = ((failure == "master" and serial == lab.clock.MASTER_SERIAL)
|
|
or (failure == "slave" and serial == lab.clock.SLAVE_SERIAL))
|
|
output = "0x0e" if bad else "0x26"
|
|
elif "-i" in command:
|
|
output = "UNKNOWN" if failure == "clkin" else "CLKIN status: clock signal detected"
|
|
elif "-r" in command:
|
|
code = int(failure == "receive")
|
|
return subprocess.CompletedProcess(command, code, output, "")
|
|
|
|
guard = lab.clock.TransmitGuard()
|
|
if failure:
|
|
with pytest.raises(RuntimeError):
|
|
lab.clock_preflight(runner, guard, sleep_function=lambda _: None)
|
|
assert not guard.tx_allowed
|
|
else:
|
|
assert lab.clock_preflight(runner, guard, sleep_function=lambda _: None)["passed"]
|
|
assert guard.tx_allowed
|
|
assert not any("-t" in command for command in commands)
|
|
for command in commands:
|
|
if Path(command[0]).name == "hackrf_transfer.exe":
|
|
assert command[command.index("-f") + 1] == "1400000000"
|
|
assert command[command.index("-l") + 1] == command[command.index("-g") + 1] == "24"
|
|
|
|
|
|
def test_authorization_precedes_all_side_effects(tmp_path):
|
|
with pytest.raises(PermissionError):
|
|
lab.capture_once(tmp_path, runner=lambda _: pytest.fail("hardware"))
|
|
assert list(tmp_path.iterdir()) == []
|
|
|
|
|
|
@pytest.mark.parametrize("failure", [None, "clock", "early_rx", "tx", "cleanup"])
|
|
def test_one_shot_order_and_no_retry(tmp_path, monkeypatch, failure):
|
|
events = []
|
|
monkeypatch.setattr(lab, "validate_prepared", lambda directory: {"tx_samples": 123})
|
|
monkeypatch.setattr(lab.phy.lab045, "_verify_device_inventory", lambda runner: {})
|
|
|
|
def preflight(runner, guard):
|
|
if failure == "clock":
|
|
raise RuntimeError("clock")
|
|
guard.allow()
|
|
return {"passed": True}
|
|
|
|
monkeypatch.setattr(lab, "clock_preflight", preflight)
|
|
monkeypatch.setattr(lab, "disable_clock_outputs", lambda runner: events.append("clocks_off") or [])
|
|
|
|
class Process:
|
|
def poll(self):
|
|
return 1 if failure == "early_rx" else None
|
|
|
|
def wait(self, timeout):
|
|
with (tmp_path / "rx_stream.cs8").open("wb") as stream:
|
|
stream.truncate(lab.RX_SAMPLES * 2)
|
|
return 0
|
|
|
|
def close_logs(self):
|
|
pass
|
|
|
|
def terminate(self):
|
|
events.append("terminate")
|
|
if failure == "cleanup":
|
|
raise RuntimeError("cleanup")
|
|
|
|
def starter(command, stdout, stderr):
|
|
events.append("rx")
|
|
stdout.write_text("")
|
|
stderr.write_text("")
|
|
return Process()
|
|
|
|
def runner(command):
|
|
assert "-t" in command and "-R" not in command
|
|
events.append("tx")
|
|
return subprocess.CompletedProcess(command, int(failure == "tx"), "", "")
|
|
|
|
kwargs = dict(authorization=True, runner=runner, process_starter=starter,
|
|
process_checker=lambda: ProcessCheck(ProcessState.CLEAR, ()),
|
|
sleep_function=lambda _: events.append("lead"))
|
|
if failure:
|
|
with pytest.raises(RuntimeError):
|
|
lab.capture_once(tmp_path, **kwargs)
|
|
else:
|
|
assert lab.capture_once(tmp_path, **kwargs)["passed"]
|
|
assert events.count("tx") == (0 if failure in ("clock", "early_rx") else 1)
|
|
if "tx" in events:
|
|
assert events.index("rx") < events.index("lead") < events.index("tx")
|
|
assert "clocks_off" in events
|
|
assert json.loads((tmp_path / "capture_result.json").read_text())["passed"] == (failure is None)
|
|
before = events.copy()
|
|
with pytest.raises(FileExistsError):
|
|
lab.capture_once(tmp_path, **kwargs)
|
|
assert events == before
|
|
|
|
|
|
def test_clock_cleanup_continues_after_timeout():
|
|
commands = []
|
|
|
|
def runner(command):
|
|
commands.append(command)
|
|
raise subprocess.TimeoutExpired(command, 1)
|
|
|
|
errors = lab.disable_clock_outputs(runner)
|
|
assert len(commands) == len(errors) == 2
|
|
assert [command[2] for command in commands] == [lab.clock.MASTER_SERIAL, lab.clock.SLAVE_SERIAL]
|
|
|
|
|
|
def test_payload_match_does_not_hide_changed_packet():
|
|
from types import SimpleNamespace
|
|
|
|
plan = lab.plans()[0].plan
|
|
packet = bytes([plan.packet[0] ^ 1]) + plan.packet[1:]
|
|
reception = SimpleNamespace(payload_bytes=plan.payload, packet_bytes=packet)
|
|
frame = SimpleNamespace(reception=reception,
|
|
post_preamble_soft=lab.phy.lab045.modulate_bits_to_ideal_soft(
|
|
__import__("numpy").concatenate((plan.phy_header_bits, plan.interleaved_bits))))
|
|
result = lab.compare_after_verdict([(None, SimpleNamespace(accepted=True, frame=frame))])
|
|
assert result["payload_matched_packets"] == 1
|
|
assert result["exact_packets"] == result["unique_exact_packets"] == 0
|
|
assert result["post_fec_errors"] == 1
|