diff --git a/src/hubblenetwork/cli.py b/src/hubblenetwork/cli.py index eccf408..5783ca0 100644 --- a/src/hubblenetwork/cli.py +++ b/src/hubblenetwork/cli.py @@ -42,12 +42,42 @@ # Symbol time length in ms SYMBOL_TIME = 8.0 -# Symbol time tolerance. With our envelope detection, we can expect the symbol +# Symbol time tolerance. With our envelope detection, we can expect the symbol # to be off by 1/2 of the length of the symbol before that envelope -# reads the adjacent symbol, Any deviation from the nominal 8ms will see a dB +# reads the adjacent symbol, Any deviation from the nominal 8ms will see a dB # loss, but the frequency will be correct TIMING_TOLERANCE = 4.0 +_TIMESTAMP_FORMAT = "%Y%m%d_%H%M%S" # used to name default --record/--analyze output files + + +def _default_iq_capture_name() -> str: + return f"iq_capture_{datetime.now().strftime(_TIMESTAMP_FORMAT)}.npy" + + +def _default_analysis_name() -> str: + return f"analysis_{datetime.now().strftime(_TIMESTAMP_FORMAT)}.txt" + + +# One entry per ``--record``/``--analyze`` mode, keyed by the sat.py function +# name to call: the in-progress status label, how to name the output file +# when --output is omitted, and the mode to open that file in. Looked up via +# getattr(sat_mod, mode) rather than storing the function directly, so tests +# that patch ``hubblenetwork.cli.sat_mod`` still take effect. +_ONE_SHOT_MODES = { + "record": { + "label": "Capturing IQ samples", + "default_name": _default_iq_capture_name, + "write_mode": "wb", + }, + "analyze": { + "label": "Recording and analyzing", + "default_name": _default_analysis_name, + "write_mode": "w", + }, +} + + def _parse_key(key_str: str) -> bytes: """Parse an encryption key from hex or base64. Returns raw bytes. @@ -3081,6 +3111,25 @@ def _docker_err_msg() -> str: ) +def _report_sat_error(exc: Exception, *, suppress_info_messages: bool = False) -> None: + """Print a ``sat`` command's DockerError/SatelliteError and exit(1). + + A ``DockerError`` with no message (Docker isn't installed at all) falls + back to the install hint. When *suppress_info_messages* is set (JSON + output mode) the error is emitted as a ``{"error": ...}`` JSON line + instead of a colored ``[ERROR]`` line to stderr. + """ + if isinstance(exc, sat_mod.DockerError): + msg = str(exc) or _docker_err_msg() + else: + msg = str(exc) + if suppress_info_messages: + click.echo(json.dumps({"error": msg})) + else: + click.secho(f"\n[ERROR] {msg}", fg="red", err=True) + sys.exit(1) + + @cli.group() def sat() -> None: """Satellite (PlutoSDR) utilities.""" @@ -3101,10 +3150,57 @@ def _run_sat_scan( counter_mode: str = UNIX_TIME, auto_detect_ctr: bool = False, show_failed_decryption: bool = False, + record: Optional[float] = None, + analyze: Optional[float] = None, + output_path: Optional[str] = None, ) -> None: """Shared implementation for ``sat scan`` and ``sat mock-scan``.""" mode_label = "mock satellite receiver" if mock else "satellite receiver" + is_one_shot = record is not None or analyze is not None + if record is not None and analyze is not None: + raise click.UsageError("--record and --analyze are mutually exclusive") + + if debug: + sat_logger = logging.getLogger("hubblenetwork.sat") + sat_logger.setLevel(logging.DEBUG) + sat_logger.addHandler(_handler) + + if is_one_shot: + try: + sat_mod.ensure_docker_available() + except sat_mod.DockerError as exc: + _report_sat_error(exc) + + click.secho(f"[INFO] Web GUI: {sat_mod.web_ui_url()}", fg="green") + + def _on_status(msg: str) -> None: + click.secho(f"[INFO] {msg}", fg="cyan", err=True) + + duration = record if record is not None else analyze + mode = "record" if record is not None else "analyze" + cfg = _ONE_SHOT_MODES[mode] + out_path = output_path or cfg["default_name"]() + + click.secho(f"[INFO] {cfg['label']} for {duration}s... (Press Ctrl+C to stop)") + try: + result = getattr(sat_mod, mode)( + duration, + mock=mock, + pluto_uri=pluto_uri, + on_status=_on_status, + ) + except (sat_mod.DockerError, sat_mod.SatelliteError) as exc: + _report_sat_error(exc) + except KeyboardInterrupt: + click.secho("\n[INFO] Cancelled.", fg="yellow", err=True) + sys.exit(130) + + with open(out_path, cfg["write_mode"]) as f: + f.write(result) + click.secho(f"[SUCCESS] Saved to {out_path}", fg="green") + return + printer_class = _SAT_STREAMING_PRINTERS.get( output_format.lower(), _SatStreamingTablePrinter ) @@ -3137,21 +3233,11 @@ def _run_sat_scan( key_len=len(decoded_key) if decoded_key is not None else 0, ) - if debug: - sat_logger = logging.getLogger("hubblenetwork.sat") - sat_logger.setLevel(logging.DEBUG) - sat_logger.addHandler(_handler) - # Fail fast: verify Docker is available before printing anything. try: sat_mod.ensure_docker_available() except sat_mod.DockerError as exc: - msg = str(exc) or _docker_err_msg() - if printer.suppress_info_messages: - click.echo(json.dumps({"error": msg})) - else: - click.secho(f"\n[ERROR] {msg}", fg="red", err=True) - sys.exit(1) + _report_sat_error(exc, suppress_info_messages=printer.suppress_info_messages) if not printer.suppress_info_messages: click.secho( @@ -3212,21 +3298,9 @@ def _on_interrupt(sig, frame): printer.print_row(pkt) if count is not None and printer.packet_count >= count: break - except sat_mod.DockerError as exc: - error_occurred = True - msg = str(exc) or _docker_err_msg() - if printer.suppress_info_messages: - click.echo(json.dumps({"error": msg})) - else: - click.secho(f"\n[ERROR] {msg}", fg="red", err=True) - sys.exit(1) - except sat_mod.SatelliteError as e: + except (sat_mod.DockerError, sat_mod.SatelliteError) as exc: error_occurred = True - if printer.suppress_info_messages: - click.echo(json.dumps({"error": str(e)})) - else: - click.secho(f"\n[ERROR] {e}", fg="red", err=True) - sys.exit(1) + _report_sat_error(exc, suppress_info_messages=printer.suppress_info_messages) except KeyboardInterrupt: pass finally: @@ -3261,6 +3335,18 @@ def _sat_scan_options(fn): help="Encoding format for packet payload"), click.option("--pluto-uri", "pluto_uri", type=str, default=None, help="PlutoSDR URI passed to the container (e.g. usb:, ip:192.168.2.1)"), + click.option("--record", type=float, default=None, show_default=False, + help="Instead of streaming packets, capture this many seconds of " + "raw IQ samples and save them to a .npy file. Mutually " + "exclusive with --analyze."), + click.option("--analyze", type=float, default=None, show_default=False, + help="Instead of streaming packets, record for this many seconds " + "and save the decoded-packet log to a text file. Mutually " + "exclusive with --record."), + click.option("--output", "output_path", type=click.Path(), default=None, + show_default=False, + help="Output file path for --record/--analyze (default: " + "auto-generated name)"), click.option("--debug", is_flag=True, default=False, help="Enable debug logging to stderr"), ]): @@ -3315,11 +3401,16 @@ def sat_scan(ctx, **kwargs) -> None: source is auto-detected (UNIX_TIME or DEVICE_UPTIME) unless --counter-mode is given explicitly. + Pass --record or --analyze (mutually exclusive) to capture raw IQ samples + or a decoded-packet log to a local file instead of streaming packets. + Example: hubblenetwork sat scan --timeout 30 hubblenetwork sat scan -o json --timeout 10 hubblenetwork sat scan -n 5 hubblenetwork sat scan --key "a562a2f7e4c62bed52ab09633878f62b" --timeout 60 + hubblenetwork sat scan --record 10 + hubblenetwork sat scan --analyze 10 --output analysis.txt """ key = kwargs.get("key") counter_mode = kwargs["counter_mode"] @@ -3350,9 +3441,13 @@ def sat_mock_scan(**kwargs) -> None: Uses simulated data -- no PlutoSDR hardware required. Useful for testing the satellite scanning interface. + Pass --record or --analyze (mutually exclusive) to capture simulated IQ + samples or a decoded-packet log to a local file instead of streaming. + Example: hubblenetwork sat mock-scan --timeout 30 hubblenetwork sat mock-scan -o json -n 5 + hubblenetwork sat mock-scan --record 10 """ _run_sat_scan(mock=True, **kwargs) diff --git a/src/hubblenetwork/sat.py b/src/hubblenetwork/sat.py index 62559c1..e6a48b7 100644 --- a/src/hubblenetwork/sat.py +++ b/src/hubblenetwork/sat.py @@ -14,8 +14,9 @@ import logging import os import time +from contextlib import contextmanager from pathlib import Path -from typing import Callable, Dict, Generator, List, Optional, Set, Tuple +from typing import Callable, Dict, Generator, Iterator, List, Optional, Set, Tuple import httpx @@ -30,6 +31,10 @@ API_PORT = 8050 _CONTAINER_INTERNAL_PORT = 8050 # fixed by the Docker image +# Extra time (beyond the recording duration itself) to allow the receiver to +# process and respond before we give up on the HTTP request. +_RECORDING_HTTP_TIMEOUT_MARGIN = 30 + def _packets_url(port: int = API_PORT) -> str: return f"http://localhost:{port}/api/packets" @@ -48,6 +53,14 @@ def _timedomain_url(port: int = API_PORT) -> str: return f"http://localhost:{port}/api/timedomain" +def _iq_capture_url(port: int = API_PORT) -> str: + return f"http://localhost:{port}/api/iq_capture" + + +def _record_analyze_url(port: int = API_PORT) -> str: + return f"http://localhost:{port}/api/record_analyze" + + # --------------------------------------------------------------------------- # Docker helpers # --------------------------------------------------------------------------- @@ -289,6 +302,36 @@ def fetch_packets(port: int = API_PORT) -> List[SatellitePacket]: return _parse_jsonl(resp.text) +def iq_capture(duration: float, port: int = API_PORT) -> bytes: + """Capture *duration* seconds of raw IQ samples, returning a ``.npy`` file body.""" + url = _iq_capture_url(port) + try: + resp = httpx.get( + url, + params={"duration": duration}, + timeout=duration + _RECORDING_HTTP_TIMEOUT_MARGIN, + ) + resp.raise_for_status() + except httpx.HTTPError as exc: + raise SatelliteError(f"Failed to capture IQ samples from {url}: {exc}") + return resp.content + + +def record_analyze(duration: float, port: int = API_PORT) -> str: + """Record for *duration* seconds and return the plaintext decoded-packet log.""" + url = _record_analyze_url(port) + try: + resp = httpx.get( + url, + params={"duration": duration}, + timeout=duration + _RECORDING_HTTP_TIMEOUT_MARGIN, + ) + resp.raise_for_status() + except httpx.HTTPError as exc: + raise SatelliteError(f"Failed to analyze recording from {url}: {exc}") + return resp.text + + # --------------------------------------------------------------------------- # High-level scanning # --------------------------------------------------------------------------- @@ -299,6 +342,64 @@ def _packet_key(pkt: SatellitePacket) -> Tuple[str, int]: return (pkt.device_id, pkt.seq_num) +@contextmanager +def _running_container( + *, + port: int, + image: str, + environment: Dict[str, str], + privileged: bool, + name: str, + wait_for_sdr: bool, + on_status: Optional[Callable[[str], None]], +) -> Iterator[None]: + """Pull, start, and wait for the receiver container to be ready; stop it on exit. + + Shared lifecycle for ``scan`` and the one-shot ``record``/``analyze`` + operations: ensure Docker is available, pull *image*, start the + container, wait for its API (and the SDR, unless *wait_for_sdr* is + ``False``), then yield. The container is guaranteed to be stopped when + the ``with`` block exits, including on exception. + """ + _emit = on_status or (lambda _msg: None) + + ensure_docker_available() + + if _image_exists_locally(image): + _emit(f"Using local image {image}...") + else: + _emit("Pulling Docker image...") + pull_image(image) + + _emit("Starting container...") + container_id = start_container( + image=image, + port=port, + environment=environment, + privileged=privileged, + name=name, + ) + try: + _emit("Waiting for receiver API to be ready...") + _wait_for_api(port=port) + if wait_for_sdr: + _emit("Waiting for PlutoSDR to connect...") + _wait_for_sdr(port=port) + yield + finally: + stop_container(container_id) + + +def _build_environment(mock: bool, pluto_uri: Optional[str]) -> Dict[str, str]: + """Build the container environment shared by ``scan`` and the one-shot ops.""" + environment: Dict[str, str] = {} + if mock: + environment["SDR_TYPE"] = "mock" + if pluto_uri is not None: + environment["PLUTO_URI"] = pluto_uri + return environment + + def scan( timeout: Optional[float] = None, poll_interval: float = 2.0, @@ -322,37 +423,19 @@ def scan( *on_status*, when provided, is called with a human-readable message at each lifecycle step (pull, start, wait, ready). """ - _emit = on_status or (lambda _msg: None) + environment = _build_environment(mock, pluto_uri) - ensure_docker_available() - - if _image_exists_locally(image): - _emit(f"Using local image {image}...") - else: - _emit("Pulling Docker image...") - pull_image(image) - - _emit("Starting container...") - container_name = MOCK_CONTAINER_NAME if mock else CONTAINER_NAME - environment: Dict[str, str] = {} - if mock: - environment["SDR_TYPE"] = "mock" - if pluto_uri is not None: - environment["PLUTO_URI"] = pluto_uri + _emit = on_status or (lambda _msg: None) - container_id = start_container( - image=image, + with _running_container( port=port, + image=image, environment=environment, privileged=not mock, - name=container_name, - ) - try: - _emit("Waiting for receiver API to be ready...") - _wait_for_api(port=port) - if not mock: - _emit("Waiting for PlutoSDR to connect...") - _wait_for_sdr(port=port) + name=MOCK_CONTAINER_NAME if mock else CONTAINER_NAME, + wait_for_sdr=not mock, + on_status=on_status, + ): _emit("Receiver ready, listening for packets...") seen: Set[Tuple[str, int]] = set() @@ -375,5 +458,92 @@ def scan( time.sleep(min(poll_interval, remaining)) else: time.sleep(poll_interval) - finally: - stop_container(container_id) + + +def _run_one_shot( + action: Callable[[int], "bytes | str"], + *, + action_status: str, + port: int = API_PORT, + image: str = DOCKER_IMAGE, + mock: bool = False, + pluto_uri: Optional[str] = None, + on_status: Optional[Callable[[str], None]] = None, +) -> "bytes | str": + """Run *action* against a freshly started receiver container, then tear it down. + + Shared lifecycle for the one-shot ``record``/``analyze`` operations, which + (unlike ``scan``) make a single blocking request instead of polling. + + When *mock* is ``True`` the container is started in mock mode, the same + as ``scan``'s *mock* parameter. + """ + environment = _build_environment(mock, pluto_uri) + + _emit = on_status or (lambda _msg: None) + + with _running_container( + port=port, + image=image, + environment=environment, + privileged=not mock, + name=MOCK_CONTAINER_NAME if mock else CONTAINER_NAME, + wait_for_sdr=not mock, + on_status=on_status, + ): + _emit(action_status) + return action(port) + + +def record( + duration: float, + port: int = API_PORT, + image: str = DOCKER_IMAGE, + *, + mock: bool = False, + pluto_uri: Optional[str] = None, + on_status: Optional[Callable[[str], None]] = None, +) -> bytes: + """Capture *duration* seconds of raw IQ samples, managing the Docker lifecycle. + + When *mock* is ``True`` the container is started in mock mode, the same + as ``scan``'s *mock* parameter. + + Returns the raw bytes of the ``.npy`` file produced by the receiver. + """ + return _run_one_shot( + lambda p: iq_capture(duration, port=p), + action_status=f"Capturing IQ samples for {duration}s...", + port=port, + image=image, + mock=mock, + pluto_uri=pluto_uri, + on_status=on_status, + ) + + +def analyze( + duration: float, + port: int = API_PORT, + image: str = DOCKER_IMAGE, + *, + mock: bool = False, + pluto_uri: Optional[str] = None, + on_status: Optional[Callable[[str], None]] = None, +) -> str: + """Record for *duration* seconds and decode packets, managing the Docker lifecycle. + + When *mock* is ``True`` the container is started in mock mode, the same + as ``scan``'s *mock* parameter. + + Returns the plaintext decoded-packet log produced by the receiver. + """ + return _run_one_shot( + lambda p: record_analyze(duration, port=p), + action_status=f"Recording and analyzing for {duration}s...", + port=port, + mock=mock, + image=image, + pluto_uri=pluto_uri, + on_status=on_status, + ) diff --git a/tests/test_sat.py b/tests/test_sat.py index d7859ef..dde9ac9 100644 --- a/tests/test_sat.py +++ b/tests/test_sat.py @@ -4,6 +4,7 @@ import base64 import json +import os from pathlib import Path from unittest.mock import MagicMock, patch @@ -419,6 +420,257 @@ def test_scan_mock_skips_sdr_check( mock_wait_sdr.assert_not_called() +# --------------------------------------------------------------------------- +# Shared container lifecycle +# --------------------------------------------------------------------------- + + +class TestRunningContainer: + @patch("hubblenetwork.sat._image_exists_locally", return_value=False) + @patch("hubblenetwork.sat.stop_container") + @patch("hubblenetwork.sat._wait_for_sdr") + @patch("hubblenetwork.sat._wait_for_api") + @patch("hubblenetwork.sat.start_container") + @patch("hubblenetwork.sat.pull_image") + @patch("hubblenetwork.sat.ensure_docker_available") + def test_pulls_starts_and_stops( + self, mock_ensure, mock_pull, mock_start, mock_wait, mock_wait_sdr, mock_stop, + mock_image_exists, + ): + mock_start.return_value = "container789" + + with sat._running_container( + port=sat.API_PORT, + image=sat.DOCKER_IMAGE, + environment={"FOO": "bar"}, + privileged=True, + name=sat.CONTAINER_NAME, + wait_for_sdr=True, + on_status=None, + ): + pass + + mock_ensure.assert_called_once() + mock_pull.assert_called_once_with(sat.DOCKER_IMAGE) + mock_start.assert_called_once_with( + image=sat.DOCKER_IMAGE, + port=sat.API_PORT, + environment={"FOO": "bar"}, + privileged=True, + name=sat.CONTAINER_NAME, + ) + mock_wait.assert_called_once_with(port=sat.API_PORT) + mock_wait_sdr.assert_called_once_with(port=sat.API_PORT) + mock_stop.assert_called_once_with("container789") + + @patch("hubblenetwork.sat._image_exists_locally", return_value=True) + @patch("hubblenetwork.sat.stop_container") + @patch("hubblenetwork.sat._wait_for_sdr") + @patch("hubblenetwork.sat._wait_for_api") + @patch("hubblenetwork.sat.start_container") + @patch("hubblenetwork.sat.pull_image") + @patch("hubblenetwork.sat.ensure_docker_available") + def test_skips_pull_when_image_exists_locally( + self, mock_ensure, mock_pull, mock_start, mock_wait, mock_wait_sdr, mock_stop, + mock_image_exists, + ): + mock_start.return_value = "container789" + + with sat._running_container( + port=sat.API_PORT, + image="sdr-docker:dev", + environment={}, + privileged=True, + name=sat.CONTAINER_NAME, + wait_for_sdr=True, + on_status=None, + ): + pass + + mock_image_exists.assert_called_once_with("sdr-docker:dev") + mock_pull.assert_not_called() + + @patch("hubblenetwork.sat._image_exists_locally", return_value=False) + @patch("hubblenetwork.sat.stop_container") + @patch("hubblenetwork.sat._wait_for_sdr") + @patch("hubblenetwork.sat._wait_for_api") + @patch("hubblenetwork.sat.start_container") + @patch("hubblenetwork.sat.pull_image") + @patch("hubblenetwork.sat.ensure_docker_available") + def test_skips_sdr_wait_when_disabled( + self, mock_ensure, mock_pull, mock_start, mock_wait, mock_wait_sdr, mock_stop, + mock_image_exists, + ): + mock_start.return_value = "container789" + + with sat._running_container( + port=sat.API_PORT, + image=sat.DOCKER_IMAGE, + environment={}, + privileged=False, + name=sat.MOCK_CONTAINER_NAME, + wait_for_sdr=False, + on_status=None, + ): + pass + + mock_wait_sdr.assert_not_called() + + @patch("hubblenetwork.sat._image_exists_locally", return_value=False) + @patch("hubblenetwork.sat.stop_container") + @patch("hubblenetwork.sat._wait_for_sdr") + @patch("hubblenetwork.sat._wait_for_api") + @patch("hubblenetwork.sat.start_container") + @patch("hubblenetwork.sat.pull_image") + @patch("hubblenetwork.sat.ensure_docker_available") + def test_stops_container_on_exception( + self, mock_ensure, mock_pull, mock_start, mock_wait, mock_wait_sdr, mock_stop, + mock_image_exists, + ): + mock_start.return_value = "container789" + + with pytest.raises(ValueError): + with sat._running_container( + port=sat.API_PORT, + image=sat.DOCKER_IMAGE, + environment={}, + privileged=True, + name=sat.CONTAINER_NAME, + wait_for_sdr=True, + on_status=None, + ): + raise ValueError("boom") + + mock_stop.assert_called_once_with("container789") + + @patch("hubblenetwork.sat._image_exists_locally", return_value=False) + @patch("hubblenetwork.sat.stop_container") + @patch("hubblenetwork.sat._wait_for_sdr") + @patch("hubblenetwork.sat._wait_for_api") + @patch("hubblenetwork.sat.start_container") + @patch("hubblenetwork.sat.pull_image") + @patch("hubblenetwork.sat.ensure_docker_available") + def test_emits_status_messages( + self, mock_ensure, mock_pull, mock_start, mock_wait, mock_wait_sdr, mock_stop, + mock_image_exists, + ): + mock_start.return_value = "container789" + messages = [] + + with sat._running_container( + port=sat.API_PORT, + image=sat.DOCKER_IMAGE, + environment={}, + privileged=True, + name=sat.CONTAINER_NAME, + wait_for_sdr=True, + on_status=messages.append, + ): + pass + + assert "Pulling Docker image..." in messages + assert "Starting container..." in messages + assert "Waiting for receiver API to be ready..." in messages + assert "Waiting for PlutoSDR to connect..." in messages + + +# --------------------------------------------------------------------------- +# sat.record / sat.analyze +# --------------------------------------------------------------------------- + + +class TestRecord: + @patch("hubblenetwork.sat._image_exists_locally", return_value=False) + @patch("hubblenetwork.sat.stop_container") + @patch("hubblenetwork.sat._wait_for_sdr") + @patch("hubblenetwork.sat._wait_for_api") + @patch("hubblenetwork.sat.start_container") + @patch("hubblenetwork.sat.pull_image") + @patch("hubblenetwork.sat.ensure_docker_available") + @patch("hubblenetwork.sat.iq_capture") + def test_captures_and_returns_bytes( + self, mock_capture, mock_ensure, mock_pull, mock_start, mock_wait, mock_wait_sdr, mock_stop, + mock_image_exists, + ): + mock_start.return_value = "container123" + mock_capture.return_value = b"\x93NUMPYfake" + + result = sat.record(10) + + assert result == b"\x93NUMPYfake" + mock_capture.assert_called_once_with(10, port=sat.API_PORT) + mock_stop.assert_called_once_with("container123") + + @patch("hubblenetwork.sat._image_exists_locally", return_value=False) + @patch("hubblenetwork.sat.stop_container") + @patch("hubblenetwork.sat._wait_for_sdr") + @patch("hubblenetwork.sat._wait_for_api") + @patch("hubblenetwork.sat.start_container") + @patch("hubblenetwork.sat.pull_image") + @patch("hubblenetwork.sat.ensure_docker_available") + @patch("hubblenetwork.sat.iq_capture") + def test_passes_pluto_uri_as_environment( + self, mock_capture, mock_ensure, mock_pull, mock_start, mock_wait, mock_wait_sdr, mock_stop, + mock_image_exists, + ): + mock_start.return_value = "container123" + mock_capture.return_value = b"" + + sat.record(10, pluto_uri="usb:1.2.3") + + call_kwargs = mock_start.call_args[1] + assert call_kwargs["environment"] == {"PLUTO_URI": "usb:1.2.3"} + assert call_kwargs["privileged"] is True + assert call_kwargs["name"] == sat.CONTAINER_NAME + + @patch("hubblenetwork.sat._image_exists_locally", return_value=False) + @patch("hubblenetwork.sat.stop_container") + @patch("hubblenetwork.sat._wait_for_sdr") + @patch("hubblenetwork.sat._wait_for_api") + @patch("hubblenetwork.sat.start_container") + @patch("hubblenetwork.sat.pull_image") + @patch("hubblenetwork.sat.ensure_docker_available") + @patch("hubblenetwork.sat.iq_capture") + def test_mock_mode_params( + self, mock_capture, mock_ensure, mock_pull, mock_start, mock_wait, mock_wait_sdr, mock_stop, + mock_image_exists, + ): + mock_start.return_value = "container123" + mock_capture.return_value = b"" + + sat.record(10, mock=True) + + call_kwargs = mock_start.call_args[1] + assert call_kwargs["environment"] == {"SDR_TYPE": "mock"} + assert call_kwargs["privileged"] is False + assert call_kwargs["name"] == sat.MOCK_CONTAINER_NAME + # Mock mode has no hardware to wait for. + mock_wait_sdr.assert_not_called() + + +class TestAnalyze: + @patch("hubblenetwork.sat._image_exists_locally", return_value=False) + @patch("hubblenetwork.sat.stop_container") + @patch("hubblenetwork.sat._wait_for_sdr") + @patch("hubblenetwork.sat._wait_for_api") + @patch("hubblenetwork.sat.start_container") + @patch("hubblenetwork.sat.pull_image") + @patch("hubblenetwork.sat.ensure_docker_available") + @patch("hubblenetwork.sat.record_analyze") + def test_analyzes_and_returns_text( + self, mock_analyze, mock_ensure, mock_pull, mock_start, mock_wait, mock_wait_sdr, mock_stop, + mock_image_exists, + ): + mock_start.return_value = "container456" + mock_analyze.return_value = "device_id=0xAA seq=1\n" + + result = sat.analyze(10) + + assert result == "device_id=0xAA seq=1\n" + mock_analyze.assert_called_once_with(10, port=sat.API_PORT) + mock_stop.assert_called_once_with("container456") + + # --------------------------------------------------------------------------- # CLI - sat scan # --------------------------------------------------------------------------- @@ -585,3 +837,141 @@ def test_docker_not_available(self, mock_sat, runner): result = runner.invoke(cli, ["sat", "mock-scan", "--timeout", "1"]) assert result.exit_code != 0 assert "Docker is not installed" in result.output + + +# --------------------------------------------------------------------------- +# CLI - sat scan/mock-scan --record / --analyze +# --------------------------------------------------------------------------- + + +class TestSatRecordAnalyzeCli: + @pytest.fixture + def runner(self): + return CliRunner() + + @pytest.mark.parametrize("subcommand", ["scan", "mock-scan"]) + @patch("hubblenetwork.cli.sat_mod") + def test_record_writes_output_file(self, mock_sat, runner, subcommand): + mock_sat.DockerError = DockerError + mock_sat.SatelliteError = SatelliteError + mock_sat.record.return_value = b"\x93NUMPYfake" + + with runner.isolated_filesystem(): + result = runner.invoke( + cli, ["sat", subcommand, "--record", "5", "--output", "capture.npy"] + ) + assert result.exit_code == 0 + with open("capture.npy", "rb") as f: + assert f.read() == b"\x93NUMPYfake" + + mock_sat.record.assert_called_once() + assert mock_sat.record.call_args[0][0] == 5.0 + + @pytest.mark.parametrize( + "subcommand,expected_mock", [("scan", False), ("mock-scan", True)] + ) + @patch("hubblenetwork.cli.sat_mod") + def test_record_passes_mock_flag_correctly( + self, mock_sat, runner, subcommand, expected_mock + ): + mock_sat.DockerError = DockerError + mock_sat.SatelliteError = SatelliteError + mock_sat.record.return_value = b"" + + with runner.isolated_filesystem(): + result = runner.invoke(cli, ["sat", subcommand, "--record", "5"]) + + assert result.exit_code == 0 + assert mock_sat.record.call_args.kwargs["mock"] is expected_mock + + @pytest.mark.parametrize( + "subcommand,expected_mock", [("scan", False), ("mock-scan", True)] + ) + @patch("hubblenetwork.cli.sat_mod") + def test_analyze_passes_mock_flag_correctly( + self, mock_sat, runner, subcommand, expected_mock + ): + mock_sat.DockerError = DockerError + mock_sat.SatelliteError = SatelliteError + mock_sat.analyze.return_value = "" + + with runner.isolated_filesystem(): + result = runner.invoke(cli, ["sat", subcommand, "--analyze", "5"]) + + assert result.exit_code == 0 + assert mock_sat.analyze.call_args.kwargs["mock"] is expected_mock + + @pytest.mark.parametrize("subcommand", ["scan", "mock-scan"]) + @patch("hubblenetwork.cli.sat_mod") + def test_record_default_output_name(self, mock_sat, runner, subcommand): + mock_sat.DockerError = DockerError + mock_sat.SatelliteError = SatelliteError + mock_sat.record.return_value = b"data" + + with runner.isolated_filesystem(): + result = runner.invoke(cli, ["sat", subcommand, "--record", "5"]) + assert result.exit_code == 0 + npy_files = [f for f in os.listdir(".") if f.endswith(".npy")] + assert len(npy_files) == 1 + assert npy_files[0].startswith("iq_capture_") + + @pytest.mark.parametrize("subcommand", ["scan", "mock-scan"]) + @patch("hubblenetwork.cli.sat_mod") + def test_analyze_writes_output_file(self, mock_sat, runner, subcommand): + mock_sat.DockerError = DockerError + mock_sat.SatelliteError = SatelliteError + mock_sat.analyze.return_value = "device_id=0xAA seq=1\n" + + with runner.isolated_filesystem(): + result = runner.invoke( + cli, ["sat", subcommand, "--analyze", "5", "--output", "analysis.txt"] + ) + assert result.exit_code == 0 + with open("analysis.txt") as f: + assert f.read() == "device_id=0xAA seq=1\n" + + mock_sat.analyze.assert_called_once() + assert mock_sat.analyze.call_args[0][0] == 5.0 + + @pytest.mark.parametrize("subcommand", ["scan", "mock-scan"]) + @patch("hubblenetwork.cli.sat_mod") + def test_analyze_default_output_name(self, mock_sat, runner, subcommand): + mock_sat.DockerError = DockerError + mock_sat.SatelliteError = SatelliteError + mock_sat.analyze.return_value = "log\n" + + with runner.isolated_filesystem(): + result = runner.invoke(cli, ["sat", subcommand, "--analyze", "5"]) + assert result.exit_code == 0 + txt_files = [f for f in os.listdir(".") if f.startswith("analysis_")] + assert len(txt_files) == 1 + + @pytest.mark.parametrize("subcommand", ["scan", "mock-scan"]) + def test_record_and_analyze_mutually_exclusive(self, runner, subcommand): + result = runner.invoke( + cli, ["sat", subcommand, "--record", "5", "--analyze", "5"] + ) + assert result.exit_code != 0 + assert "mutually exclusive" in result.output + + @pytest.mark.parametrize("subcommand", ["scan", "mock-scan"]) + @patch("hubblenetwork.cli.sat_mod") + def test_record_docker_error(self, mock_sat, runner, subcommand): + mock_sat.DockerError = DockerError + mock_sat.SatelliteError = SatelliteError + mock_sat.record.side_effect = DockerError("Docker is not installed") + + result = runner.invoke(cli, ["sat", subcommand, "--record", "5"]) + assert result.exit_code != 0 + assert "Docker is not installed" in result.output + + @pytest.mark.parametrize("subcommand", ["scan", "mock-scan"]) + @patch("hubblenetwork.cli.sat_mod") + def test_analyze_satellite_error(self, mock_sat, runner, subcommand): + mock_sat.DockerError = DockerError + mock_sat.SatelliteError = SatelliteError + mock_sat.analyze.side_effect = SatelliteError("receiver not ready") + + result = runner.invoke(cli, ["sat", subcommand, "--analyze", "5"]) + assert result.exit_code != 0 + assert "receiver not ready" in result.output