diff --git a/.github/workflows/run-unit-tests.yml b/.github/workflows/run-unit-tests.yml
index be11cf5c4..db0721123 100644
--- a/.github/workflows/run-unit-tests.yml
+++ b/.github/workflows/run-unit-tests.yml
@@ -412,6 +412,7 @@ jobs:
python3 -B test/test_cascade_release_package.py
python3 -B test/test_release_publication.py
python3 -B test/test_local_release_full_selection.py
+ python3 -B test/test_local_release_capacity_archive.py
python3 -B test/test_t096_full_memory.py
python3 -B test/test_nrf52_ble_startup.py
python3 -B test/test_nrf52_ble_trace.py
diff --git a/build_legacy.sh b/build_legacy.sh
index 5013bdb7e..5ef7a2fb3 100755
--- a/build_legacy.sh
+++ b/build_legacy.sh
@@ -6174,7 +6174,7 @@ run_logged_build_targets() {
fi
preserved_log=0
echo "Building ${env} (${profile}); log: ${log_path}"
- build_firmware "$env" > "$log_tmp" 2>&1
+ PIO_BUILD_JOBS_OVERRIDE=$pio_job_limit build_firmware "$env" > "$log_tmp" 2>&1
build_status=$?
if [ "$build_status" -eq 0 ] \
&& grep -Eq "^Skipping ${env}; (existing artifacts found|matching recipe and qualified artifacts found)" "$log_tmp" \
diff --git a/scripts/build_local_release.py b/scripts/build_local_release.py
index 61d4fbc41..a2bbfedc2 100644
--- a/scripts/build_local_release.py
+++ b/scripts/build_local_release.py
@@ -51,6 +51,145 @@ def run_logged(command: list[str], log: Path, environment: dict[str, str]) -> No
raise RuntimeError(f"build failed; see {log}")
+def capacity_rejected_attempt(manifest_path: Path, version: str, source: str) -> dict:
+ """Recognize only the successful qualification followed by portable-size rejection."""
+ work = manifest_path.parent
+ if manifest_path.is_symlink() or not manifest_path.is_file():
+ raise ValueError("manifest is not a regular non-symlink file")
+ raw = manifest_path.read_bytes()
+ manifest = json.loads(raw)
+ target = manifest.get("target")
+ if not isinstance(target, str) or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.+-]*", target):
+ raise ValueError("invalid target identity")
+ artifact_version = f"{version}-{source[:8]}"
+ stems = (f"{target}-{artifact_version}", f"{target}-ota-{artifact_version}")
+ if manifest_path.name not in {stem + ".capabilities.json" for stem in stems}:
+ raise ValueError("manifest filename does not match target/version/source")
+ stem = manifest_path.name.removesuffix(".capabilities.json")
+ # Even an unfamiliar sidecar can contain firmware or other failed-build
+ # evidence. Never consume such a package merely because its bin is absent.
+ siblings = list(work.glob(stem + "*"))
+ if siblings != [manifest_path] or list(work.rglob(stem + "*")) != [manifest_path]:
+ raise ValueError("manifest is not the sole same-stem output")
+ recipe = manifest.get("build_recipe")
+ checks = manifest.get("verification")
+ if (type(manifest.get("schema_version")) is not int or manifest["schema_version"] != 2
+ or manifest.get("platform") != "ESP32_PLATFORM"
+ or manifest.get("build_profile") != "standard"
+ or manifest.get("artifact_target") != target
+ or manifest.get("verified") is not True
+ or manifest.get("ota_update_verified") is not True
+ or manifest.get("ota_update_evidence") !=
+ "firmware fits both OTA application slots; otadata present"
+ or not isinstance(checks, list) or not checks
+ or any(not isinstance(check, dict) or check.get("present") is not True
+ for check in checks)
+ or not isinstance(recipe, dict) or set(recipe) != {"schema_version", "sha256"}
+ or type(recipe.get("schema_version")) is not int or recipe["schema_version"] != 1
+ or not isinstance(recipe.get("sha256"), str)
+ or not re.fullmatch(r"[0-9a-f]{64}", recipe["sha256"])):
+ raise ValueError("qualification or original recipe binding is incomplete")
+ log = work / "build-logs" / f"{target}-standard.log"
+ if log.parent.is_symlink() or log.is_symlink() or not log.is_file():
+ raise ValueError("original standard log is missing or unsafe")
+ log_raw = log.read_bytes()
+ lines = log_raw.decode("utf-8").splitlines()
+ overflow = re.fullmatch(
+ r"ESP32 app image is (\d+) bytes, exceeding portable LoRa-OTA "
+ r"slot 0x10000\.\.0x150000 \((\d+) bytes\) by (\d+) bytes",
+ lines[-2] if len(lines) >= 2 else "")
+ marker = (f"DEFERRED: {target} (standard) exceeds the portable OTA slot; "
+ "the expanded FULL pass is required.")
+ version_line = f' -DFIRMWARE_VERSION="{artifact_version}"'
+ qualification = re.compile(
+ r"Verified \d+ capability marker\(s\); manifest: " + re.escape(str(manifest_path)))
+ if (not overflow or lines[-1] != marker or version_line not in lines
+ or not any(qualification.fullmatch(line) for line in lines[:-2])):
+ raise ValueError("original log does not prove this exact portable-size rejection")
+ size, limit, excess = map(int, overflow.groups())
+ if limit != 0x140000 or size <= limit or excess != size - limit:
+ raise ValueError("portable overflow measurements are inconsistent")
+ return {"target": target, "manifest": manifest_path.name,
+ "manifest_sha256": hashlib.sha256(raw).hexdigest(),
+ "original_log": str(log.relative_to(work)),
+ "original_log_sha256": hashlib.sha256(log_raw).hexdigest(),
+ "build_recipe": recipe, "image_bytes": size,
+ "portable_limit_bytes": limit, "excess_bytes": excess,
+ "reason": "portable_slot_overflow"}
+
+
+def archive_capacity_rejected_attempts(work: Path, version: str, source: str,
+ *, project_root: Path = ROOT,
+ archive_parent: Path | None = None) -> dict:
+ """Preserve proven manifest-only rejections outside recursive package discovery."""
+ if not re.fullmatch(r"[0-9a-f]{40}", source):
+ raise ValueError("invalid release source commit")
+ if not re.fullmatch(r"v[0-9]+(?:\.[0-9]+){3}-halo-keymind-cascade-dev", version):
+ raise ValueError("invalid release firmware version")
+ work = Path(os.path.abspath(work))
+ parent = Path(os.path.abspath(archive_parent or work.parent))
+ if (work.resolve() != work or not work.is_dir() or parent.resolve() != parent
+ or parent == work or work in parent.parents):
+ raise ValueError("unsafe work/archive location")
+ lock_path = project_root / ".pio" / "build-sh.lock"
+ lock_path.parent.mkdir(parents=True, exist_ok=True)
+ if lock_path.parent.resolve() != lock_path.parent or lock_path.is_symlink():
+ raise ValueError("unsafe build lock location")
+ with lock_path.open("a+") as lock:
+ try:
+ fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+ except BlockingIOError as error:
+ raise RuntimeError("active build owns the checkout; no attempts were archived") from error
+ candidates = []
+ preserved = []
+ for manifest in sorted(work.glob("*.capabilities.json")):
+ try:
+ candidates.append(capacity_rejected_attempt(manifest, version, source))
+ except (ValueError, OSError, TypeError, AttributeError) as error:
+ preserved.append({"manifest": manifest.name, "reason": str(error)})
+ if not candidates:
+ return {"archived": [], "preserved": preserved, "archive": None}
+ # Complete the scan and recheck every candidate before any file moves.
+ for record in candidates:
+ if capacity_rejected_attempt(work / record["manifest"], version, source) != record:
+ raise ValueError("capacity attempt changed during archival scan")
+ archive = Path(tempfile.mkdtemp(
+ prefix=f".capacity-rejected-{version}-{source[:8]}-", dir=parent))
+ (archive / "manifests").mkdir()
+ (archive / "logs").mkdir()
+ for record in candidates:
+ original = work / record["original_log"]
+ copy = archive / "logs" / original.name
+ shutil.copy2(original, copy)
+ if digest(copy) != record["original_log_sha256"]:
+ raise ValueError("original log changed while preserving rejection evidence")
+ index = {"format": "meshcore-capacity-rejected-attempts-v1",
+ "source_commit": source, "firmware_version": version,
+ "work_directory": str(work), "state": "prepared",
+ "attempts": candidates, "moved_manifests": []}
+ def save_index() -> None:
+ temporary = archive / ".index.json.tmp"
+ temporary.write_text(json.dumps(index, indent=2, sort_keys=True,
+ ensure_ascii=True) + "\n", encoding="ascii")
+ temporary.replace(archive / "index.json")
+ save_index()
+ try:
+ for record in candidates:
+ manifest = work / record["manifest"]
+ if capacity_rejected_attempt(manifest, version, source) != record:
+ raise ValueError("capacity attempt changed before archival move")
+ manifest.rename(archive / "manifests" / manifest.name)
+ index["moved_manifests"].append(manifest.name)
+ save_index()
+ except (ValueError, OSError):
+ index["state"] = "interrupted"
+ save_index()
+ raise
+ index["state"] = "archived"
+ save_index()
+ return {"archived": candidates, "preserved": preserved, "archive": str(archive)}
+
+
def full_image_pattern(target: str, version: str, short_source: str) -> str:
return f"{target}-full-*-{version}-{short_source}.bin"
@@ -400,6 +539,15 @@ def main() -> None:
environment["PLATFORMIO_BUILD_DIR"] = str(utility_build_dir)
work.mkdir(parents=True, exist_ok=True)
environment["OUTPUT_DIR"] = str(work)
+ if args.resume:
+ archived = archive_capacity_rejected_attempts(work, args.firmware_version, source)
+ if archived["archived"]:
+ print(f'Archived {len(archived["archived"])} proven portable-capacity rejections: '
+ f'{archived["archive"]}', flush=True)
+ print(f'Preserved {len(archived["preserved"])} other package manifests; '
+ 'ordinary resume qualification still applies.', flush=True)
+ # All other packages retain the existing recipe/qualification checks
+ # in build_legacy.sh; this helper never makes them resumable.
run_logged([
"bash", "build_legacy.sh", "build-firmwares-logging-matrix",
"--firmware-version", args.firmware_version,
diff --git a/scripts/generate_mesh_america_release.py b/scripts/generate_mesh_america_release.py
index af1087652..de9cee2f4 100644
--- a/scripts/generate_mesh_america_release.py
+++ b/scripts/generate_mesh_america_release.py
@@ -81,10 +81,9 @@ def read_stage(stage):
if len(manifests) != group["target_count"]:
raise ValueError("release target count differs from manifest")
for row in manifests:
- key = row["artifact_target"]
- if key in seen or row.get("verified") is not True:
- raise ValueError("duplicate or unqualified release profile")
- seen.add(key)
+ logical_target = row["artifact_target"]
+ if row.get("verified") is not True or row.get("source_commit", source) != source:
+ raise ValueError("unqualified or wrong-source release profile")
files = []
for name in row["files"]:
if Path(name).name != name or name not in checksums:
@@ -93,12 +92,35 @@ def read_stage(stage):
if hashlib.sha256(path.read_bytes()).hexdigest() != checksums[name]:
raise ValueError(f"release asset checksum failed: {name}")
if path.suffix in SUFFIXES:
- if identity(name) != key or family not in name:
- raise ValueError("firmware filename differs from qualified profile")
+ if family not in name:
+ raise ValueError("firmware filename belongs to a different family")
files.append(path)
if not files:
raise ValueError("qualified profile contains no firmware")
- records.append((row, group["tag"], files))
+ identities = {identity(path.name) for path in files}
+ if len(identities) != 1:
+ raise ValueError("firmware filenames disagree on publication profile")
+ published_target = identities.pop()
+ # Full images keep their OTA identity while the filename adds its
+ # qualified build profile. Do not relabel the logical target.
+ infix = published_target.removeprefix(logical_target)
+ valid_infix = (published_target == logical_target
+ or (published_target.startswith(logical_target + "-") and (
+ (infix == "-ota" and "lora" in row.get("ota_update_methods", []))
+ or (row["build_profile"] == "full" and
+ re.fullmatch(r"-full(?:-usb-wifi|-logging)?(?:-ota)?", infix))
+ or (row["build_profile"] == "logging" and
+ re.fullmatch(r"-logging(?:-ota)?", infix)))))
+ qualified_stem = published_target + "-" + family
+ if any(path.stem.removesuffix("-merged") != qualified_stem for path in files):
+ raise ValueError("firmware filename differs from exact release source/version")
+ if (not valid_infix or any(qualified_stem + suffix not in row["files"]
+ for suffix in (".capabilities.json", ".memory.json"))):
+ raise ValueError("firmware filename differs from qualified profile")
+ if published_target in seen:
+ raise ValueError("duplicate release publication profile")
+ seen.add(published_target)
+ records.append(({**row, "publication_target": published_target}, group["tag"], files))
return plan, family, records
@@ -116,7 +138,7 @@ def generate(catalog, plan, family, records, profiles, repo):
by_hardware.setdefault((device["type"], hardware), set()).add(index)
result["device"][index]["firmware"] = []
for row, tag, files in records:
- target = row["artifact_target"]
+ target = row.get("publication_target", row["artifact_target"])
parsed = profiles[target]
platform = PLATFORMS[row["platform"]]
indexes = by_identity.get(target) or by_identity.get(row["target"])
@@ -201,7 +223,7 @@ def main():
args = parser.parse_args()
catalog = json.loads(args.catalog.read_text())
plan, family, records = read_stage(args.stage)
- targets = {row["artifact_target"] for row, _, _ in records}
+ targets = {row.get("publication_target", row["artifact_target"]) for row, _, _ in records}
targets.update(identity(file["name"]) for device in catalog["device"]
for firmware in device["firmware"] for version in firmware["version"].values()
for file in version["files"])
diff --git a/scripts/package_cascade_release.py b/scripts/package_cascade_release.py
index 159323c33..2c1e74f1c 100644
--- a/scripts/package_cascade_release.py
+++ b/scripts/package_cascade_release.py
@@ -290,18 +290,24 @@ def collect_local_release(directory, version, source):
if any(category(record) != group.name for record in group_records):
raise ValueError("local release firmware group disagrees with qualification")
records.extend(group_records)
- indexed = {record["manifest"]["artifact_target"]: record for record in records}
+ # Standard and expanded Full images may retain the same OTA identity.
+ # Match each independently qualified package by its exact file inventory;
+ # the logical target alone cannot distinguish those publication profiles.
+ indexed = {frozenset(str(p.relative_to(directory)) for p in record["files"]): record
+ for record in records}
entries = manifest["firmware"]
if (len(indexed) != len(records) or len(entries) != len(records)
or len(records) != manifest["firmware_target_count"]
- or len({entry["artifact_target"] for entry in entries}) != len(entries)):
+ or any(len(entry["files"]) != len(set(entry["files"])) for entry in entries)
+ or len({frozenset(entry["files"]) for entry in entries}) != len(entries)):
raise ValueError("local release target inventory disagrees with qualification")
for entry in entries:
- record = indexed.get(entry["artifact_target"])
+ record = indexed.get(frozenset(entry["files"]))
if record is None:
raise ValueError("local release contains an unqualified target")
qualified = record["manifest"]
if (qualified["target"] != entry["target"]
+ or qualified["artifact_target"] != entry["artifact_target"]
or qualified.get("source_commit", source) != source
or qualified["platform"] != entry["platform"]
or qualified["build_profile"] != entry["profile"]
@@ -404,6 +410,8 @@ def main():
summaries = []
for record in group["records"]:
manifest = record["manifest"]
+ if manifest.get("source_commit", args.commit) != args.commit:
+ raise ValueError("qualified firmware belongs to another source")
file_links = []
for path in record["files"]:
shutil.copy2(path, destination / path.name)
@@ -412,7 +420,10 @@ def main():
label = "merged.bin (USB)" if path.name.endswith("-merged.bin") else path.suffix[1:]
file_links.append(f'{html.escape(label)}')
methods = ", ".join(manifest.get("ota_update_methods", [])) or "USB"
- summaries.append({**manifest, "sensor_profile": nrf52_sensor_profile(manifest),
+ # Older capability manifests inherit source provenance from the
+ # validated completed release inventory. Preserve their raw bytes.
+ summaries.append({**manifest, "source_commit": args.commit,
+ "sensor_profile": nrf52_sensor_profile(manifest),
"files": [path.name for path in record["files"]]})
rows.append(f"
| {html.escape(manifest['artifact_target'])} | {html.escape(release_profile_label(manifest))} | {html.escape(methods)} | {' | '.join(file_links)} |
")
if group['key'] == 'utility':
diff --git a/test/test_firmware_build_recipe.py b/test/test_firmware_build_recipe.py
index 39fc07072..6902f3fa5 100644
--- a/test/test_firmware_build_recipe.py
+++ b/test/test_firmware_build_recipe.py
@@ -680,5 +680,31 @@ printf 'RESULT:%s BATCH:%s\n' "$status" "$BATCH_BUILD_MODE"
self.assertEqual(before, (original.read_bytes(), original.stat().st_ino, original.stat().st_mtime_ns))
+class CompilerJobLimitTests(unittest.TestCase):
+ def test_serial_matrix_passes_validated_job_limit_and_restores_caller_state(self):
+ for requested, expected in (("4", "4"), ("2", "2"), ("0", "8"), ("invalid", "8")):
+ with self.subTest(requested=requested), tempfile.TemporaryDirectory(prefix="mesh-compiler-jobs-") as temporary:
+ script = r'''
+source "$1/build_legacy.sh"
+OUTPUT_DIR=$2
+OPTION3_PIO_JOBS=$3
+PIO_BUILD_JOBS_OVERRIDE=3
+pio() { printf 'FORBIDDEN_PLATFORMIO\n' >&2; return 99; }
+build_firmware() {
+ printf '%s:%s:%s\n' "$1" "$PIO_BUILD_JOBS_OVERRIDE" "$BATCH_BUILD_MODE" >> "$OUTPUT_DIR/jobs-observed.txt"
+}
+run_logged_build_targets first second
+status=$?
+printf 'RESULT:%s RESTORED_JOBS:%s RESTORED_BATCH:%s\n' "$status" "$PIO_BUILD_JOBS_OVERRIDE" "$BATCH_BUILD_MODE"
+'''
+ result = subprocess.run(["bash", "-c", script, "jobs-test", str(ROOT), temporary, requested],
+ cwd=temporary, text=True, capture_output=True, timeout=10)
+ self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+ self.assertNotIn("FORBIDDEN_PLATFORMIO", result.stdout + result.stderr)
+ self.assertIn("RESULT:0 RESTORED_JOBS:3 RESTORED_BATCH:0", result.stdout)
+ self.assertEqual((Path(temporary) / "jobs-observed.txt").read_text().splitlines(),
+ ["first:" + expected + ":1", "second:" + expected + ":1"])
+
+
if __name__ == "__main__":
unittest.main()
diff --git a/test/test_local_release_capacity_archive.py b/test/test_local_release_capacity_archive.py
new file mode 100644
index 000000000..be46b13ba
--- /dev/null
+++ b/test/test_local_release_capacity_archive.py
@@ -0,0 +1,282 @@
+#!/usr/bin/env python3
+"""Exercise production archival and resume wiring without invoking PlatformIO."""
+
+import fcntl
+import hashlib
+import importlib.util
+import io
+import json
+import os
+from pathlib import Path
+import sys
+import tempfile
+import unittest
+from unittest import mock
+
+REPO = Path(os.environ.get("CAPACITY_ARCHIVE_REPO", str(Path(__file__).resolve().parents[1])))
+sys.path.insert(0, str(REPO / "scripts"))
+from firmware_build_recipe import package_occupied
+MODULE = Path(os.environ.get("CAPACITY_ARCHIVE_MODULE", str(REPO / "scripts/build_local_release.py")))
+SPEC = importlib.util.spec_from_file_location("capacity_archive_release", MODULE)
+BUILD = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(BUILD)
+VERSION = "v1.17.1.9-halo-keymind-cascade-dev"
+SOURCE = "8d31b3932de614ac38d238f1a01010bc5f362fe8"
+TARGET = "heltec_v4_repeater_lora_ota_no_external_sensors"
+
+
+class CapacityArchiveTests(unittest.TestCase):
+ def setUp(self):
+ self.temporary = tempfile.TemporaryDirectory(prefix="mesh-capacity-archive-test-")
+ self.addCleanup(self.temporary.cleanup)
+ self.root = Path(self.temporary.name)
+ self.work = self.root / ".releases" / (".build-" + VERSION + "-" + SOURCE[:8])
+ (self.work / "build-logs").mkdir(parents=True)
+
+ def attempt(self, target=TARGET, *, infix="-ota"):
+ stem = f"{target}{infix}-{VERSION}-{SOURCE[:8]}"
+ manifest = self.work / (stem + ".capabilities.json")
+ value = {"schema_version": 2, "target": target, "artifact_target": target,
+ "platformio_env": "heltec_v4_repeater", "platform": "ESP32_PLATFORM",
+ "build_profile": "standard", "verified": True,
+ "ota_update_verified": True,
+ "ota_update_evidence": "firmware fits both OTA application slots; otadata present",
+ "verification": [{"capability": "ota.update.lora", "present": True,
+ "source": "packaged application", "evidence": "OTA: status"}],
+ "build_recipe": {"schema_version": 1, "sha256": "a" * 64}}
+ manifest.write_text(json.dumps(value, indent=2) + "\n", encoding="ascii")
+ log = self.work / "build-logs" / (target + "-standard.log")
+ log.write_text(
+ f' -DFIRMWARE_VERSION="{VERSION}-{SOURCE[:8]}"\n'
+ f"Verified 1 capability marker(s); manifest: {manifest}\n"
+ "ESP32 app image is 1334184 bytes, exceeding portable LoRa-OTA "
+ "slot 0x10000..0x150000 (1310720 bytes) by 23464 bytes\n"
+ f"DEFERRED: {target} (standard) exceeds the portable OTA slot; "
+ "the expanded FULL pass is required.\n", encoding="ascii")
+ return manifest, log
+
+ def archive(self, **kwargs):
+ return BUILD.archive_capacity_rejected_attempts(
+ self.work, VERSION, SOURCE, project_root=self.root, **kwargs)
+
+ def mutate(self, manifest, **updates):
+ value = json.loads(manifest.read_text()); value.update(updates)
+ manifest.write_text(json.dumps(value) + "\n", encoding="ascii")
+
+ def assert_preserved(self, manifest, log):
+ raw_manifest, raw_log = manifest.read_bytes(), log.read_bytes()
+ before = sorted(path.name for path in self.work.parent.iterdir())
+ result = self.archive()
+ self.assertEqual(result["archived"], [])
+ self.assertEqual(manifest.read_bytes(), raw_manifest)
+ self.assertEqual(log.read_bytes(), raw_log)
+ self.assertEqual(sorted(path.name for path in self.work.parent.iterdir()), before)
+ self.assertEqual(len(result["preserved"]), 1)
+
+ def test_real_log_shape_preserves_raw_bytes_and_leaves_log_in_place(self):
+ manifest, log = self.attempt()
+ raw_manifest, raw_log = manifest.read_bytes(), log.read_bytes()
+ self.assertTrue(package_occupied(manifest.with_name(
+ manifest.name.removesuffix(".capabilities.json"))))
+ result = self.archive(); archive = Path(result["archive"])
+ self.assertEqual(len(result["archived"]), 1)
+ self.assertFalse(manifest.exists())
+ self.assertFalse(package_occupied(manifest.with_name(
+ manifest.name.removesuffix(".capabilities.json"))))
+ self.assertEqual(log.read_bytes(), raw_log)
+ self.assertEqual((archive / "manifests" / manifest.name).read_bytes(), raw_manifest)
+ self.assertEqual((archive / "logs" / log.name).read_bytes(), raw_log)
+ self.assertEqual(archive.parent, self.work.parent)
+ self.assertNotIn(archive, self.work.rglob("*"))
+ index = json.loads((archive / "index.json").read_text(encoding="ascii"))
+ self.assertEqual(index["state"], "archived")
+ self.assertEqual(index["source_commit"], SOURCE)
+ record = index["attempts"][0]
+ self.assertEqual(record["manifest_sha256"], hashlib.sha256(raw_manifest).hexdigest())
+ self.assertEqual(record["original_log_sha256"], hashlib.sha256(raw_log).hexdigest())
+ self.assertEqual(record["image_bytes"] - record["portable_limit_bytes"], 23464)
+ self.assertEqual(self.archive()["archived"], [])
+
+ def test_plain_c6_filename_and_multiple_attempts_receive_unique_archive(self):
+ self.attempt("Xiao_C6_repeater_", infix="")
+ self.attempt()
+ first = self.archive()
+ self.assertEqual(len(first["archived"]), 2)
+ self.attempt()
+ second = self.archive()
+ self.assertNotEqual(first["archive"], second["archive"])
+ self.assertTrue(Path(first["archive"]).is_dir())
+
+ def test_log_only_precollection_size_deferral_needs_no_archive(self):
+ manifest, log = self.attempt("Station_G2_repeater_lora_ota_no_external_sensors")
+ manifest.unlink(); raw_log = log.read_bytes()
+ result = self.archive()
+ self.assertEqual(result["archived"], [])
+ self.assertIsNone(result["archive"])
+ self.assertEqual(log.read_bytes(), raw_log)
+
+ def test_any_firmware_memory_license_unknown_sidecar_or_directory_is_preserved(self):
+ for suffix in (".bin", "-merged.bin", ".uf2", ".zip", ".hex", ".memory.json",
+ ".font-license.txt", ".unexpected", ".directory", ".bin-link"):
+ with self.subTest(suffix=suffix):
+ manifest, log = self.attempt()
+ sibling = self.work / (manifest.name.removesuffix(".capabilities.json") + suffix)
+ if suffix == ".directory": sibling.mkdir()
+ elif suffix == ".bin-link": sibling.symlink_to(self.root / "missing.bin")
+ else: sibling.write_bytes(b"original evidence or qualified firmware")
+ self.assert_preserved(manifest, log)
+ if sibling.is_dir(): sibling.rmdir()
+ else: sibling.unlink()
+
+ def test_unknown_failures_and_incomplete_or_nonstandard_contracts_are_preserved(self):
+ faults = [{"verified": False}, {"ota_update_verified": False},
+ {"platform": "NRF52_PLATFORM"}, {"build_profile": "full"},
+ {"artifact_target": "different"}, {"schema_version": 1},
+ {"schema_version": 2.0},
+ {"verification": []}, {"verification": [{"present": False}]},
+ {"ota_update_evidence": "firmware does not fit both OTA application slots"},
+ {"build_recipe": None}, {"build_recipe": {"schema_version": 1, "sha256": "x" * 64}},
+ {"build_recipe": {"schema_version": True, "sha256": "a" * 64}},
+ {"build_recipe": {"schema_version": 1, "sha256": "a" * 64, "unknown": 1}}]
+ for fault in faults:
+ with self.subTest(fault=fault):
+ manifest, log = self.attempt(); self.mutate(manifest, **fault)
+ self.assert_preserved(manifest, log)
+ manifest, log = self.attempt(); manifest.write_bytes(b"{malformed")
+ self.assert_preserved(manifest, log)
+
+ def test_nested_firmware_is_not_hidden_by_flat_manifest_scan(self):
+ manifest, log = self.attempt()
+ nested = self.work / "nested"; nested.mkdir()
+ (nested / (manifest.name.removesuffix(".capabilities.json") + ".bin")).write_bytes(b"firmware")
+ self.assert_preserved(manifest, log)
+
+ def test_numeric_source_identity_and_final_failure_consistency_are_required(self):
+ replacements = [("23464 bytes", "23465 bytes"), ("1310720 bytes", "1310721 bytes"),
+ ("1334184 bytes", "1310719 bytes"),
+ ("portable LoRa-OTA slot", "ota_0"),
+ (SOURCE[:8], "deadbeef"), ("(standard)", "(full)"),
+ ("Verified 1 capability", "Failed 1 capability"),
+ ("; the expanded FULL pass is required.", "; unknown failure.")]
+ for old, new in replacements:
+ with self.subTest(old=old):
+ manifest, log = self.attempt()
+ log.write_text(log.read_text().replace(old, new), encoding="ascii")
+ self.assert_preserved(manifest, log)
+ manifest, log = self.attempt()
+ with log.open("a") as output: output.write("FAILED: unrelated compiler error\n")
+ self.assert_preserved(manifest, log)
+
+ def test_wrong_source_version_or_target_filename_is_not_consumed(self):
+ for token, replacement in ((SOURCE[:8], "deadbeef"), (VERSION, VERSION.replace("1.9", "1.8")),
+ (TARGET, "different_target")):
+ with self.subTest(token=token):
+ manifest, log = self.attempt()
+ renamed = manifest.with_name(manifest.name.replace(token, replacement))
+ manifest.rename(renamed)
+ self.assert_preserved(renamed, log); renamed.unlink()
+
+ def test_symlink_manifest_log_log_directory_or_work_is_never_consumed(self):
+ manifest, log = self.attempt()
+ original = self.root / "original.capabilities.json"
+ manifest.rename(original); manifest.symlink_to(original)
+ self.assert_preserved(manifest, log)
+ manifest.unlink(); original.rename(manifest)
+ real_log = self.root / "original.log"; log.rename(real_log); log.symlink_to(real_log)
+ self.assert_preserved(manifest, log)
+ log.unlink(); real_log.rename(log)
+ real_logs = self.root / "original-logs"
+ log.parent.rename(real_logs); log.parent.symlink_to(real_logs, target_is_directory=True)
+ self.assert_preserved(manifest, log)
+ alias = self.root / "work-alias"; alias.symlink_to(self.work, target_is_directory=True)
+ with self.assertRaisesRegex(ValueError, "unsafe work/archive"):
+ BUILD.archive_capacity_rejected_attempts(alias, VERSION, SOURCE, project_root=self.root)
+
+ def test_archive_overlap_and_held_build_lock_leave_every_package_byte_untouched(self):
+ manifest, log = self.attempt(); raw_manifest, raw_log = manifest.read_bytes(), log.read_bytes()
+ for parent in (self.work, self.work / "nested"):
+ with self.subTest(parent=parent), self.assertRaisesRegex(ValueError, "unsafe work/archive"):
+ self.archive(archive_parent=parent)
+ lock_path = self.root / ".pio" / "build-sh.lock"; lock_path.parent.mkdir()
+ lock_path.write_bytes(b"original active build lock owner\n")
+ with lock_path.open("a+") as lock:
+ fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+ with self.assertRaisesRegex(RuntimeError, "active build owns"):
+ self.archive()
+ self.assertEqual(lock_path.read_bytes(), b"original active build lock owner\n")
+ self.assertEqual(manifest.read_bytes(), raw_manifest)
+ self.assertEqual(log.read_bytes(), raw_log)
+ self.assertEqual(list(self.work.parent.glob(".capacity-rejected-*")), [])
+
+ def test_complete_scan_recheck_blocks_changes_before_any_move(self):
+ first, _ = self.attempt(); second, _ = self.attempt("second_repeater")
+ original = BUILD.capacity_rejected_attempt; calls = 0
+ def racing(*args):
+ nonlocal calls
+ calls += 1
+ record = original(*args)
+ if calls == 2: self.mutate(first, verified=False)
+ return record
+ with mock.patch.object(BUILD, "capacity_rejected_attempt", side_effect=racing):
+ with self.assertRaises(ValueError): self.archive()
+ self.assertTrue(first.exists()); self.assertTrue(second.exists())
+ self.assertEqual(list(self.work.parent.glob(".capacity-rejected-*")), [])
+
+ def test_late_failure_keeps_moved_raw_evidence_and_marks_partial_archive(self):
+ first, _ = self.attempt(); second, _ = self.attempt("second_repeater")
+ raw_first, raw_second = first.read_bytes(), second.read_bytes()
+ original = BUILD.capacity_rejected_attempt; calls = 0
+ def racing(*args):
+ nonlocal calls
+ calls += 1
+ if calls == 6: raise ValueError("injected late race")
+ return original(*args)
+ with mock.patch.object(BUILD, "capacity_rejected_attempt", side_effect=racing):
+ with self.assertRaisesRegex(ValueError, "injected late race"): self.archive()
+ archive, = self.work.parent.glob(".capacity-rejected-*")
+ self.assertFalse(first.exists()); self.assertTrue(second.exists())
+ self.assertEqual((archive / "manifests" / first.name).read_bytes(), raw_first)
+ self.assertEqual(second.read_bytes(), raw_second)
+ index = json.loads((archive / "index.json").read_text())
+ self.assertEqual(index["state"], "interrupted")
+ self.assertEqual(index["moved_manifests"], [first.name])
+
+ def test_resume_main_archives_before_matrix_and_nonresume_dryrun_do_not_archive(self):
+ (self.work / "build-logs").rmdir()
+ for arguments, expected in ((["--resume"], True), ([], False), (["--resume", "--dry-run"], False)):
+ with self.subTest(arguments=arguments):
+ order = []
+ def archive(*args):
+ order.append("archive")
+ return {"archived": [], "preserved": [], "archive": None}
+ def matrix(*args):
+ order.append("matrix")
+ raise RuntimeError("stop before any external build")
+ with mock.patch.object(BUILD, "ROOT", self.root), \
+ mock.patch.object(BUILD, "git", side_effect=[SOURCE, ""]), \
+ mock.patch.object(BUILD, "archive_capacity_rejected_attempts", side_effect=archive), \
+ mock.patch.object(BUILD, "run_logged", side_effect=matrix), \
+ mock.patch.object(BUILD, "resolve_pio_build_dir", return_value=self.root / ".pio/build"), \
+ mock.patch.object(sys, "argv", ["build_local_release.py"] + arguments), \
+ mock.patch("sys.stdout", new=io.StringIO()), \
+ mock.patch("sys.stderr", new=io.StringIO()):
+ if "--dry-run" in arguments: BUILD.main()
+ else:
+ with self.assertRaisesRegex(RuntimeError, "stop before"): BUILD.main()
+ self.assertEqual(order, ["archive", "matrix"] if expected else
+ ([] if "--dry-run" in arguments else ["matrix"]))
+
+ def test_explicit_release_ci_step_runs_capacity_archive_regressions(self):
+ workflow = Path(os.environ.get("CAPACITY_ARCHIVE_WORKFLOW", str(
+ REPO / ".github/workflows/run-unit-tests.yml"))).read_text(encoding="utf-8")
+ command = "python3 -B test/test_local_release_capacity_archive.py"
+ self.assertEqual(workflow.count(command), 1)
+ # This workflow lists host suites explicitly, rather than discovering
+ # test files. Keep the new resume boundary in the existing release step.
+ self.assertIn("python3 -B test/test_release_publication.py\n"
+ " python3 -B test/test_local_release_full_selection.py\n"
+ " " + command, workflow)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/test/test_release_publication.py b/test/test_release_publication.py
index 1231066e9..75725f163 100644
--- a/test/test_release_publication.py
+++ b/test/test_release_publication.py
@@ -21,17 +21,19 @@ FAMILY = VERSION + "-" + SOURCE[:8]
class ReleasePublicationTest(unittest.TestCase):
- def record(self, directory, target, platform="ESP32_PLATFORM", sensor=""):
+ def record(self, directory, target, platform="ESP32_PLATFORM", sensor="", *,
+ publication_target=None, build_profile="auto"):
directory.mkdir(parents=True, exist_ok=True)
- stem = target + "-" + FAMILY
+ stem = (publication_target or target) + "-" + FAMILY
names = [stem + suffix for suffix in (
[".uf2", ".zip"] if platform == "NRF52_PLATFORM" else [".bin", "-merged.bin"])]
names += [stem + ".capabilities.json", stem + ".memory.json"]
for name in names:
(directory / name).write_bytes((name + "\n").encode("ascii"))
row = {"artifact_target": target, "target": target.removesuffix("-" + sensor + "-ota") if sensor else target,
- "platform": platform, "build_profile": "auto", "verified": True,
- "sensor_profile": sensor, "files": names, "reductions": []}
+ "platform": platform, "build_profile": build_profile, "verified": True,
+ "sensor_profile": sensor, "files": names, "reductions": [],
+ "source_commit": SOURCE, "ota_update_methods": ["lora"]}
return row, [directory / name for name in names]
def stage(self, directory):
@@ -140,10 +142,114 @@ class ReleasePublicationTest(unittest.TestCase):
catalogs.generate(template, plan, family, records,
catalogs.target_profiles(targets), "mikecarper/MeshCore")
+ def test_catalog_accepts_standard_and_full_with_the_same_logical_identity(self):
+ with tempfile.TemporaryDirectory() as temp:
+ directory = Path(temp)
+ plan = self.stage(directory)
+ target = "Station_G2_repeater"
+ published = target + "-full-usb-wifi-ota"
+ row, _ = self.record(directory / "full-profiles", target,
+ publication_target=published, build_profile="full")
+ folder = directory / "full-profiles"
+ (folder / "TARGET-MANIFEST.json").write_text(json.dumps([row]))
+ (folder / "SHA256SUMS.txt").write_text("".join(
+ hashlib.sha256(p.read_bytes()).hexdigest() + " " + p.name + "\n"
+ for p in sorted(folder.iterdir())))
+ plan["groups"].append({"key": "full-profiles", "tag": "full-profiles-" + FAMILY,
+ "target_count": 1})
+ (directory / "release-plan.json").write_text(json.dumps(plan))
+ plan, family, records = catalogs.read_stage(directory)
+ self.assertEqual([row["artifact_target"] for row, _, _ in records].count(target), 2)
+ targets = {row["publication_target"] for row, _, _ in records} | {
+ "Station_G2_companion_radio_full", target, "RAK_4631_repeater"}
+ result = catalogs.generate(self.template(), plan, family, records,
+ catalogs.target_profiles(targets), "mikecarper/MeshCore")
+ entries = result["device"][0]["firmware"]
+ self.assertEqual(len(entries), 3)
+ full = next(entry for entry in entries if entry["subTitle"].startswith(published))
+ self.assertIn("Full profile", full["subTitle"])
+ self.assertTrue(all("/full-profiles-" + FAMILY + "/" in file["url"]
+ for file in full["version"][FAMILY]["files"]))
+ # Matching filenames do not authorize a profile for another board
+ # or a fabricated Full infix on a standard qualified build.
+ for change in ({"artifact_target": "Other_repeater"}, {"build_profile": "auto"},
+ {"source_commit": "b" * 40}):
+ (folder / "TARGET-MANIFEST.json").write_text(json.dumps([{**row, **change}]))
+ (folder / "SHA256SUMS.txt").write_text("".join(
+ hashlib.sha256(p.read_bytes()).hexdigest() + " " + p.name + "\n"
+ for p in sorted(folder.iterdir()) if p.name != "SHA256SUMS.txt"))
+ with self.subTest(change=change), self.assertRaises(ValueError):
+ catalogs.read_stage(directory)
+ # A second source suffix cannot hide inside an otherwise familiar
+ # family filename while keeping the original cap/memory stems.
+ wrong = {**row, "files": list(row["files"])}
+ for index, name in enumerate(wrong["files"]):
+ if name.endswith(".bin"):
+ extra = name.replace(FAMILY, FAMILY + "-bbbbbbbb")
+ (folder / name).rename(folder / extra)
+ wrong["files"][index] = extra
+ (folder / "TARGET-MANIFEST.json").write_text(json.dumps([wrong]))
+ (folder / "SHA256SUMS.txt").write_text("".join(
+ hashlib.sha256(p.read_bytes()).hexdigest() + " " + p.name + "\n"
+ for p in sorted(folder.iterdir()) if p.name != "SHA256SUMS.txt"))
+ with self.assertRaisesRegex(ValueError, "exact release source/version"):
+ catalogs.read_stage(directory)
+
+ def test_local_release_distinguishes_standard_and_full_by_exact_qualified_files(self):
+ with tempfile.TemporaryDirectory() as temp:
+ directory = Path(temp)
+ target = "Station_G2_companion_radio_usb"
+ folder = directory / "firmware/companion"
+ standard, standard_files = self.record(folder, target)
+ full, full_files = self.record(folder, target,
+ publication_target=target + "-full-usb-wifi-ota", build_profile="full")
+ records = [{"manifest": row, "files": files} for row, files in
+ ((standard, standard_files), (full, full_files))]
+ manifest = {"format": "meshcore-local-release-v1", "source_commit": SOURCE,
+ "firmware_version": VERSION, "profile": "cascade", "publication": "local-only",
+ "radio": {"frequency_mhz": 910.525, "bandwidth_khz": 62.5,
+ "spreading_factor": 7, "coding_rate": 5},
+ "firmware_target_count": 2, "migration_board_role_count": 1,
+ "firmware": [{"target": row["target"], "artifact_target": row["artifact_target"],
+ "profile": row["build_profile"], "platform": row["platform"],
+ "files": [str(p.relative_to(directory)) for p in files]}
+ for row, files in ((standard, standard_files), (full, full_files))]}
+ migrations = directory / "esp32-partition-migration"
+ migrations.mkdir()
+ archive = migrations / ("example-" + FAMILY + "-migration.zip")
+ archive.write_bytes(b"qualified migration fixture")
+ bundle = directory / ("esp32-partition-migration-" + FAMILY + "-release.zip")
+ with zipfile.ZipFile(bundle, "w") as z:
+ z.writestr("release-manifest.json", "{}")
+ z.writestr(archive.name, archive.read_bytes())
+ def write_inventory():
+ (directory / "manifest.json").write_text(json.dumps(manifest))
+ (directory / "SHA256SUMS.txt").write_text("".join(
+ hashlib.sha256(p.read_bytes()).hexdigest() + " " + str(p.relative_to(directory)) + "\n"
+ for p in sorted(directory.rglob("*")) if p.is_file() and p.name != "SHA256SUMS.txt"))
+ with patch.object(package, "collect_artifacts", return_value=records), \
+ patch("package_esp32_partition_migration.BOARDS", {"example": {}}), \
+ patch("build_esp32_partition_migration.verify_archive"):
+ write_inventory()
+ self.assertEqual(len(package.collect_local_release(directory, VERSION, SOURCE)[0]), 2)
+ for field, value in (("artifact_target", "Other_companion_radio_usb"),
+ ("profile", "auto")):
+ original = manifest["firmware"][1][field]
+ manifest["firmware"][1][field] = value
+ write_inventory()
+ with self.subTest(field=field), self.assertRaisesRegex(ValueError, "manifest disagrees"):
+ package.collect_local_release(directory, VERSION, SOURCE)
+ manifest["firmware"][1][field] = original
+ manifest["firmware"][1]["files"] = manifest["firmware"][0]["files"]
+ write_inventory()
+ with self.assertRaisesRegex(ValueError, "inventory disagrees"):
+ package.collect_local_release(directory, VERSION, SOURCE)
+
def test_stable_publication_notes_and_assets_are_ascii(self):
with tempfile.TemporaryDirectory() as temp:
base = Path(temp)
row, files = self.record(base / "input", "Station_G2_companion_radio_usb")
+ row.pop("source_commit") # Actual older cap manifests inherit validated release provenance.
output = base / "stage"
argv = ["package", "--input", str(base / "input"), "--output", str(output),
"--local-release", "--stable", "--version", "1.17.1.9", "--commit", SOURCE]
@@ -153,12 +259,23 @@ class ReleasePublicationTest(unittest.TestCase):
package.main()
plan = json.loads((output / "release-plan.json").read_text())
self.assertFalse(plan["groups"][0]["prerelease"])
+ staged_rows = json.loads((output / "companion/TARGET-MANIFEST.json").read_text())
+ self.assertEqual(staged_rows[0]["source_commit"], SOURCE)
+ self.assertNotIn("source_commit", row, "raw capability record was relabeled")
notes = (output / "companion-notes.md").read_text(encoding="ascii")
self.assertIn("Stable release", notes)
self.assertNotIn("Development prerelease", notes)
for path in output.rglob("*"):
if path.is_file():
path.read_text(encoding="ascii")
+ bad = {**row, "source_commit": "b" * 40}
+ bad_argv = list(argv)
+ bad_argv[bad_argv.index(str(output))] = str(base / "wrong-source-stage")
+ with patch.object(sys, "argv", bad_argv), patch.object(package, "collect_local_release",
+ return_value=([{"manifest": bad, "files": files}], [],
+ {"frequency": "910.525", "bandwidth": "62.5", "sf": "7", "cr": "5"})), \
+ self.assertRaisesRegex(ValueError, "another source"):
+ package.main()
def test_local_release_preserves_selected_profiles_and_verifies_complete_inventory(self):
with tempfile.TemporaryDirectory() as temp: