test suite

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
 

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