diff --git a/tests/remote_control/test_remote_transmitter.py b/tests/remote_control/test_remote_transmitter.py index 9899965..9c50152 100644 --- a/tests/remote_control/test_remote_transmitter.py +++ b/tests/remote_control/test_remote_transmitter.py @@ -32,17 +32,22 @@ def _make_mock_sdr(): class TestSetRadio: + def _pluto_module(self, mock_sdr): + mod = MagicMock() + mod.Pluto = MagicMock(return_value=mock_sdr) + return mod + def test_pluto_alias(self): tx = RemoteTransmitter() mock_sdr = _make_mock_sdr() - with patch("ria_toolkit_oss.sdr.pluto.Pluto", return_value=mock_sdr): + with patch.dict("sys.modules", {"ria_toolkit_oss.sdr.pluto": self._pluto_module(mock_sdr)}): tx.set_radio("pluto", "ip:192.168.2.1") assert tx._sdr is mock_sdr def test_plutosdr_alias(self): tx = RemoteTransmitter() mock_sdr = _make_mock_sdr() - with patch("ria_toolkit_oss.sdr.pluto.Pluto", return_value=mock_sdr): + with patch.dict("sys.modules", {"ria_toolkit_oss.sdr.pluto": self._pluto_module(mock_sdr)}): tx.set_radio("PlutoSDR", "ip:192.168.2.1") assert tx._sdr is mock_sdr @@ -78,10 +83,20 @@ class TestSetRadio: tx.set_radio("blade", "") assert tx._sdr is mock_sdr + def test_bladerf_string_alias(self): + """'bladerf' string (not 'blade') must also resolve to blade.Blade.""" + tx = RemoteTransmitter() + mock_sdr = _make_mock_sdr() + mock_module = MagicMock() + mock_module.Blade = MagicMock(return_value=mock_sdr) + with patch.dict("sys.modules", {"ria_toolkit_oss.sdr.blade": mock_module}): + tx.set_radio("bladerf", "") + assert tx._sdr is mock_sdr + def test_case_insensitive(self): tx = RemoteTransmitter() mock_sdr = _make_mock_sdr() - with patch("ria_toolkit_oss.sdr.pluto.Pluto", return_value=mock_sdr): + with patch.dict("sys.modules", {"ria_toolkit_oss.sdr.pluto": self._pluto_module(mock_sdr)}): tx.set_radio("PLUTO", "ip:192.168.2.1") assert tx._sdr is mock_sdr @@ -91,8 +106,12 @@ class TestSetRadio: tx.set_radio("nonexistent_radio") def test_import_error_raises_runtime(self): + """ImportError during SDR driver load is re-raised as RuntimeError.""" tx = RemoteTransmitter() - with patch.dict("sys.modules", {"ria_toolkit_oss.sdr.pluto": None}): + # Inject a fake module whose Pluto class raises ImportError on import + bad_module = MagicMock() + bad_module.Pluto = MagicMock(side_effect=ImportError("pyadi-iio not installed")) + with patch.dict("sys.modules", {"ria_toolkit_oss.sdr.pluto": bad_module}): with pytest.raises((RuntimeError, ImportError)): tx.set_radio("pluto") @@ -209,7 +228,9 @@ class TestRunFunction: def test_set_radio_success(self): tx = RemoteTransmitter() mock_sdr = _make_mock_sdr() - with patch("ria_toolkit_oss.sdr.pluto.Pluto", return_value=mock_sdr): + mod = MagicMock() + mod.Pluto = MagicMock(return_value=mock_sdr) + with patch.dict("sys.modules", {"ria_toolkit_oss.sdr.pluto": mod}): resp = tx.run_function({"function_name": "set_radio", "radio_str": "pluto", "identifier": "ip:1.2.3.4"}) assert resp["status"] is True diff --git a/tests/remote_control/test_sdr_remote_integration.py b/tests/remote_control/test_sdr_remote_integration.py index 7335da4..2f13bed 100644 --- a/tests/remote_control/test_sdr_remote_integration.py +++ b/tests/remote_control/test_sdr_remote_integration.py @@ -389,3 +389,174 @@ class TestRunWithSdrRemote: # Both must appear assert "init_sdr" in call_order assert "init_remote_tx" in call_order + + +# --------------------------------------------------------------------------- +# Additional coverage gaps +# --------------------------------------------------------------------------- + + +class TestTransmitBufferAndTimeout: + """Verify the exact buffer and timeout constants used in start/stop.""" + + def _executor_with_ctrl(self): + from ria_toolkit_oss.orchestration.executor import CampaignExecutor + cfg = CampaignConfig.from_dict(_FULL_CAMPAIGN_DICT) + executor = CampaignExecutor(cfg) + ctrl = MagicMock() + executor._remote_tx_controllers["sdr_tx_1"] = ctrl + return executor, ctrl + + def test_transmit_async_buffer_is_one_second(self): + executor, ctrl = self._executor_with_ctrl() + tx = executor.config.transmitters[0] + step = tx.schedule[0] # duration = 10s + executor._start_transmitter(tx, step) + duration_arg = ctrl.transmit_async.call_args[0][0] + assert duration_arg == pytest.approx(step.duration + 1.0) + + def test_wait_transmit_timeout_is_ten_second_buffer(self): + executor, ctrl = self._executor_with_ctrl() + tx = executor.config.transmitters[0] + step = tx.schedule[0] # duration = 10s + executor._stop_transmitter(tx, step) + timeout = ctrl.wait_transmit.call_args[1]["timeout"] + assert timeout == pytest.approx(step.duration + 10.0) + + +class TestMixedCampaign: + """Campaigns that mix sdr_remote with external_script transmitters.""" + + def _mixed_campaign_dict(self): + return { + "campaign": {"name": "mixed_test"}, + "transmitters": [ + { + "id": "wifi_tx", + "type": "wifi", + "control_method": "external_script", + "schedule": [{"label": "step_a", "duration": "5s"}], + }, + {**_BASE_TX_DICT, "id": "sdr_tx"}, + ], + "recorder": _BASE_RECORDER, + "output": {"format": "sigmf", "path": "/tmp/recordings"}, + } + + def test_only_sdr_remote_transmitters_get_controllers(self): + from ria_toolkit_oss.orchestration.executor import CampaignExecutor + + cfg = CampaignConfig.from_dict(self._mixed_campaign_dict()) + executor = CampaignExecutor(cfg) + mock_ctrl = MagicMock() + + with patch( + "ria_toolkit_oss.remote_control.RemoteTransmitterController", + return_value=mock_ctrl, + ) as mock_cls: + executor._init_remote_tx_controllers() + + mock_cls.assert_called_once() # only the sdr_remote one + assert "sdr_tx" in executor._remote_tx_controllers + assert "wifi_tx" not in executor._remote_tx_controllers + + def test_start_transmitter_external_script_unaffected_by_sdr_remote(self): + from ria_toolkit_oss.orchestration.executor import CampaignExecutor + + cfg = CampaignConfig.from_dict(self._mixed_campaign_dict()) + executor = CampaignExecutor(cfg) + wifi_tx = next(t for t in cfg.transmitters if t.id == "wifi_tx") + step = wifi_tx.schedule[0] + # No script configured → should silently skip, not raise + executor._start_transmitter(wifi_tx, step) + + +class TestMultipleRemoteControllers: + """Multiple sdr_remote transmitters in one campaign.""" + + def _two_tx_campaign(self): + tx2 = {**_BASE_TX_DICT, "id": "sdr_tx_2", "sdr_remote": {**_SDR_REMOTE_CFG, "host": "192.168.1.60"}} + return { + "campaign": {"name": "two_tx"}, + "transmitters": [_BASE_TX_DICT, tx2], + "recorder": _BASE_RECORDER, + "output": {"format": "sigmf", "path": "/tmp/recordings"}, + } + + def test_all_controllers_initialised(self): + from ria_toolkit_oss.orchestration.executor import CampaignExecutor + + cfg = CampaignConfig.from_dict(self._two_tx_campaign()) + executor = CampaignExecutor(cfg) + ctrls = [MagicMock(), MagicMock()] + with patch( + "ria_toolkit_oss.remote_control.RemoteTransmitterController", + side_effect=ctrls, + ): + executor._init_remote_tx_controllers() + + assert len(executor._remote_tx_controllers) == 2 + assert "sdr_tx_1" in executor._remote_tx_controllers + assert "sdr_tx_2" in executor._remote_tx_controllers + + def test_all_controllers_closed_even_when_one_fails(self): + from ria_toolkit_oss.orchestration.executor import CampaignExecutor + + cfg = CampaignConfig.from_dict(self._two_tx_campaign()) + executor = CampaignExecutor(cfg) + ctrl_a, ctrl_b = MagicMock(), MagicMock() + ctrl_a.close.side_effect = RuntimeError("ssh gone") + executor._remote_tx_controllers = {"sdr_tx_1": ctrl_a, "sdr_tx_2": ctrl_b} + + executor._close_remote_tx_controllers() # must not raise + + ctrl_a.close.assert_called_once() + ctrl_b.close.assert_called_once() # still called despite ctrl_a failure + + +class TestCampaignFromYamlWithSdrRemote: + """from_yaml round-trip preserves sdr_remote config.""" + + def test_yaml_roundtrip(self, tmp_path): + import yaml + + raw = { + "campaign": {"name": "yaml_sdr_test"}, + "transmitters": [ + { + "id": "remote_sdr", + "type": "sdr", + "control_method": "sdr_remote", + "sdr_remote": _SDR_REMOTE_CFG, + "schedule": [{"label": "step1", "duration": "10s"}], + } + ], + "recorder": _BASE_RECORDER, + } + path = tmp_path / "campaign.yml" + path.write_text(yaml.dump(raw)) + cfg = CampaignConfig.from_yaml(str(path)) + tx = cfg.transmitters[0] + assert tx.control_method == "sdr_remote" + assert tx.sdr_remote["host"] == "192.168.1.50" + assert tx.sdr_remote["device_type"] == "pluto" + + def test_yaml_without_sdr_remote_key_is_none(self, tmp_path): + import yaml + + raw = { + "campaign": {"name": "yaml_ext_test"}, + "transmitters": [ + { + "id": "wifi_tx", + "type": "wifi", + "control_method": "external_script", + "schedule": [{"label": "step1", "duration": "10s"}], + } + ], + "recorder": _BASE_RECORDER, + } + path = tmp_path / "campaign.yml" + path.write_text(yaml.dump(raw)) + cfg = CampaignConfig.from_yaml(str(path)) + assert cfg.transmitters[0].sdr_remote is None