"""Unit tests for the model-free RX task watcher.""" from __future__ import annotations import json from pathlib import Path import pytest from tools.rx_agent_watcher import ( ApprovalDecision, ManualExecutor, WatcherState, handle_task, is_new_task, load_state, mark_completed, parse_task, request_approval, request_stop, save_state, ) HEAD = "1" * 40 def task_text(*, task_id: str = "WATCHER_TEST_001", rf_tx: str = "NO") -> str: hardware = "YES" if rf_tx == "YES" else "NO" risk_level = "RED" if rf_tx == "YES" else "GREEN" return f"""TASK_ID = {task_id} TASK_TITLE = Watcher test SOURCE_BRANCH = video-development EXPECTED_HEAD = {HEAD} REQUIRES_APPROVAL = YES HARDWARE_ACCESS = {hardware} RF_RX = NO RF_TX = {rf_tx} ALLOW_COMMIT = NO ALLOW_PUSH = RX_RESULT_ONLY RISK_LEVEL = {risk_level} # Body Do nothing. """ def responses(*values: str): iterator = iter(values) return lambda _prompt: next(iterator) def test_task_parsing_and_validation() -> None: task = parse_task(task_text()) assert task.task_id == "WATCHER_TEST_001" assert task.source_branch == "video-development" assert task.safety_level == "GREEN" assert not task.rf_tx with pytest.raises(ValueError, match="missing task header"): parse_task("TASK_ID = INCOMPLETE") with pytest.raises(ValueError, match="duplicate"): parse_task(task_text().replace( "TASK_TITLE = Watcher test", "TASK_TITLE = Watcher test\nTASK_TITLE = Duplicate", )) def test_state_persistence_and_completion(tmp_path: Path) -> None: path = tmp_path / "state.json" original = WatcherState(last_seen_task_id="WATCHER_TEST_001", running=True, pid=123) save_state(path, original) assert load_state(path) == original mark_completed(path, "WATCHER_TEST_001") completed = load_state(path) assert completed.last_completed_task_id == "WATCHER_TEST_001" assert completed.deferred_task_id is None with pytest.raises(ValueError, match="last seen"): mark_completed(path, "WATCHER_TEST_002") def test_approval_defer_stop_and_yellow_approve() -> None: green = parse_task(task_text()) assert request_approval(green, responses("D"), lambda _text: None) == ApprovalDecision.DEFER assert request_approval(green, responses("S"), lambda _text: None) == ApprovalDecision.STOP yellow_text = task_text().replace( "HARDWARE_ACCESS = NO", "HARDWARE_ACCESS = YES" ).replace("RISK_LEVEL = GREEN", "RISK_LEVEL = YELLOW") yellow = parse_task(yellow_text) assert yellow.safety_level == "YELLOW" assert request_approval(yellow, responses("A"), lambda _text: None) == ApprovalDecision.APPROVE def test_rf_tx_red_gate_requires_separate_phrase() -> None: red = parse_task(task_text(rf_tx="YES")) assert red.safety_level == "RED" assert request_approval( red, responses("A", "yes"), lambda _text: None ) == ApprovalDecision.BLOCKED assert request_approval( red, responses("A", "AUTHORIZE RF TX"), lambda _text: None ) == ApprovalDecision.APPROVE def test_non_rf_red_task_also_requires_separate_phrase() -> None: text = task_text(task_id="FIRMWARE_TEST_001").replace( "RISK_LEVEL = GREEN", "RISK_LEVEL = RED" ) red = parse_task(text) assert not red.rf_tx assert request_approval( red, responses("A", "AUTHORIZE RF TX"), lambda _text: None ) == ApprovalDecision.BLOCKED assert request_approval( red, responses("A", "AUTHORIZE FIRMWARE_TEST_001"), lambda _text: None ) == ApprovalDecision.APPROVE def test_duplicate_task_is_suppressed_without_prompt(tmp_path: Path) -> None: text = task_text() task = parse_task(text) state = WatcherState(last_seen_task_id=task.task_id) assert not is_new_task(task, state) def forbidden_input(_prompt: str) -> str: raise AssertionError("duplicate task prompted the user") status, returned = handle_task( text, HEAD, state, ManualExecutor(tmp_path), dry_run=True, input_function=forbidden_input, output_function=lambda _text: None, ) assert status == "DUPLICATE_SUPPRESSED" assert returned == state assert list(tmp_path.iterdir()) == [] def test_defer_persists_seen_id_and_creates_no_result(tmp_path: Path) -> None: state = WatcherState() status, state = handle_task( task_text(), HEAD, state, ManualExecutor(tmp_path), dry_run=True, input_function=responses("D"), output_function=lambda _text: None, ) assert status == "DEFERRED" assert state.last_seen_task_id == "WATCHER_TEST_001" assert state.deferred_task_id == "WATCHER_TEST_001" assert list(tmp_path.iterdir()) == [] def test_dry_run_approval_writes_prompt_and_matching_results(tmp_path: Path) -> None: status, state = handle_task( task_text(), HEAD, WatcherState(), ManualExecutor(tmp_path), dry_run=True, input_function=responses("A"), output_function=lambda _text: None, ) directory = tmp_path / "WATCHER_TEST_001" assert status == "PREPARED" assert state.last_seen_task_id == "WATCHER_TEST_001" assert state.last_completed_task_id is None assert (directory / "TASK.md").exists() markdown = (directory / "RX_RESULT.md").read_text(encoding="utf-8") result = json.loads((directory / "RX_RESULT.json").read_text(encoding="utf-8")) assert "TASK_STATUS = DRY_RUN_PREPARED" in markdown assert result["task_status"] == "DRY_RUN_PREPARED" assert result["rf_tx"] == "NO" assert result["hardware_actions"] == [] with pytest.raises(FileExistsError): ManualExecutor(tmp_path).prepare( parse_task(task_text()), task_text(), HEAD, dry_run=True ) def test_stop_request_for_idle_watcher(tmp_path: Path) -> None: state_path = tmp_path / "state.json" save_state(state_path, WatcherState(running=False, pid=None)) assert request_stop(state_path, timeout_s=0.1) assert state_path.with_suffix(".json.stop").read_text(encoding="ascii") == "STOP\n"