tools: Add the flash image builder

target/ast10x0/defs.bzl

@@ -9,7 +9,7 @@ TARGET_COMPATIBLE_WITH = select({
     "//conditions:default": ["@platforms//:incompatible"],
 })
 
-def _system_image_test_impl(ctx):
+def _system_image_test_impl(ctx, extra_runfiles = []):
     image_info = ctx.attr.image[SystemImageInfo]
     executable_symlink = ctx.actions.declare_file(ctx.label.name)
     ctx.actions.symlink(output = executable_symlink, target_file = image_info.elf)
@@ -31,15 +31,15 @@ def _system_image_test_impl(ctx):
             runfiles.merge(ctx.attr.slave_image[DefaultInfo].default_runfiles),
         )
 
+    if extra_runfiles:
+        runfiles = runfiles.merge(ctx.runfiles(files = extra_runfiles))
+
     return [DefaultInfo(
         executable = executable_symlink,
         runfiles = runfiles,
     )]
 
 def _flash_system_image_test_impl(ctx):
-    default_info = _system_image_test_impl(ctx)[0]
-    providers = [default_info]
-
     # fmc_model describes the QEMU FMC device uniformly (JEDEC ID + SFDP
     # geometry), shared by both chip selects. The qemu_runner seeds fresh
     # images at $TEST_TMPDIR/<name> and attaches each present CS image as
@@ -54,6 +54,19 @@ def _flash_system_image_test_impl(ctx):
     if ctx.attr.cs1_image:
         env["AST10X0_CS1_IMAGE"] = ctx.attr.cs1_image
         env["AST10X0_CS1_FILL"] = str(ctx.attr.cs1_fill)
+
+    # A contents file is copied into the image instead of filling it, so the
+    # device already holds something when the guest boots. The runner opens
+    # the path relative to the runfiles tree the test runs in.
+    contents = []
+    if ctx.file.cs0_contents:
+        contents.append(ctx.file.cs0_contents)
+        env["AST10X0_CS0_CONTENTS"] = ctx.file.cs0_contents.short_path
+    if ctx.file.cs1_contents:
+        contents.append(ctx.file.cs1_contents)
+        env["AST10X0_CS1_CONTENTS"] = ctx.file.cs1_contents.short_path
+
+    providers = [_system_image_test_impl(ctx, extra_runfiles = contents)[0]]
     if ctx.attr.cs0_image or ctx.attr.cs1_image:
         providers.append(RunEnvironmentInfo(environment = env))
     return providers
