builroot-based flasher images

packages/jumpstarter-cli/jumpstarter_cli/j.py

@@ -8,6 +8,7 @@
 from anyio.from_thread import BlockingPortal
 from jumpstarter_cli_common.exceptions import async_handle_exceptions, leaf_exceptions
 from jumpstarter_cli_common.signal import signal_handler
+from rich import traceback
 
 from jumpstarter.utils.env import env_async
 
@@ -43,6 +44,7 @@ async def cli():
 
 
 def j():
+    traceback.install()
     run(j_async)
 
 

packages/jumpstarter-driver-flashers/README.md

@@ -68,7 +68,9 @@ HTTP servers are used to serve images to the DUT bootloader and busybox shell.
 | cache_dir      | The directory to cache the images          | str  | no       | /var/lib/jumpstarter/flasher |
 | tftp_dir       | The directory to serve the images via TFTP | str  | no       | /var/lib/tftpboot            |
 | http_dir       | The directory to serve the images via HTTP | str  | no       | /var/www/html                |
-
+| variant        | The variant of the DUT DTB to flash to     | str  | no       | (the default defined in the manifest) |
+| manifest       | The manifest to use from the bundle. Every bundle can have multiple manifests, this is the name of the manifest to use  | str  | no       | manifest.yaml |
+| default_target | The default target to flash to if none specified | str  | no       |                              |
 
 ## BaseFlasher API
 

packages/jumpstarter-driver-flashers/jumpstarter_driver_flashers/bundle.py

@@ -1,5 +1,5 @@
 import os
-from typing import Literal, Optional
+from typing import Literal
 
 import yaml
 from pydantic import BaseModel, Field
@@ -9,12 +9,14 @@ class FileAddress(BaseModel):
     file: str
     address: str
 
-
 class DtbVariant(BaseModel):
+    bootcmd: None | str = None
+    file: None | str = None
+
+class Dtb(BaseModel):
     default: str
     address: str
-    variants: dict[str, str]
-
+    variants: dict[str, DtbVariant]
 
 class FlasherLogin(BaseModel):
     login_prompt: str
@@ -25,15 +27,15 @@ class FlasherLogin(BaseModel):
 
 class FlashBundleSpecV1Alpha1(BaseModel):
     manufacturer: str
-    link: Optional[str]
+    link: None | str
     bootcmd: str
     shelltype: Literal["busybox"] = Field(default="busybox")
     login: FlasherLogin = Field(default_factory=lambda: FlasherLogin(login_prompt="login:", prompt="#"))
     default_target: str
     targets: dict[str, str]
     kernel: FileAddress
-    initram: Optional[FileAddress] = None
-    dtb: Optional[DtbVariant] = None
+    initram: None | FileAddress = None
+    dtb: None | Dtb = None
     preflash_commands: list[str] = Field(default_factory=list)
 
 
@@ -56,10 +58,33 @@ def get_dtb_file(self, variant: str | None = None) -> str | None:
         if not self.spec.dtb:
             return None
 
+        # if no variant is provided, use the default variant name from the manifest
         if not variant:
             variant = self.spec.dtb.default
 
-        return self.spec.dtb.variants.get(variant)
+        # look for the variant struct in this manifest
+        variant_struct = self.spec.dtb.variants.get(variant)
+        if variant_struct:
+            return variant_struct.file
+        else:
+            raise ValueError(f"DTB variant {variant} not found in the manifest.")
+
+    def get_boot_cmd(self, variant: str | None = None) -> str:
+        if not self.spec.dtb:
+            return self.spec.bootcmd
+        # if no variant is provided, use the default variant name from the manifest
+        if not variant:
+            variant = self.spec.dtb.default
+        # look for the variant struct in this manifest
+        variant_struct = self.spec.dtb.variants.get(variant)
+        if variant_struct:
+            # If variant has a custom bootcmd, use it; otherwise fall back to default
+            if variant_struct.bootcmd:
+                return variant_struct.bootcmd
+            else:
+                return self.spec.bootcmd
+        else:
+            raise ValueError(f"DTB variant {variant} not found in the manifest.")
 
     def get_kernel_address(self) -> str:
         return self.spec.kernel.address

packages/jumpstarter-driver-flashers/jumpstarter_driver_flashers/client.py

