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,216 @@
+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
+
+
+# OCI Path Detection Tests
+
+
+@pytest.mark.parametrize(
+    "path,expected",
+    [
+        # OCI paths (should return True)
+        ("oci://quay.io/myimage:latest", True),
+        ("quay.io/myorg/myimage:latest", True),
+        ("ghcr.io/owner/repo:tag", True),
+        ("registry.redhat.io/rhel:9", True),
+        # Non-OCI paths (should return False)
+        ("docker://registry.example.com/image:v1", False),  # only oci:// scheme accepted
+        ("/path/to/file.img", False),
+        ("relative/path/file.img", False),
+        ("boot.img", False),
+    ],
+)
+def test_is_oci_path(ridesx_client, path, expected):
+    assert ridesx_client._is_oci_path(path) is expected
+
+
+# OCI URL from argv Tests
+
+
+@pytest.mark.parametrize(
+    "argv,expected",
+    [
+        (["cmd", "oci://quay.io/image:tag"], "oci://quay.io/image:tag"),
+        (["cmd", "quay.io/org/image:v1"], "quay.io/org/image:v1"),
+        (["cmd", "ghcr.io/owner/repo:latest"], "ghcr.io/owner/repo:latest"),
+        (["cmd", "docker.io/library/alpine:3.18"], "docker.io/library/alpine:3.18"),
+        (["cmd", "/path/to/file.img", "-t", "boot:file"], None),
+        (["cmd", "-t", "boot:file", "oci://first:tag", "oci://second:tag"], "oci://first:tag"),
+    ],
+)
+def test_oci_url_from_argv(ridesx_client, argv, expected):
+    with patch("sys.argv", argv):
+        result = ridesx_client._oci_url_from_argv()
+        assert result == expected
+
+
+# Target Mappings from argv Tests
+
+
+@pytest.mark.parametrize(
+    "argv,expected",
+    [
+        (["cmd", "-t", "boot_a:boot.img"], {"boot_a": "boot.img"}),
+        (["cmd", "--target", "system_a:rootfs.simg"], {"system_a": "rootfs.simg"}),
+        (["cmd", "--target=boot_a:boot.img"], {"boot_a": "boot.img"}),
+        (
+            ["cmd", "-t", "boot_a:boot.img", "-t", "system_a:rootfs.simg"],
+            {"boot_a": "boot.img", "system_a": "rootfs.simg"},
+        ),
+        (["cmd", "oci://image:tag"], None),
+        (
+            ["cmd", "-t", "boot_a:boot.img", "oci://image:tag", "--other", "value"],
+            {"boot_a": "boot.img"},
+        ),
+    ],
+)
+def test_target_mappings_from_argv(ridesx_client, argv, expected):
+    with patch("sys.argv", argv):
+        result = ridesx_client._target_mappings_from_argv()
+        assert result == expected
+
+
+# Resolve Flash Input Tests
+
+
+def test_resolve_flash_input_dict_path(ridesx_client):
+    """Test dict path input returns partitions directly"""
+    partitions, operators, oci_url = ridesx_client._resolve_flash_input(
+        {"boot": "/path/boot.img"}, None, None, None
+    )
+    assert partitions == {"boot": "/path/boot.img"}
+    assert operators is None
+    assert oci_url is None
+
+    # With OCI from argv
+    partitions, operators, oci_url = ridesx_client._resolve_flash_input(
+        {"boot": "boot.img"}, None, None, "oci://image:tag"
+    )
+    assert partitions == {"boot": "boot.img"}
+    assert oci_url == "oci://image:tag"
+
+
+def test_resolve_flash_input_oci_path(ridesx_client):
+    """Test OCI path input with and without target"""
+    # Without target - auto-detect mode
+    partitions, operators, oci_url = ridesx_client._resolve_flash_input(
+        "oci://quay.io/image:tag", None, None, None
+    )
+    assert partitions is None
+    assert operators is None
+    assert oci_url == "oci://quay.io/image:tag"
+
+    # With target - explicit mapping
+    partitions, operators, oci_url = ridesx_client._resolve_flash_input(
+        "oci://quay.io/image:tag", "boot_a:boot.img", None, None
+    )
+    assert partitions == {"boot_a": "boot.img"}
+    assert oci_url == "oci://quay.io/image:tag"
+
+
+def test_resolve_flash_input_oci_path_invalid_target(ridesx_client):
+    """Test OCI path with invalid target format raises error"""
+    with pytest.raises(ValueError, match="Target must be in format"):
+        ridesx_client._resolve_flash_input("oci://quay.io/image:tag", "boot_a", None, None)
+
+
+def test_resolve_flash_input_local_path(ridesx_client):
+    """Test local path input requires target"""
+    # Without target - should raise
+    with pytest.raises(ValueError, match="This driver requires a target partition"):
+        ridesx_client._resolve_flash_input("/path/to/boot.img", None, None, None)
+
+    # With target
+    partitions, operators, oci_url = ridesx_client._resolve_flash_input(
+        "/path/to/boot.img", "boot_a", None, None
+    )
+    assert partitions == {"boot_a": "/path/to/boot.img"}
+    assert oci_url is None
+
+    # With target and OCI from argv
+    partitions, operators, oci_url = ridesx_client._resolve_flash_input(
+        "/path/to/boot.img", "boot_a", None, "oci://image:tag"
+    )
+    assert partitions == {"boot_a": "/path/to/boot.img"}
+    assert oci_url == "oci://image:tag"
+
+
+# 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_normalizes_bare_registry(ridesx_client):
+    """Test that bare registry URLs are normalized with oci:// prefix"""
+    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("quay.io/org/image:tag")
+
+        assert result == {"status": "success"}
+        flash_call = mock_call.call_args_list[1]
+        assert flash_call[0][1].startswith("oci://")
+
+
+def test_flash_oci_auto_error_cases(ridesx_client):
+    """Test flash_oci_auto error handling"""
+    # Invalid URL
+    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,100 @@ 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] == "quay.io/image:tag"  # oci:// prefix stripped
+
+
+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="Only oci:// URLs are supported"):
+            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_normalizes_url(temp_storage_dir, ridesx_driver):
+    """Test that OCI URLs are normalized correctly"""
+    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 = "Done"
+                mock_result.stderr = ""
+                mock_result.returncode = 0
+                mock_subprocess.return_value = mock_result
+
+                # Test that oci:// prefix is stripped
+                client.call("flash_oci_image", "oci://quay.io/org/image:v1", None, None, None)
+
+                call_args = mock_subprocess.call_args[0][0]
+                # The URL passed to fls should not have oci:// prefix
+                assert "quay.io/org/image:v1" in call_args
+                assert "oci://quay.io/org/image:v1" not in call_args