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

@@ -59,6 +59,7 @@ 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 = [
+            None,  # boot_to_fastboot call
             {"status": "device_found", "device_id": "ABC123"},
             {"status": "success"},
         ]
@@ -67,7 +68,7 @@ def test_flash_oci_auto_success(ridesx_client):
 
         assert result == {"status": "success"}
         # Verify flash_oci_image was called with the OCI URL
-        flash_call = mock_call.call_args_list[1]
+        flash_call = mock_call.call_args_list[2]
         assert flash_call[0][0] == "flash_oci_image"
         assert flash_call[0][1] == "oci://quay.io/org/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)
+
+                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)
+
+                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)
+
+        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)
+
+                # 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)
+
+                # 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)
+
+
+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)

python/packages/jumpstarter/jumpstarter/common/fls_test.py

@@ -66,19 +66,31 @@ def test_get_fls_binary_falls_back_to_path():
 
 
 def test_download_fls_success():
-    with patch("urllib.request.urlretrieve") as mock_retrieve:
+    from unittest.mock import MagicMock, mock_open
+
+    mock_response = MagicMock()
+    mock_response.read.side_effect = [b"binary data", b""]  # Simulate chunked read
+    mock_response.__enter__ = MagicMock(return_value=mock_response)
+    mock_response.__exit__ = MagicMock(return_value=None)
+
+    with patch("urllib.request.urlopen", return_value=mock_response) as mock_urlopen:
         with patch("tempfile.mkstemp", return_value=(99, "/tmp/fls-test")):
             with patch("os.close") as mock_close:
-                with patch("pathlib.Path.chmod"):
-                    result = download_fls("https://example.com/fls")
+                with patch("pathlib.Path.chmod") as mock_chmod:
+                    with patch("os.replace") as mock_replace:
+                        with patch("builtins.open", mock_open()):
+                            with patch("os.fsync"):
+                                result = download_fls("https://example.com/fls")
 
-                    mock_close.assert_called_once_with(99)
-                    mock_retrieve.assert_called_once_with("https://example.com/fls", "/tmp/fls-test")
-                    assert result == "/tmp/fls-test"
+                                mock_close.assert_called_once_with(99)
+                                mock_urlopen.assert_called_once_with("https://example.com/fls", timeout=30.0)
+                                mock_chmod.assert_called_once_with(0o755)
+                                mock_replace.assert_called_once_with("/tmp/fls-test.part", "/tmp/fls-test")
+                                assert result == "/tmp/fls-test"
 
 
 def test_download_fls_failure():
-    with patch("urllib.request.urlretrieve", side_effect=Exception("Network error")):
+    with patch("urllib.request.urlopen", side_effect=Exception("Network error")):
         with patch("tempfile.mkstemp", return_value=(99, "/tmp/fls-test")):
             with patch("os.close"):
                 with patch("pathlib.Path.unlink"):