"""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"))