@@ -117,7 +117,7 @@ def flash(
         with self._services_up():
             with self._busybox() as console:
                 manifest = self.manifest
-                target = partition or manifest.spec.default_target
+                target = partition or self.call("get_default_target") or manifest.spec.default_target
                 if not target:
                     raise ArgumentError("No partition or default target specified")
 
@@ -401,19 +401,28 @@ def _busybox(self):
             if manifest.get_initram_file():
                 initram_filename = Path(manifest.get_initram_file()).name
                 initram_address = manifest.get_initram_address()
-                self.uboot.run_command(f"tftpboot {initram_address} {initram_filename}", timeout=120)
-
-            if manifest.get_dtb_file():
-                dtb_filename = Path(manifest.get_dtb_file()).name
-                dtb_address = manifest.get_dtb_address()
-                self.uboot.run_command(f"tftpboot {dtb_address} {dtb_filename}", timeout=120)
+                if initram_address:
+                    self.uboot.run_command(f"tftpboot {initram_address} {initram_filename}", timeout=120)
+
+            try:
+                dtb_file = manifest.get_dtb_file()
+                if dtb_file:
+                    dtb_filename = Path(dtb_file).name
+                    dtb_address = manifest.get_dtb_address()
+                    if dtb_address:
+                        self.uboot.run_command(f"tftpboot {dtb_address} {dtb_filename}", timeout=120)
+            except ValueError:
+                # DTB variant not found, skip DTB loading
+                pass
 
         with self.serial.pexpect() as console:
             if self._console_debug:
                 console.logfile_read = sys.stdout.buffer
 
-            self.logger.info(f"Running boot command: {manifest.spec.bootcmd}")
-            console.send(manifest.spec.bootcmd + "\n")
+            bootcmd = self.call("get_bootcmd")
+
+            self.logger.info(f"Running boot command: {bootcmd}")
+            console.send(bootcmd + "\n")
 
             # if manifest has login details, we need to login
             if manifest.spec.login.username:

packages/jumpstarter-driver-flashers/jumpstarter_driver_flashers/driver.py

@@ -17,6 +17,9 @@ class BaseFlasher(Driver):
     """driver for Jumpstarter"""
 
     flasher_bundle: str = field(default="quay.io/jumpstarter-dev/jumpstarter-flasher-test:latest")
+    variant: None | str = field(default=None)
+    manifest: str = field(default="manifest.yaml")
+    default_target: None | str = field(default=None)
     cache_dir: str = field(default="/var/lib/jumpstarter/flasher")
     tftp_dir: str = field(default="/var/lib/tftpboot")
     http_dir: str = field(default="/var/www/html")
@@ -56,12 +59,16 @@ def __post_init__(self):
 
         # bundles that have already been downloaded in the current session
         self._downloaded = {}
-        self._use_dtb = None  # use default dtb unless set by client
 
     @classmethod
     def client(cls) -> str:
         return "jumpstarter_driver_flashers.client.BaseFlasherClient"
 
+    @export
+    async def get_default_target(self):
+        """Return the default target"""
+        return self.default_target
+
     @export
     async def setup_flasher_bundle(self, force_flash_bundle: str | None = None):
         """Setup flasher bundle
@@ -79,13 +86,15 @@ async def setup_flasher_bundle(self, force_flash_bundle: str | None = None):
         self.logger.info(f"Setting up kernel in tftp: {kernel_path}")
         await self.tftp.storage.copy_exporter_file(kernel_path, kernel_path.name)
 
-        initram_path = await self._get_file_path(manifest.spec.initram.file) if manifest.spec.initram else None
-        if initram_path:
+        initram_file = manifest.get_initram_file()
+        if initram_file:
+            initram_path = await self._get_file_path(initram_file)
             self.logger.info(f"Setting up initram in tftp: {initram_path}")
             await self.tftp.storage.copy_exporter_file(initram_path, initram_path.name)
 
-        dtb_path = await self._get_file_path(manifest.get_dtb_file(self._use_dtb)) if manifest.spec.dtb else None
-        if dtb_path:
+        dtb_file = manifest.get_dtb_file(self.variant) if manifest.spec.dtb else None
+        if dtb_file:
+            dtb_path = await self._get_file_path(dtb_file)
             self.logger.info(f"Setting up dtb in tftp: {dtb_path}")
             await self.tftp.storage.copy_exporter_file(dtb_path, dtb_path.name)
 
@@ -98,12 +107,16 @@ def set_dtb(self, handle):
     async def use_dtb_variant(self, variant):
         """Provide a different dtb reference from the flasher bundle"""
         manifest = await self.get_flasher_manifest()
-        if manifest.get_dtb_file(variant) is None:
+        # Check if the variant exists in the manifest
+        if not manifest.spec.dtb or variant not in manifest.spec.dtb.variants:
+            variant_list = []
+            if manifest.spec.dtb:
+                variant_list = list(manifest.spec.dtb.variants.keys())
             raise ValueError(
                 f"DTB variant {variant} not found in the flasher bundle, "
-                f"available variants are: {list(manifest.spec.dtb.variants.keys())}"
+                f"available variants are: {variant_list}."
             )
-        self._use_dtb = variant
+        self.variant = variant
 
     def set_kernel(self, handle):
         """Provide a different kernel from client"""
@@ -146,17 +159,19 @@ async def _get_file_path(self, filename) -> Path:
         This function will ensure that the bundle is downloaded into cache, and
         then return the path to the requested file in the cache directory.
         """
+        if filename is None:
+            raise ValueError("filename cannot be None")
         bundle_dir = await anyio.to_thread.run_sync(self._download_to_cache)
         return Path(bundle_dir) / filename
 
     @export
     async def get_flasher_manifest_yaml(self) -> str:
         """Return the manifest yaml as a string for client side consumption"""
-        with open(await self._get_file_path("manifest.yaml")) as f:
+        with open(await self._get_file_path(self.manifest)) as f:
             return f.read()
 
     async def get_flasher_manifest(self) -> FlasherBundleManifestV1Alpha1:
-        filename = await self._get_file_path("manifest.yaml")
+        filename = await self._get_file_path(self.manifest)
         return FlasherBundleManifestV1Alpha1.from_file(filename)
 
     @export
@@ -177,7 +192,11 @@ async def get_initram_filename(self) -> str | None:
     async def get_dtb_filename(self) -> str:
         """Return the dtb filename"""
         manifest = await self.get_flasher_manifest()
-        return Path(manifest.get_dtb_file(self._use_dtb)).name
+        dtb_file = manifest.get_dtb_file(self.variant)
+        if dtb_file:
+            return Path(dtb_file).name
+        else:
+            return ""
 
     @export
     async def get_dtb_address(self) -> str:
@@ -197,6 +216,11 @@ async def get_initram_address(self) -> str:
         manifest = await self.get_flasher_manifest()
         return manifest.get_initram_address()
 
+    @export
+    async def get_bootcmd(self) -> str:
+        """Return the bootcmd"""
+        manifest = await self.get_flasher_manifest()
+        return manifest.get_boot_cmd(self.variant)
 
 @dataclass(kw_only=True)
 class TIJ784S4Flasher(BaseFlasher):

packages/jumpstarter-driver-flashers/jumpstarter_driver_flashers/driver_test.py

@@ -6,6 +6,7 @@
 from jumpstarter_driver_pyserial.driver import PySerial
 
 from .driver import BaseFlasher
+from jumpstarter.client.core import DriverInvalidArgument
 from jumpstarter.common.exceptions import ConfigurationError
 from jumpstarter.common.utils import serve
 
@@ -26,6 +27,7 @@ def temp_dirs():
 def complete_flasher(temp_dirs):
     cache, http, tftp = temp_dirs
     yield BaseFlasher(
+        flasher_bundle="quay.io/jumpstarter-dev/jumpstarter-flasher-test:new",
         cache_dir=cache,
         http_dir=http,
         tftp_dir=tftp,
@@ -35,7 +37,6 @@ def complete_flasher(temp_dirs):
         },
     )
 
