143 lines
6.4 KiB
Python
143 lines
6.4 KiB
Python
"""Lab062: программные условия и запрет TX до отдельного разрешения."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict
|
|
from pathlib import Path
|
|
import subprocess
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from experiments import lab062_full_packet_rf_preflight as lab
|
|
from tools.hackrf_process_check import ProcessCheck, ProcessState
|
|
|
|
|
|
def test_fixed_packet_and_operating_point():
|
|
assert len(lab.plans()) == 16
|
|
assert all(len(frame.plan.payload) == 256 for frame in lab.plans())
|
|
assert lab.parameters()["rf_frequency_hz"] == 1_400_000_000
|
|
assert lab.parameters()["symbol_rate"] == 500_000
|
|
assert lab.parameters()["tx_vga_db"] == 10
|
|
assert lab.parameters()["rx_lna_db"] == lab.parameters()["rx_vga_db"] == 24
|
|
assert lab.POINT.distance_m == 1.2 and lab.POINT.visibility == "LOS indoor"
|
|
|
|
|
|
def test_rx_and_tx_commands_are_fixed_and_distinct():
|
|
receive = lab.transfer_command("rx.cs8", 800_000)
|
|
transmit = lab.transfer_command("tx.cs8", 100, transmit=True)
|
|
assert "-r" in receive and "-t" not in receive
|
|
assert "-t" in transmit and "-r" not in transmit
|
|
assert receive[receive.index("-f") + 1] == "1400000000"
|
|
assert receive[receive.index("-l") + 1] == receive[receive.index("-g") + 1] == "24"
|
|
assert transmit[transmit.index("-x") + 1] == "10"
|
|
|
|
|
|
@pytest.mark.parametrize("failure", [None, "clock", "clipping", "strong", "overrun"])
|
|
def test_preflight_never_transmits_and_fails_closed(tmp_path, monkeypatch, failure):
|
|
raw, output = tmp_path / "raw", tmp_path / "processed"
|
|
raw.mkdir(); output.mkdir()
|
|
monkeypatch.setattr(lab, "RAW", raw); monkeypatch.setattr(lab, "OUTPUT", output)
|
|
(raw / "tx_stream.cs8").write_bytes(b"test")
|
|
metadata = {"tx_sha256": "TEST"}
|
|
(output / "software_gate.json").write_text('{"passed": true, "tx_sha256": "TEST"}')
|
|
monkeypatch.setattr(lab, "validate_prepared", lambda: metadata)
|
|
monkeypatch.setattr(lab.lab054.phy.lab045, "_verify_device_inventory", lambda runner: {"passed": True})
|
|
def clock(point, runner, guard, **kwargs):
|
|
return {"passed": failure != "clock", "master_reg29": 0x26,
|
|
"slave_reg29": 0x26, "slave_clkin_status": "DETECTED"}
|
|
def runner(command):
|
|
assert "-t" not in command
|
|
if "-r" in command:
|
|
value = 127 if failure == "clipping" else (110 if failure == "strong" else 1)
|
|
Path(command[command.index("-r") + 1]).write_bytes(bytes([value, value]) * 800_000)
|
|
log = "1 RX overruns" if failure == "overrun" else "0 RX overruns"
|
|
return subprocess.CompletedProcess(command, 0, log, "")
|
|
monkeypatch.setattr(lab.lab054, "disable_clock_outputs", lambda runner: [])
|
|
kwargs = dict(runner=runner, clock_checker=clock,
|
|
process_checker=lambda: ProcessCheck(ProcessState.CLEAR, ()),
|
|
sleep_function=lambda _: None)
|
|
if failure:
|
|
with pytest.raises(RuntimeError):
|
|
lab.preflight(**kwargs)
|
|
else:
|
|
assert lab.preflight(**kwargs)["passed"]
|
|
result = __import__("json").loads((raw / "preflight_result.json").read_text())
|
|
assert result["rf_tx_count"] == 0
|
|
if failure is None:
|
|
assert "symbol_energy_code2_seconds" not in result["rx_level"]
|
|
assert result["rx_level"]["symbol_energy"] == "NOT_APPLICABLE_TO_BACKGROUND"
|
|
|
|
|
|
def test_second_preflight_is_refused_before_hardware(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(lab, "RAW", tmp_path)
|
|
monkeypatch.setattr(lab, "OUTPUT", tmp_path)
|
|
(tmp_path / "preflight_attempt.json").write_text("{}")
|
|
monkeypatch.setattr(lab, "validate_prepared", lambda: {"tx_sha256": "TEST"})
|
|
(tmp_path / "software_gate.json").write_text('{"passed": true, "tx_sha256": "TEST"}')
|
|
with pytest.raises(FileExistsError):
|
|
lab.preflight(runner=lambda command: pytest.fail("hardware"))
|
|
|
|
|
|
def test_capture_refuses_missing_authorization_before_files_or_hardware(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(lab, "RAW", tmp_path)
|
|
monkeypatch.setattr(lab, "OUTPUT", tmp_path)
|
|
with pytest.raises(PermissionError):
|
|
lab.capture_once(
|
|
authorization=False,
|
|
runner=lambda command: pytest.fail("hardware"),
|
|
)
|
|
assert not (tmp_path / "hardware_attempt.json").exists()
|
|
|
|
|
|
def test_capture_is_rx_first_single_tx_and_refuses_second_attempt(tmp_path, monkeypatch):
|
|
raw, output = tmp_path / "raw", tmp_path / "processed"
|
|
raw.mkdir(); output.mkdir()
|
|
monkeypatch.setattr(lab, "RAW", raw); monkeypatch.setattr(lab, "OUTPUT", output)
|
|
tx = raw / "tx_stream.cs8"; tx.write_bytes(b"\x01\x01" * 100)
|
|
metadata = {"tx_sha256": lab.sha256(tx), "tx_samples": 100}
|
|
monkeypatch.setattr(lab, "validate_prepared", lambda: metadata)
|
|
monkeypatch.setattr(lab, "require_preflight", lambda value: {})
|
|
monkeypatch.setattr(lab, "RX_SAMPLES", 800_000)
|
|
monkeypatch.setattr(lab.lab054.phy.lab045, "_verify_device_inventory", lambda runner: {"passed": True})
|
|
monkeypatch.setattr(lab.lab054, "disable_clock_outputs", lambda runner: [])
|
|
events = []
|
|
|
|
class Process:
|
|
def poll(self): return None
|
|
def wait(self, timeout): return 0
|
|
def close_logs(self): pass
|
|
def terminate(self): pass
|
|
|
|
def starter(command, stdout, stderr):
|
|
events.append("RX")
|
|
Path(command[command.index("-r") + 1]).write_bytes(b"\x01\x01" * 800_000)
|
|
Path(stdout).write_text("")
|
|
Path(stderr).write_text("0 RX overruns")
|
|
return Process()
|
|
|
|
def runner(command):
|
|
if "-t" in command:
|
|
events.append("TX")
|
|
return subprocess.CompletedProcess(command, 0, "0 TX underruns", "")
|
|
return subprocess.CompletedProcess(command, 0, "", "")
|
|
|
|
def clock(point, runner, guard, **kwargs):
|
|
guard.tx_allowed = True
|
|
return {
|
|
"passed": True, "master_reg29": 0x26, "slave_reg29": 0x26,
|
|
"slave_clkin_status": "DETECTED",
|
|
}
|
|
|
|
result = lab.capture_once(
|
|
authorization=True, runner=runner, process_starter=starter,
|
|
process_checker=lambda: ProcessCheck(ProcessState.CLEAR, ()),
|
|
clock_checker=clock, sleep_function=lambda _: None,
|
|
)
|
|
assert events == ["RX", "TX"]
|
|
assert result["rf_tx_count"] == 1 and result["automatic_tx_repeats"] == 0
|
|
assert result["rx_overruns"] == result["tx_underruns"] == 0
|
|
assert json.loads((raw / "capture_result.json").read_text())["passed"]
|
|
with pytest.raises(FileExistsError):
|
|
lab.capture_once(authorization=True, runner=lambda command: pytest.fail("hardware"))
|