"""Tests for discover command.""" import json import re from unittest.mock import MagicMock, patch from click.testing import CliRunner from ria_toolkit_oss.ria_toolkit_oss_cli.ria_toolkit_oss.discover import ( # find_bladerf_devices,; find_thinkrf_devices,; find_uhd_devices, discover, discover_all_devices, find_hackrf_devices, find_pluto_devices, find_rtlsdr_devices, load_sdr_drivers, ) def test_discover_pluto_no_devices(): """Test PlutoSDR discovery with no devices.""" with ( patch.dict("sys.modules", {"iio": MagicMock()}) as mock_modules, patch("ria_toolkit_oss.ria_toolkit_oss_cli.ria_toolkit_oss.discover.get_usb_devices") as mock_usb, ): mock_iio = mock_modules["iio"] mock_iio.scan_contexts.return_value = {} mock_usb.return_value = [] devices = find_pluto_devices() assert devices == [] def test_discover_pluto_with_device(): """Test PlutoSDR discovery with device present.""" with patch.dict("sys.modules", {"iio": MagicMock()}) as mock_modules: mock_iio = mock_modules["iio"] mock_ctx = MagicMock() mock_ctx.attrs = {"hw_serial": "123456", "fw_version": "1.0"} mock_ctx._destroy = MagicMock() mock_iio.scan_contexts.return_value = {"ip:pluto.local": "PlutoSDR (ADALM-PLUTO)"} mock_iio.Context.return_value = mock_ctx devices = find_pluto_devices() assert len(devices) == 1 assert devices[0]["type"] == "PlutoSDR" assert devices[0]["serial"] == "123456" assert devices[0]["firmware"] == "1.0" assert devices[0]["uri"] == "ip:pluto.local" def test_discover_hackrf_no_devices(): """Test HackRF discovery with no devices.""" with patch("ria_toolkit_oss.ria_toolkit_oss_cli.ria_toolkit_oss.discover.subprocess") as mock_subprocess: mock_subprocess.check_output.return_value = "" devices = find_hackrf_devices() assert devices == [] def test_discover_hackrf_with_devices(): """Test HackRF discovery with devices present.""" with patch("ria_toolkit_oss.ria_toolkit_oss_cli.ria_toolkit_oss.discover.subprocess") as mock_subprocess: mock_subprocess.check_output.return_value = """ hackrf_info version: 2023.01.1 libhackrf version: 2023.01.1 (0.8) Found HackRF Index: 0 Serial number: serial123 Board ID Number: 2 (HackRF One) Firmware Version: v2.1.0 (API:1.08) Part ID Number: 0xa000cb3c 0x005d4761 Index: 1 Serial number: serial456 Board ID Number: 2 (HackRF One) Firmware Version: v2.1.0 (API:1.08) Part ID Number: 0xa000cb3c 0x005d4761 """ devices = find_hackrf_devices() assert len(devices) == 2 assert devices[0]["type"] == "HackRF One" assert devices[0]["serial"] == "serial123" assert devices[0]["device_index"] == 0 or devices[0]["device_index"] == "0" assert devices[1]["serial"] == "serial456" assert devices[1]["device_index"] == 1 or devices[1]["device_index"] == "1" def test_discover_rtlsdr_no_devices(): """Test RTL-SDR discovery with no devices.""" with patch("ria_toolkit_oss.ria_toolkit_oss.ria_toolkit_oss.discover.subprocess") as mock_subprocess: mock_subprocess.check_output.return_value = "" devices = find_rtlsdr_devices() assert devices == [] def test_discover_rtlsdr_with_devices(): """Test RTL-SDR discovery with devices present.""" with patch("ria_toolkit_oss.ria_toolkit_oss.ria_toolkit_oss.discover.subprocess") as mock_subprocess: mock_subprocess.check_output.return_value = """ Found 2 device(s): 0: RTLSDRBlog, Blog V4, SN: 00000001 1: RTLSDRBlog, Blog V4, SN: 00000002 Using device 0: Generic RTL2832U OEM Found Rafael Micro R828D tuner RTL-SDR Blog V4 Detected """ devices = find_rtlsdr_devices() assert len(devices) == 2 assert devices[0]["type"] == "RTL-SDR" assert devices[0]["serial"] == "00000001" assert devices[0]["device_index"] == 0 or devices[0]["device_index"] == "0" def test_discover_all_devices_filter(): """Test discovering devices with type filter.""" with ( patch("ria_toolkit_oss.ria_toolkit_oss_cli.ria_toolkit_oss.discover.find_pluto_devices") as mock_pluto, patch("ria_toolkit_oss.ria_toolkit_oss_cli.ria_toolkit_oss.discover.find_hackrf_devices") as mock_hackrf, patch("ria_toolkit_oss.ria_toolkit_oss_cli.ria_toolkit_oss.discover.find_bladerf_devices") as mock_bladerf, patch("ria_toolkit_oss.ria_toolkit_oss_cli.ria_toolkit_oss.discover.find_uhd_devices") as mock_usrp, patch("ria_toolkit_oss.ria_toolkit_oss_cli.ria_toolkit_oss.discover.find_rtlsdr_devices") as mock_rtlsdr, patch("ria_toolkit_oss.ria_toolkit_oss_cli.ria_toolkit_oss.discover.find_thinkrf_devices") as mock_thinkrf, ): mock_pluto.return_value = [{"type": "PlutoSDR", "uri": "ip:pluto.local"}] mock_hackrf.return_value = [] mock_bladerf.return_value = [] mock_usrp.return_value = [] mock_rtlsdr.return_value = [] mock_thinkrf.return_value = [] # Test filtering by pluto load_sdr_drivers(verbose=False) devices = discover_all_devices() mock_pluto.assert_called_once() mock_hackrf.assert_called_once() mock_bladerf.assert_called_once() assert len(devices["devices"]) == 1 assert len(devices["pluto_devices"]) == 1 def test_discover_all_devices_no_filter(): """Test discovering all device types.""" with ( patch("ria_toolkit_oss.ria_toolkit_oss_cli.ria_toolkit_oss.discover.find_pluto_devices") as mock_pluto, patch("ria_toolkit_oss.ria_toolkit_oss_cli.ria_toolkit_oss.discover.find_hackrf_devices") as mock_hackrf, patch("ria_toolkit_oss.ria_toolkit_oss_cli.ria_toolkit_oss.discover.find_bladerf_devices") as mock_bladerf, patch("ria_toolkit_oss.ria_toolkit_oss_cli.ria_toolkit_oss.discover.find_uhd_devices") as mock_usrp, patch("ria_toolkit_oss.ria_toolkit_oss_cli.ria_toolkit_oss.discover.find_rtlsdr_devices") as mock_rtlsdr, patch("ria_toolkit_oss.ria_toolkit_oss_cli.ria_toolkit_oss.discover.find_thinkrf_devices") as mock_thinkrf, ): mock_pluto.return_value = [{"type": "PlutoSDR", "uri": "ip:pluto.local"}] mock_hackrf.return_value = [{"type": "HackRF"}] mock_bladerf.return_value = [] mock_usrp.return_value = [] mock_rtlsdr.return_value = [] mock_thinkrf.return_value = [] load_sdr_drivers(verbose=False) devices = discover_all_devices() mock_pluto.assert_called_once() mock_hackrf.assert_called_once() mock_bladerf.assert_called_once() mock_usrp.assert_called_once() mock_rtlsdr.assert_called_once() assert len(devices["devices"]) == 2 assert len(devices["pluto_devices"]) == 1 assert len(devices["hackrf_devices"]) == 1 def test_discover_command_no_devices(): """Test discover CLI command with no devices.""" runner = CliRunner() with patch("ria_toolkit_oss.ria_toolkit_oss_cli.ria_toolkit_oss.discover.discover_all_devices") as mock_discover: mock_discover.return_value = { "loaded_drivers": [], "failed_drivers": [], "devices": [], "total_devices": 0, "uhd_devices": [], "pluto_devices": [], "rtlsdr_devices": [], "bladerf_devices": [], "hackrf_devices": [], } result = runner.invoke(discover) assert result.exit_code == 0 assert "No devices detected" in result.output def test_discover_command(): """Test discover CLI command.""" runner = CliRunner() result = runner.invoke(discover) radios = ["USRP/UHD", "PlutoSDR", "RTL-SDR", "BladeRF", "HackRF", "ThinkRF"] match = re.search(r"Detected devices: (\d+)", result.output) if match: total_devices = int(match.group(1)) else: total_devices = 0 if result.exit_code == 0: assert "Attached Devices" in result.output assert "Discovery Summary" in result.output if total_devices > 0: assert any(radio in result.output for radio in radios) else: assert not any(radio in result.output for radio in radios) else: assert result.exit_code == 1 assert isinstance(result.exception, AttributeError) assert "undefined symbol: iio_get_backends_count" in str(result.exception) def test_discover_command_json_output(): """Test discover CLI command with JSON output.""" runner = CliRunner() with patch("ria_toolkit_oss.ria_toolkit_oss_cli.ria_toolkit_oss.discover.discover_all_devices") as mock_discover: mock_discover.return_value = { "loaded_drivers": [], "failed_drivers": [], "devices": [{"type": "HackRF", "serial": "123456", "status": "available"}], "total_devices": 1, "uhd_devices": [], "pluto_devices": [], "rtlsdr_devices": [], "bladerf_devices": [], "hackrf_devices": [{"type": "HackRF", "serial": "123456", "status": "available"}], } result = runner.invoke(discover, ["--json-output"]) output_data = json.loads(result.output) assert result.exit_code == 0 assert output_data["total_devices"] == 1 assert len(output_data["devices"]) == 1 assert output_data["devices"][0]["type"] == "HackRF" def test_discover_command_verbose(): """Test discover CLI command with verbose output.""" runner = CliRunner() with patch("ria_toolkit_oss.ria_toolkit_oss_cli.ria_toolkit_oss.discover.discover_all_devices") as mock_discover: mock_discover.return_value = { "loaded_drivers": [], "failed_drivers": [], "devices": [ { "type": "PlutoSDR", "serial": "123456", "firmware": "1.0", "uri": "ip:pluto.local", "status": "available", } ], "total_devices": 1, "uhd_devices": [], "pluto_devices": [], "rtlsdr_devices": [], "bladerf_devices": [], "hackrf_devices": [ { "type": "PlutoSDR", "serial": "123456", "firmware": "1.0", "uri": "ip:pluto.local", "status": "available", } ], } result = runner.invoke(discover, ["--verbose"]) assert result.exit_code == 0 assert "RTL-SDR devices: None found" in result.output or "\n rtlsdr:" in result.output