"""Receiver tests use a fake local CLI; they do not contact model providers.""" import importlib.util import http.server import json import os from pathlib import Path import subprocess import socket import ssl import sys import tempfile import threading import unittest from unittest.mock import patch SOURCE = Path(__file__).resolve().parents[2] / "scripts" / "talk_worker.py" SPEC = importlib.util.spec_from_file_location("talk_worker", SOURCE) worker = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(worker) class ReceiverTests(unittest.TestCase): def setUp(self): self.directory = tempfile.TemporaryDirectory() self.addCleanup(self.directory.cleanup) home = patch.object(worker.Path, "home", return_value=Path(self.directory.name)) home.start() self.addCleanup(home.stop) worker.set_profile("default") self.config = {"server": "http://127.0.0.1:1234", "key": "wxk_testsecret", "worker_id": "receiver-a", "name": "x", "workspace": self.directory.name, "engine": "codex", "timeout": 5} self.task = {"id": "task-1", "worker_id": "receiver-a", "source": "owner", "title": "CCTV", "instructions": "Verify camera access; report blockers."} def test_receiver_executes_stdin_in_pinned_workspace_without_talk_key(self): def command(engine, result): program = ( "import json,os,pathlib,sys; " "prompt=sys.stdin.read(); " "assert 'CCTV' in prompt; " "assert 'WXCH_TALK_KEY' not in os.environ; " "assert pathlib.Path.cwd()==pathlib.Path(sys.argv[2]); " "pathlib.Path('verified.txt').write_text('tested'); " "pathlib.Path(sys.argv[1]).write_text(json.dumps({'status':'done','summary':'Verified test feed, not a real camera.'}))" ) return [sys.executable, "-c", program, str(result), self.directory.name] with patch.object(worker, "cli_command", command), patch.dict(os.environ, {"WXCH_TALK_KEY": "never-pass-this"}): status, note = worker.execute(self.config, self.task) self.assertEqual(status, "done") self.assertIn("not a real camera", note) self.assertTrue((Path(self.directory.name) / "verified.txt").exists()) def test_channel_notes_and_wrong_target_are_not_executable(self): for change in ({"source": "peer"}, {"worker_id": "someone-else"}): with self.subTest(change=change), self.assertRaises(worker.WorkerError): worker.execute(self.config, dict(self.task, **change)) def test_bounded_sandbox_and_schema(self): for mode in ("danger-full-access", "--yolo"): with patch.object(worker.subprocess, "Popen") as launch: status, _ = worker.execute(self.config, dict(self.task, sandbox=mode)) self.assertEqual(status, "blocked") launch.assert_not_called() with patch.object(worker.shutil, "which", return_value="codex"): command = worker.cli_command("codex", Path("result.json")) self.assertIn("--output-schema", command) self.assertNotIn("--sandbox", command) def command(engine, result): schema = json.loads(result.with_suffix(".schema.json").read_text()) self.assertFalse(schema["additionalProperties"]) return ["codex", "exec", "-"] with patch.object(worker, "cli_command", command), patch.object(worker.subprocess, "Popen", side_effect=OSError("probe")) as launch: worker.execute(self.config, dict(self.task, sandbox="workspace-write")) self.assertEqual(launch.call_args.args[0], ["codex", "exec", "--sandbox", "workspace-write", "-"]) def test_raw_cli_output_is_never_a_success_report(self): status, note = worker.parse_result("arbitrary transcript including a secret", 0, self.config) self.assertEqual(status, "blocked") self.assertNotIn("secret", note) status, note = worker.parse_result('{"status":"done","summary":"wxk_testsecret sk-sensitive rtsp://u:p@camera/stream"}', 0, self.config) self.assertEqual(status, "done") self.assertNotIn("testsecret", note) self.assertNotIn("sk-sensitive", note) self.assertNotIn("u:p", note) self.assertEqual(worker.parse_result('{"status":"done","summary":"claimed success"}', 1, self.config)[0], "blocked") def test_timeout_stops_cli_and_does_not_claim_completion(self): self.config["timeout"] = 0.1 with patch.object(worker, "cli_command", return_value=[sys.executable, "-c", "import time;time.sleep(30)"]): status, _ = worker.execute(self.config, self.task) self.assertEqual(status, "blocked") def test_pending_delivery_is_retried_without_reexecuting(self): self.config.update(active="task-1", pending={"status": "done", "note": "tested"}) calls = [] def request(config, path, data): calls.append((path, data)) if len(calls) == 1: raise worker.WorkerError("temporary network failure", transient=True) return {} with patch.object(worker, "cli_command"), patch.object(worker, "request", request), patch.object(worker, "save_config"), patch.object(worker.time, "sleep"), patch.object(worker, "execute") as execute: with self.assertRaises(worker.WorkerError): worker.receive(self.config, once=True) self.assertIn("pending", self.config) worker.receive(self.config, once=True) execute.assert_not_called() self.assertNotIn("active", self.config) def test_interrupted_task_needs_explicit_recovery_and_never_reexecutes(self): self.config["active"] = "task-1" with patch.object(worker, "cli_command"), patch.object(worker, "request") as request, patch.object(worker, "save_config"), patch.object(worker, "execute") as execute: with self.assertRaises(worker.WorkerError): worker.receive(self.config, once=True) request.assert_not_called() worker.receive(self.config, once=True, recover=True) execute.assert_not_called() self.assertEqual(request.call_args.args[2]["status"], "blocked") def test_lost_claim_response_is_reconciled_without_reexecution(self): def request(config, path, data): if path == "/api/tasks/current": return {"task": self.task} self.fail("Unexpected request: " + path) with patch.object(worker, "cli_command"), patch.object(worker, "request", request), patch.object(worker, "save_config"), patch.object(worker, "execute") as execute: with self.assertRaises(worker.WorkerError): worker.receive(self.config, once=True) execute.assert_not_called() self.assertEqual(self.config["active"], self.task["id"]) def test_persistent_start_recovers_interrupted_task_without_reexecuting(self): self.config["active"] = "task-1" with patch.object(worker, "cli_command"), patch.object(worker, "request", return_value={}) as request, \ patch.object(worker, "save_config"), patch.object(worker, "legacy_cli_running", return_value=False), \ patch.object(worker, "execute") as execute: worker.receive(self.config, once=True, recover_wait=True) execute.assert_not_called() self.assertEqual(request.call_args.args[2]["status"], "blocked") self.assertNotIn("active", self.config) def test_persistent_lost_claim_recovers_then_keeps_receiving(self): calls = [] current_calls = 0 claim_calls = 0 def request(config, path, data): nonlocal current_calls, claim_calls calls.append((path, data)) if path == "/api/tasks/current": current_calls += 1 return {"task": None if current_calls == 1 else self.task} if path == "/api/tasks/claim": claim_calls += 1 if claim_calls == 1: raise worker.WorkerError("lost response", transient=True) raise KeyboardInterrupt() return {} with patch.object(worker, "cli_command"), patch.object(worker, "request", request), \ patch.object(worker, "save_config"), patch.object(worker.time, "sleep"), \ patch.object(worker, "legacy_cli_running", return_value=False), patch.object(worker, "execute") as execute: with self.assertRaises(KeyboardInterrupt): worker.receive(self.config, recover_wait=True) execute.assert_not_called() reports = [body for path, body in calls if path == "/api/tasks/task-1"] self.assertEqual(len(reports), 1) self.assertEqual(reports[0]["status"], "blocked") self.assertEqual(claim_calls, 2) def test_recovery_waits_for_surviving_cli_and_inspection_failure(self): self.config["active"] = "task-1" with patch.object(worker, "legacy_cli_running", side_effect=[OSError("private details"), True, False]), \ patch.object(worker, "request") as request, patch.object(worker.time, "sleep") as sleep, \ patch.object(worker, "execute") as execute: worker.wait_for_interrupted_cli(self.config) self.assertEqual(sleep.call_count, 2) self.assertEqual(request.call_count, 2) execute.assert_not_called() self.assertEqual(self.config["active"], "task-1") def test_recovery_checks_orphan_children_after_pid_reuse(self): self.config.update(active="task-1", active_process={"pid": 12345, "identity": "old"}) with patch.object(worker, "process_identity", return_value="new"), \ patch.object(worker, "legacy_cli_running", side_effect=[True, False]) as inspect, \ patch.object(worker, "request"), patch.object(worker.time, "sleep"), \ patch.object(worker, "terminate_tree") as terminate: worker.wait_for_interrupted_cli(self.config) self.assertEqual(inspect.call_count, 2) terminate.assert_not_called() def test_persistent_receiver_retries_local_failure_until_user_stops(self): with patch.object(worker, "receive", side_effect=[worker.WorkerError("CLI missing"), OSError(), KeyboardInterrupt()]) as receive, \ patch.object(worker.time, "sleep") as sleep: with self.assertRaises(KeyboardInterrupt): worker.run(self.config, persistent=True) self.assertEqual(receive.call_count, 3) self.assertEqual(sleep.call_count, 2) def test_persistent_receiver_stops_on_owner_revocation(self): with patch.object(worker, "receive", side_effect=worker.WorkerError("revoked", terminal=True)), \ patch.object(worker.time, "sleep") as sleep: with self.assertRaises(worker.WorkerError): worker.run(self.config, persistent=True) sleep.assert_not_called() def test_user_cancel_persists_status_and_does_not_start_more_work(self): with patch.object(worker, "cli_command"), patch.object(worker, "save_config"), \ patch.object(worker, "request", side_effect=[{"task": None}, {"task": self.task}]) as request, \ patch.object(worker, "execute", side_effect=KeyboardInterrupt()): with self.assertRaises(KeyboardInterrupt): worker.receive(self.config) self.assertEqual(self.config["pending"]["status"], "blocked") self.assertEqual(request.call_count, 2) def test_process_identity_distinguishes_exit(self): proc = subprocess.Popen([sys.executable, "-c", "import time;time.sleep(20)"]) try: identity = worker.process_identity(proc.pid) self.assertIsNotNone(identity) self.assertEqual(identity, worker.process_identity(proc.pid)) finally: proc.terminate() proc.wait(timeout=10) self.assertIsNone(worker.process_identity(proc.pid)) def test_https_required_and_credentials_in_url_rejected(self): for url in ("http://talk.example.org", "https://u:p@talk.example.org", "https://talk.example.org/path", "https://talk.example.org?key=secret"): with self.subTest(url=url), self.assertRaises(worker.WorkerError): worker.server_url(url) self.assertEqual(worker.server_url("https://talk.example.org/"), "https://talk.example.org") self.assertEqual(worker.server_url("http://127.0.0.1:1234"), "http://127.0.0.1:1234") def test_transient_reconnect_and_permanent_denial(self): with patch.object(worker, "request", side_effect=[worker.WorkerError("offline", transient=True), {"task": None}]) as request, patch.object(worker.time, "sleep"): self.assertEqual(worker.reconnect_request(self.config, "/api/tasks/current", {}), {"task": None}) self.assertEqual(request.call_count, 2) with patch.object(worker, "request", side_effect=worker.WorkerError("revoked")) as request, self.assertRaises(worker.WorkerError): worker.reconnect_request(self.config, "/api/tasks/current", {}) self.assertEqual(request.call_count, 1) def test_network_diagnostics_identify_cause_without_exception_secrets(self): cases = [ (socket.gaierror(11001, "private-proxy-password"), "DNS"), (TimeoutError("private-proxy-password"), "timed out"), (ssl.SSLError("private-proxy-password"), "TLS"), (ConnectionRefusedError("private-proxy-password"), "refused"), (ValueError("private-proxy-password"), "invalid response"), (OSError("private-proxy-password"), "network connection failed"), ] for reason, category in cases: with self.subTest(category=category): message = worker.network_failure(worker.urllib.error.URLError(reason)) self.assertIn(category, message) self.assertNotIn("private-proxy-password", message) def test_waiting_result_is_unfinished_and_receiver_accepts_next_task(self): self.assertEqual(worker.parse_result('{"status":"waiting","summary":"Run host discovery first."}', 0, self.config), ("waiting", "Run host discovery first.")) second = dict(self.task, id="task-2") results = [] claims = iter([self.task, second]) def request(config, path, data): if path == "/api/tasks/current": return {"task": None} if path == "/api/tasks/claim": try: return {"task": next(claims)} except StopIteration: raise KeyboardInterrupt() results.append((path, data.copy())) return {} with patch.object(worker, "cli_command"), patch.object(worker, "request", side_effect=request), \ patch.object(worker, "save_config"), patch.object(worker, "execute", side_effect=[("waiting", "Host action needed"), ("done", "Independent work verified")]) as execute, \ patch("builtins.print"), self.assertRaises(KeyboardInterrupt): worker.receive(self.config) self.assertEqual(execute.call_count, 2) self.assertEqual([data["status"] for _, data in results], ["waiting", "done"]) self.assertNotIn("active", self.config) def test_retry_prints_safe_http_error_and_keeps_waiting(self): with patch.object(worker, "request", side_effect=[worker.WorkerError("Talk returned HTTP 503", transient=True), {"task": None}]), \ patch.object(worker.time, "sleep"), patch("builtins.print") as output: self.assertEqual(worker.reconnect_request(self.config, "/api/tasks/current", {}), {"task": None}) self.assertIn("HTTP 503", output.call_args.args[0]) def test_truncated_http_response_recovers_on_fresh_connection(self): calls = [] class Handler(http.server.BaseHTTPRequestHandler): def do_POST(self): self.rfile.read(int(self.headers["Content-Length"])) calls.append(self.path) body = b'{"task":null}' self.send_response(200) self.send_header("Content-Length", str(len(body) + (10 if len(calls) == 1 else 0))) self.end_headers() self.wfile.write(body) self.close_connection = True def log_message(self, *args): pass server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() try: config = dict(self.config, server="http://127.0.0.1:" + str(server.server_port)) with patch.dict(os.environ, {"no_proxy": "127.0.0.1", "NO_PROXY": "127.0.0.1"}): with self.assertRaises(worker.WorkerError) as failure: worker.request(config, "/api/tasks/current", {}) self.assertTrue(failure.exception.transient) self.assertEqual(worker.request(config, "/api/tasks/current", {}), {"task": None}) self.assertEqual(len(calls), 2) self.assertEqual(config["key"], self.config["key"]) finally: server.shutdown() server.server_close() thread.join() def test_proxy_idle_failure_switches_to_paced_short_poll_without_execution(self): waits = [] paths = [] def request(config, path, data): paths.append(path) if path == "/api/tasks/current": return {"task": None} waits.append(data["wait"]) if len(waits) == 1: raise worker.WorkerError("connection lost", transient=True) if len(waits) == 2: return {"task": None} raise KeyboardInterrupt() with patch.object(worker, "cli_command"), patch.object(worker, "request", side_effect=request), \ patch.object(worker.time, "sleep") as sleep, patch.object(worker, "execute") as execute, \ patch("builtins.print"), self.assertRaises(KeyboardInterrupt): worker.receive(self.config) self.assertEqual(waits, [10, 0, 0]) self.assertEqual(paths[:3], ["/api/tasks/current", "/api/tasks/claim", "/api/tasks/current"]) self.assertIn(unittest.mock.call(3), sleep.call_args_list) execute.assert_not_called() self.assertEqual(self.config["key"], "wxk_testsecret") def test_cli_failure_classifies_without_uploading_transcripts(self): for detail, category in [("Not inside a trusted directory", "project_trust"), ("Not logged in wxk_private", "cli_login"), ("unexpected argument --flag", "cli_version"), ("network connection failed", "provider_network"), ("sandbox permission denied", "cli_permissions")]: note = worker.cli_failure(detail, 1) self.assertIn(category, note) self.assertNotIn("wxk_private", note) with patch.object(worker, "cli_command", return_value=[sys.executable, "-c", "import sys;sys.stderr.write('Not inside a trusted directory wxk_private');sys.exit(1)"]): status, note = worker.execute(self.config, self.task) self.assertEqual(status, "waiting") self.assertIn("project_trust", note) self.assertNotIn("wxk_private", note) def test_saved_credential_denial_preserves_pairing_and_distinguishes_pair_code(self): denied = worker.urllib.error.HTTPError(self.config["server"], 401, "Unauthorized", {}, None) with patch.object(worker.urllib.request, "build_opener") as opener: opener.return_value.open.side_effect = denied with self.assertRaises(worker.WorkerError) as saved: worker.request(self.config, "/api/tasks/current", {}) self.assertIn("pairing was kept", str(saved.exception)) self.assertIn("check/enable", str(saved.exception)) self.assertFalse(saved.exception.transient) with self.assertRaises(worker.WorkerError) as pairing: worker.request({"server": self.config["server"]}, "/api/pair", {"code": "test-only"}, authenticated=False) self.assertIn("Pairing code", str(pairing.exception)) self.assertNotIn("saved credential", str(pairing.exception)) def test_profiles_preserve_original_credential_location_and_validate_names(self): original = worker.config_path() self.assertEqual(worker.profile_value(), "Session") worker.set_profile("another-cli") self.assertNotEqual(worker.config_path(), original) self.assertEqual(worker.profile_value(), "Session-another-cli") for name in ("../other", "a/b", "x;echo", "a" * 41): with self.assertRaises(worker.WorkerError): worker.set_profile(name) worker.set_profile("default") self.assertEqual(worker.config_path(), original) def test_project_lock_blocks_other_receiver_without_running_cli(self): with patch.object(worker, "receiver_lock", side_effect=worker.WorkerError("project busy")), patch.object(worker, "execute_task") as execute: self.assertEqual(worker.execute(self.config, self.task), ("blocked", "project busy")) execute.assert_not_called() @unittest.skipUnless(os.name == "nt", "Windows DPAPI only") def test_windows_credential_encryption_roundtrip(self): secret = b"generated-test-credential" protected = worker.windows_protect(secret) self.assertNotIn(secret, protected) self.assertEqual(worker.windows_protect(protected, decrypt=True), secret) if __name__ == "__main__": unittest.main()