From 7a27f845fca68ca62041ec99c9b23603cc6279e3 Mon Sep 17 00:00:00 2001 From: Gareth Morgan Date: Tue, 10 Nov 2026 14:09:26 -0500 Subject: [PATCH] sensing_control: add a mock producer that runs the management channel without cameras sensing_mock_producer plays looped H.264 files through the same management channel, retry loop and direct sender as camera_plugin_sensing, with --simulate-fault-seconds to show the unplugged-camera path, so the PC side can be tried without an Orin. An end-to-end test drives it through a real video_stage_worker. Co-Authored-By: Claude Sonnet 5.5 Signed-off-by: Gareth Morgan --- CMakeLists.txt | 2 +- src/CMakeLists.txt | 4 +- src/plugins/sensing/README.md | 3 + src/sensing_control/CMakeLists.txt | 2 + .../mock_producer/CMakeLists.txt | 13 + src/sensing_control/mock_producer/main.cpp | 310 +++++++++++++ src/sensing_control/tests/CMakeLists.txt | 14 + .../tests/test_mock_producer_e2e.py | 429 ++++++++++++++++++ 8 files changed, 774 insertions(+), 3 deletions(-) create mode 100644 src/sensing_control/mock_producer/CMakeLists.txt create mode 100644 src/sensing_control/mock_producer/main.cpp create mode 100755 src/sensing_control/tests/test_mock_producer_e2e.py diff --git a/CMakeLists.txt b/CMakeLists.txt index 48325c9bec..25b1a12a1a 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -73,7 +73,7 @@ option(BUILD_PLUGIN_WUJI_GLOVE "Build Wuji glove plugin (requires the wuji_sdk C option(BUILD_PLUGIN_SHARPA_AVATAR "Build Sharpa AvatarExo Glove plugin (Linux; skips if the Avatar SDK is missing)" OFF) option(BUILD_PLUGIN_SENSING "Build SENSING GMSL camera plugin (Jetson only; requires SIPL and NVENC)" OFF) option(BUILD_VIDEO_STAGE "Build the video stage: the Orin-to-PC video transport and the PC relay" OFF) -option(BUILD_SENSING_CONTROL "Build the camera producer management channel (no camera SDK needed; builds and tests on a PC)" OFF) +option(BUILD_SENSING_CONTROL "Build the camera producer management channel and the mock producer (on a PC, to try the producer without an Orin; built with BUILD_PLUGIN_SENSING)" OFF) option(BUILD_EXAMPLES "Build examples" ON) option(BUILD_EXAMPLE_TELEOP_ROS2 "Build only the teleop_ros2 ROS 2 reference integration (e.g. for Docker)" OFF) option(BUILD_TESTING "Build unit tests" ON) diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index a1f84fa71a..03f99d36be 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -30,8 +30,8 @@ endif() if(BUILD_VIDEO_STAGE) add_subdirectory(livekit_publisher) add_subdirectory(video_stage) -elseif(BUILD_PLUGIN_SENSING) - # The camera producer sends AVS1 records to the video stage and needs only its record codec and sender. +elseif(BUILD_PLUGIN_SENSING OR BUILD_SENSING_CONTROL) + # The camera producer and its mock send AVS1 records to the video stage and need only its record codec and sender. add_subdirectory(video_stage/wire) add_subdirectory(video_stage/sender) endif() diff --git a/src/plugins/sensing/README.md b/src/plugins/sensing/README.md index 3537abd8fb..9b15d1c86d 100644 --- a/src/plugins/sensing/README.md +++ b/src/plugins/sensing/README.md @@ -76,3 +76,6 @@ request gets an error reply with a code and a sentence. The token goes in `Autho src/sensing_control/tools/orin_control.py ORIN_HOST --token-file control.token status src/sensing_control/tools/orin_control.py ORIN_HOST --token-file control.token set-target sensing-left 192.168.1.205 5600 ``` + +`sensing_mock_producer` (on a PC, configure with `-DBUILD_SENSING_CONTROL=ON`) runs the same channel, retry loop and direct sender over looped H.264 files, with +`--simulate-fault-seconds=N` to show the unplugged-camera path, so the PC side can be tried without an Orin. diff --git a/src/sensing_control/CMakeLists.txt b/src/sensing_control/CMakeLists.txt index 4432a45d25..997b64f33b 100644 --- a/src/sensing_control/CMakeLists.txt +++ b/src/sensing_control/CMakeLists.txt @@ -47,6 +47,8 @@ target_link_libraries(sensing_control nlohmann_json::nlohmann_json ) +add_subdirectory(mock_producer) + if(BUILD_TESTING) add_subdirectory(tests) endif() diff --git a/src/sensing_control/mock_producer/CMakeLists.txt b/src/sensing_control/mock_producer/CMakeLists.txt new file mode 100644 index 0000000000..0dc2d10b0c --- /dev/null +++ b/src/sensing_control/mock_producer/CMakeLists.txt @@ -0,0 +1,13 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +# A camera producer without cameras: it plays H.264 files through the same management channel, supervisor and direct +# sender as camera_plugin_sensing, so the PC side can be exercised with no Orin. +add_executable(sensing_mock_producer main.cpp) + +target_link_libraries(sensing_mock_producer + PRIVATE + sensing::control + video_stage::sender + video_stage::wire +) diff --git a/src/sensing_control/mock_producer/main.cpp b/src/sensing_control/mock_producer/main.cpp new file mode 100644 index 0000000000..1d2db0e116 --- /dev/null +++ b/src/sensing_control/mock_producer/main.cpp @@ -0,0 +1,310 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +// A camera producer without cameras. It runs the same ControlServer, ControlHandler, ProducerSupervisor and +// DirectSender as camera_plugin_sensing and differs only in where frames come from: H.264 files looped at a fixed rate. +// +// sensing_mock_producer --control-token-file=token.txt --stream=sensing-left=left.h264,direct=127.0.0.1:5600 +// --stream=sensing-right=right.h264 +// +// --simulate-fault-seconds=N makes the first N seconds behave like a camera with no power, so the fault reporting and +// the retry can be seen without unplugging anything. + +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace plugins::sensing; + +namespace +{ + +std::atomic g_stop{ false }; + +void on_signal(int) +{ + g_stop = true; +} + +struct StreamArgs +{ + std::string track; + std::string file; + std::string direct_host; + std::uint16_t direct_port = 0; +}; + +/// "TRACK=FILE[,direct=HOST:PORT]" +StreamArgs parse_stream(const std::string& text) +{ + StreamArgs out; + const auto eq = text.find('='); + if (eq == std::string::npos || eq == 0) + throw std::runtime_error("--stream must look like TRACK=FILE[,direct=HOST:PORT]: " + text); + out.track = text.substr(0, eq); + std::string rest = text.substr(eq + 1); + const auto comma = rest.find(",direct="); + if (comma != std::string::npos) + { + const std::string target = rest.substr(comma + 8); + rest = rest.substr(0, comma); + const auto colon = target.rfind(':'); + if (colon == std::string::npos) + throw std::runtime_error("direct= must be HOST:PORT: " + target); + out.direct_host = target.substr(0, colon); + out.direct_port = static_cast(std::stoul(target.substr(colon + 1))); + } + out.file = rest; + return out; +} + +std::uint64_t monotonic_us() +{ + return static_cast( + std::chrono::duration_cast(std::chrono::steady_clock::now().time_since_epoch()).count()); +} + +/// One looped file and the direct sender that carries it. +struct MockStream +{ + std::string track; + std::vector bytes; + std::vector units; + std::size_t cursor = 0; + std::uint32_t frame_id = 0; + std::uint64_t frames = 0; + std::unique_ptr sender; +}; + +class MockSession : public CameraSession +{ +public: + MockSession(const std::vector& streams, const DirectTargets& targets, std::uint32_t fps) + : m_period(std::chrono::microseconds(1'000'000 / fps)), m_next(std::chrono::steady_clock::now()) + { + for (const auto& args : streams) + { + auto stream = std::make_unique(); + stream->track = args.track; + std::ifstream file(args.file, std::ios::binary); + if (!file) + throw std::runtime_error("cannot read " + args.file); + stream->bytes.assign(std::istreambuf_iterator(file), std::istreambuf_iterator()); + stream->units = video_stage::split_access_units(stream->bytes.data(), stream->bytes.size()); + if (stream->units.empty()) + throw std::runtime_error(args.file + " has no H.264 access units"); + + DirectSenderConfig config; + config.fps = fps; + const auto found = targets.find(args.track); + if (found != targets.end()) + { + config.host = found->second.host; + config.port = found->second.port; + } + stream->sender = std::make_unique(config); + m_streams.push_back(std::move(stream)); + } + } + + void update() override + { + const auto now = std::chrono::steady_clock::now(); + while (now >= m_next) + { + m_next += m_period; + for (auto& stream : m_streams) + emit(*stream); + } + } + + void flush() override + { + } + + std::vector streams() const override + { + std::vector out; + for (const auto& stream : m_streams) + { + StreamStatus entry; + entry.track = stream->track; + entry.frames = stream->frames; + entry.published = stream->frames; + entry.direct_host = stream->sender->host(); + entry.direct_port = stream->sender->port(); + entry.direct_connected = stream->sender->connected(); + entry.direct_error = stream->sender->last_error(); + out.push_back(std::move(entry)); + } + return out; + } + + void set_direct_target(const std::string& track, const std::string& host, std::uint16_t port) override + { + for (auto& stream : m_streams) + if (stream->track == track) + stream->sender->set_target(host, port); + } + +private: + void emit(MockStream& stream) + { + // A keyframe request is answered at the file's next IDR, as an encoder would answer it. + if (stream.sender->take_keyframe_request()) + { + for (std::size_t k = 0; k < stream.units.size(); ++k) + { + const std::size_t index = (stream.cursor + k) % stream.units.size(); + if (stream.units[index].idr) + { + stream.cursor = index; + break; + } + } + } + const auto& unit = stream.units[stream.cursor]; + stream.cursor = (stream.cursor + 1) % stream.units.size(); + + EncodedVideoFrame frame; + frame.data = stream.bytes.data() + unit.offset; + frame.size = unit.size; + frame.keyframe = unit.idr; + frame.timestamp_us = monotonic_us(); + frame.user_timestamp_us = frame.timestamp_us; + frame.frame_id = ++stream.frame_id; + stream.sender->submit(frame); + ++stream.frames; + } + + std::chrono::microseconds m_period; + std::chrono::steady_clock::time_point m_next; + std::vector> m_streams; +}; + +void print_usage(const char* program) +{ + std::cout << "Usage: " << program << " --control-token-file=PATH --stream=TRACK=FILE[,direct=HOST:PORT] ...\n" + << " --control-port=N management channel port (default: 5610; 0 picks a free one)\n" + << " --control-bind=ADDR IPv4 address to listen on (default: 127.0.0.1)\n" + << " --fps=N frames per second per stream (default: 30)\n" + << " --retry-seconds=N wait between start attempts (default: 2)\n" + << " --simulate-fault-seconds=N behave like an unpowered camera for the first N seconds\n" + << " --release=ID release id to report (default: mock)\n"; +} + +} // namespace + +int main(int argc, char** argv) +try +{ + std::vector streams; + ControlServerConfig control; + control.bind_address = "127.0.0.1"; + std::string token_file; + std::uint32_t fps = 30; + std::uint32_t retry_seconds = 2; + std::uint32_t fault_seconds = 0; + std::string release = "mock"; + + for (int i = 1; i < argc; ++i) + { + const std::string arg = argv[i]; + const auto value = [&](const char* name) { return arg.substr(std::string(name).size()); }; + if (arg == "--help" || arg == "-h") + { + print_usage(argv[0]); + return 0; + } + else if (arg.rfind("--stream=", 0) == 0) + streams.push_back(parse_stream(value("--stream="))); + else if (arg.rfind("--control-token-file=", 0) == 0) + token_file = value("--control-token-file="); + else if (arg.rfind("--control-port=", 0) == 0) + control.port = static_cast(std::stoul(value("--control-port="))); + else if (arg.rfind("--control-bind=", 0) == 0) + control.bind_address = value("--control-bind="); + else if (arg.rfind("--fps=", 0) == 0) + fps = static_cast(std::stoul(value("--fps="))); + else if (arg.rfind("--retry-seconds=", 0) == 0) + retry_seconds = static_cast(std::stoul(value("--retry-seconds="))); + else if (arg.rfind("--simulate-fault-seconds=", 0) == 0) + fault_seconds = static_cast(std::stoul(value("--simulate-fault-seconds="))); + else if (arg.rfind("--release=", 0) == 0) + release = value("--release="); + else + { + std::cerr << "Unknown option: " << arg << std::endl; + print_usage(argv[0]); + return 1; + } + } + if (streams.empty() || token_file.empty() || fps == 0) + { + print_usage(argv[0]); + return 1; + } + control.token = read_secret_file(token_file); + + std::signal(SIGINT, on_signal); + std::signal(SIGTERM, on_signal); + + ProducerStatus status(release); + std::vector configured; + for (const auto& args : streams) + { + StreamStatus entry; + entry.track = args.track; + entry.direct_host = args.direct_host; + entry.direct_port = args.direct_port; + configured.push_back(entry); + } + + const auto started = std::chrono::steady_clock::now(); + const SessionFactory factory = [&](const DirectTargets& targets) -> std::unique_ptr + { + if (std::chrono::steady_clock::now() - started < std::chrono::seconds(fault_seconds)) + throw std::runtime_error("INvSIPLCamera::Init failed with SIPLStatus 10"); + return std::make_unique(streams, targets, fps); + }; + + std::unique_ptr server; + SupervisorConfig supervisor_config; + supervisor_config.retry_interval = std::chrono::seconds(retry_seconds); + supervisor_config.poll_interval = std::chrono::milliseconds(2); + ProducerSupervisor supervisor( + status, configured, factory, + [&server](const std::string& text) + { + if (server) + server->broadcast(text); + }, + supervisor_config); + ControlHandler handler(status, supervisor); + server = + std::make_unique(control, [&handler](const std::string& text) { return handler.handle(text); }); + std::cout << "Mock producer: management channel on " << control.bind_address << ":" << server->port() << std::endl; + + supervisor.run([] { return g_stop.load(); }); + return 0; +} +catch (const std::exception& e) +{ + std::cerr << argv[0] << ": " << e.what() << std::endl; + return 1; +} diff --git a/src/sensing_control/tests/CMakeLists.txt b/src/sensing_control/tests/CMakeLists.txt index 0581b44eb7..6db52fd2dc 100644 --- a/src/sensing_control/tests/CMakeLists.txt +++ b/src/sensing_control/tests/CMakeLists.txt @@ -19,3 +19,17 @@ target_link_libraries(sensing_control_tests ) catch_discover_tests(sensing_control_tests) + +# The mock producer end to end: needs the built binaries, so it is skipped (not failed) when they are absent, and it +# needs the PC relay, which exists only in a build with BUILD_VIDEO_STAGE. +find_package(Python3 COMPONENTS Interpreter QUIET) +if(Python3_Interpreter_FOUND AND TARGET video_stage_worker) + add_test( + NAME sensing_mock_producer_e2e + COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/test_mock_producer_e2e.py + ) + set_tests_properties(sensing_mock_producer_e2e PROPERTIES + ENVIRONMENT "SENSING_MOCK_PRODUCER=$;VIDEO_STAGE_WORKER=$" + TIMEOUT 120 + ) +endif() diff --git a/src/sensing_control/tests/test_mock_producer_e2e.py b/src/sensing_control/tests/test_mock_producer_e2e.py new file mode 100755 index 0000000000..13d91df9bb --- /dev/null +++ b/src/sensing_control/tests/test_mock_producer_e2e.py @@ -0,0 +1,429 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""sensing_mock_producer end to end, through the management channel and a real video_stage_worker. + +Binaries come from SENSING_MOCK_PRODUCER and VIDEO_STAGE_WORKER, else tmp/build-vs under the repository root; the +input stream from SENSING_E2E_INPUT, else tmp/vs/orin_like.h264. The test is skipped when any of them is missing. +The worker writes to files (no LiveKit), on ports 5730 and 5731. +""" + +from __future__ import annotations + +import os +import shutil +import signal +import subprocess +import sys +import tempfile +import time +import unittest +from pathlib import Path + +HERE = Path(__file__).resolve().parent +REPO = HERE.parents[2] +sys.path.insert(0, str(HERE.parent / "tools")) + +import orin_control # noqa: E402 + +LEFT_PORT, RIGHT_PORT = 5730, 5731 +CLOSED_PORT = 5799 # nothing listens here + + +def _binary(env: str, default: str) -> Path: + return Path(os.environ.get(env) or REPO / "tmp" / "build-vs" / default) + + +PRODUCER = _binary( + "SENSING_MOCK_PRODUCER", "src/sensing_control/mock_producer/sensing_mock_producer" +) +WORKER = _binary("VIDEO_STAGE_WORKER", "src/video_stage/app/video_stage_worker") +INPUT = Path( + os.environ.get("SENSING_E2E_INPUT") or REPO / "tmp" / "vs" / "orin_like.h264" +) + + +def _missing() -> str | None: + for what, path in ( + ("sensing_mock_producer", PRODUCER), + ("video_stage_worker", WORKER), + ("input stream", INPUT), + ): + if not path.is_file(): + return f"{what} not found at {path} (set SENSING_MOCK_PRODUCER, VIDEO_STAGE_WORKER, SENSING_E2E_INPUT)" + return None + + +def _wait(predicate, timeout: float, interval: float = 0.1): + """Poll until predicate() returns something truthy; return it, or None on timeout.""" + deadline = time.monotonic() + timeout + while True: + value = predicate() + if value: + return value + if time.monotonic() >= deadline: + return None + time.sleep(interval) + + +def _size(path: Path) -> int: + return path.stat().st_size if path.exists() else 0 + + +@unittest.skipIf(_missing() is not None, _missing() or "") +class MockProducerEndToEnd(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.work = Path(tempfile.mkdtemp(prefix="sensing_e2e_", dir=REPO / "tmp")) + cls.token = "e2e-token-" + os.urandom(8).hex() + token_file = cls.work / "token.txt" + token_file.write_text(cls.token + "\n") + token_file.chmod(0o600) + cls.token_file = token_file + cls.out = { + "left": cls.work / "e2e_left.h264", + "right": cls.work / "e2e_right.h264", + } + cls.procs: list[subprocess.Popen] = [] + cls.started = time.monotonic() + + producer_log = (cls.work / "producer.log").open("w") + cls.producer = subprocess.Popen( + [ + str(PRODUCER), + "--control-port=0", + f"--control-token-file={token_file}", + f"--stream=sensing-left={INPUT}", + f"--stream=sensing-right={INPUT}", + "--simulate-fault-seconds=3", + "--retry-seconds=1", + ], + stdout=producer_log, + stderr=subprocess.STDOUT, + ) + cls.procs.append(cls.producer) + + def announced_port() -> int | None: + for line in ( + (cls.work / "producer.log").read_text(errors="replace").splitlines() + ): + if "management channel on" in line: + return int(line.rsplit(":", 1)[1]) + return None + + port = _wait(announced_port, 10) + if port is None: + cls.tearDownClass() + raise RuntimeError( + "the producer did not announce its management channel:\n" + + cls.log("producer.log") + ) + cls.port = port + + worker_log = (cls.work / "worker.log").open("w") + cls.worker = subprocess.Popen( + [ + str(WORKER), + f"--stream={LEFT_PORT}:file={cls.out['left']}", + f"--stream={RIGHT_PORT}:file={cls.out['right']}", + ], + stdout=worker_log, + stderr=subprocess.STDOUT, + start_new_session=True, + ) + cls.procs.append(cls.worker) + + @classmethod + def tearDownClass(cls) -> None: + for proc in cls.procs: + if proc.poll() is None: + proc.send_signal(signal.SIGINT) + for proc in cls.procs: + try: + proc.wait(timeout=10) + except subprocess.TimeoutExpired: + proc.terminate() + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait() + + @classmethod + def log(cls, name: str) -> str: + path = cls.work / name + return path.read_text(errors="replace")[-3000:] if path.exists() else "" + + # -- helpers ---------------------------------------------------------------------------------------------- + + def connect(self, token: str | None = None) -> orin_control.Connection: + return orin_control.Connection("127.0.0.1", self.port, token or self.token) + + def request(self, message: dict) -> dict: + conn = self.connect() + try: + return conn.request(message) + finally: + conn._sock.close() + + def status(self) -> dict: + reply = self.request({"type": "get_status"}) + self.assertTrue(reply["ok"], reply) + return reply["status"] + + def set_target(self, stream: str, host: str, port: int = 0) -> dict: + return self.request( + {"type": "set_direct_target", "stream": stream, "host": host, "port": port} + ) + + def wait_streaming(self) -> dict: + status = _wait(lambda: (s := self.status())["state"] == "streaming" and s, 30) + self.assertIsNotNone( + status, "the producer never reached streaming:\n" + self.log("producer.log") + ) + return status + + def stream_of(self, status: dict, name: str) -> dict: + return next(s for s in status["streams"] if s["stream"] == name) + + def wait_stable(self, path: Path) -> int: + """The size once it stops changing.""" + last = [-1] + + def stable() -> int | None: + size = _size(path) + if size == last[0]: + return size or None + last[0] = size + time.sleep(0.5) + return None + + value = _wait(stable, 15, interval=0.0) + self.assertIsNotNone(value, f"{path.name} never stopped growing") + return value + + # -- the scenario, in order ------------------------------------------------------------------------------- + + def test_1_fault_then_streaming(self) -> None: + # The simulated unpowered camera is reported in plain words first. + fault = _wait(lambda: (s := self.status())["state"] == "camera_fault" and s, 3) + if fault is None and time.monotonic() - self.started > 3: + self.skipTest("the simulated fault window had already passed") + self.assertIsNotNone( + fault, "never saw camera_fault:\n" + self.log("producer.log") + ) + self.assertIn("power cable", fault["fault"]["message"]) + self.assertEqual(fault["fault"]["code"], "camera_start_failed") + self.assertEqual( + [s["stream"] for s in fault["streams"]], ["sensing-left", "sensing-right"] + ) + + streaming = self.wait_streaming() + self.assertNotIn("fault", streaming) + self.assertGreaterEqual( + time.monotonic() - self.started, + 2.5, + "the fault window was shorter than asked", + ) + + def test_2_set_target_streams_to_the_worker(self) -> None: + self.wait_streaming() + left = self.out["left"] + reply = self.set_target("sensing-left", "127.0.0.1", LEFT_PORT) + self.assertTrue(reply["ok"], reply) + target = self.stream_of(reply["status"], "sensing-left")["direct"] + self.assertEqual((target["host"], target["port"]), ("127.0.0.1", LEFT_PORT)) + + grew = _wait(lambda: _size(left) > 100_000, 20) + self.assertTrue( + grew, "the worker's file did not grow:\n" + self.log("worker.log") + ) + before = _size(left) + self.assertTrue( + _wait(lambda: _size(left) > before + 100_000, 10), + "the file stopped growing", + ) + + connected = _wait( + lambda: self.stream_of(self.status(), "sensing-left")["direct"][ + "connected" + ], + 10, + ) + self.assertTrue( + connected, "the status never reported the direct link as connected" + ) + self.assertFalse( + self.stream_of(self.status(), "sensing-right")["direct"]["host"] + ) + + if shutil.which("ffprobe"): + probe = subprocess.run( + [ + "ffprobe", + "-v", + "error", + "-count_frames", + "-select_streams", + "v:0", + "-show_entries", + "stream=nb_read_frames", + "-of", + "csv=p=0", + str(left), + ], + capture_output=True, + check=False, + text=True, + timeout=60, + ) + self.assertEqual(probe.returncode, 0, probe.stderr) + self.assertGreater(int(probe.stdout.strip().strip(",") or 0), 0) + + def test_3_second_stream_is_independent(self) -> None: + self.wait_streaming() + right = self.out["right"] + self.assertTrue(self.set_target("sensing-right", "127.0.0.1", RIGHT_PORT)["ok"]) + self.assertTrue( + _wait(lambda: _size(right) > 100_000, 20), self.log("worker.log") + ) + + def test_4_empty_host_stops_the_link(self) -> None: + self.wait_streaming() + left = self.out["left"] + reply = self.set_target("sensing-left", "") + self.assertTrue(reply["ok"], reply) + direct = self.stream_of(reply["status"], "sensing-left")["direct"] + self.assertEqual((direct["host"], direct["port"]), ("", 0)) + + settled = self.wait_stable(left) + time.sleep(1.5) + self.assertEqual( + _size(left), settled, "the file kept growing after the link was turned off" + ) + self.assertFalse( + self.stream_of(self.status(), "sensing-left")["direct"]["connected"] + ) + # The other stream is unaffected. + right = self.out["right"] + before = _size(right) + self.assertTrue( + _wait(lambda: _size(right) > before, 10), "the right stream stopped too" + ) + + def test_5_errors(self) -> None: + self.wait_streaming() + with self.assertRaises(orin_control.ChannelError) as refused: + self.connect(token="not-the-token") + self.assertIn("401", str(refused.exception)) + + reply = self.set_target("sensing-up", "127.0.0.1", LEFT_PORT) + self.assertFalse(reply["ok"]) + self.assertEqual(reply["error"]["code"], "unknown_stream") + self.assertIn("sensing-left", reply["error"]["message"]) + + self.assertEqual( + self.set_target("sensing-left", "bad host", 1)["error"]["code"], + "invalid_host", + ) + self.assertEqual( + self.set_target("sensing-left", "127.0.0.1", 0)["error"]["code"], + "invalid_port", + ) + + # The command-line tool agrees. + cli = [ + sys.executable, + str(HERE.parent / "tools" / "orin_control.py"), + f"127.0.0.1:{self.port}", + "--token-file", + str(self.token_file), + ] + ok = subprocess.run( + cli + ["status"], capture_output=True, text=True, timeout=20, check=False + ) + self.assertEqual(ok.returncode, 0, ok.stderr) + self.assertIn("state: streaming", ok.stdout) + bad = subprocess.run( + cli + ["set-target", "nope", "127.0.0.1", "5730"], + capture_output=True, + check=False, + text=True, + timeout=20, + ) + self.assertEqual(bad.returncode, 1) + self.assertIn("unknown_stream", bad.stderr) + + def test_6_unreachable_target_reports_the_error(self) -> None: + status = self.wait_streaming() + # Every stream carries the key, empty while nothing is wrong. + for stream in status["streams"]: + self.assertIn("error", stream["direct"]) + self.assertEqual(self.stream_of(status, "sensing-right")["direct"]["error"], "") + + watcher = self.connect() + reply = self.set_target("sensing-left", "127.0.0.1", CLOSED_PORT) + self.assertTrue(reply["ok"], reply) + self.assertIn( + "error", self.stream_of(reply["status"], "sensing-left")["direct"] + ) + + # The first error to appear after the retarget must already name the new destination. + wanted = f"cannot connect to 127.0.0.1:{CLOSED_PORT}" + + def error_text() -> str | None: + error = self.stream_of(self.status(), "sensing-left")["direct"]["error"] + return error or None + + error = _wait(error_text, 15) + self.assertIsNotNone( + error, "direct.error never appeared:\n" + self.log("producer.log") + ) + self.assertTrue(error.startswith(wanted), error) + direct = self.stream_of(self.status(), "sensing-left")["direct"] + self.assertFalse(direct["connected"]) + self.assertEqual((direct["host"], direct["port"]), ("127.0.0.1", CLOSED_PORT)) + # The other stream is unaffected. + self.assertEqual( + self.stream_of(self.status(), "sensing-right")["direct"]["error"], "" + ) + # The failure is also pushed once to connected clients as an error event. + events = [] + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + try: + message = watcher.receive(timeout=1.0) + except (TimeoutError, OSError): + continue + if message.get("event") == "error": + events.append(message) + if message["error"]["code"] == "direct_link_failed": + break + watcher._sock.close() + direct_events = [ + e for e in events if e["error"]["code"] == "direct_link_failed" + ] + self.assertEqual(len(direct_events), 1, events) + self.assertIn("sensing-left", direct_events[0]["error"]["message"]) + self.assertIn("cannot connect to", direct_events[0]["error"]["message"]) + + # Turning the link off clears the error. + self.assertTrue(self.set_target("sensing-left", "")["ok"]) + self.assertTrue( + _wait( + lambda: ( + self.stream_of(self.status(), "sensing-left")["direct"]["error"] + == "" + ), + 5, + ), + "the error was not cleared", + ) + + def test_9_producer_stops_on_sigint(self) -> None: + self.producer.send_signal(signal.SIGINT) + self.assertEqual(self.producer.wait(timeout=15), 0, self.log("producer.log")) + + +if __name__ == "__main__": + unittest.main(verbosity=2)