"""Software-only tests for the fixed-command RX HTTP agent.""" from __future__ import annotations import argparse from datetime import datetime, timedelta, timezone import hashlib import json from pathlib import Path import shutil from typing import Sequence import pytest import tools.rf001_agent_update as update_module import tools.rf001_rx_http_agent as agent_module from tools.rf001_http_control import PROTOCOL_VERSION, ProtocolError from tools.rf001_rx_http_agent import ( CommandResult, collect_inventory, execute_task, normalized_agent_id, read_git_metadata, ) def fake_runner(arguments: Sequence[str], timeout: float) -> CommandResult: del timeout command = list(arguments) if command == ["hostname"]: stdout = "User-PC" elif command == ["whoami"]: stdout = "user-pc\\professional" elif command[0] == "git": stdout = "fake-git-read-only-result" elif command[0].lower().endswith("hackrf_info.exe"): stdout = "hackrf_info version: 2024.02.1" else: # pragma: no cover - protects the test allowlist raise AssertionError(f"unexpected command: {command}") return CommandResult(command, 0, stdout, "") def test_agent_id_is_normalized_without_shell_content() -> None: assert normalized_agent_id("User-PC") == "user-pc" assert normalized_agent_id(" RX Laptop! ") == "rx-laptop" def test_inventory_runs_only_fixed_read_only_commands(tmp_path: Path) -> None: repository = tmp_path / "repo" repository.mkdir() hackrf_info = tmp_path / "hackrf_info.exe" hackrf_info.write_bytes(b"") inventory, errors = collect_inventory( repository, include_hackrf_info=True, hackrf_info_path=hackrf_info, command_runner=fake_runner, ) assert errors == [] assert inventory["rf_tx"] == 0 assert inventory["hostname_command"]["stdout"] == "User-PC" assert inventory["whoami_command"]["stdout"] == "user-pc\\professional" assert inventory["hackrf_info"]["executed"] is True assert inventory["hackrf_info"]["rf_tx"] == 0 def test_agent_rejects_unknown_task_and_rf_tx(tmp_path: Path) -> None: base_task: dict[str, object] = { "protocol_version": PROTOCOL_VERSION, "task_id": "INVENTORY_TEST_001", "kind": "INVENTORY", "agent_id": "rx-user-pc", "created_at": "2026-09-23T00:00:00+00:00", "include_hackrf_info": False, "rf_tx": 0, } unknown = dict(base_task) unknown["kind"] = "SHELL" with pytest.raises(ProtocolError, match="unsupported task kind"): execute_task( unknown, agent_id="rx-user-pc", repository=tmp_path, hackrf_info_path=None, command_runner=fake_runner, ) transmit = dict(base_task) transmit["rf_tx"] = 1 with pytest.raises(ProtocolError, match="rf_tx = 0"): execute_task( transmit, agent_id="rx-user-pc", repository=tmp_path, hackrf_info_path=None, command_runner=fake_runner, ) def test_agent_dispatches_fixed_capture_diagnostics_without_hardware( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: calls: list[tuple[Path, str, str, str]] = [] def diagnostics( work_root: Path, session_id: str, tx_started_at: str, tx_finished_at: str, ) -> tuple[dict[str, object], list[str]]: calls.append((work_root, session_id, tx_started_at, tx_finished_at)) return { "session_id": session_id, "hardware_accessed": False, "rf_tx": 0, "report": {"detected_preambles": 0}, }, [] monkeypatch.setattr(agent_module, "read_capture_diagnostics", diagnostics) session_id = "RF001A_20260925T104504Z_5FA2C83B" task = { "protocol_version": PROTOCOL_VERSION, "task_id": "DIAG_TEST_001", "kind": "CAPTURE_DIAGNOSTICS", "agent_id": "rx-user-pc", "created_at": "2026-09-25T10:51:00+00:00", "session_id": session_id, "tx_started_at": "2026-09-25T10:46:54+00:00", "tx_finished_at": "2026-09-25T10:46:56+00:00", "rf_tx": 0, } result = execute_task( task, agent_id="rx-user-pc", repository=tmp_path, hackrf_info_path=None, work_root=tmp_path / "runtime", command_runner=fake_runner, ) assert result["status"] == "COMPLETE" assert result["result"]["hardware_accessed"] is False assert result["rf_tx"] == 0 assert calls == [ ( tmp_path / "runtime", session_id, "2026-09-25T10:46:54+00:00", "2026-09-25T10:46:56+00:00", ) ] def _write_git_metadata(repository: Path, head: str) -> None: git_dir = repository / ".git" ref = git_dir / "refs" / "heads" / "video-development" ref.parent.mkdir(parents=True) (git_dir / "HEAD").write_text( "ref: refs/heads/video-development\n", encoding="ascii", ) ref.write_text(head + "\n", encoding="ascii") def test_read_git_metadata_without_git_executable(tmp_path: Path) -> None: repository = tmp_path / "repo" repository.mkdir() head = "a94f2db4ee8377201a97263166cb09431df5f5d6" _write_git_metadata(repository, head) metadata = read_git_metadata(repository) assert metadata["branch"] == "video-development" assert metadata["head"] == head assert metadata["status"] == "NOT_RUN_NO_GIT_EXECUTABLE" def test_read_git_metadata_from_linked_worktree(tmp_path: Path) -> None: repository = tmp_path / "worktree" repository.mkdir() common = tmp_path / "main.git" worktree_git = common / "worktrees" / "worktree" worktree_git.mkdir(parents=True) (repository / ".git").write_text( f"gitdir: {worktree_git}\n", encoding="utf-8", ) (worktree_git / "commondir").write_text("../..\n", encoding="ascii") (worktree_git / "HEAD").write_text( "ref: refs/heads/video-development\n", encoding="ascii", ) head = "a94f2db4ee8377201a97263166cb09431df5f5d6" ref = common / "refs" / "heads" / "video-development" ref.parent.mkdir(parents=True) ref.write_text(head + "\n", encoding="ascii") metadata = read_git_metadata(repository) assert metadata["git_dir"] == str(worktree_git.resolve()) assert metadata["common_dir"] == str(common.resolve()) assert metadata["branch"] == "video-development" assert metadata["head"] == head def test_offline_verify_is_fixed_software_only_task(tmp_path: Path) -> None: repository = tmp_path / "repo" repository.mkdir() head = "a94f2db4ee8377201a97263166cb09431df5f5d6" _write_git_metadata(repository, head) (repository / "tools").mkdir() (repository / "configs" / "rf").mkdir(parents=True) for relative in ( Path("tools/rf001a_prepare.py"), Path("tools/qpsk_rf_rx.py"), Path("configs/rf/rf001a_truth_metadata.json"), ): (repository / relative).write_text("fixture\n", encoding="utf-8") commands: list[list[str]] = [] def offline_runner(arguments: Sequence[str], timeout: float) -> CommandResult: del timeout command = list(arguments) commands.append(command) if "rf001a_prepare.py" in command[4]: output = Path(command[command.index("--output-dir") + 1]) output.mkdir() (output / "tx_waveform.cs8").write_bytes(b"fixture") (output / "tx_metadata.json").write_text("{}", encoding="utf-8") elif "rf001_offline_decode.py" in command[4]: report = Path(command[command.index("--report") + 1]) report.write_text( '{"crc_valid_packets": 100, "exact_packets": 100}', encoding="utf-8", ) else: # pragma: no cover - protects the fixed command contract raise AssertionError(command) return CommandResult(command, 0, "PASS", "") task = { "protocol_version": PROTOCOL_VERSION, "task_id": "OFFLINE_TEST_001", "kind": "OFFLINE_VERIFY", "agent_id": "rx-user-pc", "created_at": "2026-09-23T00:00:00+00:00", "expected_head": head, "rf_tx": 0, } result = execute_task( task, agent_id="rx-user-pc", repository=repository, hackrf_info_path=None, work_root=tmp_path / "runtime", command_runner=offline_runner, ) assert result["status"] == "COMPLETE" assert result["result"]["hardware_accessed"] is False assert result["result"]["crc_valid_packets"] == 100 assert result["result"]["exact_packets"] == 100 assert "--repository" in commands[1] assert all("hackrf" not in Path(command[0]).name.lower() for command in commands) assert all("-t" not in command for command in commands) assert result["rf_tx"] == 0 def _sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def _staircase_fixture( tmp_path: Path, ) -> tuple[Path, dict[str, object], Path, Path]: work_root = tmp_path / "runtime" session_id = "RF001A_20260925T125829Z_227EADB4" session_directory = work_root / "rx_capture" / session_id session_directory.mkdir(parents=True) capture = session_directory / "rf001a_ota_capture.cs8" capture.write_bytes(b"fixed-rx-capture") reference = work_root / "references" / "lab065_reference_staircase.cs8" reference.parent.mkdir(parents=True) reference.write_bytes(b"fixed-staircase-reference") task: dict[str, object] = { "protocol_version": PROTOCOL_VERSION, "task_id": "STAIRCASE_TEST_001", "kind": "STAIRCASE_SEARCH", "agent_id": "rx-user-pc", "created_at": "2026-09-28T12:00:00+00:00", "session_id": session_id, "expected_rx_sha256": _sha256(capture), "expected_reference_sha256": _sha256(reference), "hardware_accessed": False, "automatic_retry": False, "rf_tx": 0, } return work_root, task, capture, reference def _write_staircase_report(arguments: Sequence[str]) -> Path: command = list(arguments) report_path = Path(command[command.index("--report") + 1]) report = { "analyzer_version": "test", "mode": "OFFLINE_REFERENCE_STAIRCASE_SEARCH", "capture_sha256": command[command.index("--capture-sha256") + 1], "reference_sha256": _sha256( Path(command[command.index("--reference") + 1]) ), "search": { "coverage_start_sample": 0, "coverage_stop_sample": 8, "coverage_seconds": 0.000001, "coverage_complete": True, "source_samples_processed": 8, "frequency_grid_hz": [-3.0, 0.0, 3.0], "background_score_p99": 0.01, "candidates": [], }, "verifications": [], "verdict": "NOT_DETECTED", "verdict_reasons": ["NO_CANDIDATES"], "analysis_seconds": 1.25, "peak_process_memory_bytes": 1234, "peak_additional_memory_bytes": 456, "hardware_accessed": False, "rf_tx": 0, } report_path.write_text(json.dumps(report), encoding="utf-8") return report_path def test_staircase_task_rejects_invalid_session_and_hashes(tmp_path: Path) -> None: work_root, task, _capture, _reference = _staircase_fixture(tmp_path) invalid_session = dict(task, session_id="..\\outside") with pytest.raises(ProtocolError, match="session_id"): execute_task( invalid_session, agent_id="rx-user-pc", repository=tmp_path, hackrf_info_path=None, work_root=work_root, ) for field in ("expected_rx_sha256", "expected_reference_sha256"): invalid_hash = dict(task) invalid_hash[field] = "invalid" with pytest.raises(ProtocolError, match=field): execute_task( invalid_hash, agent_id="rx-user-pc", repository=tmp_path, hackrf_info_path=None, work_root=work_root, ) def test_staircase_hash_mismatch_does_not_start_child(tmp_path: Path) -> None: work_root, task, _capture, _reference = _staircase_fixture(tmp_path) calls: list[list[str]] = [] def forbidden_runner(arguments: Sequence[str], timeout: float) -> CommandResult: del timeout calls.append(list(arguments)) raise AssertionError("child process must not run after a hash mismatch") for field in ("expected_rx_sha256", "expected_reference_sha256"): changed = dict(task) changed[field] = "0" * 64 result = execute_task( changed, agent_id="rx-user-pc", repository=tmp_path, hackrf_info_path=None, work_root=work_root, command_runner=forbidden_runner, ) assert result["status"] == "FAILED" assert result["result"]["reference_detection"] == "EXECUTION_FAILED" assert calls == [] def test_staircase_repeat_reuses_existing_report_without_child( tmp_path: Path, ) -> None: work_root, task, _capture, _reference = _staircase_fixture(tmp_path) calls: list[list[str]] = [] def analyzer_runner(arguments: Sequence[str], timeout: float) -> CommandResult: assert timeout == 43_200.0 calls.append(list(arguments)) _write_staircase_report(arguments) assert "hackrf" not in " ".join(arguments).lower() return CommandResult(list(arguments), 0, "REFERENCE STAIRCASE = NOT_DETECTED", "") first = execute_task( task, agent_id="rx-user-pc", repository=tmp_path, hackrf_info_path=None, work_root=work_root, command_runner=analyzer_runner, ) second = execute_task( task, agent_id="rx-user-pc", repository=tmp_path, hackrf_info_path=None, work_root=work_root, command_runner=lambda *_args: (_ for _ in ()).throw( AssertionError("existing report must not start another child") ), ) assert first["status"] == "COMPLETE" assert first["result"]["hardware_accessed"] is False assert first["result"]["rf_tx"] == 0 assert first["result"]["reused_existing_report"] is False assert second["status"] == "COMPLETE" assert second["result"]["reused_existing_report"] is True assert len(calls) == 1 def test_staircase_child_failure_is_execution_failed(tmp_path: Path) -> None: work_root, task, _capture, _reference = _staircase_fixture(tmp_path) def failed_runner(arguments: Sequence[str], timeout: float) -> CommandResult: del timeout return CommandResult(list(arguments), 9, "", "fixed failure") result = execute_task( task, agent_id="rx-user-pc", repository=tmp_path, hackrf_info_path=None, work_root=work_root, command_runner=failed_runner, ) assert result["status"] == "FAILED" assert result["result"]["reference_detection"] == "EXECUTION_FAILED" assert result["result"]["hardware_accessed"] is False assert result["result"]["rf_tx"] == 0 assert "exited with 9" in result["errors"][0] def _update_fixture( tmp_path: Path, ) -> tuple[ Path, bytes, dict[str, object], dict[str, object], argparse.Namespace, ]: source = tmp_path / "source" for index, relative in enumerate(update_module.RELEASE_SOURCE_FILES.values()): path = source / relative path.parent.mkdir(parents=True, exist_ok=True) content = f"# fixed update fixture {index}\n" if relative == update_module.RELEASE_SOURCE_FILES[ "app/tools/rf001_rx_http_agent.py" ]: content += 'AGENT_VERSION = "2026.09.29.2"\n' path.write_text(content, encoding="utf-8") release_id = "rf001-agent-2026.09.29.2" package = tmp_path / f"{release_id}.zip" inspection = update_module.build_release_archive( source, package, release_id=release_id, agent_version="2026.09.29.2", protocol_version=PROTOCOL_VERSION, ) manifest: dict[str, object] = { **inspection.descriptor(), "package_kind": "AGENT_RELEASE", } token = b"fixed-test-token-0123456789abcdef" task: dict[str, object] = { "protocol_version": PROTOCOL_VERSION, "task_id": "UPDATE_TEST_001", "kind": "UPDATE_AGENT", "agent_id": "rx-user-pc", "created_at": datetime.now(timezone.utc).isoformat(), "release_id": release_id, "agent_version": "2026.09.29.2", "package_sha256": inspection.package_sha256, "package_bytes": inspection.package_bytes, "content_manifest_sha256": inspection.content_manifest_sha256, "compatible_protocol_version": PROTOCOL_VERSION, "approval_expires_at": ( datetime.now(timezone.utc) + timedelta(hours=1) ).isoformat(), "automatic_retry": False, "hardware_accessed": False, "rf_tx": 0, } install_root = tmp_path / "install" install_root.mkdir() launcher_status = install_root / "launcher_status.json" launcher_status.write_text( json.dumps( { "active_release_id": "rf001-agent-2026.09.29.1", "active_agent_version": "2026.09.29.1", "rf_tx": 0, } ), encoding="utf-8", ) arguments = argparse.Namespace( agent_id="rx-user-pc", server="http://127.0.0.1:8765", timeout=1.0, install_root=install_root, launcher_request=install_root / "launcher_request.json", launcher_status=launcher_status, release_id="rf001-agent-2026.09.29.1", ) return package, token, task, manifest, arguments def _successful_self_test( arguments: Sequence[str], timeout: float ) -> CommandResult: assert timeout == 120.0 assert "hackrf" not in " ".join(arguments).lower() return CommandResult(list(arguments), 0, "SELF TEST = PASS", "") def test_agent_stages_authenticated_update_and_reuses_it( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: package, token, task, manifest, arguments = _update_fixture(tmp_path) downloads: list[Path] = [] monkeypatch.setattr( agent_module, "request_json", lambda *_args, **_kwargs: { "manifest": manifest, "manifest_signature": update_module.sign_update_manifest(token, manifest), }, ) def download( _server: str, _token: bytes, _target: str, destination: Path, **_kwargs: object, ) -> None: downloads.append(destination) shutil.copyfile(package, destination) monkeypatch.setattr(agent_module, "request_file", download) result, errors, request = agent_module.stage_agent_update( task, arguments=arguments, token=token, command_runner=_successful_self_test, ) assert errors == [] assert result["update_status"] == "STAGED" assert result["hardware_accessed"] is False assert result["rf_tx"] == 0 assert request is not None assert len(downloads) == 1 result, errors, repeated_request = agent_module.stage_agent_update( task, arguments=arguments, token=token, command_runner=lambda *_args: (_ for _ in ()).throw( AssertionError("an already staged release must not run again") ), ) assert errors == [] assert result["update_status"] == "ALREADY_STAGED" assert repeated_request is not None assert { key: value for key, value in repeated_request.items() if key != "requested_at" } == {key: value for key, value in request.items() if key != "requested_at"} assert len(downloads) == 1 def test_agent_rejects_bad_update_signature_sha_and_protocol( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: _package, token, task, manifest, arguments = _update_fixture(tmp_path) network_downloads: list[str] = [] monkeypatch.setattr( agent_module, "request_json", lambda *_args, **_kwargs: { "manifest": manifest, "manifest_signature": "0" * 64, }, ) monkeypatch.setattr( agent_module, "request_file", lambda *_args, **_kwargs: network_downloads.append("called"), ) result, errors, request = agent_module.stage_agent_update( task, arguments=arguments, token=token, command_runner=_successful_self_test, ) assert result["update_status"] == "EXECUTION_FAILED" assert any("signature mismatch" in error for error in errors) assert request is None assert network_downloads == [] incompatible = dict(task, compatible_protocol_version=PROTOCOL_VERSION + 1) _result, errors, _request = agent_module.stage_agent_update( incompatible, arguments=arguments, token=token, command_runner=_successful_self_test, ) assert any("incompatible" in error for error in errors) changed_sha = dict(task, package_sha256="0" * 64) signed = update_module.sign_update_manifest(token, manifest) monkeypatch.setattr( agent_module, "request_json", lambda *_args, **_kwargs: { "manifest": manifest, "manifest_signature": signed, }, ) _result, errors, _request = agent_module.stage_agent_update( changed_sha, arguments=arguments, token=token, command_runner=_successful_self_test, ) assert any("field mismatch: package_sha256" in error for error in errors) assert network_downloads == [] def test_agent_retries_one_interrupted_download_then_stages( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: package, token, task, manifest, arguments = _update_fixture(tmp_path) monkeypatch.setattr( agent_module, "request_json", lambda *_args, **_kwargs: { "manifest": manifest, "manifest_signature": update_module.sign_update_manifest(token, manifest), }, ) calls = 0 def interrupted_then_complete( _server: str, _token: bytes, _target: str, destination: Path, **_kwargs: object, ) -> None: nonlocal calls calls += 1 if calls == 1: destination.write_bytes(b"partial") raise OSError("fixed interrupted transfer") shutil.copyfile(package, destination) monkeypatch.setattr(agent_module, "request_file", interrupted_then_complete) result, errors, request = agent_module.stage_agent_update( task, arguments=arguments, token=token, command_runner=_successful_self_test, ) assert errors == [] assert result["update_status"] == "STAGED" assert request is not None assert calls == 2 state = json.loads( ( arguments.install_root / "staging" / str(task["task_id"]) / "download_state.json" ).read_text(encoding="utf-8") ) assert state["network_attempts"] == 2 def test_agent_does_not_exceed_persisted_download_attempt_limit( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: _package, token, task, manifest, arguments = _update_fixture(tmp_path) monkeypatch.setattr( agent_module, "request_json", lambda *_args, **_kwargs: { "manifest": manifest, "manifest_signature": update_module.sign_update_manifest(token, manifest), }, ) calls = 0 def always_interrupted(*_args: object, **_kwargs: object) -> None: nonlocal calls calls += 1 raise OSError("fixed interrupted transfer") monkeypatch.setattr(agent_module, "request_file", always_interrupted) for _ in range(2): result, errors, request = agent_module.stage_agent_update( task, arguments=arguments, token=token, command_runner=_successful_self_test, ) assert result["update_status"] == "EXECUTION_FAILED" assert errors assert request is None assert calls == 2 def test_agent_restart_does_not_duplicate_background_analyzer( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: work_root, task, _capture, _reference = _staircase_fixture(tmp_path) starts: list[list[str]] = [] class FakeProcess: pid = 4242 def fake_popen(arguments: Sequence[str], **_kwargs: object) -> FakeProcess: starts.append(list(arguments)) return FakeProcess() monkeypatch.setattr(agent_module.subprocess, "Popen", fake_popen) pointer = agent_module.start_staircase_background_job( task, agent_id="rx-user-pc", work_root=work_root, ) assert pointer["pid"] == 4242 with pytest.raises(RuntimeError, match="already active"): agent_module.start_staircase_background_job( task, agent_id="rx-user-pc", work_root=work_root, ) assert len(starts) == 1 assert pointer["hardware_accessed"] is False assert pointer["rf_tx"] == 0 def test_completed_background_result_survives_network_loss( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: work_root, task, _capture, _reference = _staircase_fixture(tmp_path) directory = work_root / "agent_jobs" / str(task["task_id"]) directory.mkdir(parents=True) pointer = { "schema_version": 1, "task": task, "pid": 999999, "started_at": datetime.now(timezone.utc).isoformat(), "automatic_retry": False, "hardware_accessed": False, "rf_tx": 0, } agent_module.atomic_write_json( work_root / "agent_jobs" / "active_job.json", pointer ) result = { "protocol_version": PROTOCOL_VERSION, "task_id": task["task_id"], "agent_id": "rx-user-pc", "status": "COMPLETE", "result": {"hardware_accessed": False, "rf_tx": 0}, "errors": [], "rf_tx": 0, } agent_module.atomic_write_json(directory / "result.json", result) arguments = argparse.Namespace( work_root=work_root, server="http://127.0.0.1:8765", timeout=1.0, ) monkeypatch.setattr( agent_module, "request_json", lambda *_args, **_kwargs: (_ for _ in ()).throw( OSError("fixed network loss") ), ) with pytest.raises(OSError, match="network loss"): agent_module.deliver_background_result(arguments, b"token", pointer) assert (work_root / "agent_jobs" / "active_job.json").is_file() assert (directory / "result.json").is_file() monkeypatch.setattr( agent_module, "request_json", lambda *_args, **_kwargs: {"accepted": True, "rf_tx": 0}, ) assert agent_module.deliver_background_result(arguments, b"token", pointer) assert not (work_root / "agent_jobs" / "active_job.json").exists() assert (directory / "delivered_job.json").is_file() assert (directory / "result.json").is_file()