diff --git a/packages/jumpstarter-driver-ridesx/jumpstarter_driver_ridesx/client.py b/packages/jumpstarter-driver-ridesx/jumpstarter_driver_ridesx/client.py index 82ffb25d9..c40a996d8 100644 --- a/packages/jumpstarter-driver-ridesx/jumpstarter_driver_ridesx/client.py +++ b/packages/jumpstarter-driver-ridesx/jumpstarter_driver_ridesx/client.py @@ -121,14 +121,19 @@ def cli(self): generic_cli = FlasherClient.cli(self) @driver_click_group(self) - def storage(): + def base(): """RideSX storage operations""" pass for name, cmd in generic_cli.commands.items(): - storage.add_command(cmd, name=name) + base.add_command(cmd, name=name) - return storage + @base.command() + def boot_to_fastboot(): + """Boot to fastboot""" + self.boot_to_fastboot() + + return base @dataclass(kw_only=True) diff --git a/packages/jumpstarter-driver-ridesx/jumpstarter_driver_ridesx/driver.py b/packages/jumpstarter-driver-ridesx/jumpstarter_driver_ridesx/driver.py index 41781dd8d..7898f4aae 100644 --- a/packages/jumpstarter-driver-ridesx/jumpstarter_driver_ridesx/driver.py +++ b/packages/jumpstarter-driver-ridesx/jumpstarter_driver_ridesx/driver.py @@ -14,7 +14,9 @@ @dataclass(kw_only=True) class RideSXDriver(Driver): """RideSX Driver""" - + decompression_timeout: int = field(default=15 * 60) # 15 minutes + flash_timeout: int = field(default=30 * 60) # 30 minutes + continue_timeout: int = field(default=20 * 60) # 20 minutes storage_dir: str = field(default="/var/lib/jumpstarter/ridesx") def __post_init__(self): @@ -74,7 +76,7 @@ def _decompress_file(self, compressed_file: Path) -> Path: stderr=subprocess.PIPE, text=False, check=True, - timeout=600, + timeout=self.decompression_timeout, ) if result.stderr: @@ -168,7 +170,7 @@ def flash_with_fastboot(self, device_id: str, partitions: Dict[str, str]): self.logger.debug(f"Running command: {' '.join(cmd)}") try: - result = subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=800) + result = subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=self.flash_timeout) self.logger.info(f"Successfully flashed {partition_name}") self.logger.debug(f"Flash stdout: {result.stdout}") if result.stderr: @@ -186,7 +188,7 @@ def flash_with_fastboot(self, device_id: str, partitions: Dict[str, str]): cmd = ["fastboot", "-s", device_id, "continue"] self.logger.debug(f"Running command: {' '.join(cmd)}") try: - result = subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=300) + result = subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=self.continue_timeout) self.logger.debug(f"Fastboot continue stdout: {result.stdout}") self.logger.debug(f"Fastboot continue stderr: {result.stderr}") self.logger.info("Fastboot continue completed successfully") diff --git a/packages/jumpstarter-driver-ridesx/jumpstarter_driver_ridesx/driver_test.py b/packages/jumpstarter-driver-ridesx/jumpstarter_driver_ridesx/driver_test.py new file mode 100644 index 000000000..acfeb8f8b --- /dev/null +++ b/packages/jumpstarter-driver-ridesx/jumpstarter_driver_ridesx/driver_test.py @@ -0,0 +1,454 @@ +import subprocess +import tempfile +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from jumpstarter_driver_pyserial.driver import PySerial + +from .driver import RideSXDriver, RideSXPowerDriver +from jumpstarter.common.exceptions import ConfigurationError +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(scope="session") +def ridesx_power_driver(): + yield RideSXPowerDriver( + children={ + "serial": PySerial(url="loop://"), + }, + ) + + +# Configuration Tests + + +def test_missing_serial(temp_storage_dir): + with pytest.raises(ConfigurationError): + RideSXDriver(storage_dir=temp_storage_dir, children={}) + + +# Fastboot Detection Tests + + +def test_detect_fastboot_device_found(ridesx_driver): + with serve(ridesx_driver) as client: + with patch("subprocess.run") as mock_subprocess: + mock_result = MagicMock() + mock_result.stdout = "ABC123456789 fastboot\n" + mock_result.returncode = 0 + mock_subprocess.return_value = mock_result + + result = client.call("detect_fastboot_device", 1, 0.1) + + assert result["status"] == "device_found" + assert result["device_id"] == "ABC123456789" + mock_subprocess.assert_called_once() + + +def test_detect_fastboot_device_not_found(ridesx_driver): + with serve(ridesx_driver) as client: + with patch("subprocess.run") as mock_subprocess: + mock_result = MagicMock() + mock_result.stdout = "" + mock_result.returncode = 0 + mock_subprocess.return_value = mock_result + + result = client.call("detect_fastboot_device", 2, 0.01) + + assert result["status"] == "no_device_found" + assert result["device_id"] is None + # Driver makes max_attempts calls plus one final attempt + assert mock_subprocess.call_count >= 2 + + +def test_detect_fastboot_device_timeout(ridesx_driver): + with serve(ridesx_driver) as client: + with patch("subprocess.run") as mock_subprocess: + mock_subprocess.side_effect = subprocess.TimeoutExpired("fastboot", 10) + + result = client.call("detect_fastboot_device", 2, 0.01) + + assert result["status"] == "no_device_found" + assert result["device_id"] is None + + +def test_detect_fastboot_device_not_found_error(ridesx_driver): + with serve(ridesx_driver) as client: + with patch("subprocess.run", side_effect=FileNotFoundError("fastboot not found")): + # When called through client, RuntimeError becomes DriverError + from jumpstarter.client.core import DriverError + + with pytest.raises(DriverError, match="fastboot command not found"): + client.call("detect_fastboot_device", 1, 0.1) + + +def test_detect_fastboot_device_retry_logic(ridesx_driver): + with serve(ridesx_driver) as client: + with patch("subprocess.run") as mock_subprocess: + # First two attempts return empty, third returns device + mock_results = [ + MagicMock(stdout="", returncode=0), + MagicMock(stdout="", returncode=0), + MagicMock(stdout="ABC123456789 fastboot\n", returncode=0), + ] + mock_subprocess.side_effect = mock_results + + result = client.call("detect_fastboot_device", 3, 0.01) + + assert result["status"] == "device_found" + assert result["device_id"] == "ABC123456789" + assert mock_subprocess.call_count == 3 + + +# File Decompression Tests + + +def test_needs_decompression_gz(ridesx_driver): + assert ridesx_driver._needs_decompression("file.gz") is True + # Note: endswith is case-sensitive, so .GZ won't match + # This is expected behavior - filenames should use lowercase extensions + + +def test_needs_decompression_xz(ridesx_driver): + assert ridesx_driver._needs_decompression("file.xz") is True + + +def test_needs_decompression_gzip(ridesx_driver): + assert ridesx_driver._needs_decompression("file.gzip") is True + + +def test_needs_decompression_no_compression(ridesx_driver): + assert ridesx_driver._needs_decompression("file.img") is False + assert ridesx_driver._needs_decompression("file.bin") is False + assert ridesx_driver._needs_decompression("file") is False + + +def test_get_decompression_command(ridesx_driver): + assert ridesx_driver._get_decompression_command("file.gz") == "zcat" + assert ridesx_driver._get_decompression_command("file.gzip") == "zcat" + assert ridesx_driver._get_decompression_command("file.xz") == "xzcat" + assert ridesx_driver._get_decompression_command("file.img") == "cat" + + +def test_decompress_file_gz(temp_storage_dir, ridesx_driver): + compressed_file = Path(temp_storage_dir) / "test.gz" + compressed_file.write_bytes(b"compressed data") + + with patch("subprocess.run") as mock_subprocess: + mock_result = MagicMock() + mock_result.stderr = b"" + mock_result.returncode = 0 + mock_subprocess.return_value = mock_result + + # Mock the file existence and size check + with patch.object(Path, "exists") as mock_exists, patch.object(Path, "stat") as mock_stat: + mock_exists.return_value = True + mock_stat.return_value = MagicMock(st_size=100) # Non-zero size + + decompressed = ridesx_driver._decompress_file(compressed_file) + + assert decompressed.name == "test" + mock_subprocess.assert_called_once() + call_args = mock_subprocess.call_args + assert call_args[0][0] == ["zcat", str(compressed_file)] + + +def test_decompress_file_xz(temp_storage_dir, ridesx_driver): + compressed_file = Path(temp_storage_dir) / "test.xz" + compressed_file.write_bytes(b"compressed data") + + with patch("subprocess.run") as mock_subprocess: + mock_result = MagicMock() + mock_result.stderr = b"" + mock_result.returncode = 0 + mock_subprocess.return_value = mock_result + + # Mock the file existence and size check + with patch.object(Path, "exists") as mock_exists, patch.object(Path, "stat") as mock_stat: + mock_exists.return_value = True + mock_stat.return_value = MagicMock(st_size=100) # Non-zero size + + decompressed = ridesx_driver._decompress_file(compressed_file) + + assert decompressed.name == "test" + mock_subprocess.assert_called_once() + call_args = mock_subprocess.call_args + assert call_args[0][0] == ["xzcat", str(compressed_file)] + + +def test_decompress_file_failure(temp_storage_dir, ridesx_driver): + compressed_file = Path(temp_storage_dir) / "test.gz" + compressed_file.write_bytes(b"compressed data") + + with patch("subprocess.run") as mock_subprocess: + mock_subprocess.side_effect = subprocess.CalledProcessError(1, "zcat", stderr=b"error") + + with pytest.raises(RuntimeError, match="failed to decompress"): + ridesx_driver._decompress_file(compressed_file) + + +def test_decompress_file_timeout(temp_storage_dir, ridesx_driver): + compressed_file = Path(temp_storage_dir) / "test.gz" + compressed_file.write_bytes(b"compressed data") + + with patch("subprocess.run") as mock_subprocess: + mock_subprocess.side_effect = subprocess.TimeoutExpired("zcat", 10) + + with pytest.raises(RuntimeError, match="decompression timeout"): + ridesx_driver._decompress_file(compressed_file) + + +# Fastboot Flashing Tests + + +def test_flash_with_fastboot_single_partition(temp_storage_dir, ridesx_driver): + # Create test image file + image_file = Path(temp_storage_dir) / "boot.img" + image_file.write_bytes(b"boot image data") + + with serve(ridesx_driver) as client: + with patch("subprocess.run") as mock_subprocess: + # Mock flash command + flash_result = MagicMock() + flash_result.stdout = "Flashing boot..." + flash_result.stderr = "" + flash_result.returncode = 0 + + # Mock continue command + continue_result = MagicMock() + continue_result.stdout = "Continuing..." + continue_result.stderr = "" + continue_result.returncode = 0 + + mock_subprocess.side_effect = [flash_result, continue_result] + + client.call("flash_with_fastboot", "ABC123", {"boot": "boot.img"}) + + assert mock_subprocess.call_count == 2 + # Check flash command + flash_call = mock_subprocess.call_args_list[0] + assert flash_call[0][0] == ["fastboot", "-s", "ABC123", "flash", "boot", str(image_file)] + # Check continue command + continue_call = mock_subprocess.call_args_list[1] + assert continue_call[0][0] == ["fastboot", "-s", "ABC123", "continue"] + + +def test_flash_with_fastboot_multiple_partitions(temp_storage_dir, ridesx_driver): + # Create test image files + boot_file = Path(temp_storage_dir) / "boot.img" + boot_file.write_bytes(b"boot image data") + system_file = Path(temp_storage_dir) / "system.img" + system_file.write_bytes(b"system image data") + + with serve(ridesx_driver) as client: + with patch("subprocess.run") as mock_subprocess: + flash_result = MagicMock() + flash_result.stdout = "Flashing..." + flash_result.stderr = "" + flash_result.returncode = 0 + + continue_result = MagicMock() + continue_result.stdout = "Continuing..." + continue_result.stderr = "" + continue_result.returncode = 0 + + mock_subprocess.side_effect = [flash_result, flash_result, continue_result] + + client.call("flash_with_fastboot", "ABC123", {"boot": "boot.img", "system": "system.img"}) + + assert mock_subprocess.call_count == 3 + # Verify both partitions were flashed + flash_calls = [call[0][0] for call in mock_subprocess.call_args_list[:2]] + assert ["fastboot", "-s", "ABC123", "flash", "boot", str(boot_file)] in flash_calls + assert ["fastboot", "-s", "ABC123", "flash", "system", str(system_file)] in flash_calls + + +def test_flash_with_fastboot_compressed_file(temp_storage_dir, ridesx_driver): + # Create compressed file + compressed_file = Path(temp_storage_dir) / "boot.img.gz" + compressed_file.write_bytes(b"compressed data") + + # Create decompressed file that will be "created" by decompression + decompressed_file = Path(temp_storage_dir) / "boot.img" + decompressed_file.write_bytes(b"decompressed data") + + with serve(ridesx_driver) as client: + with patch.object(ridesx_driver, "_decompress_file", return_value=decompressed_file): + with patch("subprocess.run") as mock_subprocess: + flash_result = MagicMock() + flash_result.stdout = "Flashing..." + flash_result.stderr = "" + flash_result.returncode = 0 + + continue_result = MagicMock() + continue_result.stdout = "Continuing..." + continue_result.stderr = "" + continue_result.returncode = 0 + + mock_subprocess.side_effect = [flash_result, continue_result] + + client.call("flash_with_fastboot", "ABC123", {"boot": "boot.img.gz"}) + + # Verify decompression was called + ridesx_driver._decompress_file.assert_called_once_with(compressed_file) + # Verify flash used decompressed file + flash_call = mock_subprocess.call_args_list[0] + assert str(decompressed_file) in flash_call[0][0] + + +def test_flash_with_fastboot_file_not_found(temp_storage_dir, ridesx_driver): + with serve(ridesx_driver) as client: + from jumpstarter.client.core import DriverError + + # When called through client, FileNotFoundError becomes DriverError + with pytest.raises(DriverError, match="Image not found in storage"): + client.call("flash_with_fastboot", "ABC123", {"boot": "nonexistent.img"}) + + +def test_flash_with_fastboot_empty_partitions(ridesx_driver): + with serve(ridesx_driver) as client: + with pytest.raises(ValueError, match="At least one partition must be provided"): + client.call("flash_with_fastboot", "ABC123", {}) + + +def test_flash_with_fastboot_flash_failure(temp_storage_dir, ridesx_driver): + image_file = Path(temp_storage_dir) / "boot.img" + image_file.write_bytes(b"boot image data") + + with serve(ridesx_driver) as client: + from jumpstarter.client.core import DriverError + + with patch("subprocess.run") as mock_subprocess: + mock_subprocess.side_effect = subprocess.CalledProcessError(1, "fastboot", stderr=b"flash failed") + + # When called through client, RuntimeError becomes DriverError + with pytest.raises(DriverError, match="Failed to flash"): + client.call("flash_with_fastboot", "ABC123", {"boot": "boot.img"}) + + +def test_flash_with_fastboot_flash_timeout(temp_storage_dir, ridesx_driver): + image_file = Path(temp_storage_dir) / "boot.img" + image_file.write_bytes(b"boot image data") + + with serve(ridesx_driver) as client: + from jumpstarter.client.core import DriverError + + with patch("subprocess.run") as mock_subprocess: + mock_subprocess.side_effect = subprocess.TimeoutExpired("fastboot", 20 * 60) + + # When called through client, RuntimeError becomes DriverError + with pytest.raises(DriverError, match="timeout while flashing"): + client.call("flash_with_fastboot", "ABC123", {"boot": "boot.img"}) + + +def test_flash_with_fastboot_continue_success(temp_storage_dir, ridesx_driver): + image_file = Path(temp_storage_dir) / "boot.img" + image_file.write_bytes(b"boot image data") + + with serve(ridesx_driver) as client: + with patch("subprocess.run") as mock_subprocess: + flash_result = MagicMock() + flash_result.stdout = "Flashing..." + flash_result.stderr = "" + flash_result.returncode = 0 + + continue_result = MagicMock() + continue_result.stdout = "Continuing..." + continue_result.stderr = "" + continue_result.returncode = 0 + + mock_subprocess.side_effect = [flash_result, continue_result] + + client.call("flash_with_fastboot", "ABC123", {"boot": "boot.img"}) + + # Verify continue was called + continue_call = mock_subprocess.call_args_list[1] + assert continue_call[0][0] == ["fastboot", "-s", "ABC123", "continue"] + + +def test_flash_with_fastboot_continue_failure(temp_storage_dir, ridesx_driver): + image_file = Path(temp_storage_dir) / "boot.img" + image_file.write_bytes(b"boot image data") + + with serve(ridesx_driver) as client: + with patch("subprocess.run") as mock_subprocess: + flash_result = MagicMock() + flash_result.stdout = "Flashing..." + flash_result.stderr = "" + flash_result.returncode = 0 + + # First call succeeds (flash), second call fails (continue) + mock_subprocess.side_effect = [ + flash_result, + subprocess.CalledProcessError(1, "fastboot", stderr=b"continue failed"), + ] + + # Should not raise, just log warning + client.call("flash_with_fastboot", "ABC123", {"boot": "boot.img"}) + + # Verify both flash and continue were called + assert mock_subprocess.call_count == 2 + + +def test_power_missing_serial(): + with pytest.raises(ConfigurationError): + RideSXPowerDriver(children={}) + + +def test_power_on_exported(ridesx_power_driver): + """Test that power on method is properly exported""" + with serve(ridesx_power_driver): + # Verify the method exists and is exported + # Full execution requires proper serial responses + assert hasattr(ridesx_power_driver, "on") + import inspect + + assert inspect.iscoroutinefunction(ridesx_power_driver.on) + + +def test_power_off_exported(ridesx_power_driver): + """Test that power off method is properly exported""" + with serve(ridesx_power_driver): + # Verify the method exists and is exported + assert hasattr(ridesx_power_driver, "off") + import inspect + + assert inspect.iscoroutinefunction(ridesx_power_driver.off) + + +@pytest.mark.asyncio +async def test_power_cycle(ridesx_power_driver): + """Test power cycle calls off, waits, then on""" + with patch.object(ridesx_power_driver, "off", new_callable=AsyncMock) as mock_off: + with patch.object(ridesx_power_driver, "on", new_callable=AsyncMock) as mock_on: + with patch("asyncio.sleep", new_callable=AsyncMock) as mock_sleep: + await ridesx_power_driver.cycle(delay=0.1) + + mock_off.assert_called_once() + mock_on.assert_called_once() + mock_sleep.assert_called_once_with(0.1) + + +def test_power_rescue(ridesx_power_driver): + """Test that rescue raises NotImplementedError""" + with serve(ridesx_power_driver) as client: + with pytest.raises(NotImplementedError, match="Rescue mode not available"): + client.call("rescue") + diff --git a/packages/jumpstarter-driver-ridesx/pyproject.toml b/packages/jumpstarter-driver-ridesx/pyproject.toml index 576e10858..567d25c97 100644 --- a/packages/jumpstarter-driver-ridesx/pyproject.toml +++ b/packages/jumpstarter-driver-ridesx/pyproject.toml @@ -8,7 +8,9 @@ authors = [{ name = "Benny Zlotnik", email = "bzlotnik@redhat.com" }] requires-python = ">=3.11" dependencies = [ "jumpstarter", + "jumpstarter-driver-composite", "jumpstarter-driver-opendal", + "jumpstarter-driver-power", "jumpstarter-driver-pyserial", ] @@ -25,16 +27,19 @@ addopts = "--cov --cov-report=html --cov-report=xml" log_cli = true log_cli_level = "INFO" testpaths = ["jumpstarter_driver_ridesx"] +asyncio_mode = "auto" -[tool.uv.sources] -#asyncio_mode = "auto" [build-system] requires = ["hatchling", "hatch-vcs", "hatch-pin-jumpstarter"] build-backend = "hatchling.build" [dependency-groups] -dev = ["pytest-cov>=6.0.0", "pytest>=8.3.3"] +dev = [ + "pytest-cov>=6.0.0", + "pytest>=8.3.3", + "pytest-asyncio>=0.0.0", +] [tool.hatch.build.hooks.pin_jumpstarter] name = "pin_jumpstarter" diff --git a/uv.lock b/uv.lock index e6febc9e6..4e10b4097 100644 --- a/uv.lock +++ b/uv.lock @@ -1907,26 +1907,32 @@ name = "jumpstarter-driver-ridesx" source = { editable = "packages/jumpstarter-driver-ridesx" } dependencies = [ { name = "jumpstarter" }, + { name = "jumpstarter-driver-composite" }, { name = "jumpstarter-driver-opendal" }, + { name = "jumpstarter-driver-power" }, { name = "jumpstarter-driver-pyserial" }, ] [package.dev-dependencies] dev = [ { name = "pytest" }, + { name = "pytest-asyncio" }, { name = "pytest-cov" }, ] [package.metadata] requires-dist = [ { name = "jumpstarter", editable = "packages/jumpstarter" }, + { name = "jumpstarter-driver-composite", editable = "packages/jumpstarter-driver-composite" }, { name = "jumpstarter-driver-opendal", editable = "packages/jumpstarter-driver-opendal" }, + { name = "jumpstarter-driver-power", editable = "packages/jumpstarter-driver-power" }, { name = "jumpstarter-driver-pyserial", editable = "packages/jumpstarter-driver-pyserial" }, ] [package.metadata.requires-dev] dev = [ { name = "pytest", specifier = ">=8.3.3" }, + { name = "pytest-asyncio", specifier = ">=0.0.0" }, { name = "pytest-cov", specifier = ">=6.0.0" }, ]