@@ -83,6 +96,11 @@ flash_system_image_test = rule(
     implementation = _flash_system_image_test_impl,
     test = True,
     attrs = {
+        "cs0_contents": attr.label(
+            doc = "File copied into cs0_image at offset 0 instead of filling " +
+                  "it. The rest stays erased. Longer than flash_size is an error.",
+            allow_single_file = True,
+        ),
         "cs0_fill": attr.int(
             doc = "Byte value the qemu_runner seeds cs0_image with (default 0xFF, erased).",
             default = 0xFF,
@@ -92,6 +110,11 @@ flash_system_image_test = rule(
                   "attached as FMC CS0 flash (if=mtd, index=0).",
             default = "",
         ),
+        "cs1_contents": attr.label(
+            doc = "File copied into cs1_image at offset 0 instead of filling " +
+                  "it. The rest stays erased. Longer than flash_size is an error.",
+            allow_single_file = True,
+        ),
         "cs1_fill": attr.int(
             doc = "Byte value the qemu_runner seeds cs1_image with (default 0xFF, erased).",
             default = 0xFF,

target/ast10x0/harness/BUILD.bazel

@@ -3,6 +3,7 @@
 
 load("@rules_platform//platform_data:defs.bzl", "platform_data")
 load("@rules_python//python:py_binary.bzl", "py_binary")
+load("@rules_python//python:py_test.bzl", "py_test")
 
 py_binary(
     name = "qemu_runner_bin",
@@ -25,6 +26,21 @@ platform_data(
     visibility = ["//visibility:public"],
 )
 
+py_test(
+    name = "qemu_runner_test",
+    srcs = [
+        "qemu_runner.py",
+        "qemu_runner_test.py",
+    ],
+    imports = ["."],
+    main = "qemu_runner_test.py",
+    deps = [
+        "@@pigweed++cipd+pigweed.qemu//:qemu-system-arm-runfiles",
+        "@pigweed//pw_tokenizer/py:detokenize",
+        "@rules_python//python/runfiles",
+    ],
+)
+
 py_binary(
     name = "test_runner_bin",
     srcs = [

target/ast10x0/harness/qemu_runner.py

@@ -131,35 +131,64 @@ def _sentinel_watcher(
         print(f"Exception watching sentinel: {e}", file=sys.stderr)
 
 
-def _seed_flash_image(path: str, size: int, fill: int = 0xFF) -> None:
-    """Create/overwrite `path` with `size` bytes of `fill` (0xFF = erased)."""
+def _seed_flash_image(
+    path: str, size: int, fill: int = 0xFF, contents: str = ""
+) -> None:
+    """Create/overwrite `path` with `size` bytes.
+
+    Without `contents` the image is `size` bytes of `fill` (0xFF = erased).
+    With it, the file is copied in at offset 0 and the remainder is left
+    erased, so a test can hand the device a real image to read.
+
+    A `contents` file longer than `size` is an error rather than a truncation:
+    a half-written image still parses far enough to fail verification, which
+    would make a verify-failure scenario pass for the wrong reason.
+    """
+    if not contents:
+        with open(path, "wb") as f:
+            f.write(bytes([fill & 0xFF]) * size)
+        return
+
+    data = Path(contents).read_bytes()
+    if len(data) > size:
+        raise ValueError(
+            f"{contents} is {len(data)} bytes, larger than the {size}-byte "
+            f"flash it seeds"
+        )
     with open(path, "wb") as f:
-        f.write(bytes([fill & 0xFF]) * size)
+        f.write(data)
+        f.write(b"\xff" * (size - len(data)))
 
 
 def _resolve_flash_drives(args):
-    """Return a list of (index, path, size, fill) FMC backing images.
+    """Return a list of (index, path, size, fill, contents) FMC backing images.
 
     index 0 -> FMC CS0, index 1 -> FMC CS1. Each image is re-seeded on every
     run so tests start from a known device state. An explicit --flash-image
     (manual runs) attaches at CS1. A flash_system_image_test sets
     AST10X0_CS0_IMAGE / AST10X0_CS1_IMAGE (basenames) plus AST10X0_FLASH_SIZE
     and per-CS AST10X0_CS0_FILL / AST10X0_CS1_FILL, resolved against
     $TEST_TMPDIR so each run gets private, freshly-seeded images.
+
+    AST10X0_CS0_CONTENTS / AST10X0_CS1_CONTENTS name a file to copy in at
+    offset 0 instead of filling, for a test that needs the device to already
+    hold an image. The path is a runfile, resolved by the test rule.
     """
     base = os.environ.get("TEST_TMPDIR", tempfile.gettempdir())
     size = int(os.environ.get("AST10X0_FLASH_SIZE", str(args.flash_size)))
     drives = []
     if args.flash_image:
-        drives.append((1, args.flash_image, size, 0xFF))
+        drives.append((1, args.flash_image, size, 0xFF, ""))
     cs0 = os.environ.get("AST10X0_CS0_IMAGE")
     if cs0:
         fill = int(os.environ.get("AST10X0_CS0_FILL", "255"))
-        drives.append((0, os.path.join(base, cs0), size, fill))
+        contents = os.environ.get("AST10X0_CS0_CONTENTS", "")
+        drives.append((0, os.path.join(base, cs0), size, fill, contents))
     cs1 = os.environ.get("AST10X0_CS1_IMAGE")
     if cs1:
         fill = int(os.environ.get("AST10X0_CS1_FILL", "255"))
-        drives.append((1, os.path.join(base, cs1), size, fill))
+        contents = os.environ.get("AST10X0_CS1_CONTENTS", "")
+        drives.append((1, os.path.join(base, cs1), size, fill, contents))
     return drives
 
 
@@ -191,8 +220,8 @@ def _main(args) -> None:
         args.image,
     ]
 
-    for index, path, size, fill in drives:
-        _seed_flash_image(path, size, fill)
+    for index, path, size, fill, contents in drives:
+        _seed_flash_image(path, size, fill, contents)
         qemu_args += [
             "-drive",
             f"file={path},format=raw,if=mtd,index={index}",

target/ast10x0/harness/qemu_runner_test.py

@@ -0,0 +1,55 @@
+# Licensed under the Apache-2.0 license
+# SPDX-License-Identifier: Apache-2.0
+"""Tests for the flash seeding in qemu_runner."""
+
+import tempfile
+import unittest
+
+from pathlib import Path
+
+from qemu_runner import _seed_flash_image
+
+
+class SeedFlashImageTest(unittest.TestCase):
+    """A seeded image is always exactly flash_size bytes."""
+
+    def setUp(self) -> None:
+        self._dir = tempfile.TemporaryDirectory()
+        self.addCleanup(self._dir.cleanup)
+        self.image = str(Path(self._dir.name) / "cs1.img")
+
+    def _contents_file(self, data: bytes) -> str:
+        path = Path(self._dir.name) / "contents.bin"
+        path.write_bytes(data)
+        return str(path)
+
+    def test_without_contents_the_image_is_all_fill(self) -> None:
+        _seed_flash_image(self.image, 32, fill=0xFF)
+        self.assertEqual(Path(self.image).read_bytes(), b"\xff" * 32)
+
+    def test_fill_is_honoured(self) -> None:
+        _seed_flash_image(self.image, 8, fill=0x00)
+        self.assertEqual(Path(self.image).read_bytes(), b"\x00" * 8)
+
+    def test_contents_land_at_offset_zero_and_the_rest_is_erased(self) -> None:
+        _seed_flash_image(self.image, 16, contents=self._contents_file(b"openprot"))
+        self.assertEqual(Path(self.image).read_bytes(), b"openprot" + b"\xff" * 8)
+
+    def test_contents_override_fill(self) -> None:
+        """Fill describes the erased remainder, not the seeded bytes."""
+        _seed_flash_image(self.image, 4, fill=0x00, contents=self._contents_file(b"ab"))
+        self.assertEqual(Path(self.image).read_bytes(), b"ab\xff\xff")
+
+    def test_contents_may_fill_the_whole_image(self) -> None:
+        _seed_flash_image(self.image, 4, contents=self._contents_file(b"abcd"))
+        self.assertEqual(Path(self.image).read_bytes(), b"abcd")
+
+    def test_contents_larger_than_the_flash_is_an_error(self) -> None:
+        """Truncating would hand the guest an image that fails for the
+        wrong reason."""
+        with self.assertRaises(ValueError):
+            _seed_flash_image(self.image, 4, contents=self._contents_file(b"abcde"))
+
+
+if __name__ == "__main__":
+    unittest.main()

tools/flash_image/BUILD.bazel

@@ -0,0 +1,40 @@
+# Licensed under the Apache-2.0 license
+# SPDX-License-Identifier: Apache-2.0
+
+load("@rules_rust//rust:defs.bzl", "rust_binary", "rust_test")
+load(":defs.bzl", "flash_image")
+
+# Host build tool: makes the flash image a QEMU test hands the device.
+rust_binary(
+    name = "flash_image",
+    srcs = [
+        "main.rs",
+        "tests.rs",
+    ],
+    edition = "2024",
+    visibility = ["//visibility:public"],
+    deps = ["//services/orchestrator/config:orchestrator_config"],
+)
+
+rust_test(
+    name = "flash_image_test",
+    crate = ":flash_image",
+)
+
+# Proves the rule wiring end to end: the builder runs under Bazel, both
+# payloads land at their bases, and the result is usable as cs0_contents or
+# cs1_contents on a flash_system_image_test.
+flash_image(
+    name = "example_image",
+    flash_size = 64,
+    slots = {
+        "0:0x0:0x20": "testdata/slot_a.bin",
+        "1:0x20:0x20": "testdata/slot_b.bin",
+    },
+)
+
+sh_test(
+    name = "example_image_test",
+    srcs = ["example_image_test.sh"],
+    data = [":example_image"],
+)

tools/flash_image/defs.bzl

@@ -0,0 +1,67 @@
+# Licensed under the Apache-2.0 license
+# SPDX-License-Identifier: Apache-2.0
+"""Builds a flash image from a slot layout, for QEMU tests to hand the device."""
+
+def _flash_image_impl(ctx):
+    image = ctx.actions.declare_file(ctx.label.name + ".img")
+
+    args = ctx.actions.args()
+    args.add("--flash-size", ctx.attr.flash_size)
+    args.add("--output", image)
+
+    payloads = []
+    for spec, target in ctx.attr.slots.items():
+        files = target.files.to_list()
+        if len(files) != 1:
+            fail("slot {}: {} is not a single file".format(spec, target.label))
+        payloads.append(files[0])
+        args.add("--slot", "{}={}".format(spec, files[0].path))
+
+    if ctx.attr.golden:
+        files = ctx.attr.golden.files.to_list()
+        if len(files) != 1:
+            fail("golden: {} is not a single file".format(ctx.attr.golden.label))
+        payloads.append(files[0])
+        args.add("--golden", "{}={}".format(ctx.attr.golden_region, files[0].path))
+
+    ctx.actions.run(
+        mnemonic = "FlashImage",
+        executable = ctx.executable._builder,
+        arguments = [args],
+        inputs = payloads,
+        outputs = [image],
+    )
+
+    return [DefaultInfo(files = depset([image]))]
+
+flash_image = rule(
+    implementation = _flash_image_impl,
+    doc = "A flash image with each payload written at its slot's base, the " +
+          "rest left erased. The layout goes through the same constructors " +
+          "the board tables use, so a layout the orchestrator would refuse " +
+          "fails the build here.",
+    attrs = {
+        "flash_size": attr.int(
+            doc = "Size in bytes of the image, matching the test's flash_size.",
+            mandatory = True,
+        ),
+        "golden": attr.label(
+            doc = "Payload for the golden image, if the layout has one.",
+            allow_single_file = True,
+        ),
+        "golden_region": attr.string(
+            doc = "Where the golden image lives, as base:len.",
+            default = "",
+        ),
+        "slots": attr.string_keyed_label_dict(
+            doc = "Maps id:base:len to the file that fills that slot. Base " +
+                  "and len take decimal or 0x hex.",
+            allow_files = True,
+        ),
+        "_builder": attr.label(
+            default = "//tools/flash_image:flash_image",
+            executable = True,
+            cfg = "exec",
+        ),
+    },
+)

tools/flash_image/example_image_test.sh

@@ -0,0 +1,36 @@
+#!/usr/bin/env bash
+# Licensed under the Apache-2.0 license
+# SPDX-License-Identifier: Apache-2.0
+#
+# Checks the image the flash_image rule built: both payloads at their slot
+# bases, erased everywhere else, and exactly flash_size bytes. This covers
+# the Bazel wiring; the layout rules themselves are covered by
+# //tools/flash_image:flash_image_test.
+
+set -euo pipefail
+
+image="${TEST_SRCDIR}/_main/tools/flash_image/example_image.img"
+
+# wc rather than stat: runfiles entries are symlinks, and GNU stat reports
+# the link itself unless told otherwise.
+actual_size=$(wc -c < "${image}")
+if [ "${actual_size}" -ne 64 ]; then
+    echo "expected a 64-byte image, got ${actual_size}" >&2
+    exit 1
+fi
+
+# od keeps this readable: slot A at 0x00, slot B at 0x20, 0xFF between.
+expected=$(
+    cat <<'EOF'
+0000000   S   L   O   T   -   A   -   C   O   N   T   E   N   T 377 377
+0000020 377 377 377 377 377 377 377 377 377 377 377 377 377 377 377 377
+0000040   S   L   O   T   -   B 377 377 377 377 377 377 377 377 377 377
+0000060 377 377 377 377 377 377 377 377 377 377 377 377 377 377 377 377
+0000100
+EOF
+)
+
+if ! diff <(od -c "${image}") <(echo "${expected}"); then
+    echo "image contents do not match the declared layout" >&2
+    exit 1
+fi