158 lines
7.8 KiB
Python
158 lines
7.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Host-only contracts for the observer recipes imported from 33f1b1096.
|
|
|
|
Resolve inheritance and interpolation without starting PlatformIO or a build.
|
|
The existing board recipes remain authoritative for hardware and OTA policy.
|
|
"""
|
|
import configparser
|
|
import fnmatch
|
|
from pathlib import Path
|
|
import re
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
BOARDS = {
|
|
"ebyte_eora_s3": ("Ebyte_EoRa-S3",),
|
|
"heltec_e213": ("Heltec_E213",),
|
|
"heltec_e290": ("Heltec_E290",),
|
|
"heltec_rc32": ("heltec_rc32", "heltec_rc32_without_display"),
|
|
"heltec_tracker": ("Heltec_Wireless_Tracker",),
|
|
"heltec_wireless_paper": ("Heltec_Wireless_Paper",),
|
|
"lilygo_t3s3_sx1276": ("LilyGo_T3S3_sx1276",),
|
|
"lilygo_teth_elite": ("LilyGo_TETH_Elite_sx1262",),
|
|
"meshnology_w12": ("meshnology_w12",),
|
|
"xiao_s3": ("Xiao_S3",),
|
|
}
|
|
NAMES = tuple(
|
|
f"{board}_{role}_observer_mqtt"
|
|
for boards in BOARDS.values() for board in boards
|
|
for role in ("repeater", "room_server")
|
|
)
|
|
|
|
|
|
class Recipes:
|
|
def __init__(self):
|
|
self.config = configparser.ConfigParser(interpolation=None, strict=True)
|
|
self.config.read([ROOT / "platformio.ini", *sorted(ROOT.glob("variants/*/platformio.ini"))])
|
|
|
|
def value(self, section, key, seen=()):
|
|
pair = (section, key)
|
|
if pair in seen:
|
|
raise AssertionError(f"cyclic recipe reference: {pair}")
|
|
seen = (*seen, pair)
|
|
if not self.config.has_option(section, key):
|
|
for parent in self.config.get(section, "extends", fallback="").split(","):
|
|
parent = parent.strip()
|
|
if parent:
|
|
value = self.value(parent, key, seen)
|
|
if value:
|
|
return value
|
|
return ""
|
|
value = self.config.get(section, key)
|
|
return re.sub(r"\$\{([^}.]+)\.([^}]+)\}",
|
|
lambda m: self.value(m[1], m[2], seen), value)
|
|
|
|
def includes(self, section, source):
|
|
included = False
|
|
for sign, pattern in re.findall(r"([+-])<([^>]+)>", self.value(section, "build_src_filter")):
|
|
# PlatformIO's file glob does not let a root *.cpp select a file
|
|
# outside src. Match path components, then handle directory filters.
|
|
source_parts, pattern_parts = source.split("/"), pattern.split("/")
|
|
file_match = len(source_parts) == len(pattern_parts) and all(
|
|
fnmatch.fnmatchcase(part, glob)
|
|
for part, glob in zip(source_parts, pattern_parts)
|
|
)
|
|
if file_match or source.startswith(pattern.rstrip("/") + "/"):
|
|
included = sign == "+"
|
|
return included
|
|
|
|
|
|
class UpstreamObserverRecipeTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.recipes = Recipes()
|
|
|
|
def test_all_22_named_additions_exist_once(self):
|
|
self.assertEqual(len(NAMES), 22)
|
|
for family, boards in BOARDS.items():
|
|
text = (ROOT / "variants" / family / "platformio.ini").read_text()
|
|
for board in boards:
|
|
for role in ("repeater", "room_server"):
|
|
name = f"{board}_{role}_observer_mqtt"
|
|
with self.subTest(env=name):
|
|
self.assertEqual(text.count(f"[env:{name}]"), 1)
|
|
self.assertTrue(self.recipes.config.has_section(f"env:{name}"))
|
|
|
|
def test_observers_keep_inherited_ota_and_complete_tls_hooks(self):
|
|
expected_hooks = (
|
|
"generate_webconfig_html.py", "meshcore_image_identity.py",
|
|
"portable_esp32_link.py", "esp32_ota_heap_context.py", "merge-bin.py",
|
|
"tools/mota/pio_endf.py", "check_esp32_dram.py", "check_firmware_ram.py",
|
|
"generate_cert_bundle.py",
|
|
)
|
|
for name in NAMES:
|
|
section = f"env:{name}"
|
|
with self.subTest(env=name):
|
|
flags = self.recipes.value(section, "build_flags")
|
|
for flag in ("ESP32_PLATFORM", "ENABLE_OTA=1", "OTA_FLASH_STORE=1",
|
|
"OTA_FOLDER_SERIAL", "WITH_MQTT_BRIDGE=1", "MQTT_MAX_PACKET_SIZE=1024"):
|
|
self.assertIn(flag, flags)
|
|
hooks = self.recipes.value(section, "extra_scripts")
|
|
for hook in expected_hooks:
|
|
self.assertIn(hook, hooks)
|
|
self.assertEqual(self.recipes.value(section, "board_ssl_cert_source"), "adafruit-full")
|
|
self.assertEqual(self.recipes.value(section, "board_build.embed_files"),
|
|
"src/certs/x509_crt_bundle.bin")
|
|
self.assertTrue((ROOT / "scripts/generate_cert_bundle.py").is_file())
|
|
|
|
def test_mqtt_sources_dependencies_and_role_entrypoints(self):
|
|
for name in NAMES:
|
|
section = f"env:{name}"
|
|
role = "simple_room_server" if "_room_server_" in name else "simple_repeater"
|
|
other = "simple_repeater" if role == "simple_room_server" else "simple_room_server"
|
|
with self.subTest(env=name):
|
|
for source in ("helpers/bridges/MQTTBridge.cpp", "helpers/MQTTMessageBuilder.cpp",
|
|
"helpers/JWTHelper.cpp", f"../examples/{role}/main.cpp",
|
|
"helpers/ota/OtaTinf.c"):
|
|
self.assertTrue(self.recipes.includes(section, source), source)
|
|
self.assertFalse(self.recipes.includes(section, f"../examples/{other}/main.cpp"))
|
|
deps = self.recipes.value(section, "lib_deps")
|
|
for dependency in ("PsychicMqttClient", "ArduinoJson @ 7.4.3", "NTPClient",
|
|
"JChristensen/Timezone", "paulstoffregen/Time@1.6.1"):
|
|
self.assertIn(dependency, deps)
|
|
if "WITH_SNMP=1" in self.recipes.value(section, "build_flags"):
|
|
self.assertTrue(self.recipes.includes(section, "helpers/SNMPAgent.cpp"))
|
|
self.assertIn("0neblock/SNMP_Agent", deps)
|
|
if "_room_server_" in name:
|
|
self.assertIn("ROOM_PASSWORD=", self.recipes.value(section, "build_flags"))
|
|
|
|
def test_small_flash_and_non_psram_profiles_keep_upstream_limits(self):
|
|
for board in ("Ebyte_EoRa-S3", "LilyGo_T3S3_sx1276"):
|
|
for role in ("repeater", "room_server"):
|
|
section = f"env:{board}_{role}_observer_mqtt"
|
|
self.assertEqual(self.recipes.value(section, "board_build.partitions"), "min_spiffs.csv")
|
|
self.assertNotIn("WITH_SNMP", self.recipes.value(section, "build_flags"))
|
|
for board in ("Heltec_Wireless_Tracker", "Heltec_Wireless_Paper"):
|
|
for role in ("repeater", "room_server"):
|
|
section = f"env:{board}_{role}_observer_mqtt"
|
|
self.assertIn("MQTT_NEIGHBORS_WITHOUT_PSRAM=1", self.recipes.value(section, "build_flags"))
|
|
|
|
def test_rc32_local_startup_platform_and_safety_remain_inherited(self):
|
|
for board in ("heltec_rc32", "heltec_rc32_without_display"):
|
|
for role in ("repeater", "room_server"):
|
|
section = f"env:{board}_{role}_observer_mqtt"
|
|
flags = self.recipes.value(section, "build_flags")
|
|
self.assertIn("ESP32_POST_BOOT_CPU_FREQ=160", flags)
|
|
self.assertIn("RC32_PERIPHERAL_WARMUP_MS=100", flags)
|
|
self.assertIn("-UENV_INCLUDE_GPS", flags)
|
|
self.assertNotIn("-D ESP32_CPU_FREQ=", flags)
|
|
self.assertNotIn("PIN_GPS_EN=45", flags)
|
|
self.assertIn("55.03.311", self.recipes.value(section, "platform"))
|
|
self.assertIn("3.3.11", self.recipes.value(section, "platform_packages"))
|
|
self.assertEqual(self.recipes.value(section, "board_build.partitions"), "default_16MB.csv")
|
|
self.assertEqual("HELTEC_RC32_WITH_DISPLAY" in flags, board == "heltec_rc32")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|