Skip to content
Merged
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 @@ -95,12 +95,6 @@ def flash_images(self, partitions: Dict[str, str], operators: Optional[Dict[str,

return flash_result

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://"))
)

def _validate_partition_mappings(self, partitions: Dict[str, str] | None) -> None:
"""Validate partition mappings; raise ValueError if any path is empty."""
if partitions is None:
Expand Down Expand Up @@ -188,7 +182,7 @@ def flash(
operators_dict = operator if isinstance(operator, dict) else None
return self.flash_local(path, operators_dict, power_off=power_off)

elif isinstance(path, str) and (path.startswith("oci://") or self._is_oci_path(path)):
elif isinstance(path, str) and path.startswith("oci://"):
# OCI mode: auto-detect partitions or use target as partition->filename mapping
if target and ":" in target:
# Target is "partition:filename" format for OCI explicit mapping
Expand All @@ -202,7 +196,21 @@ def flash(
else:
# Traditional single file mode
if target is None:
raise ValueError(
if isinstance(path, str) and ":" in path:
before_colon, after_colon = path.split(":", 1)
if "/" in before_colon:
# registry/path:tag — likely an OCI ref missing the oci:// prefix
raise click.ClickException(
f"OCI URLs must start with oci://, got: {path}\n"
f"Usage: j storage flash oci://{path}"
)
# partition:something — likely a partition:path mapping
raise click.ClickException(
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('{after_colon}', target='{before_colon}')"
)
raise click.ClickException(
"This driver requires a target partition for non-OCI paths.\n"
"Usage: client.flash('/path/to/file.img', target='boot_a')\n"
"For OCI: client.flash('oci://registry.com/image:tag')\n"
Expand Down Expand Up @@ -337,14 +345,11 @@ def flash_oci_auto(
partitions: Optional mapping of partition -> filename inside OCI image
power_off: Whether to power off the device after flashing (default: True)
"""
# Normalize OCI URL
if not oci_url.startswith("oci://"):
if "://" in oci_url:
raise ValueError(f"Only oci:// URLs are supported, got: {oci_url}")
if ":" in oci_url and "/" in oci_url:
oci_url = f"oci://{oci_url}"
else:
raise ValueError(f"Invalid OCI URL format: {oci_url}")
raise ValueError(
f"OCI URL must start with oci://, got: {oci_url}\n"
f"Usage: j storage flash oci://{oci_url}"
)

if partitions:
self.logger.info(f"Flashing OCI image with explicit mapping: {list(partitions.keys())}")
Expand Down Expand Up @@ -395,6 +400,17 @@ def _execute_flash_command(self, path, target_specs, power_off: bool = True):

if mapping:
if path:
# In multi-target mode, positional path must be an oci:// reference.
if not path.startswith("oci://"):
t_flags = " ".join(f"-t {s}" for s in target_specs)
raise click.ClickException(
f"'{path}' is missing the -t flag.\n"
f"Each partition mapping needs its own -t flag.\n\n"
f"Example:\n"
f" j storage flash -t {path} {t_flags}\n\n"
f"For OCI images, use the oci:// prefix:\n"
f" j storage flash oci://registry.com/image:tag {t_flags}"
)
# 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 @@ -76,16 +76,88 @@ def test_flash_oci_auto_success(ridesx_client):
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"):
with pytest.raises(ValueError, match="OCI URL must start with oci://"):
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")
# Bare registry URL without oci:// prefix
with pytest.raises(ValueError, match="OCI URL must start with oci://"):
ridesx_client.flash_oci_auto("quay.io/org/image:tag")

# 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(click.ClickException, match="No fastboot devices found"):
ridesx_client.flash_oci_auto("oci://image:tag")


# _execute_flash_command Tests


@pytest.mark.parametrize("invalid_path", [
"boot_a:/path/to/boot.img", # partition:absolute_path
"boot_a:./boot_a.simg", # partition:relative_path
"boot_a:boot.img", # partition:filename
"./boot_a.simg", # local file path
"quay.io/org/image:tag", # bare registry URL missing oci://
])
def test_execute_flash_command_rejects_non_oci_positional_with_targets(ridesx_client, invalid_path):
"""Non-oci:// positional paths should be rejected in multi-target mode"""
with pytest.raises(click.ClickException, match="missing the -t flag"):
ridesx_client._execute_flash_command(
invalid_path,
("system_a:/path/to/system.img",),
)


def test_execute_flash_command_error_shows_all_target_specs(ridesx_client):
"""Error example should include all -t specs, not just the first"""
with pytest.raises(click.ClickException) as exc_info:
ridesx_client._execute_flash_command(
"boot_a:boot.img",
("system_a:/path/to/system.img", "vendor_a:/path/to/vendor.img"),
)
msg = str(exc_info.value)
assert "-t system_a:/path/to/system.img" in msg
assert "-t vendor_a:/path/to/vendor.img" in msg


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"""
# Absolute path after colon
with pytest.raises(click.ClickException, match="looks like a partition:path mapping"):
ridesx_client.flash("boot_a:/path/to/boot.img")

# Bare filename after colon
with pytest.raises(click.ClickException, match="looks like a partition:path mapping"):
ridesx_client.flash("boot_a:boot.img")


def test_flash_hints_oci_missing_prefix(ridesx_client):
"""Bare registry URL without oci:// should suggest adding the prefix"""
with pytest.raises(click.ClickException, match="OCI URLs must start with oci://") as exc_info:
ridesx_client.flash("quay.io/org/image:tag")
assert "oci://quay.io/org/image:tag" in str(exc_info.value)


def test_flash_no_target_no_partition_spec(ridesx_client):
"""Non-OCI path without colon or target should give a generic helpful error"""
with pytest.raises(click.ClickException, match="requires a target partition"):
ridesx_client.flash("/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(("/", "./", "../", "~")):
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Duplicated path-detection logic

This after.startswith(("/", "./", "../", "~")) check mirrors _looks_like_partition_spec in client.py. Since these live in separate classes (client vs driver) the duplication is acceptable, but a comment here noting the parallel would help keep them in sync if the set of path prefixes changes.

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