Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -97,9 +97,18 @@ def flash_images(self, partitions: Dict[str, str], operators: Optional[Dict[str,

def _is_oci_path(self, path: str) -> bool:
"""Return True if path looks like an OCI image reference."""
return path.startswith("oci://") or (
":" in path and "/" in path and not path.startswith("/") and not path.startswith(("http://", "https://"))
)
if path.startswith("oci://"):
return True
if ":" not in path or "/" not in path:
return False
if path.startswith(("/", "http://", "https://", "./", "../")):
return False
# Split on first colon: OCI tags are short labels (e.g. "latest", "v1"),
# while partition:path specs have paths after the colon.
_, after_colon = path.split(":", 1)
if after_colon.startswith(("/", "./", "../", "~")):
return False
return True

def _validate_partition_mappings(self, partitions: Dict[str, str] | None) -> None:
"""Validate partition mappings; raise ValueError if any path is empty."""
Expand Down Expand Up @@ -202,6 +211,14 @@ def flash(
else:
# Traditional single file mode
if target is None:
# Detect partition:path patterns and give a specific hint
if isinstance(path, str) and self._looks_like_partition_spec(path):
partition, filepath = path.split(":", 1)
raise ValueError(
f"'{path}' looks like a partition:path mapping.\n"
f"Use the -t flag: j storage flash -t {path}\n"
f"Or use the API: client.flash('{filepath}', target='{partition}')"
)
raise ValueError(
"This driver requires a target partition for non-OCI paths.\n"
"Usage: client.flash('/path/to/file.img', target='boot_a')\n"
Expand Down Expand Up @@ -387,6 +404,13 @@ def _parse_and_validate_targets(self, target_specs: tuple[str, ...]):

return mapping, single_target

def _looks_like_partition_spec(self, value: str) -> bool:
"""Check if a string looks like a partition:filepath spec rather than an OCI URL."""
if ":" not in value:
return False
_, after_colon = value.split(":", 1)
return after_colon.startswith(("/", "./", "../", "~"))

def _execute_flash_command(self, path, target_specs, power_off: bool = True):
"""Execute flash command logic with proper argument handling."""
# Parse target specifications
Expand All @@ -395,6 +419,15 @@ def _execute_flash_command(self, path, target_specs, power_off: bool = True):

if mapping:
if path:
# In multi-target mode, positional path must be OCI.
# Any non-OCI value likely means a missing -t mapping.
if not self._is_oci_path(path):
raise click.ClickException(
f"'{path}' is not an OCI reference and is missing the -t flag.\n"
f"Each image must be passed as partition:path with its own -t.\n\n"
f"Example:\n"
f" j storage flash -t {path} -t {next(iter(target_specs))}"
)
# Multi-partition mode with path: extract specific files from OCI image
self.flash_with_targets(path, mapping, power_off=power_off)
else:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -89,3 +89,98 @@ def test_flash_oci_auto_error_cases(ridesx_client):

with pytest.raises(click.ClickException, match="No fastboot devices found"):
ridesx_client.flash_oci_auto("oci://image:tag")


# _is_oci_path Tests


def test_is_oci_path_rejects_partition_specs(ridesx_client):
"""Partition:path specs should not be detected as OCI references"""
assert not ridesx_client._is_oci_path("boot_a:./boot_a.simg")
assert not ridesx_client._is_oci_path("boot_a:/path/to/boot_a.simg")
assert not ridesx_client._is_oci_path("system_a:../images/system.img")
assert not ridesx_client._is_oci_path("boot_a:~/images/boot.img")


def test_is_oci_path_accepts_oci_references(ridesx_client):
"""Valid OCI references should be detected"""
assert ridesx_client._is_oci_path("oci://quay.io/org/image:tag")
assert ridesx_client._is_oci_path("quay.io/org/image:tag")
assert ridesx_client._is_oci_path("registry.com/repo:latest")


def test_is_oci_path_rejects_non_oci(ridesx_client):
"""Non-OCI paths should not be detected"""
assert not ridesx_client._is_oci_path("/absolute/path/to/file.img")
assert not ridesx_client._is_oci_path("./relative/path")
assert not ridesx_client._is_oci_path("http://example.com/file.img")
assert not ridesx_client._is_oci_path("https://example.com/file.img")
assert not ridesx_client._is_oci_path("just-a-filename")


# _execute_flash_command Tests


def test_execute_flash_command_detects_missing_t_flag(ridesx_client):
"""When a partition:path is passed as positional arg with other -t specs, suggest -t"""
with pytest.raises(click.ClickException, match="missing the -t flag"):
ridesx_client._execute_flash_command(
"boot_a:/path/to/boot.img",
("system_a:/path/to/system.img",),
)


def test_execute_flash_command_detects_missing_t_flag_relative(ridesx_client):
"""Relative partition:path should also be caught"""
with pytest.raises(click.ClickException, match="missing the -t flag"):
ridesx_client._execute_flash_command(
"boot_a:./boot_a.simg",
("system_a:./system_a.simg",),
)


def test_execute_flash_command_rejects_non_oci_positional_with_targets(ridesx_client):
"""Non-OCI positional paths should be rejected in multi-target mode"""
# partition:filename without path prefix (previously slipped through)
with pytest.raises(click.ClickException, match="not an OCI reference"):
ridesx_client._execute_flash_command(
"boot_a:boot.img",
("system_a:/path/to/system.img",),
)

# Local file path without colon
with pytest.raises(click.ClickException, match="not an OCI reference"):
ridesx_client._execute_flash_command(
"./boot_a.simg",
("system_a:/path/to/system.img",),
)

# Plain filename
with pytest.raises(click.ClickException, match="not an OCI reference"):
ridesx_client._execute_flash_command(
"boot.img",
("system_a:/path/to/system.img",),
)


def test_execute_flash_command_allows_oci_positional_with_targets(ridesx_client):
"""OCI positional paths should pass the guard in multi-target mode"""
with patch.object(ridesx_client, "flash_with_targets") as mock_flash:
ridesx_client._execute_flash_command(
"oci://quay.io/org/image:tag",
("boot_a:boot.img",),
)
mock_flash.assert_called_once_with(
"oci://quay.io/org/image:tag",
{"boot_a": "boot.img"},
power_off=True,
)


# flash() partition:path hint Tests


def test_flash_hints_partition_spec_without_target(ridesx_client):
"""Passing partition:/path directly to flash() should give a helpful error"""
with pytest.raises(ValueError, match="looks like a partition:path mapping"):
ridesx_client.flash("boot_a:/path/to/boot.img")
Original file line number Diff line number Diff line change
Expand Up @@ -227,6 +227,24 @@ def flash_with_fastboot(self, device_id: str, partitions: Dict[str, str]):
self.logger.warning(f"stdout: {e.stdout}")
self.logger.warning(f"stderr: {e.stderr}")

@staticmethod
def _validate_oci_url(oci_url: str):
"""Validate that the URL is a proper OCI reference."""
if oci_url.startswith("oci://"):
return
hint = ""
if ":" in oci_url:
_, after = oci_url.split(":", 1)
if after.startswith(("/", "./", "../", "~")):
hint = (
f"\n\nIt looks like '{oci_url}' is a partition:path mapping, not an OCI reference.\n"
f"For local files, use: j storage flash -t {oci_url}"
)
raise ValueError(
f"OCI URL must start with oci://, got: {oci_url}"
f"{hint}"
)

def _build_fls_command(self, oci_url, partitions):
"""Build FLS fastboot command and environment."""
fls_binary = get_fls_binary(
Expand Down Expand Up @@ -264,8 +282,7 @@ def flash_oci_image(
oci_username: Registry username for OCI authentication
oci_password: Registry password for OCI authentication
"""
if not oci_url.startswith("oci://"):
raise ValueError(f"OCI URL must start with oci://, got: {oci_url}")
self._validate_oci_url(oci_url)

if bool(oci_username) != bool(oci_password):
raise ValueError("OCI authentication requires both --username and --password")
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -600,3 +600,58 @@ def test_flash_oci_image_requires_oci_scheme(temp_storage_dir, ridesx_driver):
# 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)


# OCI URL Validation Tests (direct unit tests for _validate_oci_url)


def test_validate_oci_url_accepts_valid():
"""Valid oci:// URLs should pass without error"""
RideSXDriver._validate_oci_url("oci://quay.io/org/image:tag")
RideSXDriver._validate_oci_url("oci://registry.example.com/repo:latest")


def test_validate_oci_url_rejects_non_oci():
"""Non-oci:// URLs should raise ValueError"""
with pytest.raises(ValueError, match="OCI URL must start with oci://"):
RideSXDriver._validate_oci_url("docker://image:tag")


def test_validate_oci_url_rejects_bare_registry():
"""Bare registry URLs (with tag colon) should be rejected without hint"""
with pytest.raises(ValueError, match="OCI URL must start with oci://") as exc_info:
RideSXDriver._validate_oci_url("quay.io/org/image:v1")
# Tag suffix is not a path, so no hint should be shown
assert "partition:path" not in str(exc_info.value)


def test_validate_oci_url_hint_absolute_path():
"""partition:/absolute/path should produce a helpful hint"""
with pytest.raises(ValueError, match="partition:path mapping") as exc_info:
RideSXDriver._validate_oci_url("boot:/path/to/boot.img")
assert "j storage flash -t boot:/path/to/boot.img" in str(exc_info.value)


def test_validate_oci_url_hint_relative_path():
"""partition:./relative/path should produce a helpful hint"""
with pytest.raises(ValueError, match="partition:path mapping"):
RideSXDriver._validate_oci_url("system:./images/system.img")


def test_validate_oci_url_hint_parent_path():
"""partition:../parent/path should produce a helpful hint"""
with pytest.raises(ValueError, match="partition:path mapping"):
RideSXDriver._validate_oci_url("boot:../other/boot.img")


def test_validate_oci_url_hint_home_path():
"""partition:~/home/path should produce a helpful hint"""
with pytest.raises(ValueError, match="partition:path mapping"):
RideSXDriver._validate_oci_url("boot:~/images/boot.img")


def test_validate_oci_url_no_colon():
"""URL with no colon should be rejected without hint"""
with pytest.raises(ValueError, match="OCI URL must start with oci://") as exc_info:
RideSXDriver._validate_oci_url("just-a-string")
assert "partition:path" not in str(exc_info.value)
Loading