ridesx: add tests for fls integration

Signed-off-by: Benny Zlotnik <bzlotnik@redhat.com>

Author: bennyz

python/packages/jumpstarter-driver-ridesx/jumpstarter_driver_ridesx/client_test.py

@@ -0,0 +1,89 @@
+import tempfile
+from unittest.mock import patch
+
+import pytest
+from jumpstarter_driver_pyserial.driver import PySerial
+
+from .driver import RideSXDriver
+from jumpstarter.common.utils import serve
+
+
+@pytest.fixture(scope="session")
+def temp_storage_dir():
+    with tempfile.TemporaryDirectory() as temp_dir:
+        yield temp_dir
+
+
+@pytest.fixture(scope="session")
+def ridesx_driver(temp_storage_dir):
+    yield RideSXDriver(
+        storage_dir=temp_storage_dir,
+        children={
+            "serial": PySerial(url="loop://"),
+        },
+    )
+
+
+@pytest.fixture
+def ridesx_client(ridesx_driver):
+    """Create a client instance for testing client-side methods"""
+    with serve(ridesx_driver) as client:
+        yield client
+
+
+# Validate Partition Mappings Tests
+
+
+def test_validate_partition_mappings(ridesx_client):
+    """Test partition mapping validation"""
+    # None is valid (auto-detect mode)
+    ridesx_client._validate_partition_mappings(None)
+
+    # Valid mapping
+    ridesx_client._validate_partition_mappings({"boot": "/path/to/boot.img"})
+
+    # Empty path raises
+    with pytest.raises(ValueError, match="has an empty file path"):
+        ridesx_client._validate_partition_mappings({"boot": ""})
+
+    # Whitespace-only path raises
+    with pytest.raises(ValueError, match="has an empty file path"):
+        ridesx_client._validate_partition_mappings({"boot": "   "})
+
+
+# Flash OCI Auto Tests
+
+
+def test_flash_oci_auto_success(ridesx_client):
+    """Test successful flash_oci_auto call"""
+    with patch.object(ridesx_client, "call") as mock_call:
+        mock_call.side_effect = [
+            {"status": "device_found", "device_id": "ABC123"},
+            {"status": "success"},
+        ]
+
+        result = ridesx_client.flash_oci_auto("oci://quay.io/org/image:tag")
+
+        assert result == {"status": "success"}
+        # Verify flash_oci_image was called with the OCI URL
+        flash_call = mock_call.call_args_list[1]
+        assert flash_call[0][0] == "flash_oci_image"
+        assert flash_call[0][1] == "oci://quay.io/org/image:tag"
+
+
+def test_flash_oci_auto_error_cases(ridesx_client):
+    """Test flash_oci_auto error handling"""
+    # URL without oci:// scheme
+    with pytest.raises(ValueError, match="Only oci:// URLs are supported"):
+        ridesx_client.flash_oci_auto("docker://image:tag")
+
+    # Invalid URL format
+    with pytest.raises(ValueError, match="Invalid OCI URL format"):
+        ridesx_client.flash_oci_auto("invalid-url")
+
+    # No device found
+    with patch.object(ridesx_client, "call") as mock_call:
+        mock_call.return_value = {"status": "no_device_found", "device_id": None}
+
+        with pytest.raises(RuntimeError, match="No fastboot devices found"):
+            ridesx_client.flash_oci_auto("oci://image:tag")

python/packages/jumpstarter-driver-ridesx/jumpstarter_driver_ridesx/driver_test.py

@@ -452,3 +452,90 @@ def test_power_rescue(ridesx_power_driver):
         with pytest.raises(NotImplementedError, match="Rescue mode not available"):
             client.call("rescue")
 
+
+# Flash OCI Image Tests
+# Note: FLS download utilities are tested in jumpstarter.common.fls_test
+
+
+def test_flash_oci_image_success(temp_storage_dir, ridesx_driver):
+    with serve(ridesx_driver) as client:
+        with patch("jumpstarter_driver_ridesx.driver.get_fls_binary", return_value="/usr/local/bin/fls"):
+            with patch("subprocess.run") as mock_subprocess:
+                mock_result = MagicMock()
+                mock_result.stdout = "Flashing complete"
+                mock_result.stderr = ""
+                mock_result.returncode = 0
+                mock_subprocess.return_value = mock_result
+
+                result = client.call("flash_oci_image", "oci://quay.io/image:tag", None, None, None)
+
+                assert result["status"] == "success"
+                mock_subprocess.assert_called_once()
+                call_args = mock_subprocess.call_args[0][0]
+                assert call_args[0] == "/usr/local/bin/fls"
+                assert call_args[1] == "fastboot"
+                assert call_args[2] == "oci://quay.io/image:tag"
+
+
+def test_flash_oci_image_with_partitions(temp_storage_dir, ridesx_driver):
+    with serve(ridesx_driver) as client:
+        with patch("jumpstarter_driver_ridesx.driver.get_fls_binary", return_value="fls"):
+            with patch("subprocess.run") as mock_subprocess:
+                mock_result = MagicMock()
+                mock_result.stdout = "Flashing complete"
+                mock_result.stderr = ""
+                mock_result.returncode = 0
+                mock_subprocess.return_value = mock_result
+
+                partitions = {"boot_a": "boot.img", "system_a": "rootfs.simg"}
+                result = client.call("flash_oci_image", "oci://image:tag", partitions, None, None)
+
+                assert result["status"] == "success"
+                call_args = mock_subprocess.call_args[0][0]
+                # Check that -t flags are present for partitions
+                assert "-t" in call_args
+                assert "boot_a:boot.img" in call_args
+                assert "system_a:rootfs.simg" in call_args
+
+
+def test_flash_oci_image_error_cases(temp_storage_dir, ridesx_driver):
+    """Test flash_oci_image error handling for various failure modes"""
+    from jumpstarter.client.core import DriverError
+
+    with serve(ridesx_driver) as client:
+        # Reject non-oci:// schemes
+        with pytest.raises(DriverError, match="OCI URL must start with oci://"):
+            client.call("flash_oci_image", "docker://image:tag", None, None, None)
+
+        with patch("jumpstarter_driver_ridesx.driver.get_fls_binary", return_value="fls"):
+            with patch("subprocess.run") as mock_subprocess:
+                # CalledProcessError
+                error = subprocess.CalledProcessError(1, "fls")
+                error.stdout = ""
+                error.stderr = "Flash failed"
+                mock_subprocess.side_effect = error
+
+                with pytest.raises(DriverError, match="FLS fastboot auto-detection failed"):
+                    client.call("flash_oci_image", "oci://image:tag", None, None, None)
+
+                # TimeoutExpired
+                mock_subprocess.side_effect = subprocess.TimeoutExpired("fls", 1800)
+
+                with pytest.raises(DriverError, match="FLS fastboot auto-detection timeout"):
+                    client.call("flash_oci_image", "oci://image:tag", None, None, None)
+
+                # FileNotFoundError
+                mock_subprocess.side_effect = FileNotFoundError("fls not found")
+
+                with pytest.raises(DriverError, match="FLS command not found"):
+                    client.call("flash_oci_image", "oci://image:tag", None, None, None)
+
+
+def test_flash_oci_image_requires_oci_scheme(temp_storage_dir, ridesx_driver):
+    """Test that only oci:// URLs are accepted"""
+    from jumpstarter.client.core import DriverError
+
+    with serve(ridesx_driver) as client:
+        # Bare registry URL should be rejected
+        with pytest.raises(DriverError, match="OCI URL must start with oci://"):
+            client.call("flash_oci_image", "quay.io/org/image:v1", None, None, None)