dandri/HaloKeymind
1
0 Fork 0
mirror of https://github.com/mikecarper/MeshCore.git synced 2026-10-23 06:22:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

Bind resumed firmware packages to their effective build recipe

Browse Source
This commit is contained in:
mikecarper committed 2026-10-03 20:48:49 -07:00
1 parent eab6676316
commit e368b52f7d
5 files changed
+1092 -15
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.github/workflows/run-unit-tests.yml
+1
View File
@@ -296,6 +296,7 @@ jobs:
python3 -B test/test_full_build_flag_dedup.py
python3 -B test/test_full_logging_compile.py
python3 -B test/test_full_logging_resume.py
python3 -B test/test_firmware_build_recipe.py -v
python3 -B test/test_firmware_capabilities.py
MESHCORE_TEST_REAL_PIO_PATHS=1 python3 -B test/test_esp32_partition_migration_build_dir.py -v
bash test/test_build_profiles.sh
build_legacy.sh
+108 -14
View File
@@ -107,7 +107,7 @@ FALLBACK_VERSION_PREFIX="dev"
FALLBACK_VERSION_DATE_FORMAT='+%Y-%m-%d-%H-%M'
# External programs invoked by this script:
# bash, cat, cp, date, env, find, flock, git, grep, head, mkdir, mv, pgrep,
# bash, cat, cp, date, env, find, flock, git, grep, head, mkdir, mktemp, mv, pgrep,
# pio, python3, rm, sed, sleep, sort, systemctl, systemd-run, tee, wc
# Keep this list in sync when adding or removing non-builtin command usage.
@@ -4332,7 +4332,10 @@ write_build_capability_manifest() {
checker_args+=(--expect-application "$item")
done
python3 scripts/check_firmware_capabilities.py "${checker_args[@]}"
python3 scripts/check_firmware_capabilities.py "${checker_args[@]}" || return $?
python3 scripts/firmware_build_recipe.py attach \
"${OUTPUT_DIR}/${firmware_filename}.capabilities.json" \
"$BUILD_RECIPE_SHA256"
}
collect_esp32_artifacts() {
@@ -4414,6 +4417,26 @@ output_artifact_exists() {
[ -s "${OUTPUT_DIR}/$1" ]
}
compute_build_recipe_digest() {
local env_name=$1 env_platform=$2 pio_env_name=$3 embedded_version=$4
local original_flags=$5 full_source_commit=$6 artifact_target checkout_dir
artifact_target=${FIRMWARE_OUTPUT_ENV_NAME:-$env_name}
if [ -n "${NRF52_OTA_SENSOR_PROFILE:-}" ] \
&& is_nrf52_sensor_ota_pair_target "$env_name"; then
artifact_target+="-${NRF52_OTA_SENSOR_PROFILE}-ota"
fi
checkout_dir=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd -P) || return 1
MESHCORE_RECIPE_ORIGINAL_FLAGS="$original_flags" \
python3 "${checkout_dir}/scripts/firmware_build_recipe.py" digest \
--root "$checkout_dir" --output "$OUTPUT_DIR" \
--source-commit "$full_source_commit" \
--target "$env_name" --artifact-target "$artifact_target" \
--pio-env "$pio_env_name" --platform "$env_platform" \
--embedded-version "$embedded_version" --profile "$BUILD_PROFILE_FOR_TARGET" \
--sensor-profile "${NRF52_OTA_SENSOR_PROFILE:-}" \
--ota-policy "${REQUIRE_OTA_UPDATES:-}" <<<"$PIO_CONFIG_JSON"
}
nrf52_sensor_profile_manifest_matches() {
python3 - "$1" "$2" "$3" "$4" <<'PY'
import json, sys
@@ -4676,6 +4699,7 @@ build_firmware_one_profile() {
local pio_env_name
local env_platform
local commit_hash
local full_source_commit
local firmware_build_date
local firmware_build_epoch
local firmware_version
@@ -4694,6 +4718,9 @@ build_firmware_one_profile() {
local had_platformio_build_src_filter=0
local had_platformio_extra_scripts=0
local build_status
local BUILD_RECIPE_SHA256
local recipe_build_flags
local completed_recipe
local platformio_package_lock_fd=""
local -a pio_run_args=()
local -a BUILD_CAPABILITIES=()
@@ -4798,8 +4825,8 @@ build_firmware_one_profile() {
# Publication tools bind artifacts to an explicit eight-character prefix
# of the full source commit. Git's adaptive --short length can grow (or be
# configured differently), which otherwise makes valid releases look stale.
commit_hash=$(git rev-parse HEAD)
commit_hash=${commit_hash:0:8}
full_source_commit=$(git rev-parse HEAD) || return 1
commit_hash=${full_source_commit:0:8}
firmware_build_date=$(date -u '+%d-%b-%Y')
firmware_build_epoch=$(date -u '+%s')
firmware_version=${FIRMWARE_VERSION:-}
@@ -4845,11 +4872,6 @@ build_firmware_one_profile() {
local embedded_version_string="${firmware_version}${embedded_build_suffix}${embedded_variant_tag}-${commit_hash}"
declare_full_logging_application_contract "$env_name"
if [ "$RESUME_BUILD_OUTPUT" == "1" ] && build_artifacts_exist "$env_name" "$env_platform" "$firmware_filename"; then
echo "Skipping ${env_name}; existing artifacts found for ${firmware_filename}."
return 0
fi
if [ "${PLATFORMIO_BUILD_FLAGS+x}" ]; then
had_platformio_build_flags=1
original_platformio_build_flags=$PLATFORMIO_BUILD_FLAGS
@@ -4875,7 +4897,23 @@ build_firmware_one_profile() {
original_platformio_extra_scripts=""
fi
export PLATFORMIO_BUILD_FLAGS="${original_platformio_build_flags} -DFIRMWARE_BUILD_DATE='\"${firmware_build_date}\"' -DFIRMWARE_BUILD_EPOCH=${firmware_build_epoch} -DFIRMWARE_VERSION='\"${embedded_version_string}\"' -DOTA_VARIANT='\"${env_name}\"'${mota_target_flag}${mota_migration_flag}"
# Scope the ambient values and their export attributes to this target. In
# particular, a resume skip/conflict must not leak its computed profile into
# the next target or convert an unset/unexported caller variable to exported.
local PLATFORMIO_BUILD_FLAGS="$original_platformio_build_flags"
local PLATFORMIO_BUILD_UNFLAGS="$original_platformio_build_unflags"
local PLATFORMIO_BUILD_SRC_FILTER="$original_platformio_build_src_filter"
local PLATFORMIO_EXTRA_SCRIPTS="$original_platformio_extra_scripts"
local MESHCORE_ESP32_FULL_BUILD="${MESHCORE_ESP32_FULL_BUILD-}"
local MESHCORE_REQUIRE_PACKET_LOGGING="${MESHCORE_REQUIRE_PACKET_LOGGING-}"
local MESHCORE_COMPANION_RADIO_FULL="${MESHCORE_COMPANION_RADIO_FULL-}"
local MESHCORE_ESP32_FULL_PARTITION_TABLE="${MESHCORE_ESP32_FULL_PARTITION_TABLE-}"
local MESHCORE_NRF52_INTERNAL_BOOTLOADER_UPDATE="${MESHCORE_NRF52_INTERNAL_BOOTLOADER_UPDATE-}"
local MESHCORE_FORCE_LORA_OTA="${MESHCORE_FORCE_LORA_OTA-}"
# Only builder-generated wall-clock stamps are excluded from the recipe.
# User-supplied flags remain verbatim, including any timestamp definitions.
export PLATFORMIO_BUILD_FLAGS="${original_platformio_build_flags} -DFIRMWARE_VERSION='\"${embedded_version_string}\"' -DOTA_VARIANT='\"${env_name}\"'${mota_target_flag}${mota_migration_flag}"
disable_debug_flags "$env_name"
apply_debug_overrides "$env_name"
apply_mqtt_bridge_override "$env_name"
@@ -4943,6 +4981,29 @@ build_firmware_one_profile() {
append_platformio_extra_script "post:scripts/check_firmware_ram.py"
fi
recipe_build_flags=$PLATFORMIO_BUILD_FLAGS
BUILD_RECIPE_SHA256=$(compute_build_recipe_digest "$env_name" "$env_platform" \
"$pio_env_name" "$embedded_version_string" "$original_platformio_build_flags" \
"$full_source_commit") || return 1
if [ "$RESUME_BUILD_OUTPUT" == "1" ]; then
if python3 scripts/firmware_build_recipe.py occupied "${OUTPUT_DIR}/${firmware_filename}"; then
if python3 scripts/firmware_build_recipe.py matches \
"${OUTPUT_DIR}/${firmware_filename}.capabilities.json" "$BUILD_RECIPE_SHA256" \
>/dev/null 2>&1 \
&& build_artifacts_exist "$env_name" "$env_platform" "$firmware_filename"; then
echo "Skipping ${env_name}; matching recipe and qualified artifacts found for ${firmware_filename}."
return 0
fi
echo "Cannot resume ${firmware_filename}: existing package has a missing/different recipe or failed qualification." >&2
echo "Existing files were preserved. Use a fresh OUTPUT_DIR or a different --firmware-version." >&2
return 1
elif [ "$?" -ne 1 ]; then
echo "Cannot inspect the existing package safely; refusing to build over OUTPUT_DIR." >&2
return 1
fi
fi
export PLATFORMIO_BUILD_FLAGS="${recipe_build_flags} -DFIRMWARE_BUILD_DATE='\"${firmware_build_date}\"' -DFIRMWARE_BUILD_EPOCH=${firmware_build_epoch}"
print_build_flags "$pio_env_name" "$env_name"
build_status=0
if [ "$env_platform" = "ESP32_PLATFORM" ]; then
@@ -4978,6 +5039,16 @@ build_firmware_one_profile() {
run_pio_with_size_detection "${pio_run_args[@]}"
build_status=$?
fi
if [ "$build_status" -eq 0 ]; then
completed_recipe=$(PLATFORMIO_BUILD_FLAGS="$recipe_build_flags" \
compute_build_recipe_digest "$env_name" "$env_platform" "$pio_env_name" \
"$embedded_version_string" "$original_platformio_build_flags" \
"$full_source_commit") || build_status=1
if [ "$build_status" -eq 0 ] && [ "$completed_recipe" != "$BUILD_RECIPE_SHA256" ]; then
echo "Source or build inputs changed during compilation; refusing to publish this package." >&2
build_status=1
fi
fi
if [ "$build_status" -eq 0 ]; then
collect_build_artifacts "$env_name" "$env_platform" "$pio_env_name" "$firmware_filename"
build_status=$?
@@ -5926,15 +5997,26 @@ run_logged_build_targets() {
for env in "${targets[@]}"; do
log_path="${log_dir}/${env}-${profile}.log"
log_tmp="${log_path}.tmp"
if [ "$RESUME_BUILD_OUTPUT" = "1" ]; then
log_tmp=$(mktemp "${log_path%.log}-resume-attempt.XXXXXX.log") || {
BATCH_BUILD_MODE=$previous_batch_build_mode
return 1
}
fi
preserved_log=0
echo "Building ${env} (${profile}); log: ${log_path}"
build_firmware "$env" > "$log_tmp" 2>&1
build_status=$?
if [ "$build_status" -eq 0 ] \
&& grep -q "^Skipping ${env}; existing artifacts found" "$log_tmp" \
&& grep -Eq "^Skipping ${env}; (existing artifacts found|matching recipe and qualified artifacts found)" "$log_tmp" \
&& [ -s "$log_path" ]; then
rm -f -- "$log_tmp"
preserved_log=1
elif [ "$build_status" -ne 0 ] \
&& [ "$RESUME_BUILD_OUTPUT" = "1" ] && [ -s "$log_path" ]; then
# Keep the original compile log. This already-unique attempt file is
# the failure's real diagnostic log, including in summary/report paths.
log_path=$log_tmp
else
mv -f -- "$log_tmp" "$log_path"
fi
@@ -5948,7 +6030,7 @@ run_logged_build_targets() {
elif [ "$build_status" -ne 0 ]; then
overall_status=1
LOGGING_MATRIX_FAILURES+=("${env} (${profile}) -> ${log_path}")
echo "FAILED: ${env} (${profile}), status ${build_status}"
echo "FAILED: ${env} (${profile}), status ${build_status}; log: ${log_path}"
echo "FAILED: ${env} (${profile}), status ${build_status}" >> "$log_path"
else
echo "SUCCEEDED: ${env} (${profile})"
@@ -5993,6 +6075,15 @@ run_logged_build_targets() {
pio_env_key=$candidate_key
log_path="${log_dir}/${env}-${profile}.log"
log_tmp="${log_path}.tmp"
if [ "$RESUME_BUILD_OUTPUT" = "1" ]; then
log_tmp=$(mktemp "${log_path%.log}-resume-attempt.XXXXXX.log") || {
for interrupt_pid in "${running_pids[@]}"; do terminate_process_tree "$interrupt_pid"; done
for interrupt_pid in "${running_pids[@]}"; do wait "$interrupt_pid" 2>/dev/null || true; done
BATCH_BUILD_MODE=$previous_batch_build_mode
trap - INT TERM
return 1
}
fi
echo "Building ${env} (${profile}); log: ${log_path}"
(
BATCH_BUILD_MODE=1
@@ -6050,10 +6141,13 @@ run_logged_build_targets() {
log_tmp=${job_tmp_by_pid[$pid]}
preserved_log=0
if [ "$build_status" -eq 0 ] \
&& grep -q "^Skipping ${env}; existing artifacts found" "$log_tmp" \
&& grep -Eq "^Skipping ${env}; (existing artifacts found|matching recipe and qualified artifacts found)" "$log_tmp" \
&& [ -s "$log_path" ]; then
rm -f -- "$log_tmp"
preserved_log=1
elif [ "$build_status" -ne 0 ] \
&& [ "$RESUME_BUILD_OUTPUT" = "1" ] && [ -s "$log_path" ]; then
log_path=$log_tmp
else
mv -f -- "$log_tmp" "$log_path"
fi
@@ -6067,7 +6161,7 @@ run_logged_build_targets() {
elif [ "$build_status" -ne 0 ]; then
overall_status=1
LOGGING_MATRIX_FAILURES+=("${env} (${profile}) -> ${log_path}")
echo "FAILED: ${env} (${profile}), status ${build_status}"
echo "FAILED: ${env} (${profile}), status ${build_status}; log: ${log_path}"
echo "FAILED: ${env} (${profile}), status ${build_status}" >> "$log_path"
else
echo "SUCCEEDED: ${env} (${profile})"
scripts/firmware_build_recipe.py
+264
View File
@@ -0,0 +1,264 @@
#!/usr/bin/env python3
"""Bind resumable firmware to its effective inputs without publishing secrets."""
import argparse
import configparser
import glob
import hashlib
import json
import os
from pathlib import Path
import re
import stat
import subprocess
import sys
SCHEMA = 1
RECIPE_ENVIRONMENT = (
"PLATFORMIO_BUILD_FLAGS", "PLATFORMIO_BUILD_UNFLAGS",
"PLATFORMIO_BUILD_SRC_FILTER", "PLATFORMIO_EXTRA_SCRIPTS",
"MESHCORE_ESP32_FULL_BUILD", "MESHCORE_REQUIRE_PACKET_LOGGING",
"MESHCORE_COMPANION_RADIO_FULL", "MESHCORE_ESP32_FULL_PARTITION_TABLE",
"MESHCORE_NRF52_INTERNAL_BOOTLOADER_UPDATE", "MESHCORE_FORCE_LORA_OTA",
"MESHCORE_NRF52_LEGACY_FLASH_OTA", "MESHCORE_REDUCED_TLS",
)
PACKAGE_SUFFIXES = (".bin", "-merged.bin", ".uf2", ".zip", ".hex",
".capabilities.json", ".memory.json", ".font-license.txt")
def canonical_digest(value):
payload = json.dumps(value, sort_keys=True, separators=(",", ":"),
ensure_ascii=True).encode("ascii")
return hashlib.sha256(payload).hexdigest()
def git(root, *arguments):
return subprocess.check_output(["git", "-C", str(root), *arguments],
stderr=subprocess.PIPE)
def file_digest(path):
result = hashlib.sha256()
with path.open("rb") as source:
for block in iter(lambda: source.read(1024 * 1024), b""):
result.update(block)
return result.hexdigest()
def source_digest(root, output, ancestors=(), expected_commit=None):
"""Read actual tracked bytes, including staged/unstaged and missing files."""
root = root.resolve()
output = Path(os.path.abspath(output))
output_paths = (output, output.resolve())
if any(selected == root or selected in root.parents for selected in output_paths):
raise ValueError("output overlaps the source checkout or its ancestors")
if root in ancestors:
raise ValueError("recursive source checkout")
commit = git(root, "rev-parse", "HEAD").decode("ascii").strip()
if expected_commit is not None and commit != expected_commit:
raise ValueError("source commit no longer matches artifact identity")
entries = {}
for record in git(root, "ls-files", "--stage", "-z").split(b"\0"):
if not record:
continue
metadata, name = record.split(b"\t", 1)
mode, object_id, stage = metadata.split()
if stage != b"0":
raise ValueError("unresolved source merge")
entries[os.fsdecode(name)] = (mode.decode("ascii"), object_id.decode("ascii"))
for name in git(root, "ls-files", "--others", "--exclude-standard", "-z").split(b"\0"):
if name:
entries.setdefault(os.fsdecode(name), ("untracked", ""))
# The project imports this Git-ignored file. Its absence is an input too;
# cached resolved options alone cannot see edits during a long compile.
entries.setdefault("platformio.local.ini", ("local-config", ""))
content = []
for name, (mode, object_id) in sorted(entries.items()):
path = root / name
# Only generated, untracked output may be excluded. Tracked sources
# and imported configuration must always bind their actual bytes,
# even if a caller incorrectly places output within their subtree.
if mode == "untracked" and any(path == selected or selected in path.parents
for selected in output_paths):
continue
item = [name, mode]
if mode == "160000":
item.append(object_id)
if path.is_dir() and (path / ".git").exists():
item.append(source_digest(path, output, (*ancestors, root)))
else:
if path.is_dir() and any(path.iterdir()):
raise ValueError("nonempty source checkout lacks Git metadata")
item.append("uninitialized")
elif path.is_symlink():
item.extend(("symlink", os.readlink(path)))
if path.is_file():
item.append(file_digest(path))
elif path.exists():
raise ValueError("source symlink is not a regular file")
else:
item.append("missing-target")
elif path.is_file():
item.extend(("file", bool(path.stat().st_mode & stat.S_IXUSR), file_digest(path)))
elif not path.exists():
item.append("missing")
else:
raise ValueError("source is not a regular file")
content.append(item)
if git(root, "rev-parse", "HEAD").decode("ascii").strip() != commit:
raise ValueError("source commit changed during snapshot")
return canonical_digest({"commit": commit, "files": content})
def configuration_inputs(root, options):
"""Bind config imports, not package caches; unresolved imports fail closed."""
patterns = options.get("extra_configs", [])
if isinstance(patterns, str):
patterns = [line.strip() for line in patterns.splitlines() if line.strip()]
if not isinstance(patterns, list) or not all(isinstance(value, str) for value in patterns):
raise ValueError("invalid imported configuration paths")
# Resolved options may contain only the final imported setting. Include
# literal imports from the project and each loaded file too, so an
# intermediate external config cannot disappear from the receipt.
pending = ["platformio.ini", "platformio.local.ini", *patterns]
seen_patterns = set()
parsed_files = set()
inputs = []
while pending:
pattern = pending.pop(0)
expanded = os.path.expanduser(os.path.expandvars(pattern))
if "$" in expanded or "%(" in expanded:
raise ValueError("unresolved imported configuration path")
absolute = str(root.resolve() / expanded)
if absolute in seen_patterns:
continue
seen_patterns.add(absolute)
files = []
for name in sorted(glob.glob(absolute, recursive=True)):
path = Path(name)
if not path.is_file():
raise ValueError("imported configuration is not a regular file")
files.append({"path": name, "resolved": str(path.resolve()),
"link": os.readlink(path) if path.is_symlink() else None,
"sha256": file_digest(path)})
resolved = path.resolve()
if resolved not in parsed_files:
parsed_files.add(resolved)
parser = configparser.RawConfigParser(inline_comment_prefixes=("#", ";"))
with path.open(encoding="utf-8") as stream:
parser.read_file(stream)
imports = parser.get("platformio", "extra_configs", fallback="")
if imports:
pending.extend(value.strip() for value in imports.split(
"\n" if "\n" in imports else ", ") if value.strip())
# An unmatched optional file/glob is retained, so later creation or
# removal is visible even when the resolved PIO options are cached.
inputs.append({"pattern": pattern, "expanded": absolute, "files": files})
return inputs
def recipe_digest(args, options):
if not isinstance(options, list):
raise ValueError("invalid resolved PlatformIO configuration")
selected = {}
for value in options:
if not isinstance(value, list) or len(value) != 2:
raise ValueError("invalid PlatformIO section")
section, pairs = value
if section not in ("platformio", "env:" + args.pio_env):
continue
if section in selected or not isinstance(pairs, list):
raise ValueError("invalid PlatformIO options")
selected[section] = {}
for pair in pairs:
if not isinstance(pair, list) or len(pair) != 2 or not isinstance(pair[0], str):
raise ValueError("invalid PlatformIO option")
key, setting = pair
if key in selected[section]:
raise ValueError("duplicate PlatformIO option")
selected[section][key] = setting
if "env:" + args.pio_env not in selected:
raise ValueError("missing resolved PlatformIO environment")
names = set(RECIPE_ENVIRONMENT) | {name for name in os.environ
if name.startswith(("PLATFORMIO_", "MESHCORE_"))}
names.discard("MESHCORE_RECIPE_ORIGINAL_FLAGS")
environment = {name: os.environ.get(name) for name in names}
return canonical_digest({
"schema_version": SCHEMA,
"source": source_digest(args.root, args.output, expected_commit=args.source_commit),
"target": args.target, "artifact_target": args.artifact_target,
"platformio_env": args.pio_env, "platform": args.platform,
"embedded_version": args.embedded_version, "profile": args.profile,
"sensor_profile": args.sensor_profile, "ota_policy": args.ota_policy,
"pio_options": selected, "environment": environment,
"config_inputs": configuration_inputs(args.root, selected.get("platformio", {})),
"original_flags": os.environ.get("MESHCORE_RECIPE_ORIGINAL_FLAGS", ""),
})
def recipe_record(digest):
if not re.fullmatch(r"[0-9a-f]{64}", digest):
raise ValueError("invalid recipe digest")
return {"schema_version": SCHEMA, "sha256": digest}
def attach_recipe(manifest_path, digest):
manifest = json.loads(manifest_path.read_text())
if not isinstance(manifest, dict):
raise ValueError("invalid capability manifest")
manifest["build_recipe"] = recipe_record(digest)
temporary = manifest_path.with_name(f".{manifest_path.name}.{os.getpid()}.tmp")
try:
temporary.write_text(json.dumps(manifest, indent=2, sort_keys=True, ensure_ascii=True) + "\n")
temporary.replace(manifest_path)
finally:
temporary.unlink(missing_ok=True)
def matches_recipe(manifest_path, digest):
manifest = json.loads(manifest_path.read_text())
record = manifest.get("build_recipe")
return (isinstance(record, dict) and type(record.get("schema_version")) is int
and record == recipe_record(digest))
def package_occupied(stem):
return any(os.path.lexists(str(stem) + suffix) for suffix in PACKAGE_SUFFIXES)
def main():
parser = argparse.ArgumentParser(description=__doc__)
commands = parser.add_subparsers(dest="command", required=True)
digest = commands.add_parser("digest")
digest.add_argument("--root", required=True, type=Path)
digest.add_argument("--output", required=True, type=Path)
for name in ("target", "artifact-target", "pio-env", "platform", "source-commit", "embedded-version",
"profile", "sensor-profile", "ota-policy"):
digest.add_argument("--" + name, required=True)
for name in ("attach", "matches"):
action = commands.add_parser(name)
action.add_argument("manifest", type=Path)
action.add_argument("digest")
occupied = commands.add_parser("occupied")
occupied.add_argument("stem", type=Path)
args = parser.parse_args()
try:
if args.command == "digest":
print(recipe_digest(args, json.load(sys.stdin)))
elif args.command == "attach":
attach_recipe(args.manifest, args.digest)
elif args.command == "matches":
return 0 if matches_recipe(args.manifest, args.digest) else 1
else:
return 0 if package_occupied(args.stem) else 1
except (OSError, RuntimeError, ValueError, TypeError, AttributeError,
configparser.Error, subprocess.CalledProcessError):
# Raw options, flags, paths and nested Git diagnostics can contain
# credentials. Report a fixed diagnostic, never their exception text.
print("firmware build recipe: cannot qualify recipe inputs or manifest", file=sys.stderr)
return 2
return 0
if __name__ == "__main__":
raise SystemExit(main())
test/test_firmware_build_recipe.py
+684
View File
@@ -0,0 +1,684 @@
#!/usr/bin/env python3
"""Exercise real package qualification and safe recipe-bound resume, without PIO."""
import argparse
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import struct
import subprocess
import tempfile
import unittest
from unittest import mock
ROOT = Path(__file__).resolve().parents[1]
SPEC = importlib.util.spec_from_file_location("firmware_build_recipe", ROOT / "scripts/firmware_build_recipe.py")
RECIPE = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(RECIPE)
SECRET = "private-credential-not-for-manifests"
TARGET = "test_companion_radio_usb"
OPTIONS = [["env:" + TARGET, [["build_flags", ["-DESP32_PLATFORM"]],
["board", "test-board"], ["extra_scripts", []]]]]
def sha256(data):
return hashlib.sha256(data).hexdigest()
class RecipeTests(unittest.TestCase):
def source_fixture(self, root, *, untracked=(), stage=b"0", tracked="source.cpp"):
def git(_root, *arguments):
if arguments == ("rev-parse", "HEAD"):
return b"a" * 40 + b"\n"
if arguments == ("ls-files", "--stage", "-z"):
return b"100644 " + b"b" * 40 + b" " + stage + b"\t" + os.fsencode(tracked) + b"\0"
if arguments == ("ls-files", "--others", "--exclude-standard", "-z"):
return b"\0".join(os.fsencode(name) for name in untracked)
raise AssertionError(arguments)
return mock.patch.object(RECIPE, "git", side_effect=git)
def test_actual_tracked_bytes_deletions_and_untracked_source_change_digest(self):
with tempfile.TemporaryDirectory(prefix="mesh-recipe-source-") as temporary:
root = Path(temporary)
path = root / "source.cpp"; path.write_bytes(b"original tracked source")
with self.source_fixture(root):
original = RECIPE.source_digest(root, root / "output")
path.write_bytes(b"changed without changing HEAD or index")
self.assertNotEqual(original, RECIPE.source_digest(root, root / "output"))
path.unlink()
self.assertNotEqual(original, RECIPE.source_digest(root, root / "output"))
path.write_bytes(b"original tracked source")
extra = root / "added.cpp"; extra.write_bytes(b"uncommitted source")
with self.source_fixture(root, untracked=(extra.name,)):
self.assertNotEqual(original, RECIPE.source_digest(root, root / "output"))
with self.source_fixture(root, stage=b"2"):
with self.assertRaises(ValueError):
RECIPE.source_digest(root, root / "output")
def test_source_commit_must_match_filename_and_remain_stable_during_snapshot(self):
with tempfile.TemporaryDirectory(prefix="mesh-recipe-source-") as temporary:
root = Path(temporary)
(root / "source.cpp").write_bytes(b"source")
with self.source_fixture(root):
with self.assertRaises(ValueError):
RECIPE.source_digest(root, root / "output", expected_commit="d" * 40)
commits = iter((b"a" * 40, b"b" * 40))
def git(_root, *arguments):
return next(commits) if arguments == ("rev-parse", "HEAD") else b""
with mock.patch.object(RECIPE, "git", side_effect=git):
with self.assertRaises(ValueError):
RECIPE.source_digest(root, root / "output", expected_commit="a" * 40)
def test_output_root_aliases_and_ancestors_are_rejected_instead_of_hiding_source(self):
with tempfile.TemporaryDirectory(prefix="mesh-recipe-source-") as temporary:
root = Path(temporary)
(root / "source.cpp").write_bytes(b"source")
link = root / "root-link"; link.symlink_to(root, target_is_directory=True)
with self.source_fixture(root):
for output in (root, root / "child/..", root.parent, Path("/"), link):
with self.subTest(output=output), self.assertRaises(ValueError):
RECIPE.source_digest(root, output)
def test_tracked_sources_inside_output_are_hashed_but_untracked_artifacts_are_not(self):
with tempfile.TemporaryDirectory(prefix="mesh-recipe-source-") as temporary:
root = Path(temporary)
output = root / "src"; output.mkdir()
source = output / "source.cpp"; source.write_bytes(b"tracked source")
artifact = output / "generated.bin"; artifact.write_bytes(b"old artifact")
with self.source_fixture(root, tracked="src/source.cpp", untracked=("src/generated.bin",)):
first = RECIPE.source_digest(root, output)
artifact.write_bytes(b"new artifact")
self.assertEqual(first, RECIPE.source_digest(root, output))
source.write_bytes(b"changed tracked source")
self.assertNotEqual(first, RECIPE.source_digest(root, output))
def test_ignored_local_config_absence_presence_and_bytes_are_source_inputs(self):
with tempfile.TemporaryDirectory(prefix="mesh-recipe-source-") as temporary:
root = Path(temporary)
(root / "source.cpp").write_bytes(b"source")
local = root / "platformio.local.ini"
with self.source_fixture(root):
absent = RECIPE.source_digest(root, root / "output")
local.write_text("[env:test]\nbuild_flags=-DONE=1\n")
present = RECIPE.source_digest(root, root / "output")
self.assertNotEqual(absent, present)
local.write_text("[env:test]\nbuild_flags=-DONE=2\n")
self.assertNotEqual(present, RECIPE.source_digest(root, root / "output"))
def test_external_config_imports_globs_nested_files_and_missing_matches_are_bound(self):
with tempfile.TemporaryDirectory(prefix="mesh-recipe-config-") as temporary:
directory = Path(temporary)
root = directory / "project"; root.mkdir()
external = directory / "external.ini"
nested = directory / "nested.ini"
overrides = directory / "overrides"; overrides.mkdir()
external.write_text("[platformio]\nextra_configs = ../nested.ini\n")
nested.write_text("[env:test]\nbuild_flags=-DPRIVATE=" + SECRET + "\n")
# An intermediate external file is not necessarily listed in the
# final resolved extra_configs value, but remains an input.
(root / "platformio.ini").write_text("[platformio]\nextra_configs=../external.ini\n")
options = {"extra_configs": [str(directory / "overrides/*.ini"), "../nested.ini"]}
before = RECIPE.canonical_digest(RECIPE.configuration_inputs(root, options))
external.write_text("[platformio]\nextra_configs=../nested.ini\n[env:test]\nboard=changed\n")
after = RECIPE.canonical_digest(RECIPE.configuration_inputs(root, options))
self.assertNotEqual(before, after)
(overrides / "added.ini").write_text("[env:test]\nboard=added\n")
self.assertNotEqual(after, RECIPE.canonical_digest(RECIPE.configuration_inputs(root, options)))
for pattern in ("${sysenv.UNRESOLVED_RECIPE_CONFIG}", "%(unresolved_config)s"):
with self.subTest(pattern=pattern), self.assertRaises(ValueError):
RECIPE.configuration_inputs(root, {"extra_configs": [pattern]})
def test_custom_output_and_output_link_do_not_hash_generated_files(self):
with tempfile.TemporaryDirectory(prefix="mesh-recipe-source-") as temporary:
root = Path(temporary)
(root / "source.cpp").write_bytes(b"source")
output = root / "unignored-artifacts"; output.mkdir()
generated = output / "firmware.bin"; generated.write_bytes(b"old artifact")
with self.source_fixture(root, untracked=("unignored-artifacts/firmware.bin",)):
initial = RECIPE.source_digest(root, output)
generated.write_bytes(b"new artifact")
self.assertEqual(initial, RECIPE.source_digest(root, output))
link = root / "output-link"; link.symlink_to(output, target_is_directory=True)
with self.source_fixture(root, untracked=("output-link", "unignored-artifacts/firmware.bin")):
self.assertEqual(initial, RECIPE.source_digest(root, link))
def test_gitlinks_and_source_symlink_targets_do_not_hide_local_changes(self):
with tempfile.TemporaryDirectory(prefix="mesh-recipe-source-") as temporary:
root = Path(temporary)
nested = root / "nested"; nested.mkdir()
gitlink = b"160000 " + b"b" * 40 + b" 0\tnested\0"
def git(where, *arguments):
if arguments == ("rev-parse", "HEAD"):
return b"a" * 40 + b"\n"
if arguments == ("ls-files", "--stage", "-z"):
return gitlink if where == root else b"100644 " + b"c" * 40 + b" 0\tfile.cpp\0"
return b""
with mock.patch.object(RECIPE, "git", side_effect=git):
RECIPE.source_digest(root, root / "output")
source = nested / "file.cpp"; source.write_bytes(b"local source")
with self.assertRaises(ValueError):
RECIPE.source_digest(root, root / "output")
(nested / ".git").write_text("fixture metadata marker")
first = RECIPE.source_digest(root, root / "output")
source.write_bytes(b"changed local source")
self.assertNotEqual(first, RECIPE.source_digest(root, root / "output"))
external = root / "external.txt"; external.write_bytes(b"initial target")
(root / "source.cpp").symlink_to(external)
with self.source_fixture(root):
first = RECIPE.source_digest(root, root / "output")
external.write_bytes(b"changed target")
self.assertNotEqual(first, RECIPE.source_digest(root, root / "output"))
def test_every_package_suffix_and_broken_symlink_counts_as_occupied(self):
with tempfile.TemporaryDirectory(prefix="mesh-recipe-occupied-") as temporary:
stem = Path(temporary) / "firmware"
self.assertFalse(RECIPE.package_occupied(stem))
for suffix in RECIPE.PACKAGE_SUFFIXES:
path = Path(str(stem) + suffix)
path.write_bytes(b"")
self.assertTrue(RECIPE.package_occupied(stem))
path.unlink()
path.symlink_to(stem.parent / "missing")
self.assertTrue(RECIPE.package_occupied(stem))
def test_selected_options_are_canonical_but_ordered_flags_and_source_are_bound(self):
args = argparse.Namespace(root=ROOT, output=ROOT / "out", target=TARGET,
artifact_target=TARGET, pio_env=TARGET, platform="ESP32_PLATFORM",
source_commit="a" * 40,
embedded_version="vtest.12-abcdef00", profile="auto",
sensor_profile="", ota_policy="0")
environment = {"PLATFORMIO_BUILD_FLAGS": "-DONE=1 -UTWO -DTWO=2",
"MESHCORE_RECIPE_ORIGINAL_FLAGS": "-DWIFI_PWD=" + SECRET}
with mock.patch.object(RECIPE, "source_digest", return_value="c" * 64), \
mock.patch.dict(os.environ, environment, clear=True):
first = RECIPE.recipe_digest(args, OPTIONS)
reordered = [[OPTIONS[0][0], list(reversed(OPTIONS[0][1]))],
["env:unrelated", [["build_flags", ["not compiled"]]]]]
self.assertEqual(first, RECIPE.recipe_digest(args, reordered))
os.environ["PLATFORMIO_BUILD_FLAGS"] = "-DONE=1 -DTWO=2 -UTWO"
self.assertNotEqual(first, RECIPE.recipe_digest(args, OPTIONS))
os.environ.update(environment)
args.embedded_version = "vtest.13-abcdef00"
self.assertNotEqual(first, RECIPE.recipe_digest(args, OPTIONS))
args.embedded_version = "vtest.12-abcdef00"
os.environ["MESHCORE_NEW_RECIPE_OPTION"] = "1"
self.assertNotEqual(first, RECIPE.recipe_digest(args, OPTIONS))
def test_stored_recipe_is_digest_only_and_has_strict_schema(self):
with tempfile.TemporaryDirectory(prefix="mesh-recipe-manifest-") as temporary:
manifest = Path(temporary) / "capabilities.json"
manifest.write_text(json.dumps({"verified": True}))
digest = RECIPE.canonical_digest({"credentials": SECRET})
RECIPE.attach_recipe(manifest, digest)
text = manifest.read_text()
self.assertNotIn(SECRET, text)
self.assertTrue(RECIPE.matches_recipe(manifest, digest))
self.assertEqual(json.loads(text)["build_recipe"],
{"schema_version": 1, "sha256": digest})
for record in (None, {}, {"schema_version": True, "sha256": digest},
{"schema_version": 1, "sha256": "broken"},
{"schema_version": 1, "sha256": digest, "flags": SECRET}):
manifest.write_text(json.dumps({"build_recipe": record}))
self.assertFalse(RECIPE.matches_recipe(manifest, digest))
def test_malformed_inputs_cannot_echo_credentials_in_diagnostics(self):
command = ["python3", "-B", str(ROOT / "scripts/firmware_build_recipe.py"), "digest",
"--root", str(ROOT), "--output", str(ROOT / "out"), "--target", TARGET,
"--artifact-target", TARGET, "--pio-env", TARGET, "--platform", "ESP32_PLATFORM",
"--source-commit", "a" * 40,
"--embedded-version", "vtest", "--profile", "auto", "--sensor-profile", "",
"--ota-policy", "0"]
result = subprocess.run(command, input=SECRET, capture_output=True, text=True, timeout=10)
self.assertNotEqual(result.returncode, 0)
self.assertNotIn(SECRET, result.stdout + result.stderr)
self.assertEqual(result.stdout, "")
class ResumeTests(unittest.TestCase):
def setUp(self):
self.temporary = tempfile.TemporaryDirectory(prefix="mesh-recipe-resume-")
self.addCleanup(self.temporary.cleanup)
self.work = Path(self.temporary.name)
(self.work / "scripts").symlink_to(ROOT / "scripts", target_is_directory=True)
self.output = self.work / "artifacts"; self.output.mkdir()
self.build_dir = self.work / "current-builds"
build = self.build_dir / TARGET; build.mkdir(parents=True)
image = (b"qualified fixture\0ERR usage: tempradio\0OTA: status\0"
b"image_hash MISMATCH after decode\0")
(build / "firmware.elf").write_bytes(image)
(build / "firmware.bin").write_bytes(image)
(build / "firmware-merged.bin").write_bytes(b"merged fixture\0" + image)
table = struct.pack("<HBBII16sI", 0x50AA, 0, 0x10, 0x10000, 0x140000, b"ota_0", 0)
(build / "partitions.bin").write_bytes(table)
report = dict(schema_version=1, passed=True, target=TARGET,
available_internal_bytes=80000, required_heap_bytes=50000,
largest_internal_region_bytes=80000, required_contiguous_bytes=5120,
elf_sha256=sha256(image))
(build / "firmware.memory.json").write_text(json.dumps(report))
def build(self, *, version="v1.17.1.9", settings="", succeed=False, ambient="exported",
clock="", after_build="", logged=False, source_root=None):
script = r'''
source "$1/build_legacy.sh"
recipe_root=$1
git() { command git -C "$recipe_root" "$@"; }
pio() { printf 'FORBIDDEN_PLATFORMIO\n' >&2; return 99; }
prepare_esp32_arduino3_framework() { :; }
print_build_flags() { :; }
PIO_ENV_PLATFORM_BY_NAME[test_companion_radio_usb]=ESP32_PLATFORM
PIO_CONFIG_JSON=$4
OUTPUT_DIR=$2
PIO_BUILD_DIR_OVERRIDE=$3
FIRMWARE_VERSION=$5
RESUME_BUILD_OUTPUT=1
RADIO_FREQ_OVERRIDE=868.1
RADIO_BW_OVERRIDE=125
RADIO_SF_OVERRIDE=8
RADIO_CR_OVERRIDE=6
FIRMWARE_PROFILE_OVERRIDE=default
REQUIRE_OTA_UPDATES=0
recipe_source=$6
if [ -n "$recipe_source" ]; then
fixture_git_dir=$(command git -C "$recipe_root" rev-parse --absolute-git-dir)
compute_build_recipe_digest() {
GIT_DIR="$fixture_git_dir" GIT_WORK_TREE="$recipe_source" \
MESHCORE_RECIPE_ORIGINAL_FLAGS="$5" \
command python3 "$recipe_root/scripts/firmware_build_recipe.py" digest \
--root "$recipe_source" --output "$OUTPUT_DIR" --source-commit "$6" \
--target "$1" --artifact-target "$1" --pio-env "$3" --platform "$2" \
--embedded-version "$4" --profile "$BUILD_PROFILE_FOR_TARGET" \
--sensor-profile "${NRF52_OTA_SENSOR_PROFILE:-}" --ota-policy "$REQUIRE_OTA_UPDATES" \
<<<"$PIO_CONFIG_JSON"
}
fi
'''
values = {"PLATFORMIO_BUILD_FLAGS": "-DWIFI_PWD='\\\"" + SECRET + "\\\"'",
"PLATFORMIO_BUILD_UNFLAGS": "-DOLD_FLAG",
"PLATFORMIO_BUILD_SRC_FILTER": "+<original/*.cpp>",
"PLATFORMIO_EXTRA_SCRIPTS": "pre:scripts/original.py"}
if ambient == "unset":
script += "unset " + " ".join(values) + "\n"
else:
prefix = "export " if ambient == "exported" else ""
for name, value in values.items():
script += prefix + name + "=" + self.shell_quote(value) + "\n"
script += settings + "\n" + clock + "\n"
script += r'''
ambient_before=$(for name in PLATFORMIO_BUILD_FLAGS PLATFORMIO_BUILD_UNFLAGS PLATFORMIO_BUILD_SRC_FILTER PLATFORMIO_EXTRA_SCRIPTS BUILD_RECIPE_SHA256 recipe_build_flags completed_recipe; do declare -p "$name" 2>/dev/null || :; done)
run_pio_with_size_detection() {
printf 'STUB_BUILD_REQUIRED\n'
'''
script += after_build + "\nreturn " + ("0" if succeed else "89") + "\n}\n"
script += r'''
build_firmware_one_profile test_companion_radio_usb
build_result=$?
ambient_after=$(for name in PLATFORMIO_BUILD_FLAGS PLATFORMIO_BUILD_UNFLAGS PLATFORMIO_BUILD_SRC_FILTER PLATFORMIO_EXTRA_SCRIPTS BUILD_RECIPE_SHA256 recipe_build_flags completed_recipe; do declare -p "$name" 2>/dev/null || :; done)
if [ "$ambient_before" != "$ambient_after" ]; then printf 'AMBIENT_CHANGED\n' >&2; exit 77; fi
exit "$build_result"
'''
if logged:
script = script.replace("build_firmware_one_profile test_companion_radio_usb\n",
"run_logged_build_targets test_companion_radio_usb\n")
script = script.replace('exit "$build_result"\n',
'for failure in "${LOGGING_MATRIX_FAILURES[@]}"; do '
'printf "RECORDED_FAILURE:%s\\n" "$failure"; done\n'
'exit "$build_result"\n')
result = subprocess.run(["bash", "-c", script, "recipe-test", str(ROOT), str(self.output),
str(self.build_dir), json.dumps(OPTIONS), version,
str(source_root) if source_root is not None else ""],
cwd=self.work, text=True, capture_output=True, timeout=45)
self.assertNotIn("FORBIDDEN_PLATFORMIO", result.stdout + result.stderr)
self.assertNotIn(SECRET, result.stdout + result.stderr)
self.assertNotIn("AMBIENT_CHANGED", result.stdout + result.stderr)
return result
@staticmethod
def shell_quote(value):
return "'" + value.replace("'", "'\"'\"'") + "'"
def qualified_fixture(self, **kwargs):
result = self.build(succeed=True, **kwargs)
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
self.assertIn("STUB_BUILD_REQUIRED", result.stdout)
capabilities = next(self.output.glob("*.capabilities.json"))
self.stem = Path(str(capabilities).removesuffix(".capabilities.json"))
self.capabilities = capabilities
self.memory = Path(str(self.stem) + ".memory.json")
self.assertNotIn(SECRET, capabilities.read_text() + self.memory.read_text())
self.validate_fixture()
def validate_fixture(self):
result = subprocess.run(["python3", "-B", str(ROOT / "scripts/firmware_memory_manifest.py"),
"validate-package", str(self.stem)], cwd=ROOT,
text=True, capture_output=True, timeout=10)
self.assertEqual(result.returncode, 0, result.stderr)
def snapshot(self):
return {str(path.relative_to(self.output)):
(os.readlink(path) if path.is_symlink() else path.read_bytes() if path.is_file() else None,
path.lstat().st_ino, path.lstat().st_mtime_ns)
for path in self.output.rglob("*")}
def reseal_capabilities(self):
report = json.loads(self.memory.read_text())
report["files"][self.capabilities.name] = sha256(self.capabilities.read_bytes())
self.memory.write_text(json.dumps(report))
self.validate_fixture()
def assert_conflict(self, **kwargs):
before = self.snapshot()
result = self.build(**kwargs)
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
self.assertNotIn("STUB_BUILD_REQUIRED", result.stdout)
self.assertNotIn("Skipping", result.stdout)
self.assertIn("fresh OUTPUT_DIR", result.stderr)
self.assertEqual(before, self.snapshot())
def test_fresh_build_attaches_recipe_and_exact_qualified_match_skips(self):
self.qualified_fixture()
before = self.snapshot()
result = self.build()
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
self.assertIn("matching recipe and qualified artifacts", result.stdout)
self.assertNotIn("STUB_BUILD_REQUIRED", result.stdout)
self.assertEqual(before, self.snapshot())
def test_generated_clock_changes_do_not_invalidate_recipe(self):
self.qualified_fixture()
result = self.build(clock=r'''date() { case "$*" in *+%s*) printf '9999999999\n';; *) printf '01-Jan-2099\n';; esac; }''')
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
self.assertIn("Skipping", result.stdout)
def test_real_logged_recipe_resume_preserves_original_compile_log(self):
self.qualified_fixture()
logs = self.output / "build-logs"; logs.mkdir()
log = logs / (TARGET + "-standard.log")
log.write_bytes(b"original compiler output and memory qualification\n")
before = (log.read_bytes(), log.stat().st_ino, log.stat().st_mtime_ns)
artifacts = {name: value for name, value in self.snapshot().items() if not name.startswith("build-logs")}
result = self.build(logged=True)
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
self.assertNotIn("STUB_BUILD_REQUIRED", result.stdout + log.read_text())
self.assertEqual(before, (log.read_bytes(), log.stat().st_ino, log.stat().st_mtime_ns))
self.assertEqual(artifacts, {name: value for name, value in self.snapshot().items()
if not name.startswith("build-logs")})
self.assertFalse(Path(str(log) + ".tmp").exists())
def test_real_logged_resume_conflict_preserves_prior_log_and_points_to_new_attempt(self):
self.qualified_fixture()
logs = self.output / "build-logs"; logs.mkdir()
log = logs / (TARGET + "-standard.log")
log.write_bytes(b"original compiler output and memory qualification\n")
before = (log.read_bytes(), log.stat().st_ino, log.stat().st_mtime_ns)
artifacts = {name: value for name, value in self.snapshot().items() if not name.startswith("build-logs")}
result = self.build(logged=True, settings="RADIO_FREQ_OVERRIDE=910.525")
self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
self.assertEqual(before, (log.read_bytes(), log.stat().st_ino, log.stat().st_mtime_ns))
self.assertEqual(artifacts, {name: value for name, value in self.snapshot().items()
if not name.startswith("build-logs")})
attempts = list(logs.glob(TARGET + "-standard-resume-attempt.*.log"))
self.assertEqual(len(attempts), 1)
attempt = attempts[0]
self.assertIn("Cannot resume", attempt.read_text())
self.assertNotIn("STUB_BUILD_REQUIRED", result.stdout + attempt.read_text())
self.assertNotIn(SECRET, attempt.read_text())
self.assertIn("status 1; log: " + str(attempt), result.stdout)
self.assertIn("RECORDED_FAILURE:" + TARGET + " (standard) -> " + str(attempt), result.stdout)
def test_changed_radio_profile_build_number_and_ambient_options_stop_without_overwrite(self):
self.qualified_fixture()
for settings in ("RADIO_FREQ_OVERRIDE=910.525", "RADIO_BW_OVERRIDE=62.5",
"RADIO_SF_OVERRIDE=7", "RADIO_CR_OVERRIDE=5",
"FIRMWARE_PROFILE_OVERRIDE=cascade", "FIRMWARE_BUILD_NUMBER=2",
"export PLATFORMIO_BUILD_FLAGS+=' -DCHANGED_OPTION=1'",
"export PLATFORMIO_BUILD_UNFLAGS+=' -DANOTHER_OPTION'",
"export PLATFORMIO_BUILD_SRC_FILTER+=' +<changed/*.cpp>'",
"export PLATFORMIO_EXTRA_SCRIPTS+=' post:scripts/changed.py'",
"export MESHCORE_REDUCED_TLS=1",
"PIO_CONFIG_JSON=" + self.shell_quote(json.dumps(
[[OPTIONS[0][0], OPTIONS[0][1] + [["board_build.partitions", "other.csv"]]]]))):
with self.subTest(settings=settings):
self.assert_conflict(settings=settings)
def test_missing_and_malformed_recipes_stop_even_when_package_hashes_are_valid(self):
self.qualified_fixture()
original = json.loads(self.capabilities.read_text())
for recipe in (None, {}, {"schema_version": True, "sha256": "a" * 64},
{"schema_version": 1, "sha256": "broken"},
{"schema_version": 1, "sha256": "a" * 64}):
with self.subTest(recipe=recipe):
manifest = dict(original)
if recipe is None:
manifest.pop("build_recipe")
else:
manifest["build_recipe"] = recipe
self.capabilities.write_text(json.dumps(manifest))
self.reseal_capabilities()
self.assert_conflict()
def test_corrupt_and_partial_occupied_packages_stop_without_pio_or_writes(self):
self.qualified_fixture()
application = Path(str(self.stem) + ".bin")
application.write_bytes(b"tampered application")
self.assert_conflict()
for path in list(self.output.iterdir()):
if path != application:
path.unlink()
self.assert_conflict()
application.write_bytes(b"")
self.assert_conflict()
application.unlink()
application.symlink_to(self.work / "missing-image")
self.assert_conflict()
def test_memory_seal_prevents_recipe_marker_tampering(self):
self.qualified_fixture()
old_output = self.output
self.output = self.work / "other-artifacts"; self.output.mkdir()
result = self.build(succeed=True, settings="RADIO_FREQ_OVERRIDE=910.525")
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
new_manifest = json.loads(next(self.output.glob("*.capabilities.json")).read_text())
self.output = old_output
manifest = json.loads(self.capabilities.read_text())
# Marker now matches the changed settings, but cannot be substituted
# into the old image's sealed capability manifest to bypass the gate.
manifest["build_recipe"] = new_manifest["build_recipe"]
self.capabilities.write_text(json.dumps(manifest))
self.assert_conflict(settings="RADIO_FREQ_OVERRIDE=910.525")
def test_ambient_unset_unexported_and_exported_state_is_restored_on_skip_and_conflict(self):
for ambient in ("unset", "unexported", "exported"):
with self.subTest(ambient=ambient):
self.qualified_fixture(ambient=ambient, version="vtest-" + ambient)
result = self.build(ambient=ambient, version="vtest-" + ambient)
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
self.assertIn("Skipping", result.stdout)
self.assert_conflict(ambient=ambient, version="vtest-" + ambient,
settings="RADIO_FREQ_OVERRIDE=910.525")
def test_empty_different_stem_may_build_without_overwriting_previous_version(self):
self.qualified_fixture()
before = self.snapshot()
result = self.build(version="v1.17.1.10")
self.assertEqual(result.returncode, 89, result.stdout + result.stderr)
self.assertIn("STUB_BUILD_REQUIRED", result.stdout)
self.assertEqual(before, self.snapshot())
def test_input_change_during_stub_compile_prevents_artifact_publication(self):
result = self.build(succeed=True, after_build="export PLATFORMIO_BUILD_UNFLAGS+=' -DCHANGED_DURING_BUILD'")
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
self.assertIn("changed during compilation", result.stderr)
self.assertEqual(self.snapshot(), {})
def test_ignored_local_config_change_during_stub_compile_prevents_publication(self):
source = self.work / "source-fixture"; source.mkdir()
# Read-only Git metadata from the real checkout plus a scratch working
# tree avoids initializing/changing Git or touching the real private
# local config. The helper reads only Git inventory/HEAD commands.
(source / ".gitignore").write_bytes((ROOT / ".gitignore").read_bytes())
result = self.build(succeed=True, source_root=source, after_build=r'''
python3 - "$recipe_source/platformio.local.ini" <<'PY'
from pathlib import Path
import sys
Path(sys.argv[1]).write_text("[env:test]\nbuild_flags=-DLOCAL_RECIPE_REPRO=1\n")
PY
''')
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
self.assertIn("changed during compilation", result.stderr)
self.assertEqual(self.snapshot(), {})
def test_external_config_change_during_stub_compile_prevents_publication(self):
source = self.work / "source-fixture"; source.mkdir()
(source / ".gitignore").write_bytes((ROOT / ".gitignore").read_bytes())
(source / "platformio.ini").write_text("[platformio]\nextra_configs=../external.ini\n")
external = self.work / "external.ini"
external.write_text("[env:test]\nbuild_flags=-DEXTERNAL_RECIPE_REPRO=1\n")
result = self.build(succeed=True, source_root=source, after_build=r'''
python3 - "$recipe_source/../external.ini" <<'PY'
from pathlib import Path
import sys
Path(sys.argv[1]).write_text("[env:test]\nbuild_flags=-DEXTERNAL_RECIPE_REPRO=2\n")
PY
''')
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
self.assertIn("changed during compilation", result.stderr)
self.assertEqual(self.snapshot(), {})
def test_occupied_inspection_error_fails_closed_before_stub_compile(self):
result = self.build(settings=r'''
python3() {
if [ "$1" = scripts/firmware_build_recipe.py ] && [ "$2" = occupied ]; then return 2; fi
command python3 "$@"
}
''')
self.assertNotEqual(result.returncode, 0)
self.assertNotIn("STUB_BUILD_REQUIRED", result.stdout)
self.assertIn("Cannot inspect", result.stderr)
self.assertEqual(self.snapshot(), {})
class LogPreservationTests(unittest.TestCase):
def test_serial_and_dormant_worker_consumers_preserve_new_and_legacy_skip_logs(self):
for parallel in (False, True):
for message in ("existing artifacts found", "matching recipe and qualified artifacts found"):
with self.subTest(parallel=parallel, message=message), \
tempfile.TemporaryDirectory(prefix="mesh-recipe-logs-") as temporary:
work = Path(temporary)
output = work / "artifacts"
logs = output / "build-logs"; logs.mkdir(parents=True)
paths = [logs / (name + "-standard.log") for name in ("first", "second")]
for path in paths:
path.write_bytes(b"original compile log\n")
before = [(path.read_bytes(), path.stat().st_ino, path.stat().st_mtime_ns)
for path in paths]
script = r'''
source "$1/build_legacy.sh"
OUTPUT_DIR=$2
skip_message=$3
pio() { printf 'FORBIDDEN_PLATFORMIO\n' >&2; return 99; }
build_firmware() { printf 'Skipping %s; %s for qualified-fixture.\n' "$1" "$skip_message"; }
if [ "$4" = parallel ]; then
# Exercise the retained consumer code using harmless shell-only workers.
# The production policy stays at one worker; no PIO process is launched.
consumer=$(declare -f run_logged_build_targets)
consumer=${consumer/local worker_limit=1/local worker_limit=2}
eval "$consumer"
fi
run_logged_build_targets first second
'''
result = subprocess.run(["bash", "-c", script, "log-test", str(ROOT), str(output),
message, "parallel" if parallel else "serial"],
cwd=work, text=True, capture_output=True, timeout=15)
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
self.assertNotIn("FORBIDDEN_PLATFORMIO", result.stdout + result.stderr)
self.assertEqual(before, [(path.read_bytes(), path.stat().st_ino, path.stat().st_mtime_ns)
for path in paths])
self.assertEqual(list(logs.glob("*.tmp")), [])
def test_failed_resume_attempts_get_unique_logs_but_other_build_results_keep_old_policy(self):
for parallel in (False, True):
for resume, status in ((True, 1), (False, 1), (True, 0), (False, 0)):
with self.subTest(parallel=parallel, resume=resume, status=status), \
tempfile.TemporaryDirectory(prefix="mesh-recipe-log-failures-") as temporary:
work = Path(temporary)
output = work / "artifacts"
logs = output / "build-logs"; logs.mkdir(parents=True)
paths = [logs / (name + "-standard.log") for name in ("first", "second")]
for path in paths:
path.write_bytes(b"original compile log\n")
before = [(path.read_bytes(), path.stat().st_ino, path.stat().st_mtime_ns)
for path in paths]
script = r'''
source "$1/build_legacy.sh"
OUTPUT_DIR=$2
RESUME_BUILD_OUTPUT=$3
stub_status=$4
pio() { printf 'FORBIDDEN_PLATFORMIO\n' >&2; return 99; }
build_firmware() { printf 'current attempt diagnostics\n'; return "$stub_status"; }
if [ "$5" = parallel ]; then
consumer=$(declare -f run_logged_build_targets)
consumer=${consumer/local worker_limit=1/local worker_limit=2}
eval "$consumer"
fi
run_logged_build_targets first second
first_result=$?
run_logged_build_targets first second
second_result=$?
for failure in "${LOGGING_MATRIX_FAILURES[@]}"; do printf 'RECORDED_FAILURE:%s\n' "$failure"; done
printf 'RESULTS:%s/%s\n' "$first_result" "$second_result"
'''
result = subprocess.run(["bash", "-c", script, "failed-log-test", str(ROOT), str(output),
"1" if resume else "0", str(status),
"parallel" if parallel else "serial"],
cwd=work, text=True, capture_output=True, timeout=15)
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
self.assertNotIn("FORBIDDEN_PLATFORMIO", result.stdout + result.stderr)
self.assertIn(f"RESULTS:{status}/{status}", result.stdout)
attempts = list(logs.glob("*-resume-attempt.*.log"))
failed_paths = {line.split(" -> ", 1)[1] for line in result.stdout.splitlines()
if line.startswith("RECORDED_FAILURE:")}
if resume and status:
self.assertEqual(before, [(path.read_bytes(), path.stat().st_ino, path.stat().st_mtime_ns)
for path in paths])
self.assertEqual(len(attempts), 4)
self.assertEqual(failed_paths, set(map(str, attempts)))
for attempt in attempts:
self.assertIn("current attempt diagnostics", attempt.read_text())
self.assertIn("FAILED:", attempt.read_text())
self.assertIn("status 1; log: " + str(attempt), result.stdout)
else:
self.assertEqual(attempts, [])
for path in paths:
self.assertIn("current attempt diagnostics", path.read_text())
self.assertNotIn("original compile log", path.read_text())
self.assertEqual(failed_paths, set(map(str, paths)) if status else set())
self.assertEqual(list(logs.glob("*.tmp")), [])
def test_failed_attempt_log_allocation_preserves_prior_log_and_never_starts_build(self):
with tempfile.TemporaryDirectory(prefix="mesh-recipe-log-allocation-") as temporary:
work = Path(temporary)
output = work / "artifacts"
logs = output / "build-logs"; logs.mkdir(parents=True)
original = logs / "first-standard.log"; original.write_bytes(b"original compile log\n")
before = (original.read_bytes(), original.stat().st_ino, original.stat().st_mtime_ns)
script = r'''
source "$1/build_legacy.sh"
OUTPUT_DIR=$2
RESUME_BUILD_OUTPUT=1
mktemp() { return 70; }
build_firmware() { printf 'SHOULD_NOT_BUILD\n'; }
run_logged_build_targets first second
status=$?
printf 'RESULT:%s BATCH:%s\n' "$status" "$BATCH_BUILD_MODE"
'''
result = subprocess.run(["bash", "-c", script, "allocation-log-test", str(ROOT), str(output)],
cwd=work, text=True, capture_output=True, timeout=10)
self.assertNotIn("SHOULD_NOT_BUILD", result.stdout)
self.assertIn("RESULT:1 BATCH:0", result.stdout)
self.assertEqual(before, (original.read_bytes(), original.stat().st_ino, original.stat().st_mtime_ns))
if __name__ == "__main__":
unittest.main()
test/test_full_infrastructure_selection.sh
+35 -1
View File
@@ -8,6 +8,36 @@ fail() { echo "test_full_infrastructure_selection: $*" >&2; exit 1; }
# Resolve real board inheritance, but never compile or flash firmware.
init_project_context >/dev/null
# No fixture assertion is allowed to fall through into a firmware build. The
# sole PlatformIO operation above resolves board metadata; all later calls are
# errors, including dependency preflight before the compiler entry point.
pio() { fail "selection fixture attempted PlatformIO after metadata resolution"; }
prepare_esp32_arduino3_framework() { fail "selection fixture reached build preflight"; }
run_pio_with_size_detection() { fail "selection fixture attempted compilation"; }
collect_build_artifacts() { fail "selection fixture attempted artifact publication"; }
use_qualified_resume_fixture() {
fixture_output_dir=$(mktemp -d "${TMPDIR:-/tmp}/mesh-full-selection.XXXXXX")
OUTPUT_DIR=$fixture_output_dir
trap 'rm -rf -- "$fixture_output_dir"' EXIT
# These tests mock package qualification to inspect target selection. Supply
# a real, schema-valid receipt for the new recipe gate too; its actual
# occupied/matches helpers run normally. Recipe hashing and package sealing
# are exercised independently by test_firmware_build_recipe.py.
compute_build_recipe_digest() {
python3 - "${OUTPUT_DIR}/${firmware_filename}.capabilities.json" "$1" "$3" <<'PY'
import json
from pathlib import Path
import sys
digest = "a" * 64
manifest = {"target": sys.argv[2], "platformio_env": sys.argv[3],
"build_recipe": {"schema_version": 1, "sha256": digest}}
Path(sys.argv[1]).write_text(json.dumps(manifest) + "\n")
print(digest)
PY
}
}
assert_auto_unified() (
local plain=$1 expected=$2
BUILD_PROFILE_OVERRIDE=auto
@@ -180,8 +210,9 @@ done
# Direct observer promotion must use the matrix's exact output recipe even if
# interactive flags previously requested MQTT without USB logging. Stop at the
# existing-artifact check to exercise the real build entry point without pio run.
# matching-recipe/qualified-artifact check without entering compilation.
(
use_qualified_resume_fixture
ESP32_FULL_BUILD=0
BUILD_PROFILE_EFFECTIVE=standard
BUILD_PROFILE_OVERRIDE=auto
@@ -211,6 +242,7 @@ done
# A canonical bulk build promotes an audited target in-place, retaining its
# environment/mOTA identity. Explicit standard builds remain untouched above.
(
use_qualified_resume_fixture
ESP32_FULL_BUILD=0
BUILD_PROFILE_EFFECTIVE=standard
BUILD_PROFILE_OVERRIDE=auto
@@ -232,6 +264,7 @@ done
|| fail "FULL-only bulk target lost its exact identity"
)
(
use_qualified_resume_fixture
ESP32_FULL_BUILD=0
BUILD_PROFILE_EFFECTIVE=standard
BUILD_PROFILE_OVERRIDE=auto
@@ -250,6 +283,7 @@ done
|| fail "V4 Full did not keep its identity and unified feature recipe"
)
(
use_qualified_resume_fixture
ESP32_FULL_BUILD=0
BUILD_PROFILE_EFFECTIVE=standard
BUILD_PROFILE_OVERRIDE=standard