-
 def test_missing_serial(temp_dirs):
     cache, http, tftp = temp_dirs
     with pytest.raises(ConfigurationError):
@@ -88,3 +89,55 @@ def test_drivers_flashers_addresses(complete_flasher):
         assert client.call("get_kernel_address") == "0x82000000"
         assert client.call("get_initram_address") == "0x83000000"
         assert client.call("get_dtb_address") == "0x84000000"
+
+
+def test_drivers_flashers_get_bootcmd_default(complete_flasher):
+    """Test getting the default boot command"""
+    with serve(complete_flasher) as client:
+        bootcmd = client.call("get_bootcmd")
+        assert bootcmd == "booti 0x82000000 - 0x84000000"
+
+
+def test_drivers_flashers_get_bootcmd_with_dtb_variant(complete_flasher):
+    """Test getting boot command with DTB variant that has custom bootcmd"""
+    with serve(complete_flasher) as client:
+        # Switch to variant with custom boot command
+        client.call("use_dtb_variant", "othercmd")
+        bootcmd = client.call("get_bootcmd")
+        assert bootcmd == "bootm"
+
+
+def test_drivers_flashers_get_bootcmd_with_dtb_variant_no_custom_cmd(complete_flasher):
+    """Test getting boot command with DTB variant that has no custom bootcmd (should use default)"""
+    with serve(complete_flasher) as client:
+        # Switch to variant without custom boot command
+        client.call("use_dtb_variant", "alternate")
+        bootcmd = client.call("get_bootcmd")
+        assert bootcmd == "booti 0x82000000 - 0x84000000"
+
+
+def test_drivers_flashers_get_bootcmd_variant_switching(complete_flasher):
+    """Test that boot command changes when switching between DTB variants"""
+    with serve(complete_flasher) as client:
+        # Start with default variant
+        bootcmd = client.call("get_bootcmd")
+        assert bootcmd == "booti 0x82000000 - 0x84000000"
+
+        # Switch to variant with custom boot command
+        client.call("use_dtb_variant", "othercmd")
+        bootcmd = client.call("get_bootcmd")
+        assert bootcmd == "bootm"
+
+        # Switch back to default variant
+        client.call("use_dtb_variant", "test-dtb")
+        bootcmd = client.call("get_bootcmd")
+        assert bootcmd == "booti 0x82000000 - 0x84000000"
+
+
+def test_drivers_flashers_get_bootcmd_invalid_variant(complete_flasher):
+    """Test that get_bootcmd raises DriverInvalidArgument for invalid DTB variant"""
+    with serve(complete_flasher) as client:
+        # Set an invalid variant
+        with pytest.raises(DriverInvalidArgument):
+            client.call("use_dtb_variant", "noexists")
+

