Preserve qualified release artifacts across resumed builds
This commit is contained in:
1 parent
101adf0d9e
commit
f9d1e80aee
8 files changed
+625
-18
No files matched your search
@@ -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()
|
||||
@@ -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()
|
||||
@@ -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:
|
||||
|
||||
Reference in new issue
Block a user