239 lines
14 KiB
Python
239 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""The ordinary release has one ESP32 Full OTA identity per board and role."""
|
|
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
|
from build_local_release import select_ordinary_full_records
|
|
|
|
|
|
def record(target: str, profile: str = "full", platform: str = "ESP32_PLATFORM",
|
|
capabilities: tuple[str, ...] = ()) -> dict:
|
|
return {"manifest": {"target": target, "build_profile": profile, "platform": platform,
|
|
"verified": True, "capabilities": list(capabilities),
|
|
"verification": [{"capability": capability, "present": True,
|
|
"source": "linked image", "evidence": capability}
|
|
for capability in capabilities]},
|
|
"files": []}
|
|
|
|
|
|
class ReleaseFullSelectionTest(unittest.TestCase):
|
|
def test_meshtower_sd_primary_filters_stale_internal_images_without_relabeling(self):
|
|
primary = "Heltec_tower_v2_sdcard_repeater_lora_ota_no_external_sensors"
|
|
variants = []
|
|
for sensor in ("full", "reduced"):
|
|
item = record(primary, "auto", "NRF52_PLATFORM")
|
|
item["manifest"].update(artifact_target=primary + "-" + sensor + "-ota",
|
|
capabilities=["sensor.profile." + sensor])
|
|
variants.append(item)
|
|
companion = record("Heltec_tower_v2_companion_radio_full", platform="NRF52_PLATFORM")
|
|
room = record("Heltec_tower_v2_room_server_lora_ota_no_external_sensors",
|
|
platform="NRF52_PLATFORM")
|
|
legacy = [record(target, "auto", "NRF52_PLATFORM") for target in (
|
|
"Heltec_tower_v2_repeater",
|
|
"Heltec_tower_v2_repeater_lora_ota_no_external_sensors",
|
|
)]
|
|
selected = select_ordinary_full_records(legacy + variants + [companion, room])
|
|
self.assertEqual(selected, variants + [companion, room])
|
|
for before, after in zip(variants, selected):
|
|
self.assertIs(before, after)
|
|
self.assertEqual(after["manifest"]["target"], primary)
|
|
|
|
def test_meshtower_internal_images_cannot_substitute_for_missing_primary(self):
|
|
for target in ("Heltec_tower_v2_repeater",
|
|
"Heltec_tower_v2_repeater_lora_ota_no_external_sensors"):
|
|
with self.subTest(target=target):
|
|
with self.assertRaisesRegex(ValueError, "requires the SD primary"):
|
|
select_ordinary_full_records([record(target, "auto", "NRF52_PLATFORM")])
|
|
|
|
def test_nrf52_full_and_reduced_ota_pair_is_never_collapsed(self):
|
|
for role in ('repeater', 'room_server', 'sensor'):
|
|
target = 'RAK_3401_' + role
|
|
variants = []
|
|
for sensor in ('full', 'reduced'):
|
|
item = record(target, 'auto', 'NRF52_PLATFORM')
|
|
item['manifest'].update(artifact_target=target + '-' + sensor + '-ota',
|
|
capabilities=['sensor.profile.' + sensor])
|
|
variants.append(item)
|
|
self.assertEqual(select_ordinary_full_records(variants), variants)
|
|
|
|
def test_resume_filters_sibling_full_images_only(self):
|
|
inputs = [
|
|
record("Heltec_v3_repeater_observer_mqtt"),
|
|
record("Heltec_v3_repeater_bridge_espnow"),
|
|
record("Heltec_v3_repeater", capabilities=("bridge.espnow",)),
|
|
record("Heltec_v3_repeater", "standard"),
|
|
record("Heltec_v3_room_server"),
|
|
record("Station_G2_repeater"),
|
|
record("Station_G2_repeater_observer_mqtt"),
|
|
record("Tbeam_SX1262_repeater_bridge_espnow"),
|
|
record("Tbeam_SX1262_repeater_observer_mqtt", capabilities=("bridge.espnow",)),
|
|
record("LilyGo_TLora_V2_1_1_6_repeater"),
|
|
record("LilyGo_TLora_V2_1_1_6_repeater_observer_mqtt_"),
|
|
record("Meshadventurer_sx1262_repeater"),
|
|
record("Meshadventurer_sx1262_repeater_bridge_espnow"),
|
|
record("RAK_4631_repeater", platform="NRF52_PLATFORM"),
|
|
]
|
|
selected = select_ordinary_full_records(inputs)
|
|
self.assertEqual({item["manifest"]["target"] for item in selected}, {
|
|
"Heltec_v3_repeater", "Heltec_v3_room_server",
|
|
"Station_G2_repeater_observer_mqtt",
|
|
"Tbeam_SX1262_repeater_observer_mqtt",
|
|
"LilyGo_TLora_V2_1_1_6_repeater_observer_mqtt_",
|
|
"Meshadventurer_sx1262_repeater_bridge_espnow",
|
|
"RAK_4631_repeater",
|
|
})
|
|
self.assertEqual(len(selected), 8) # standard V3 remains alongside Full V3
|
|
|
|
def test_mke_combined_repeater_filters_stale_bridge_identities_without_relabeling(self):
|
|
primary = record("MKE_s3_repeater", capabilities=("bridge.rs232", "bridge.espnow"))
|
|
legacy = [record("MKE_s3_repeater_bridge_" + bridge, profile)
|
|
for bridge in ("rs232", "espnow") for profile in ("full", "standard")]
|
|
for inputs in ([primary], legacy + [primary], [primary] + legacy):
|
|
with self.subTest(primary_first=inputs[0] is primary):
|
|
selected = select_ordinary_full_records(inputs)
|
|
self.assertEqual(selected, [primary])
|
|
self.assertIs(selected[0], primary)
|
|
self.assertEqual(selected[0]["manifest"]["target"], "MKE_s3_repeater")
|
|
|
|
def test_mke_legacy_images_require_a_qualified_combined_primary(self):
|
|
incomplete = []
|
|
incomplete.append([])
|
|
incomplete.append([record("MKE_s3_repeater", "standard",
|
|
capabilities=("bridge.rs232", "bridge.espnow"))])
|
|
for capability in ("bridge.rs232", "bridge.espnow"):
|
|
incomplete.append([record("MKE_s3_repeater", capabilities=(capability,))])
|
|
unqualified = record("MKE_s3_repeater", capabilities=("bridge.rs232", "bridge.espnow"))
|
|
unqualified["manifest"]["verified"] = False
|
|
incomplete.append([unqualified])
|
|
for capability in ("bridge.rs232", "bridge.espnow"):
|
|
missing = record("MKE_s3_repeater", capabilities=("bridge.rs232", "bridge.espnow"))
|
|
next(check for check in missing["manifest"]["verification"]
|
|
if check["capability"] == capability)["present"] = False
|
|
incomplete.append([missing])
|
|
unproven = record("MKE_s3_repeater", capabilities=("bridge.rs232", "bridge.espnow"))
|
|
unproven["manifest"]["verification"] = []
|
|
incomplete.append([unproven])
|
|
for bridge in ("rs232", "espnow"):
|
|
for normal in incomplete:
|
|
with self.subTest(bridge=bridge, normal=normal):
|
|
with self.assertRaisesRegex(ValueError, "qualified combined repeater"):
|
|
select_ordinary_full_records(normal + [record("MKE_s3_repeater_bridge_" + bridge)])
|
|
|
|
def test_mke_full_primary_requires_verified_bridges_without_legacy_images(self):
|
|
incomplete = [record("MKE_s3_repeater", capabilities=capabilities)
|
|
for capabilities in ((), ("bridge.rs232",), ("bridge.espnow",))]
|
|
unqualified = record("MKE_s3_repeater", capabilities=("bridge.rs232", "bridge.espnow"))
|
|
unqualified["manifest"]["verified"] = False
|
|
incomplete.append(unqualified)
|
|
unproven = record("MKE_s3_repeater", capabilities=("bridge.rs232", "bridge.espnow"))
|
|
unproven["manifest"]["verification"] = []
|
|
incomplete.append(unproven)
|
|
wrong_platform = record("MKE_s3_repeater", platform="NRF52_PLATFORM",
|
|
capabilities=("bridge.rs232", "bridge.espnow"))
|
|
incomplete.append(wrong_platform)
|
|
for capability in ("bridge.rs232", "bridge.espnow"):
|
|
missing = record("MKE_s3_repeater", capabilities=("bridge.rs232", "bridge.espnow"))
|
|
next(check for check in missing["manifest"]["verification"]
|
|
if check["capability"] == capability)["present"] = False
|
|
incomplete.append(missing)
|
|
for primary in incomplete:
|
|
with self.subTest(primary=primary):
|
|
with self.assertRaisesRegex(ValueError, "qualified combined repeater"):
|
|
select_ordinary_full_records([primary])
|
|
|
|
def test_mke_full_contract_does_not_apply_to_other_platforms_or_standard_profiles(self):
|
|
other_platform = record("RAK_4631_repeater", platform="NRF52_PLATFORM")
|
|
other_platform["manifest"].update(verified=None, capabilities=None, verification=None)
|
|
standard_mke = record("MKE_s3_repeater", "standard")
|
|
inputs = [other_platform, standard_mke]
|
|
self.assertEqual(select_ordinary_full_records(inputs), inputs)
|
|
|
|
def test_dedicated_espnow_survives_when_plain_image_lacks_the_capability(self):
|
|
inputs = [record("heltec_v4_tft_repeater"),
|
|
record("heltec_v4_tft_repeater_bridge_espnow", capabilities=("bridge.espnow",))]
|
|
for ordered in (inputs, list(reversed(inputs))):
|
|
self.assertEqual(select_ordinary_full_records(ordered), ordered)
|
|
|
|
def test_qualified_combined_plain_image_replaces_only_its_matching_espnow_image(self):
|
|
primary = record("Heltec_v3_repeater", capabilities=("bridge.espnow",))
|
|
other_board = record("heltec_v4_tft_repeater_bridge_espnow", capabilities=("bridge.espnow",))
|
|
for profile in ("full", "standard"):
|
|
legacy = record("Heltec_v3_repeater_bridge_espnow", profile, capabilities=("bridge.espnow",))
|
|
for inputs in ([primary, legacy, other_board], [legacy, primary, other_board]):
|
|
with self.subTest(profile=profile, primary_first=inputs[0] is primary):
|
|
self.assertEqual(select_ordinary_full_records(inputs), [primary, other_board])
|
|
|
|
def test_matching_dedicated_images_survive_all_missing_driver_proof_cases(self):
|
|
for profile in ("full", "standard"):
|
|
for fault in ("missing", "false", "wrong_capability", "false_verified"):
|
|
primary = record("heltec_v4_tft_repeater", capabilities=("bridge.espnow",))
|
|
if fault == "missing":
|
|
primary["manifest"]["verification"] = []
|
|
elif fault == "false":
|
|
primary["manifest"]["verification"][0]["present"] = False
|
|
elif fault == "wrong_capability":
|
|
primary["manifest"]["verification"][0]["capability"] = "bridge.rs232"
|
|
else:
|
|
primary["manifest"]["verified"] = False
|
|
legacy = record("heltec_v4_tft_repeater_bridge_espnow", profile,
|
|
capabilities=("bridge.espnow",))
|
|
for inputs in ([primary, legacy], [legacy, primary]):
|
|
with self.subTest(profile=profile, fault=fault, primary_first=inputs[0] is primary):
|
|
self.assertCountEqual(select_ordinary_full_records(inputs), inputs)
|
|
|
|
def test_stale_plain_cannot_displace_the_observer_proving_the_combined_driver(self):
|
|
for profile in ("full", "standard"):
|
|
for fault in ("no_capability", "no_proof", "failed_proof", "unverified"):
|
|
stale = record("heltec_v4_tft_repeater", capabilities=("bridge.espnow",))
|
|
if fault == "no_capability":
|
|
stale["manifest"]["capabilities"] = []
|
|
elif fault == "no_proof":
|
|
stale["manifest"]["verification"] = []
|
|
elif fault == "failed_proof":
|
|
stale["manifest"]["verification"][0]["present"] = False
|
|
else:
|
|
stale["manifest"]["verified"] = False
|
|
observer = record("heltec_v4_tft_repeater_observer_mqtt",
|
|
capabilities=("bridge.espnow",))
|
|
legacy = record("heltec_v4_tft_repeater_bridge_espnow", profile,
|
|
capabilities=("bridge.espnow",))
|
|
for inputs in ([stale, observer, legacy], [legacy, observer, stale],
|
|
[observer, legacy, stale], [legacy, stale, observer]):
|
|
with self.subTest(profile=profile, fault=fault, first=inputs[0]["manifest"]["target"]):
|
|
self.assertEqual(select_ordinary_full_records(inputs), [observer])
|
|
|
|
def test_new_non_mqtt_full_repeater_replaces_case_and_underscore_legacy_names(self):
|
|
for primary_name, legacy_name in (
|
|
("heltec_v4_tft_repeater", "heltec_v4_tft_repeater_bridge_espnow"),
|
|
("ThinkNode_M2_repeater", "ThinkNode_M2_Repeater_bridge_espnow"),
|
|
("nibble_zero_connect_repeater_", "nibble_zero_connect_repeater_bridge_espnow_"),
|
|
("Meshadventurer_sx1262_repeater", "Meshadventurer_sx1262_repeater_bridge_espnow"),
|
|
):
|
|
primary = record(primary_name, capabilities=("bridge.espnow",))
|
|
for profile in ("full", "standard"):
|
|
legacy = record(legacy_name, profile, capabilities=("bridge.espnow",))
|
|
for inputs in ([primary, legacy], [legacy, primary]):
|
|
with self.subTest(primary=primary_name, profile=profile):
|
|
self.assertEqual(select_ordinary_full_records(inputs), [primary])
|
|
|
|
def test_combined_observer_with_trailing_underscore_replaces_matching_espnow(self):
|
|
observer = record("LilyGo_TLora_V2_1_1_6_repeater_observer_mqtt_",
|
|
capabilities=("bridge.espnow",))
|
|
legacy = record("LilyGo_TLora_V2_1_1_6_repeater_bridge_espnow",
|
|
capabilities=("bridge.espnow",))
|
|
for inputs in ([observer, legacy], [legacy, observer]):
|
|
self.assertEqual(select_ordinary_full_records(inputs), [observer])
|
|
|
|
def test_unqualified_plain_image_cannot_replace_dedicated_espnow(self):
|
|
primary = record("heltec_v4_tft_repeater", capabilities=("bridge.espnow",))
|
|
primary["manifest"]["verified"] = False
|
|
legacy = record("heltec_v4_tft_repeater_bridge_espnow", capabilities=("bridge.espnow",))
|
|
self.assertEqual(select_ordinary_full_records([primary, legacy]), [primary, legacy])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|