packages/jumpstarter-driver-flashers/jumpstarter_driver_flashers/test_bundle.py

@@ -1,5 +1,7 @@
 from pathlib import Path
 
+import pytest
+
 from . import bundle
 
 
@@ -13,3 +15,40 @@ def test_bundle_read():
         "usd": "/sys/class/block#4fb0000",
         "emmc": "/sys/class/block#4f80000",
     }
+
+
+def test_bundle_get_boot_cmd_default():
+    """Test getting default boot command from test bundle"""
+    manifest_file = Path(__file__).parent / "../oci_bundles/test/manifest.yaml"
+    flasher_bundle = bundle.FlasherBundleManifestV1Alpha1.from_file(manifest_file)
+
+    # Test default boot command (no variant specified)
+    bootcmd = flasher_bundle.get_boot_cmd()
+    assert bootcmd == "booti 0x82000000 - 0x84000000"
+
+
+def test_bundle_get_boot_cmd_with_variant():
+    """Test getting boot command with specific DTB variant"""
+    manifest_file = Path(__file__).parent / "../oci_bundles/test/manifest.yaml"
+    flasher_bundle = bundle.FlasherBundleManifestV1Alpha1.from_file(manifest_file)
+
+    # Test variant with custom boot command
+    bootcmd = flasher_bundle.get_boot_cmd("othercmd")
+    assert bootcmd == "bootm"
+
+    # Test variant without custom boot command (should use default)
+    bootcmd = flasher_bundle.get_boot_cmd("alternate")
+    assert bootcmd == "booti 0x82000000 - 0x84000000"
+
+    # Test default variant explicitly
+    bootcmd = flasher_bundle.get_boot_cmd("test-dtb")
+    assert bootcmd == "booti 0x82000000 - 0x84000000"
+
+
+def test_bundle_get_boot_cmd_invalid_variant():
+    """Test that get_boot_cmd raises ValueError for invalid variant"""
+    manifest_file = Path(__file__).parent / "../oci_bundles/test/manifest.yaml"
+    flasher_bundle = bundle.FlasherBundleManifestV1Alpha1.from_file(manifest_file)
+
+    with pytest.raises(ValueError, match="DTB variant noexists not found in the manifest"):
+        flasher_bundle.get_boot_cmd("noexists")