Merge remote-tracking branch 'origin/keymindCascade' into keymindCascade
This commit is contained in:
mikecarper committed
commit
41405ee01b
54 files changed
+3962
-477
No files matched your search
+11
@@ -138,3 +138,14 @@ public:
|
||||
uint32_t bytesPerCluster() const { return sd_test::medium.cluster_bytes; }
|
||||
SdCard* card() { return sd_test::medium.connected ? &_card : nullptr; }
|
||||
};
|
||||
|
||||
// The optional RAK15002 STORE profile intentionally restricts the card to
|
||||
// FAT16/FAT32. Its production aliases select SdFat32/FatFile; map those
|
||||
// names to distinct classes over the same in-memory filesystem boundary.
|
||||
// These must remain classes, matching the real library's forward declarations.
|
||||
class SdFat32 : public SdFs {};
|
||||
class FatFile : public FsFile {
|
||||
public:
|
||||
FatFile() = default;
|
||||
FatFile(const FsFile& file) : FsFile(file) {}
|
||||
};
|
||||
Vendored
+9
-1
@@ -18,7 +18,7 @@ using namespace mesh::ota;
|
||||
// hardware boundary: no boot authorization, flash writes, or apply call exists
|
||||
// in this executable. The real archive implementation is compiled separately.
|
||||
namespace mesh { namespace ota {
|
||||
OtaStoreSdNrf52::OtaStoreSdNrf52() : _sd(new SdFs), _file(new FsFile) {}
|
||||
OtaStoreSdNrf52::OtaStoreSdNrf52() : _sd(new OtaSdFs), _file(new OtaSdFile) {}
|
||||
OtaStoreSdNrf52::~OtaStoreSdNrf52() { delete _file; delete _sd; }
|
||||
bool OtaStoreSdNrf52::mount() {
|
||||
if (_mounted) return true;
|
||||
@@ -26,6 +26,7 @@ bool OtaStoreSdNrf52::mount() {
|
||||
if (!_mounted) fail("SD mount failed");
|
||||
return _mounted;
|
||||
}
|
||||
bool OtaStoreSdNrf52::busAccessAllowed() const { return _mounted && sd_test::medium.connected; }
|
||||
void OtaStoreSdNrf52::fail(const char* text) { snprintf(_error, sizeof(_error), "%s", text); }
|
||||
bool OtaStoreSdNrf52::begin(uint32_t) { assert(false); return false; }
|
||||
bool OtaStoreSdNrf52::write(uint32_t, const uint8_t*, uint32_t) { assert(false); return false; }
|
||||
@@ -69,6 +70,13 @@ static void published(OtaCacheSdNrf52& cache) {
|
||||
assert(cache.describe(0, d) && !cache.describe(1, d));
|
||||
assert(d.target_id == SIM_TARGET_ID && d.total_size == length && d.block_count == 2);
|
||||
assert(d.block_size_log2 == 11 && d.flags == (MFLAG_FULL | MFLAG_SIGNED));
|
||||
MotaManifest manifest;
|
||||
assert(mota_parse_manifest(bytes + 8, MOTA_MFL, manifest));
|
||||
assert(!memcmp(d.mid, manifest.merkle_root, 4));
|
||||
assert(d.fw_version == manifest.fw_version && d.codec_id == manifest.codec_id);
|
||||
assert(d.source_caps == 0 && d.leaves_off == 8 + MOTA_MFL);
|
||||
assert(d.payload_off == d.leaves_off + manifest.block_count * 4);
|
||||
assert(d.payload_size == manifest.payload_size);
|
||||
std::vector<uint8_t> loaded(length);
|
||||
assert(cache.read(0, 0, loaded.data(), length));
|
||||
assert(!memcmp(loaded.data(), bytes, length));
|
||||
|
||||
@@ -281,6 +281,7 @@ int main(){
|
||||
# implementation is already exercised above. Reuse its method builder.
|
||||
methods = "\n".join(extract_braced(cpp,sig) for sig in (
|
||||
"bool RAK13800EthernetInterface::begin()",
|
||||
"bool RAK13800EthernetInterface::beginHardware()",
|
||||
"bool RAK13800EthernetInterface::terminalShutdown(",
|
||||
"bool RAK13800EthernetInterface::shutdownForPowerOff(",
|
||||
"void RAK13800EthernetInterface::disconnectClient()",
|
||||
@@ -301,6 +302,7 @@ int main(){
|
||||
cls = extract_braced(header,"class RAK13800EthernetInterface")+";"
|
||||
signatures = (
|
||||
"bool RAK13800EthernetInterface::begin()",
|
||||
"bool RAK13800EthernetInterface::beginHardware()",
|
||||
"bool RAK13800EthernetInterface::terminalShutdown(",
|
||||
"bool RAK13800EthernetInterface::shutdownForPowerOff(",
|
||||
"void RAK13800EthernetInterface::disconnectClient()",
|
||||
|
||||
@@ -751,6 +751,48 @@ int main() {
|
||||
with self.assertRaises(ValueError):
|
||||
ram.requirements(platform, definitions, "companion")
|
||||
|
||||
def test_optional_rak_sd_counts_lazy_volume_and_files_without_hiding_source_scratch(self):
|
||||
flags = {"ENABLE_OTA": 1, "OTA_SD_STORE": 1}
|
||||
base = ram.requirements("NRF52_PLATFORM", flags, "RAK_4631_repeater")
|
||||
flags["OTA_RAK15002_SD_STORE"] = 1
|
||||
sd = ram.requirements("NRF52_PLATFORM", flags, "RAK_4631_repeater")
|
||||
self.assertEqual(sd["required_heap_bytes"] - base["required_heap_bytes"], 2048)
|
||||
self.assertEqual(sd["components"]["sd_fat_volume_and_files"], 2048)
|
||||
self.assertEqual(sd["components"]["ota_generic_source_buffers"],
|
||||
base["components"]["ota_generic_source_buffers"])
|
||||
|
||||
def test_optional_sd_room_ethernet_counts_exclusive_worker_but_companion_has_no_worker(self):
|
||||
flags = {"ENABLE_OTA": 1, "OTA_SD_STORE": 1, "OTA_RAK15002_SD_STORE": 1,
|
||||
"ETHERNET_ENABLED": 1, "RAK_4631": 1, "OTA_RAK_AUTO_STORE": 1,
|
||||
"MESHCORE_OTA_DEVICE_DEFLATE": 1}
|
||||
room = ram.requirements("NRF52_PLATFORM", flags, "RAK_4631_room_server")
|
||||
parts = room["components"]
|
||||
self.assertIn("ethernet_or_external_ota_workspaces", parts)
|
||||
self.assertEqual(parts["ethernet_or_external_ota_workspaces"], 15440)
|
||||
self.assertEqual(parts["sd_fat_volume_and_files"], 2048)
|
||||
self.assertNotIn("ota_generic_source_buffers", parts)
|
||||
large = ram.requirements("NRF52_PLATFORM", {**flags, "RAK4631_ETHERNET_TASK_STACK_WORDS": 4096},
|
||||
"RAK_4631_room_server")
|
||||
self.assertEqual(large["components"]["ethernet_or_external_ota_workspaces"], 17216)
|
||||
companion = ram.requirements("NRF52_PLATFORM", {**flags, "OTA_SEEDER_ONLY": 1},
|
||||
"RAK_4631_companion_radio_full")
|
||||
self.assertNotIn("ethernet_or_external_ota_workspaces", companion["components"])
|
||||
self.assertIn("ota_generic_source_buffers", companion["components"])
|
||||
|
||||
def test_only_rak_sd_full_companion_counts_large_name_lookup(self):
|
||||
flags = {"ENABLE_OTA": 1, "OTA_SD_STORE": 1, "OTA_RAK15002_SD_STORE": 1,
|
||||
"OTA_TARGET_NAME_FRONT_CODED": 1, "COMPANION_RADIO_FULL": 1, "OTA_SEEDER_ONLY": 1}
|
||||
full = ram.requirements("NRF52_PLATFORM", flags, "RAK_4631_companion_radio_full")
|
||||
self.assertEqual(full["components"]["ota_target_name_lookup"], 4112)
|
||||
for missing in ("OTA_RAK15002_SD_STORE", "OTA_TARGET_NAME_FRONT_CODED"):
|
||||
other = ram.requirements("NRF52_PLATFORM", {key: value for key, value in flags.items() if key != missing},
|
||||
"RAK_4631_companion_radio_full")
|
||||
self.assertNotIn("ota_target_name_lookup", other["components"])
|
||||
receiver = ram.requirements("NRF52_PLATFORM", {key: value for key, value in flags.items()
|
||||
if key not in ("COMPANION_RADIO_FULL", "OTA_SEEDER_ONLY")},
|
||||
"RAK_4631_repeater_unified_lora_ota")
|
||||
self.assertNotIn("ota_target_name_lookup", receiver["components"])
|
||||
|
||||
def test_queue_sharing_applies_to_qualified_direct_full_builds_only(self):
|
||||
source = '#include "src/helpers/ota/OtaMemoryPolicy.h"\n#ifdef OTA_SHARED_COMPANION_QUEUE\nSHARING_ENABLED\n#endif\n'
|
||||
for flags, expected in (
|
||||
|
||||
@@ -224,6 +224,47 @@ class Nrf52EndfHexTests(unittest.TestCase):
|
||||
stamp(action, path, env)
|
||||
self.assertEqual(path.read_bytes(), before)
|
||||
|
||||
def test_optional_rak_sd_preserves_adaptive_hybrid_and_companion_extrafs_layouts(self):
|
||||
app_base = ml.NRF52_APP_BASE_S140_V6
|
||||
for companion in (False, True):
|
||||
with self.subTest(companion=companion), tempfile.TemporaryDirectory() as directory:
|
||||
env = BuildEnvironment(app_base)
|
||||
env["CPPDEFINES"].extend(["RAK_4631", "OTA_SD_STORE", "OTA_RAK15002_SD_STORE"])
|
||||
if companion:
|
||||
env.source_filter = "+<examples/companion_radio>"
|
||||
env.maximum_size = ml.NRF52_EXTRAFS_START - app_base
|
||||
env["CPPDEFINES"].extend(["OTA_SEEDER_ONLY", "EXTRAFS"])
|
||||
expected_flags = ml.NRF52_LAYOUT_FLAG_INTERNAL_EXTRAFS
|
||||
expected_ceiling = ml.NRF52_EXTRAFS_START
|
||||
else:
|
||||
env["CPPDEFINES"].extend(["OTA_RAK_AUTO_STORE", "OTA_QSPI_STORE", "OTA_FLASH_STORE",
|
||||
"OTA_HYBRID_RAM_STORE", "OTA_INTERNAL_BOOTLOADER_UPDATE"])
|
||||
env.linker = str(ROOT / "boards/nrf52840_s140_v6_mota64.ld")
|
||||
env["LDSCRIPT_PATH"] = env.linker
|
||||
expected_flags = ml.NRF52_LAYOUT_FLAG_AUTO_STORE | ml.NRF52_LAYOUT_FLAG_HYBRID_RAM
|
||||
expected_ceiling = ml.NRF52_APP_END
|
||||
load_post_script(env)
|
||||
path = Path(directory) / "firmware.hex"
|
||||
original = IntelHex()
|
||||
original.frombytes(bytes(range(256)), offset=app_base)
|
||||
original.write_hex_file(str(path))
|
||||
action = env.post_actions["$BUILD_DIR/${PROGNAME}.hex"]
|
||||
stamp(action, path, env)
|
||||
layout = ml.parse_nrf52_layout(read_application(path))
|
||||
self.assertEqual(layout.flags, expected_flags)
|
||||
self.assertEqual(layout.stage_ceiling, expected_ceiling)
|
||||
before = path.read_bytes()
|
||||
for missing in ("RAK_4631", "OTA_SD_STORE"):
|
||||
env["CPPDEFINES"].remove(missing)
|
||||
with self.assertRaisesRegex(RuntimeError, "optional RAK15002"):
|
||||
stamp(action, path, env)
|
||||
self.assertEqual(path.read_bytes(), before)
|
||||
env["CPPDEFINES"].append(missing)
|
||||
env["CPPDEFINES"].remove("OTA_SEEDER_ONLY" if companion else "OTA_RAK_AUTO_STORE")
|
||||
with self.assertRaisesRegex(RuntimeError, "optional RAK15002"):
|
||||
stamp(action, path, env)
|
||||
self.assertEqual(path.read_bytes(), before)
|
||||
|
||||
def test_hex_exact_size_limit_and_smaller_profile_revalidated_after_stamp(self):
|
||||
app_base = ml.NRF52_APP_BASE_S140_V7
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
|
||||
@@ -14,6 +14,7 @@ FIXTURE = ROOT / "test/fixtures/ota_sd_cache"
|
||||
|
||||
|
||||
class OtaSdCacheTests(unittest.TestCase):
|
||||
extra_defines = ()
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++")
|
||||
@@ -27,6 +28,7 @@ class OtaSdCacheTests(unittest.TestCase):
|
||||
compiler, "-std=c++17", "-O1", "-Wall", "-Wextra", "-Werror",
|
||||
"-Wno-unused-parameter", "-Wno-sign-compare", *SANITIZER_FLAGS,
|
||||
"-DNRF52_PLATFORM=1", "-DOTA_SD_STORE=1", "-DOTA_SD_CACHE_MAX=16",
|
||||
*["-D" + define for define in cls.extra_defines],
|
||||
"-I", str(FIXTURE / "mocks"), "-I", str(ROOT / "src"),
|
||||
"-I", str(ROOT / "test/mocks"), str(FIXTURE / "test.cpp"),
|
||||
*[str(ROOT / "src/helpers/ota" / name) for name in source_names],
|
||||
@@ -59,5 +61,11 @@ class OtaSdCacheTests(unittest.TestCase):
|
||||
def test_full_index_refuses_seventeenth_without_losing_resumable_partial(self): self.run_case(10)
|
||||
|
||||
|
||||
class RakOtaSdCacheTests(OtaSdCacheTests):
|
||||
# Run all production capture/serve/reboot/error boundaries against the
|
||||
# lossless bounded RAK descriptor index as well as the original index.
|
||||
extra_defines = ("OTA_RAK15002_SD_STORE=1",)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -91,10 +91,23 @@ class OtaTargetNameTest(unittest.TestCase):
|
||||
def test_cpp_standard_and_compact_match_all_names(self):
|
||||
ids = [target for target, _ in self.rows] + [0, 0xffffffff]
|
||||
expected = [name for _, name in self.rows] + ["?", "?"]
|
||||
for flags in ([], ["-DOTA_TARGET_NAME_FRONT_CODED=1"]):
|
||||
for flags in ([], ["-DOTA_TARGET_NAME_FRONT_CODED=1"],
|
||||
["-DOTA_TARGET_NAME_FRONT_CODED=1", "-DOTA_RAK15002_SD_STORE=1", "-DCOMPANION_RADIO_FULL=1"]):
|
||||
with self.subTest(flags=flags):
|
||||
self.assertEqual(self.compile_and_lookup(flags, ids), expected)
|
||||
|
||||
def test_rak_full_lookup_fails_closed_without_heap_and_keeps_every_name(self):
|
||||
flags = ["-DOTA_TARGET_NAME_FRONT_CODED=1", "-DOTA_RAK15002_SD_STORE=1", "-DCOMPANION_RADIO_FULL=1"]
|
||||
data, offsets, lengths, firsts = gen_targets.compressed_name_chunks(
|
||||
self.rows, gen_targets.RAK_FULL_NAME_DECODE_CHUNK_SIZE)
|
||||
ordinary = gen_targets.compressed_name_chunks(self.rows)
|
||||
self.assertGreater(len(ordinary[0]) - len(data), 1500)
|
||||
self.assertLessEqual(max(lengths), 4096)
|
||||
self.assertEqual(firsts[-1], len(self.rows))
|
||||
self.assertEqual(offsets[-1], len(data))
|
||||
no_heap = '#include <stdlib.h>\n#define malloc(bytes) nullptr\n' + self.text
|
||||
self.assertEqual(self.compile_and_lookup(flags, [self.rows[0][0], 0], no_heap), ["?", "?"])
|
||||
|
||||
def test_table_disabled_preserves_optional_local_name(self):
|
||||
local_id, local_name = self.rows[0]
|
||||
ids = [local_id, self.rows[1][0], 0]
|
||||
@@ -128,8 +141,9 @@ class OtaTargetNameTest(unittest.TestCase):
|
||||
lambda match: match[1] + f"\n 0, {len(stream)}," + match[2],
|
||||
altered, flags=re.S)
|
||||
with self.subTest(stream=stream.hex()):
|
||||
self.assertEqual(self.compile_and_lookup(
|
||||
["-DOTA_TARGET_NAME_FRONT_CODED=1"], [1], altered), ["?"])
|
||||
for flags in (["-DOTA_TARGET_NAME_FRONT_CODED=1"],
|
||||
["-DOTA_TARGET_NAME_FRONT_CODED=1", "-DOTA_RAK15002_SD_STORE=1", "-DCOMPANION_RADIO_FULL=1"]):
|
||||
self.assertEqual(self.compile_and_lookup(flags, [1], altered), ["?"])
|
||||
|
||||
def test_generator_rejects_unencodable_names(self):
|
||||
for name in ("x\0y", "x" * 256):
|
||||
|
||||
@@ -104,6 +104,24 @@ class Rak15002SdSeederTests(unittest.TestCase):
|
||||
recipes = Recipes()
|
||||
self.assertEqual(recipes.config.get(ENV, "extends"), BASE)
|
||||
flags = recipes.value(ENV, "build_flags")
|
||||
unflags = recipes.value(ENV, "build_unflags")
|
||||
|
||||
# The ordinary reduced receiver is now adaptive internal/NOR/SD, but
|
||||
# this historical exact profile must still compile as its distinct
|
||||
# CACHE-only source. Check the effective compiler macro set rather
|
||||
# than treating a parent recipe's library dependency as a child store.
|
||||
def macros(text, directive):
|
||||
return set(re.findall(
|
||||
r"(?:^|\s)-" + directive + r"\s*([A-Za-z_][A-Za-z0-9_]*)",
|
||||
text,
|
||||
))
|
||||
|
||||
effective = macros(flags, "D")
|
||||
effective -= macros(flags, "U")
|
||||
# PlatformIO's build_unflags removes inherited compiler arguments;
|
||||
# -D entries here deliberately fence an adaptive parent macro.
|
||||
effective -= macros(unflags, "D")
|
||||
effective -= macros(unflags, "U")
|
||||
for flag in ("OTA_SEEDER_ONLY=1", "OTA_SD_STORE=1", "OTA_RAK15002_SD_CACHE=1",
|
||||
"OTA_SD_CS_PIN=26", "OTA_SD_SCK_MHZ=4", "OTA_SD_CACHE_MAX=16",
|
||||
"OTA_MAX_SERVE=24", "OTA_HEAP_CONTEXT=1"):
|
||||
@@ -114,13 +132,20 @@ class Rak15002SdSeederTests(unittest.TestCase):
|
||||
self.assertIn(flag, flags)
|
||||
for flag in ("OTA_QSPI_STORE", "OTA_RAK_AUTO_STORE", "OTA_TOWER_AUTO_STORE",
|
||||
"OTA_SD_BOOTLOADER_UPDATE", "COMPANION_RADIO_FULL", "ETHERNET_ENABLED"):
|
||||
self.assertNotRegex(flags, r"-D\s*" + flag + r"(?:=|\s|$)")
|
||||
self.assertNotIn(flag, effective)
|
||||
self.assertNotIn("OTA_RAK15002_SD_STORE", effective)
|
||||
self.assertNotIn("OTA_FLASH_STORE", effective)
|
||||
for flag in ("OTA_SEEDER_ONLY", "OTA_SD_STORE", "OTA_RAK15002_SD_CACHE",
|
||||
"OTA_HEAP_CONTEXT"):
|
||||
self.assertIn(flag, effective)
|
||||
libraries = recipes.value(ENV, "lib_deps")
|
||||
self.assertIn("SdFat-beta.git#a249cfa", libraries)
|
||||
self.assertIn("u-blox", libraries)
|
||||
self.assertIn("SSD1306", libraries)
|
||||
self.assertNotIn("OTA_RAK15002_SD_CACHE", recipes.value(BASE, "build_flags"))
|
||||
self.assertNotIn("SdFat", recipes.value(BASE, "lib_deps"))
|
||||
base_flags = recipes.value(BASE, "build_flags")
|
||||
self.assertIn("OTA_RAK15002_SD_STORE=1", base_flags)
|
||||
self.assertNotIn("OTA_RAK15002_SD_CACHE", base_flags)
|
||||
self.assertIn("SdFat", recipes.value(BASE, "lib_deps"))
|
||||
|
||||
def test_real_bus_keeps_radio_pins_and_cs_safe(self):
|
||||
build(self, BUS_TEST, DEFINES, {"SPI.h": SPI_MOCK})
|
||||
@@ -167,6 +192,7 @@ class EnvironmentSensorManager { public: uint8_t detectWisBlockStorageHolds(); }
|
||||
int main() { EnvironmentSensorManager sensors; assert(sensors.detectWisBlockStorageHolds() == EXPECT_HOLD); }
|
||||
'''.replace("@METHOD@", method)
|
||||
for defines in (["OTA_SD_STORE", "OTA_RAK15002_SD_CACHE", "EXPECT_HOLD=0"],
|
||||
["OTA_SD_STORE", "OTA_RAK15002_SD_STORE", "EXPECT_HOLD=0"],
|
||||
["OTA_SD_STORE", "EXPECT_HOLD=mesh::gps::ExternalStorage"],
|
||||
["OTA_QSPI_SHARED_WISBLOCK_SPI", "EXPECT_HOLD=mesh::gps::ExternalStorage"],
|
||||
["EXPECT_HOLD=0"]):
|
||||
|
||||
@@ -0,0 +1,458 @@
|
||||
"""Execute RAK15002 SD/NOR/Ethernet lifetime guards without hardware or PIO."""
|
||||
from pathlib import Path
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
from test_replay_reset_integration import extract_braced
|
||||
from test_ethernet_terminal_shutdown import SYNC
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
FLAGS = ["NRF52_PLATFORM", "RAK_4631", "OTA_SD_STORE", "OTA_RAK15002_SD_STORE",
|
||||
"SPI_INTERFACES_COUNT=1", "PIN_SPI_MISO=29", "PIN_SPI_SCK=3", "PIN_SPI_MOSI=30",
|
||||
"WB_IO6=10", "RAK4631_COMBINED_ETHERNET"]
|
||||
SPI = r'''
|
||||
#pragma once
|
||||
#include <cassert>
|
||||
#include <cstdint>
|
||||
#include <vector>
|
||||
constexpr int INPUT=0,OUTPUT=1,INPUT_PULLUP=2,LOW=0,HIGH=1;
|
||||
struct Peripheral { unsigned id; };
|
||||
inline Peripheral periph1{1},periph2{2},periph3{3};
|
||||
#define NRF_SPIM1 (&periph1)
|
||||
#define NRF_SPIM2 (&periph2)
|
||||
#define NRF_SPIM3 (&periph3)
|
||||
inline bool inserted=false;
|
||||
inline unsigned begin_count[4]={},end_count[4]={};
|
||||
inline std::vector<int> touched;
|
||||
inline void pinMode(int pin,int){touched.push_back(pin);}
|
||||
inline void digitalWrite(int pin,int){touched.push_back(pin);}
|
||||
inline int digitalRead(int pin){assert(pin==10);return inserted?LOW:HIGH;}
|
||||
struct SPIClass {
|
||||
Peripheral* peripheral; int miso,sck,mosi;
|
||||
SPIClass(Peripheral* p,int i,int c,int o):peripheral(p),miso(i),sck(c),mosi(o){}
|
||||
void begin(){++begin_count[peripheral->id];}
|
||||
void end(){++end_count[peripheral->id];}
|
||||
};
|
||||
inline SPIClass SPI(NRF_SPIM3,45,43,44),SPI1(NRF_SPIM2,9,10,11);
|
||||
'''
|
||||
PRELUDE = r'''
|
||||
#include <helpers/ota/OtaSdSpiBusNrf52.h>
|
||||
#include <cassert>
|
||||
#include <cstring>
|
||||
#include <new>
|
||||
#include <initializer_list>
|
||||
using namespace mesh::ota;
|
||||
'''
|
||||
|
||||
|
||||
def run_native(test, source, files=None, defines=FLAGS, negative=False):
|
||||
compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++")
|
||||
if not compiler:
|
||||
test.skipTest("host C++17 compiler required")
|
||||
with tempfile.TemporaryDirectory(prefix="rak-shared-spi-") as folder:
|
||||
work = Path(folder)
|
||||
sources = {"main.cpp": source, "SPI.h": SPI, "Arduino.h": '#pragma once\n#include "SPI.h"\n'}
|
||||
sources.update(files or {})
|
||||
for name, content in sources.items():
|
||||
(work / name).write_text(content)
|
||||
sanitizer = (["-fsanitize=address,undefined", "-fno-sanitize-recover=all", "-fno-pie", "-no-pie"]
|
||||
if sys.platform.startswith("linux") else [])
|
||||
command = [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror", "-Wno-unused-parameter",
|
||||
*sanitizer, *["-D" + item for item in defines], "-I" + str(work),
|
||||
"-I" + str(ROOT / "src"), *[str(work / name) for name in sources if name.endswith(".cpp")],
|
||||
"-o", str(work / "case")]
|
||||
built = subprocess.run(command, text=True, capture_output=True, timeout=60)
|
||||
test.assertEqual(built.returncode, 0, built.stdout + built.stderr)
|
||||
ran = subprocess.run([str(work / "case")], text=True, capture_output=True, timeout=15)
|
||||
if negative:
|
||||
test.assertNotEqual(ran.returncode, 0)
|
||||
else:
|
||||
test.assertEqual(ran.returncode, 0, ran.stdout + ran.stderr)
|
||||
|
||||
|
||||
class Rak15002SharedSpiTests(unittest.TestCase):
|
||||
def test_shared_nor_ethernet_compile_gate_accepts_only_admitted_runtime_profiles(self):
|
||||
header = (ROOT / "src/helpers/ota/OtaStoreQspiNrf52.h").read_text()
|
||||
guards = header.split("#if defined(OTA_QSPI_SHARED_WISBLOCK_SPI) && defined(RAK_3401)", 1)[0]
|
||||
base = {"NRF52_PLATFORM", "RAK_4631", "OTA_RAK_AUTO_STORE",
|
||||
"OTA_QSPI_SHARED_WISBLOCK_SPI", "ETHERNET_ENABLED"}
|
||||
compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++")
|
||||
if not compiler:
|
||||
self.skipTest("host C++17 compiler required")
|
||||
for extra, missing, admitted in (
|
||||
({"RAK4631_COMBINED_ETHERNET"}, set(), True),
|
||||
({"OTA_RAK15002_SD_STORE", "OTA_SD_STORE"}, set(), True),
|
||||
(set(), set(), False),
|
||||
({"OTA_RAK15002_SD_STORE"}, {"RAK_4631"}, False),
|
||||
({"OTA_RAK15002_SD_STORE"}, {"OTA_RAK_AUTO_STORE"}, False),
|
||||
({"OTA_RAK15002_SD_STORE"}, {"NRF52_PLATFORM"}, False)):
|
||||
with self.subTest(extra=extra, missing=missing):
|
||||
built = subprocess.run([compiler, "-x", "c++", "-fsyntax-only",
|
||||
*["-D" + value for value in sorted((base | extra) - missing)], "-"],
|
||||
input=guards + "\nint main() {}\n", capture_output=True, text=True, timeout=15)
|
||||
self.assertEqual(built.returncode == 0, admitted, built.stderr)
|
||||
|
||||
def test_owner_shared_across_translation_units_no_legacy_setter_steal(self):
|
||||
app = PRELUDE + r'''
|
||||
void* secondOwner();
|
||||
int main(){
|
||||
assert(secondOwner()==&rak4631_shared_spi_owner());
|
||||
assert(!rak4631_try_acquire_spi(RakSharedSpiOwner::None));
|
||||
for(auto device:{RakSharedSpiOwner::Sd,RakSharedSpiOwner::Nor}){
|
||||
assert(rak4631_try_acquire_spi(device));
|
||||
assert(rak4631_try_acquire_spi(device));
|
||||
rak4631_set_ethernet_spi_owner(true);assert(!rak4631_ethernet_owns_spi());
|
||||
rak4631_set_ethernet_spi_owner(false);assert(rak4631_spi_owner()==device);
|
||||
assert(!rak4631_try_acquire_spi(RakSharedSpiOwner::Ethernet));
|
||||
rak4631_release_spi(RakSharedSpiOwner::Ethernet);assert(rak4631_spi_owner()==device);
|
||||
rak4631_release_spi(device);assert(rak4631_spi_owner()==RakSharedSpiOwner::None);
|
||||
}
|
||||
rak4631_set_ethernet_spi_owner(true);assert(rak4631_ethernet_owns_spi());
|
||||
assert(!rak4631_try_acquire_spi(RakSharedSpiOwner::Sd));
|
||||
rak4631_set_ethernet_spi_owner(false);assert(!rak4631_ethernet_owns_spi());
|
||||
}
|
||||
'''
|
||||
other = PRELUDE + "void* secondOwner(){return &rak4631_shared_spi_owner();}"
|
||||
run_native(self, app, {"second.cpp": other})
|
||||
header = (ROOT / "src/helpers/nrf52/Rak4631SharedSpi.h").read_text()
|
||||
broken = header.replace("inline std::atomic<RakSharedSpiOwner>& rak4631_shared_spi_owner()",
|
||||
"static std::atomic<RakSharedSpiOwner>& rak4631_shared_spi_owner()")
|
||||
self.assertNotEqual(header, broken)
|
||||
local_app = app.replace("<helpers/ota/OtaSdSpiBusNrf52.h>", '"Owner.h"')
|
||||
local_other = other.replace("<helpers/ota/OtaSdSpiBusNrf52.h>", '"Owner.h"')
|
||||
run_native(self, local_app, {"second.cpp": local_other, "Owner.h": broken}, negative=True)
|
||||
|
||||
def test_sd_bus_and_active_low_card_detect_never_reconfigure_radio(self):
|
||||
run_native(self, PRELUDE + r'''
|
||||
int main(){
|
||||
assert(&ota_sd_spi_bus()!=&SPI&&&ota_sd_spi_bus()!=&SPI1);
|
||||
assert(ota_sd_spi_bus().peripheral==NRF_SPIM2);
|
||||
assert(ota_sd_spi_bus().miso==29&&ota_sd_spi_bus().sck==3&&ota_sd_spi_bus().mosi==30);
|
||||
assert(SPI.peripheral==NRF_SPIM3&&SPI.miso==45&&SPI.sck==43&&SPI.mosi==44);
|
||||
assert(!ota_rak15002_sd_card_present());inserted=true;assert(ota_rak15002_sd_card_present());
|
||||
for(auto pin:touched)assert(pin==10);
|
||||
assert(!begin_count[1]&&!begin_count[2]&&!begin_count[3]);
|
||||
}
|
||||
''')
|
||||
|
||||
def test_rak_store_uses_fat32_aliases_without_changing_legacy_sd_types(self):
|
||||
# The new optional RAK15002 STORE profile intentionally excludes
|
||||
# exFAT. Existing MeshTower and CACHE-only profiles retain the wider
|
||||
# SdFs/FsFile type pair, so both selections are compiled explicitly.
|
||||
source = (ROOT / "src/helpers/ota/OtaStoreSdNrf52.cpp").read_text()
|
||||
bound_start = source.index("static_assert(sizeof(")
|
||||
bound = source[bound_start:source.index(");", bound_start) + 2]
|
||||
run_native(self, r'''
|
||||
#include <helpers/ota/OtaSdTypesNrf52.h>
|
||||
class SdFat32 {};
|
||||
class FatFile {};
|
||||
#include <type_traits>
|
||||
static_assert(std::is_same<mesh::ota::OtaSdFs, SdFat32>::value, "RAK store must be FAT16/FAT32 only");
|
||||
static_assert(std::is_same<mesh::ota::OtaSdFile, FatFile>::value, "RAK store must use FatFile");
|
||||
@BOUND@
|
||||
int main() {}
|
||||
'''.replace("@BOUND@", bound), {"SdFat.h": '#error "SD public headers must not leak SdFat File into LittleFS callers"\n'},
|
||||
defines=["NRF52_PLATFORM", "OTA_SD_STORE", "OTA_RAK15002_SD_STORE"])
|
||||
run_native(self, r'''
|
||||
#include <helpers/ota/OtaSdTypesNrf52.h>
|
||||
class SdFs {};
|
||||
class FsFile {};
|
||||
#include <type_traits>
|
||||
static_assert(std::is_same<mesh::ota::OtaSdFs, SdFs>::value, "legacy SD must retain SdFs");
|
||||
static_assert(std::is_same<mesh::ota::OtaSdFile, FsFile>::value, "legacy SD must retain FsFile");
|
||||
int main() {}
|
||||
''', defines=["NRF52_PLATFORM", "OTA_SD_STORE"])
|
||||
|
||||
def test_user_gpio_reserves_exact_sd_bus_and_detect_only_for_new_profile(self):
|
||||
run_native(self, r'''
|
||||
#include <helpers/UserGpioPinPolicy.h>
|
||||
#include <cassert>
|
||||
int main(){
|
||||
for(uint8_t pin=0;pin<48;++pin){
|
||||
const bool claimed=pin==3||pin==10||pin==26||pin==29||pin==30;
|
||||
assert(UserGpioPinPolicy::isFirmwareReserved(pin)==claimed);
|
||||
}
|
||||
}
|
||||
''')
|
||||
run_native(self, r'''
|
||||
#include <helpers/UserGpioPinPolicy.h>
|
||||
#include <cassert>
|
||||
int main(){for(uint8_t pin=0;pin<48;++pin)assert(!UserGpioPinPolicy::isFirmwareReserved(pin));}
|
||||
''', defines=[])
|
||||
|
||||
def test_real_sd_mount_retains_owner_failed_acquire_has_no_spi_and_pins_release(self):
|
||||
source = (ROOT / "src/helpers/ota/OtaStoreSdNrf52.cpp").read_text()
|
||||
signatures = ("OtaStoreSdNrf52::OtaStoreSdNrf52()", "OtaStoreSdNrf52::~OtaStoreSdNrf52()",
|
||||
"bool OtaStoreSdNrf52::acquireBus()", "bool OtaStoreSdNrf52::busAccessAllowed() const",
|
||||
"void OtaStoreSdNrf52::releaseBusClaim()", "bool OtaStoreSdNrf52::releaseBus()",
|
||||
"void OtaStoreSdNrf52::fail(", "void OtaStoreSdNrf52::resetStoreState()",
|
||||
"bool OtaStoreSdNrf52::beginCardOnly()", "bool OtaStoreSdNrf52::mount()",
|
||||
"bool OtaStoreSdNrf52::write(", "bool OtaStoreSdNrf52::read(",
|
||||
"uint32_t OtaStoreSdNrf52::capacity() const", "void OtaStoreSdNrf52::checkpoint()")
|
||||
receiver = source[source.index("bool OtaStoreSdNrf52::locate_file()") :]
|
||||
install = ("bool OtaStoreSdNrf52::write(", "bool OtaStoreSdNrf52::read(",
|
||||
"void OtaStoreSdNrf52::checkpoint()")
|
||||
methods = "\n".join(extract_braced(receiver if name in install else source, name) for name in signatures)
|
||||
mock = r'''
|
||||
inline unsigned fs_allocations=0,file_allocations=0,io=0;
|
||||
inline bool mount_ok=true;
|
||||
inline void allowed(){assert(rak4631_spi_owner()==RakSharedSpiOwner::Sd);++io;}
|
||||
struct SdSpiConfig{SdSpiConfig(int,int,int,SPIClass*){}};
|
||||
constexpr int DEDICATED_SPI=0;
|
||||
#define SD_SCK_MHZ(x) (x)
|
||||
#define OTA_SD_SCK_MHZ 4
|
||||
constexpr uint32_t MOTA_SD_AUTH_SECTOR_SIZE=512;
|
||||
struct Card{uint32_t sectorCount(){allowed();return 2048;}bool syncDevice(){allowed();return true;}};
|
||||
class SdFs{public: SdFs(){++fs_allocations;} bool begin(SdSpiConfig){allowed();ota_sd_spi_bus().begin();return mount_ok;}
|
||||
bool cardBegin(SdSpiConfig c){return begin(c);}void end(){allowed();}Card* card(){allowed();static Card c;return &c;}};
|
||||
class FsFile{public:bool opened=false;FsFile(){++file_allocations;}explicit operator bool()const{return opened;}
|
||||
void close(){allowed();opened=false;}bool seekSet(uint32_t){allowed();return true;}
|
||||
size_t write(const uint8_t*,uint32_t n){allowed();return n;}int read(uint8_t*,uint32_t n){allowed();return n;}
|
||||
bool sync(){allowed();return true;}};
|
||||
using SdFat32=SdFs;using FatFile=FsFile;
|
||||
namespace mesh{namespace ota{using OtaSdFs=::SdFat32;using OtaSdFile=::FatFile;}}
|
||||
static void clear_sd_auth_record(){}
|
||||
'''
|
||||
header = (ROOT / "src/helpers/ota/OtaStoreSdNrf52.h").read_text()
|
||||
cls = extract_braced(header, "class OtaStoreSdNrf52 : public OtaStore").replace("private:", "public:") + ";"
|
||||
stubs = r'''
|
||||
bool OtaStoreSdNrf52::begin(uint32_t){return false;}
|
||||
bool OtaStoreSdNrf52::discard(){return false;}void OtaStoreSdNrf52::clear(){}
|
||||
bool OtaStoreSdNrf52::set_meta_size(uint32_t){return false;}
|
||||
bool OtaStoreSdNrf52::finalize(){return false;}bool OtaStoreSdNrf52::reopen(){return false;}
|
||||
bool OtaStoreSdNrf52::plan_layout(bool,uint32_t,uint32_t,uint32_t,bool){return false;}
|
||||
'''
|
||||
runner = r'''
|
||||
int main(){
|
||||
OtaStoreSdNrf52 store;assert(!fs_allocations&&!file_allocations);inserted=true;
|
||||
for(auto peer:{RakSharedSpiOwner::Ethernet,RakSharedSpiOwner::Nor}){
|
||||
assert(rak4631_try_acquire_spi(peer));touched.clear();const unsigned before=io;
|
||||
assert(!store.probeMedia());assert(io==before&&!fs_allocations&&!begin_count[2]);
|
||||
for(auto pin:touched)assert(pin==10);
|
||||
rak4631_release_spi(peer);
|
||||
}
|
||||
inserted=false;assert(!store.probeMedia()&&!fs_allocations);
|
||||
inserted=true;mount_ok=false;assert(!store.probeMedia());
|
||||
assert(rak4631_spi_owner()==RakSharedSpiOwner::None&&begin_count[2]==1&&end_count[2]==1);
|
||||
mount_ok=true;assert(store.probeMedia());assert(rak4631_spi_owner()==RakSharedSpiOwner::Sd);
|
||||
assert(fs_allocations==1&&file_allocations==1&&!begin_count[3]&&!end_count[3]);
|
||||
store._file->opened=true;store._total=20;uint8_t bytes[20]={};
|
||||
assert(store.read(0,bytes,20)&&store.write(0,bytes,20));
|
||||
assert(!store.releaseBus()&&rak4631_spi_owner()==RakSharedSpiOwner::Sd);
|
||||
const unsigned before=io;inserted=false;assert(!store.read(0,bytes,20)&&!store.write(0,bytes,20));
|
||||
store.checkpoint();assert(!store.capacity()&&io==before);
|
||||
store._total=0;assert(store.releaseBus());assert(rak4631_spi_owner()==RakSharedSpiOwner::None);
|
||||
assert(!store._mounted&&!store._file->opened&&end_count[2]==2);
|
||||
assert(rak4631_try_acquire_spi(RakSharedSpiOwner::Ethernet));
|
||||
const unsigned ended=end_count[2];assert(store.releaseBus()&&end_count[2]==ended);
|
||||
assert(rak4631_ethernet_owns_spi());rak4631_set_ethernet_spi_owner(false);
|
||||
}
|
||||
'''
|
||||
code = PRELUDE + '#include <helpers/ota/OtaStore.h>\n' + mock
|
||||
code += "namespace mesh{class MainBoard;namespace ota{" + cls + methods + stubs + "}}\n" + runner
|
||||
run_native(self, code)
|
||||
|
||||
def test_actual_nor_discovery_skips_card_eth_and_sd_owners_without_touching_pads(self):
|
||||
source = (ROOT / "src/helpers/ota/OtaStoreQspiNrf52.cpp").read_text()
|
||||
methods = "\n".join(extract_braced(source, name) for name in (
|
||||
"uint8_t OtaStoreQspiNrf52::autoDetect()",
|
||||
"bool OtaStoreQspiNrf52::headerW25Detected()",
|
||||
"void OtaStoreQspiNrf52::autoProbeIds("))
|
||||
harness = PRELUDE + r'''
|
||||
namespace mesh { namespace ota {
|
||||
static uint8_t auto_cs_pin=0xff,auto_detection=0xff;
|
||||
static uint32_t auto_jedec_id=0,auto_rak_probe_id=0,auto_w25_probe_id=0;
|
||||
static bool auto_header_w25=false;
|
||||
constexpr uint8_t HEADER_SCK_PIN=16,HEADER_IO0_PIN=17,HEADER_IO1_PIN=15;
|
||||
static unsigned gpio=0,probes=0;
|
||||
uint8_t arduino_to_physical(uint32_t pin){return pin;}
|
||||
void nrf_gpio_pin_set(uint32_t){++gpio;}
|
||||
void nrf_gpio_cfg_output(uint32_t){++gpio;}
|
||||
void nrf_gpio_cfg_default(uint32_t){++gpio;}
|
||||
uint32_t probe_nor(uint8_t cs){
|
||||
assert(!inserted&&rak4631_spi_owner()==RakSharedSpiOwner::Nor);++probes;
|
||||
return cs==31?0xef4015:0;
|
||||
}
|
||||
class OtaStoreQspiNrf52 {public:
|
||||
static uint8_t autoDetect();static bool headerW25Detected();
|
||||
static void autoProbeIds(uint32_t&,uint32_t&);
|
||||
};
|
||||
@METHODS@
|
||||
} }
|
||||
int main(){
|
||||
inserted=true;assert(!OtaStoreQspiNrf52::autoDetect()&&!gpio&&!probes);
|
||||
inserted=false;
|
||||
for(auto other:{RakSharedSpiOwner::Ethernet,RakSharedSpiOwner::Sd}){
|
||||
assert(rak4631_try_acquire_spi(other));
|
||||
assert(!OtaStoreQspiNrf52::autoDetect()&&!gpio&&!probes&&auto_detection==0xff);
|
||||
rak4631_release_spi(other);
|
||||
}
|
||||
assert(OtaStoreQspiNrf52::autoDetect()==2&&probes==2&&auto_cs_pin==31&&auto_jedec_id==0xef4015);
|
||||
assert(rak4631_spi_owner()==RakSharedSpiOwner::None);
|
||||
const auto done=gpio;inserted=true;
|
||||
assert(!OtaStoreQspiNrf52::autoDetect()&&gpio==done&&probes==2);
|
||||
uint32_t rak=0,w25=0;OtaStoreQspiNrf52::autoProbeIds(rak,w25);
|
||||
assert(rak==0&&w25==0xef4015&&gpio==done&&probes==2);
|
||||
}
|
||||
'''
|
||||
run_native(self, harness.replace("@METHODS@", methods),
|
||||
defines=FLAGS + ["OTA_RAK_AUTO_STORE"])
|
||||
|
||||
def test_real_archive_handles_recheck_owner_even_when_initialized(self):
|
||||
source = (ROOT / "src/helpers/ota/OtaCacheSdNrf52.cpp").read_text()
|
||||
methods = "\n".join(extract_braced(source, name) for name in (
|
||||
"bool OtaCacheSdNrf52::ready()", "bool OtaCacheSdNrf52::initialize()",
|
||||
"bool OtaCacheSdNrf52::write(uint32_t", "bool OtaCacheSdNrf52::read(uint32_t",
|
||||
"uint32_t OtaCacheSdNrf52::capacity() const", "void OtaCacheSdNrf52::checkpoint()"))
|
||||
harness = PRELUDE + r'''
|
||||
namespace mesh { namespace ota {
|
||||
static unsigned calls=0;
|
||||
static void access(){assert(inserted&&rak4631_spi_owner()==RakSharedSpiOwner::Sd);++calls;}
|
||||
struct FsFile{explicit operator bool()const{return true;}
|
||||
bool seekSet(uint32_t){access();return true;}size_t write(const uint8_t*,uint32_t n){access();return n;}
|
||||
int read(uint8_t*,uint32_t n){access();return n;}bool sync(){access();return true;}};
|
||||
struct Card{uint32_t sectorCount(){access();return 2048;}bool syncDevice(){access();return true;}};
|
||||
struct SdFs{Card* card(){access();static Card c;return &c;}bool exists(const char*){access();return false;}};
|
||||
struct OtaStoreSdNrf52{bool _mounted=true;SdFs fs;SdFs* _sd=&fs;
|
||||
bool busAccessAllowed()const{return inserted&&rak4631_spi_owner()==RakSharedSpiOwner::Sd;}
|
||||
bool mount(){return busAccessAllowed();}const char* last_error()const{return "busy";}};
|
||||
class OtaCacheSdNrf52{public:
|
||||
OtaStoreSdNrf52* _owner;FsFile* _file;uint32_t _total=20;bool _initialized=true,_enabled=true;bool failed=false;char _error[80]={};
|
||||
static constexpr const char* DISABLED_PATH="/mota/cache.off";
|
||||
bool ready();bool initialize();bool write(uint32_t,const uint8_t*,uint32_t);
|
||||
bool read(uint32_t,uint8_t*,uint32_t)const;uint32_t capacity()const;void checkpoint();
|
||||
void fail(const char*){failed=true;}bool ensureDir(){access();return true;}bool rescan(){access();return true;}
|
||||
};
|
||||
@METHODS@
|
||||
} }
|
||||
int main(){
|
||||
OtaStoreSdNrf52 store;FsFile file;OtaCacheSdNrf52 cache;cache._owner=&store;cache._file=&file;
|
||||
inserted=true;uint8_t bytes[20]={};
|
||||
assert(rak4631_try_acquire_spi(RakSharedSpiOwner::Sd));
|
||||
assert(cache.initialize()&&cache.write(0,bytes,20)&&cache.read(0,bytes,20)&&cache.capacity());
|
||||
cache.checkpoint();rak4631_release_spi(RakSharedSpiOwner::Sd);
|
||||
const auto before=calls;
|
||||
for(auto peer:{RakSharedSpiOwner::Ethernet,RakSharedSpiOwner::Nor}){
|
||||
assert(rak4631_try_acquire_spi(peer));
|
||||
assert(!cache.initialize()&&!cache.write(0,bytes,20)&&!cache.read(0,bytes,20)&&!cache.capacity());
|
||||
cache.checkpoint();assert(calls==before);rak4631_release_spi(peer);
|
||||
}
|
||||
assert(rak4631_try_acquire_spi(RakSharedSpiOwner::Sd));inserted=false;
|
||||
assert(!cache.initialize()&&!cache.read(0,bytes,20)&&!cache.capacity());cache.checkpoint();assert(calls==before);
|
||||
}
|
||||
'''
|
||||
for seeder in (False, True):
|
||||
with self.subTest(source_only=seeder):
|
||||
run_native(self, harness.replace("@METHODS@", methods),
|
||||
defines=FLAGS + (["OTA_SEEDER_ONLY"] if seeder else []))
|
||||
|
||||
def test_source_only_volume_cannot_stage_write_resume_or_approve_installation(self):
|
||||
source = (ROOT / "src/helpers/ota/OtaStoreSdNrf52.cpp").read_text()
|
||||
# Preserve the actual preprocessor seam, not extracted method bodies:
|
||||
# capacity must stay available to BOTH vtables, not hidden in #else.
|
||||
gate = source.index("#if defined(OTA_SEEDER_ONLY)\n// The source-only context")
|
||||
start = source.rfind("uint32_t OtaStoreSdNrf52::capacity() const", 0, gate)
|
||||
self.assertGreaterEqual(start, 0, "volume capacity must be outside the receiver-only branch")
|
||||
end = source.index("#endif // OTA_SEEDER_ONLY", gate) + len("#endif // OTA_SEEDER_ONLY")
|
||||
methods = source[start:end]
|
||||
header = (ROOT / "src/helpers/ota/OtaStoreSdNrf52.h").read_text()
|
||||
cls = extract_braced(header, "class OtaStoreSdNrf52 : public OtaStore").replace("private:", "public:") + ";"
|
||||
code = r'''
|
||||
#include <helpers/ota/OtaStore.h>
|
||||
#include <cassert>
|
||||
#include <cstring>
|
||||
static unsigned card_queries=0;
|
||||
struct Card{uint32_t sectorCount(){++card_queries;return 2048;}};
|
||||
class SdFs{public:Card card_value;Card* card(){return &card_value;}};
|
||||
class FsFile;
|
||||
using SdFat32=SdFs;using FatFile=FsFile;
|
||||
namespace mesh{namespace ota{using OtaSdFs=::SdFat32;using OtaSdFile=::FatFile;}}
|
||||
namespace mesh {class MainBoard;namespace ota {
|
||||
static bool ready=false;
|
||||
constexpr uint32_t MOTA_SD_AUTH_SECTOR_SIZE=512;
|
||||
@CLASS@
|
||||
OtaStoreSdNrf52::OtaStoreSdNrf52()=default;
|
||||
OtaStoreSdNrf52::~OtaStoreSdNrf52()=default;
|
||||
bool OtaStoreSdNrf52::busAccessAllowed()const{return ready;}
|
||||
void OtaStoreSdNrf52::fail(const char* message){strncpy(_error,message,sizeof(_error)-1);}
|
||||
@METHODS@
|
||||
} }
|
||||
int main(){
|
||||
mesh::ota::OtaStoreSdNrf52 volume;uint8_t bytes[32]={};
|
||||
assert(!volume.capacity());
|
||||
assert(!volume.plan_layout(true,100,0,100,false));assert(!volume.begin(100));
|
||||
assert(!volume.set_meta_size(8)&&!volume.write(0,bytes,32)&&!volume.read(0,bytes,32));
|
||||
assert(!volume.finalize()&&!volume.reopen()&&!volume.discard());
|
||||
volume.clear();volume.checkpoint();assert(!volume.staged_size());
|
||||
assert(!volume.approve_for_bootloader(bytes,bytes));
|
||||
assert(strstr(volume.last_error(),"source-only"));
|
||||
assert(!card_queries);SdFs fs;volume._sd=&fs;volume._mounted=true;
|
||||
mesh::ota::ready=true;assert(volume.capacity()==1048576&&card_queries==1);
|
||||
}
|
||||
'''
|
||||
# No SdFat, GPIO, retained-memory or flash functions are declared: an
|
||||
# accidental filesystem/hardware side effect is a compile failure.
|
||||
run_native(self, code.replace("@CLASS@", cls).replace("@METHODS@", methods),
|
||||
defines=FLAGS + ["OTA_SEEDER_ONLY"])
|
||||
|
||||
def test_actual_companion_eth_is_opt_in_and_busy_handoff_never_touches_spi(self):
|
||||
stem = ROOT / "src/helpers/ethernet/RAK13800/RAK13800EthernetInterface"
|
||||
cls = extract_braced(stem.with_suffix(".h").read_text(), "class RAK13800EthernetInterface") + ";"
|
||||
cpp = stem.with_suffix(".cpp").read_text()
|
||||
signatures = ("bool RAK13800EthernetInterface::begin()", "bool RAK13800EthernetInterface::beginHardware()",
|
||||
"bool RAK13800EthernetInterface::tryEnable()", "bool RAK13800EthernetInterface::tryDisable()",
|
||||
"bool RAK13800EthernetInterface::terminalShutdown(", "bool RAK13800EthernetInterface::shutdownForPowerOff(",
|
||||
"void RAK13800EthernetInterface::disconnectClient()", "bool RAK13800EthernetInterface::isConnected() const",
|
||||
"int RAK13800EthernetInterface::available()", "int RAK13800EthernetInterface::read()",
|
||||
"size_t RAK13800EthernetInterface::write(", "void RAK13800EthernetInterface::loop()")
|
||||
methods = "\n".join(extract_braced(cpp, name) for name in signatures)
|
||||
methods = methods.replace("uint16_t remotePort = newClient.remotePort();", "(void)newClient.remotePort();")
|
||||
source = SYNC.split("int main(", 1)[0]
|
||||
source = source.replace("#include <Arduino.h>", "#include <Arduino.h>\n#include <helpers/ota/OtaSdSpiBusNrf52.h>")
|
||||
source = source.replace("void pinMode(int,int mode){if(mode==INPUT){assert(!spi);++pads;}}", "")
|
||||
source = source.replace("void digitalWrite(int,int){}", "")
|
||||
source = source.replace("struct SPIClass {void begin(){assert(!spi);spi=true;++begins;} void end(){assert(spi);spi=false;++ends;}} ETHERNET_SPI_PORT;",
|
||||
"struct EthSpi{void begin(){assert(!spi);spi=true;++begins;}void end(){assert(spi);spi=false;++ends;}} ETHERNET_SPI_PORT;")
|
||||
source = source.replace("void init(SPIClass&,int)", "void init(EthSpi&,int)")
|
||||
source = source.replace("void disable(){disconnectClient();++cancels;}", "virtual void enable(){}\n virtual void disable(){disconnectClient();++cancels;}")
|
||||
source = source.replace("@CLASS@", cls).replace("@METHODS@", methods)
|
||||
source = source.replace("assert(spi);++closes;", "assert(spi&&!inserted);++closes;")
|
||||
source += r'''
|
||||
using namespace mesh::ota;
|
||||
static bool busy=false,release_busy=false;
|
||||
static unsigned prepared=0,released=0;
|
||||
bool prepare(void*){++prepared;return !busy;}
|
||||
bool canRelease(void*){return !release_busy;}
|
||||
bool release(void*){assert(!spi&&!rak4631_ethernet_owns_spi());++released;return true;}
|
||||
int main(){
|
||||
RAK13800EthernetInterface owner;assert(owner.begin());owner.enable();
|
||||
assert(!owner.hardwareActive()&&!spi&&!begins&&!owner.tryEnable());
|
||||
owner.setSpiOwnershipHooks(prepare,canRelease,release,nullptr);
|
||||
inserted=true;assert(!owner.tryEnable()&&!prepared&&!begins);
|
||||
inserted=false;busy=true;assert(!owner.tryEnable()&&prepared==1&&!begins);
|
||||
busy=false;assert(rak4631_try_acquire_spi(RakSharedSpiOwner::Sd));
|
||||
assert(!owner.tryEnable()&&!begins&&rak4631_spi_owner()==RakSharedSpiOwner::Sd);
|
||||
rak4631_release_spi(RakSharedSpiOwner::Sd);
|
||||
assert(owner.tryEnable()&&owner.hardwareActive()&&rak4631_ethernet_owns_spi());
|
||||
release_busy=true;assert(!owner.tryDisable()&&owner.hardwareActive()&&ends==0);
|
||||
release_busy=false;assert(owner.tryDisable()&&!owner.hardwareActive()&&ends==1&&closes==4);
|
||||
assert(!rak4631_ethernet_owns_spi()&&released==2);
|
||||
Ethernet.result=0;assert(!owner.tryEnable()&&!owner.hardwareActive()&&ends==2&&!rak4631_ethernet_owns_spi());
|
||||
Ethernet.result=1;assert(owner.tryEnable());inserted=true;const auto calls=reads;
|
||||
assert(owner.available()==0&&owner.read()==-1&&owner.write(nullptr,0)==0&&reads==calls);
|
||||
owner.loop();assert(!owner.hardwareActive()&&rak4631_ethernet_owns_spi()&&reads==calls);
|
||||
assert(!owner.tryEnable());assert(owner.tryDisable()&&!rak4631_ethernet_owns_spi());
|
||||
}
|
||||
'''
|
||||
# The existing fixture's pad counter is irrelevant to the new mapped
|
||||
# GPIO mock; ownership/SPI/socket traffic remains production code.
|
||||
source = source.replace("ends=0,pads=0,cancels", "ends=0,cancels")
|
||||
run_native(self, source)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,341 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Keep the RAK4631 adaptive RAK15002 build recipes scoped and bounded.
|
||||
|
||||
These are recipe-only checks. They source shell helpers with PlatformIO
|
||||
stubbed, so a configuration regression cannot quietly start a firmware build.
|
||||
"""
|
||||
|
||||
import configparser
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import shlex
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
ROOT_CONFIG = ROOT / "platformio.ini"
|
||||
RAK_CONFIG = ROOT / "variants/rak4631/platformio.ini"
|
||||
BUILD_SCRIPT = ROOT / "build_legacy.sh"
|
||||
SEPARATOR = "\n__RAK15002_UNFLAGS__\n"
|
||||
|
||||
STORE_DEFINES = (
|
||||
"OTA_RAK15002_SD_STORE=1",
|
||||
"OTA_SD_STORE=1",
|
||||
"OTA_SD_CS_PIN=26",
|
||||
"OTA_SD_SCK_MHZ=4",
|
||||
"OTA_SD_CACHE_MAX=16",
|
||||
"OTA_MAX_SERVE=24",
|
||||
)
|
||||
RECEIVER_NOR_DEFINES = (
|
||||
"OTA_RAK_AUTO_STORE=1",
|
||||
"OTA_QSPI_STORE=1",
|
||||
"OTA_FLASH_STORE=1",
|
||||
"OTA_QSPI_SHARED_WISBLOCK_SPI=1",
|
||||
"OTA_QSPI_SCK_ARDUINO_PIN=3",
|
||||
"OTA_QSPI_CS_ARDUINO_PIN=31",
|
||||
"OTA_QSPI_IO0_ARDUINO_PIN=30",
|
||||
"OTA_QSPI_IO1_ARDUINO_PIN=29",
|
||||
"OTA_QSPI_IO2_NOT_CONNECTED=1",
|
||||
"OTA_QSPI_IO3_NOT_CONNECTED=1",
|
||||
"OTA_QSPI_SCK_FREQUENCY=NRF_QSPI_FREQ_32MDIV4",
|
||||
"OTA_QSPI_EXPECTED_SIZE=2097152UL",
|
||||
)
|
||||
RECEIVERS = (
|
||||
"RAK_4631_repeater_unified_lora_ota",
|
||||
"RAK_4631_repeater_lora_ota_no_external_sensors",
|
||||
"RAK_4631_room_server_lora_ota_no_external_sensors",
|
||||
"RAK_4631_room_server_ethernet_lora_ota_no_external_sensors",
|
||||
"RAK_4631_sensor_lora_ota_no_external_sensors",
|
||||
)
|
||||
FULL_COMPANION = "RAK_4631_companion_radio_full"
|
||||
INTERNAL_BOOTLOADER_TARGETS = (
|
||||
"RAK_4631_repeater_unified_lora_ota",
|
||||
"RAK_4631_repeater_lora_ota_no_external_sensors",
|
||||
"RAK_4631_room_server_lora_ota_no_external_sensors",
|
||||
"RAK_4631_room_server_ethernet_lora_ota_no_external_sensors",
|
||||
"RAK_4631_sensor_lora_ota_no_external_sensors",
|
||||
)
|
||||
|
||||
|
||||
def read_config(path):
|
||||
config = configparser.ConfigParser(interpolation=None, inline_comment_prefixes=(";",))
|
||||
config.read(path, encoding="utf-8")
|
||||
return config
|
||||
|
||||
|
||||
def has_define(flags, define):
|
||||
return re.search(r"(?:^|\s)-D\s*" + re.escape(define) + r"(?:\s|$)", flags) is not None
|
||||
|
||||
|
||||
def run_shell(script):
|
||||
# Windows' App Execution Alias can resolve `bash` to the WSL launcher,
|
||||
# which accepts the command but loses shell-local assignments. Prefer the
|
||||
# Git-for-Windows Bash used by the build scripts there; ordinary POSIX
|
||||
# runners keep their PATH-provided Bash.
|
||||
git_bash = Path(os.environ.get("ProgramFiles", r"C:\\Program Files")) / "Git/bin/bash.exe"
|
||||
bash = str(git_bash) if os.name == "nt" and git_bash.is_file() else shutil.which("bash")
|
||||
if not bash:
|
||||
raise unittest.SkipTest("Bash is required for build recipe checks")
|
||||
completed = subprocess.run(
|
||||
[bash, "-c", script, "rak15002-recipe-test"],
|
||||
cwd=ROOT,
|
||||
text=True,
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
if completed.returncode:
|
||||
raise AssertionError(
|
||||
"recipe shell failed with %d:\nstdout:\n%s\nstderr:\n%s"
|
||||
% (completed.returncode, completed.stdout, completed.stderr)
|
||||
)
|
||||
return completed.stdout
|
||||
|
||||
|
||||
def effective_flags(target):
|
||||
"""Exercise the real overlay order without consulting/running PlatformIO."""
|
||||
script = r'''
|
||||
set -euo pipefail
|
||||
source ./build_legacy.sh
|
||||
target=@TARGET@
|
||||
# Every configuration query is deliberately false in this isolated test. The
|
||||
# target-specific helper below must supply dynamic room/sensor OTA recipes.
|
||||
pio_env_option_contains() { return 1; }
|
||||
PIO_ENV_PLATFORM_BY_NAME[$target]=NRF52_PLATFORM
|
||||
PIO_ENV_OTA_BY_NAME[$target]=1
|
||||
PIO_ENV_SD_OTA_BY_NAME[$target]=0
|
||||
PIO_ENV_QSPI_OTA_BY_NAME[$target]=0
|
||||
BUILD_PROFILE_EFFECTIVE=standard
|
||||
BUILD_PROFILE_FOR_TARGET=standard
|
||||
SKIP_DECLARED_REDUCTIONS=0
|
||||
NRF52_OTA_SENSOR_PROFILE=reduced
|
||||
PLATFORMIO_BUILD_FLAGS=-DKEEP_RECIPE_FLAG=1
|
||||
PLATFORMIO_BUILD_UNFLAGS=
|
||||
PLATFORMIO_BUILD_SRC_FILTER=
|
||||
BUILD_REDUCTIONS=()
|
||||
if is_companion_radio_full_target "$target"; then
|
||||
apply_companion_radio_full_profile "$target" "$target"
|
||||
else
|
||||
apply_lora_ota_override "$target"
|
||||
fi
|
||||
apply_lora_ota_no_external_sensors_profile "$target"
|
||||
apply_rak4631_rak15002_sd_store_profile "$target"
|
||||
printf '%s\n__RAK15002_UNFLAGS__\n%s' "$PLATFORMIO_BUILD_FLAGS" "$PLATFORMIO_BUILD_UNFLAGS"
|
||||
'''
|
||||
result = run_shell(script.replace("@TARGET@", shlex.quote(target))).split(SEPARATOR)
|
||||
if len(result) != 2:
|
||||
raise AssertionError("unexpected shell recipe result: %r" % result)
|
||||
return result
|
||||
|
||||
|
||||
class Rak4631Rak15002SdRecipeTests(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.root = read_config(ROOT_CONFIG)
|
||||
cls.rak = read_config(RAK_CONFIG)
|
||||
|
||||
def assert_store_defines(self, flags):
|
||||
for define in STORE_DEFINES:
|
||||
with self.subTest(define=define):
|
||||
self.assertTrue(has_define(flags, define), flags)
|
||||
|
||||
def test_root_storage_helpers_pin_the_exact_optional_store(self):
|
||||
store = self.root["nrf52_rak15002_sd_store"]
|
||||
self.assert_store_defines(store["build_flags"])
|
||||
self.assertIn("SdFat-beta.git#a249cfa121e25a96ee14d8813abf01755f1f6a25",
|
||||
store["lib_deps"])
|
||||
|
||||
unified = self.root["nrf52_rak4631_unified_ota"]
|
||||
self.assertIn("${nrf52_wisblock_auto_ota.build_flags}", unified["build_flags"])
|
||||
self.assertIn("${nrf52_rak15002_sd_store.build_flags}", unified["build_flags"])
|
||||
|
||||
def test_static_repeater_and_full_companion_recipes_use_the_right_helper(self):
|
||||
self.assertIn("${nrf52_rak4631_unified_ota.build_flags}",
|
||||
self.rak["env:RAK_4631_repeater_unified_lora_ota"]["build_flags"])
|
||||
self.assertIn("${nrf52_rak4631_unified_ota.build_flags}",
|
||||
self.rak["env:RAK_4631_repeater_lora_ota_no_external_sensors"]
|
||||
["build_flags"])
|
||||
|
||||
companion = self.rak["env:" + FULL_COMPANION]
|
||||
self.assertIn("${nrf52_rak15002_sd_store.build_flags}", companion["build_flags"])
|
||||
for retained in ("ETHERNET_ENABLED=1", "ETHERNET_USE_RAK13800",
|
||||
"ETHERNET_CLASS=RAK13800EthernetInterface",
|
||||
"MESHCORE_REQUIRES_COMPANION_RADIO_FULL=1"):
|
||||
with self.subTest(retained=retained):
|
||||
self.assertIn(retained, companion["build_flags"])
|
||||
self.assertNotIn("OTA_HEAP_CONTEXT", companion["build_flags"])
|
||||
|
||||
# Generated room/sensor pairs compile their ordinary RAK4631 role
|
||||
# environment. Keep the dependency scoped to the identities that can
|
||||
# receive the STORE overlay instead of all RAK4631 roles.
|
||||
for section, helper in (
|
||||
("env:RAK_4631_repeater_unified_lora_ota",
|
||||
"${nrf52_rak4631_unified_ota.lib_deps}"),
|
||||
("env:RAK_4631_repeater_lora_ota_no_external_sensors",
|
||||
"${nrf52_rak4631_unified_ota.lib_deps}"),
|
||||
("env:RAK_4631_room_server", "${nrf52_rak15002_sd_store.lib_deps}"),
|
||||
("env:RAK_4631_room_server_ethernet",
|
||||
"${nrf52_rak15002_sd_store.lib_deps}"),
|
||||
("env:RAK_4631_sensor", "${nrf52_rak15002_sd_store.lib_deps}"),
|
||||
("env:" + FULL_COMPANION, "${nrf52_rak15002_sd_store.lib_deps}")):
|
||||
with self.subTest(section=section):
|
||||
self.assertIn(helper,
|
||||
self.rak[section]["lib_deps"])
|
||||
|
||||
def test_legacy_cache_only_seeder_fences_off_adaptive_receiver_flags(self):
|
||||
seeder = self.rak["env:RAK_4631_repeater_rak15002_sd_seeder_only"]
|
||||
flags = seeder["build_flags"]
|
||||
self.assertTrue(has_define(flags, "OTA_RAK15002_SD_CACHE=1"))
|
||||
self.assertTrue(has_define(flags, "OTA_HEAP_CONTEXT=1"))
|
||||
self.assertNotIn("OTA_RAK15002_SD_STORE", flags)
|
||||
unflags = seeder["build_unflags"]
|
||||
self.assertIn("${nrf52_base.build_unflags}", unflags)
|
||||
for define in ("OTA_RAK_AUTO_STORE=1", "OTA_QSPI_STORE=1",
|
||||
"OTA_RAK15002_SD_STORE=1"):
|
||||
with self.subTest(define=define):
|
||||
self.assertTrue(has_define(unflags, define), unflags)
|
||||
|
||||
def test_all_canonical_receiver_pair_identities_gain_adaptive_store_flags(self):
|
||||
for target in RECEIVERS:
|
||||
with self.subTest(target=target):
|
||||
flags, unflags = effective_flags(target)
|
||||
self.assert_store_defines(flags)
|
||||
for define in RECEIVER_NOR_DEFINES:
|
||||
self.assertTrue(has_define(flags, define), flags)
|
||||
self.assertNotIn("OTA_SEEDER_ONLY", flags)
|
||||
self.assertNotIn("OTA_HEAP_CONTEXT", flags)
|
||||
self.assertNotIn("OTA_RAK15002_SD_STORE", unflags)
|
||||
|
||||
def test_full_companion_is_sd_source_only_without_losing_existing_features(self):
|
||||
flags, unflags = effective_flags(FULL_COMPANION)
|
||||
self.assert_store_defines(flags)
|
||||
for define in ("OTA_SEEDER_ONLY=1", "COMPANION_RADIO_FULL=1",
|
||||
"COMPANION_FEATURE_TEMP_RADIO=1", "COMPANION_FEATURE_OTA_CLI=1",
|
||||
"OTA_SHARED_COMPANION_QUEUE=1"):
|
||||
with self.subTest(define=define):
|
||||
self.assertTrue(has_define(flags, define), flags)
|
||||
self.assertNotIn("OTA_RAK_AUTO_STORE", flags)
|
||||
self.assertNotIn("OTA_QSPI_STORE", flags)
|
||||
self.assertNotIn("OTA_HEAP_CONTEXT", flags)
|
||||
self.assertNotIn("OTA_RAK15002_SD_STORE", unflags)
|
||||
self.assertNotIn("OTA_SD_STORE=1", unflags)
|
||||
|
||||
def test_noncanonical_or_legacy_targets_do_not_receive_store_overlay(self):
|
||||
script = r'''
|
||||
set -euo pipefail
|
||||
source ./build_legacy.sh
|
||||
target=@TARGET@
|
||||
PIO_ENV_PLATFORM_BY_NAME[$target]=NRF52_PLATFORM
|
||||
PLATFORMIO_BUILD_FLAGS=-DKEEP_RECIPE_FLAG=1
|
||||
apply_rak4631_rak15002_sd_store_profile "$target"
|
||||
printf '%s' "$PLATFORMIO_BUILD_FLAGS"
|
||||
'''
|
||||
for target in (
|
||||
"RAK_4631_repeater", "RAK_4631_room_server", "RAK_4631_sensor",
|
||||
"RAK_4631_repeater_bridge_rs232_serial1_lora_ota_no_external_sensors",
|
||||
"RAK_4631_repeater_rak15001_slot_c_lora_ota",
|
||||
"RAK_4631_repeater_rak15002_sd_seeder_only",
|
||||
"RAK_4631_companion_radio_usb"):
|
||||
with self.subTest(target=target):
|
||||
flags = run_shell(script.replace("@TARGET@", shlex.quote(target)))
|
||||
self.assertEqual(flags, "-DKEEP_RECIPE_FLAG=1")
|
||||
|
||||
def test_qualified_repeater_keeps_hybrid_bootloader_linker_request(self):
|
||||
script = r'''
|
||||
set -euo pipefail
|
||||
source ./build_legacy.sh
|
||||
target=RAK_4631_repeater_lora_ota_no_external_sensors
|
||||
PIO_ENV_PLATFORM_BY_NAME[$target]=NRF52_PLATFORM
|
||||
PIO_ENV_QSPI_OTA_BY_NAME[$target]=1
|
||||
PIO_ENV_SD_OTA_BY_NAME[$target]=0
|
||||
is_lora_ota_build() { return 0; }
|
||||
pio_env_option_contains() {
|
||||
[ "$1" = "$target" ] && [ "$2" = build_flags ] && [ "$3" = OTA_RAK_AUTO_STORE ]
|
||||
}
|
||||
unset MESHCORE_NRF52_INTERNAL_BOOTLOADER_UPDATE
|
||||
apply_nrf52_lora_ota_build_recipe "$target" "$target"
|
||||
printf '%s' "${MESHCORE_NRF52_INTERNAL_BOOTLOADER_UPDATE:-0}"
|
||||
'''
|
||||
# The pre-script converts this request into OTA_INTERNAL_BOOTLOADER_UPDATE
|
||||
# and the nrf52840_s140_v6_mota64 linker. The optional RAK15002 must
|
||||
# not demote this already-qualified internal/NOR receiver contract.
|
||||
self.assertEqual(run_shell(script), "1")
|
||||
|
||||
def test_dynamic_sd_receivers_keep_the_frozen_hybrid_request_before_overlay(self):
|
||||
"""Synthetic role aliases must request mota64 before STORE flags exist."""
|
||||
inventory = {
|
||||
line.strip()
|
||||
for line in (ROOT / "tools/mota/nrf52_internal_bootloader_targets.txt")
|
||||
.read_text(encoding="utf-8").splitlines()
|
||||
if line.strip() and not line.lstrip().startswith("#")
|
||||
}
|
||||
self.assertTrue(set(INTERNAL_BOOTLOADER_TARGETS).issubset(inventory))
|
||||
|
||||
script = r'''
|
||||
set -euo pipefail
|
||||
source ./build_legacy.sh
|
||||
target=@TARGET@
|
||||
# These aliases resolve to an ordinary room/sensor PlatformIO base. At the
|
||||
# hybrid-linker decision point it has neither AUTO nor STORE flags yet.
|
||||
PIO_ENV_PLATFORM_BY_NAME[$target]=NRF52_PLATFORM
|
||||
PIO_ENV_QSPI_OTA_BY_NAME[$target]=0
|
||||
PIO_ENV_SD_OTA_BY_NAME[$target]=0
|
||||
PIO_ENV_OTA_BY_NAME[$target]=1
|
||||
NRF52_OTA_SENSOR_PROFILE=reduced
|
||||
BUILD_PROFILE_EFFECTIVE=standard
|
||||
BUILD_PROFILE_FOR_TARGET=standard
|
||||
pio_env_option_contains() { return 1; }
|
||||
unset MESHCORE_NRF52_INTERNAL_BOOTLOADER_UPDATE
|
||||
apply_nrf52_lora_ota_build_recipe "$target" "$target"
|
||||
printf '%s' "${MESHCORE_NRF52_INTERNAL_BOOTLOADER_UPDATE:-0}"
|
||||
'''
|
||||
for target in INTERNAL_BOOTLOADER_TARGETS[2:]:
|
||||
with self.subTest(target=target):
|
||||
# The pre-script turns this request into the exact existing
|
||||
# nrf52840_s140_v6_mota64 linker plus both hybrid flags. The
|
||||
# later optional SD overlay must never weaken that contract.
|
||||
self.assertEqual(
|
||||
run_shell(script.replace("@TARGET@", shlex.quote(target))), "1")
|
||||
|
||||
def test_adaptive_store_is_not_classified_as_an_sd_primary_profile(self):
|
||||
source = BUILD_SCRIPT.read_text(encoding="utf-8")
|
||||
match = re.search(
|
||||
r"python3 -c '\n(.*?)\n' \"\$SUPPORTED_PLATFORM_PATTERN\"",
|
||||
source,
|
||||
flags=re.DOTALL,
|
||||
)
|
||||
self.assertIsNotNone(match, "build context parser missing")
|
||||
config = [
|
||||
["env:RAK_4631_repeater_lora_ota_no_external_sensors", [
|
||||
["board", "rak4631"],
|
||||
["build_flags", ["-D NRF52_PLATFORM", "-D OTA_RAK_AUTO_STORE=1",
|
||||
"-D OTA_QSPI_STORE=1", "-D OTA_SD_STORE=1",
|
||||
"-D OTA_RAK15002_SD_STORE=1"]],
|
||||
["build_src_filter", ["+<helpers/ota/*.cpp>"]],
|
||||
]],
|
||||
["env:RAK_4631_repeater_rak15002_sd_seeder_only", [
|
||||
["board", "rak4631"],
|
||||
["build_flags", ["-D NRF52_PLATFORM", "-D OTA_SD_STORE=1",
|
||||
"-D OTA_RAK15002_SD_CACHE=1"]],
|
||||
["build_src_filter", ["+<helpers/ota/*.cpp>"]],
|
||||
]],
|
||||
]
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-c", match.group(1), "ESP32_PLATFORM|NRF52_PLATFORM"],
|
||||
input=json.dumps(config), text=True, capture_output=True, check=False,
|
||||
)
|
||||
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
||||
rows = {line.split("\t", 1)[0]: line.split("\t")
|
||||
for line in result.stdout.strip().splitlines()}
|
||||
receiver = rows["RAK_4631_repeater_lora_ota_no_external_sensors"]
|
||||
cache = rows["RAK_4631_repeater_rak15002_sd_seeder_only"]
|
||||
self.assertEqual((receiver[4], receiver[5]), ("0", "1"))
|
||||
self.assertEqual(cache[4], "1")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -78,7 +78,7 @@ int main() {
|
||||
def test_context_handoff_frees_idle_self_serve_and_preserves_busy_ota(self):
|
||||
header = (ROOT / "src/helpers/ota/OtaContext.h").read_text()
|
||||
start = header.index(" // Only the main thread may change ownership.")
|
||||
end = header.index("#endif\n\n#if defined(NRF52_PLATFORM) && defined(OTA_SD_STORE)", start)
|
||||
end = header.index("#endif\n\n#if defined(OTA_RAK15002_SD_STORE)", start)
|
||||
body = header[start:end]
|
||||
source = r'''
|
||||
#include <cassert>
|
||||
|
||||
@@ -0,0 +1,183 @@
|
||||
"""Execute the Full RAK SD/Ethernet context handoff without hardware."""
|
||||
from pathlib import Path
|
||||
import unittest
|
||||
|
||||
from test_rak15002_sd_seeder import build
|
||||
from test_replay_reset_integration import extract_braced
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
HARNESS = r'''
|
||||
#include <cassert>
|
||||
#include <cstdlib>
|
||||
#include <cstdint>
|
||||
#include <initializer_list>
|
||||
struct OtaManager { enum { IDLE, FETCHING, COMPLETE, PAUSED }; };
|
||||
struct Store {
|
||||
uint32_t staged = 0;
|
||||
unsigned releases = 0, selections = 0;
|
||||
uint32_t staged_size() const { return staged; }
|
||||
bool releaseBus() { ++releases; return staged == 0; }
|
||||
bool resetSelection() { ++selections; return staged == 0; }
|
||||
};
|
||||
struct Cache { unsigned resets = 0; void resetMedia() { ++resets; } };
|
||||
struct Manager {
|
||||
int state = OtaManager::IDLE;
|
||||
bool own = true, interest = true, accept_full = true;
|
||||
unsigned serve_jobs = 0, manifest_jobs = 0, removals = 0, clears = 0;
|
||||
bool servingPrimaryManifest(const uint8_t*) const { return own; }
|
||||
int fetchState() const { return state; }
|
||||
unsigned pendingServeJobs() const { return serve_jobs; }
|
||||
unsigned pendingManifestJobs() const { return manifest_jobs; }
|
||||
void remove_source(Cache*) { ++removals; }
|
||||
void set_accept_full(bool value) { accept_full = value; }
|
||||
void set_archive_interest(bool value) { interest = value; }
|
||||
void clear_primary() { ++clears; }
|
||||
};
|
||||
struct Context {
|
||||
Manager manager;
|
||||
Store fetch_store, card;
|
||||
Cache sd_cache;
|
||||
uint8_t serve_self_manifest[1] = {};
|
||||
bool apply_pending = false, bootloader_apply_pending = false, _sd_cache_fetching = false;
|
||||
bool serving = true, folder_active = false, fetch_to_folder = false;
|
||||
bool _sd_cache_source_attached = true;
|
||||
void* _folder_source = nullptr;
|
||||
void* folder_dest = nullptr;
|
||||
void* disconnected_folder_dest = nullptr;
|
||||
uint32_t serve_expected = 0, _sd_cache_init_retry_at = 999;
|
||||
uint8_t* serve_self_leaves = nullptr;
|
||||
uint8_t* serve_self_proof = nullptr;
|
||||
unsigned buffer_releases = 0;
|
||||
Store& sdStagingStore() { return card; }
|
||||
void releaseServeBuffer() { ++buffer_releases; }
|
||||
@METHODS@
|
||||
};
|
||||
int main() {
|
||||
for (unsigned busy = 0; busy < 14; ++busy) {
|
||||
Context c;
|
||||
switch (busy) {
|
||||
case 0: c.manager.state = OtaManager::FETCHING; break;
|
||||
case 1: c.manager.state = OtaManager::COMPLETE; break;
|
||||
case 2: c.manager.state = OtaManager::PAUSED; break;
|
||||
case 3: c.fetch_store.staged = 1; break;
|
||||
case 4: c.apply_pending = true; break;
|
||||
case 5: c.bootloader_apply_pending = true; break;
|
||||
case 6: c._sd_cache_fetching = true; break;
|
||||
case 7: c.manager.serve_jobs = 1; break;
|
||||
case 8: c.manager.manifest_jobs = 1; break;
|
||||
case 9: c.manager.own = false; break;
|
||||
case 10: c.serve_expected = 1; break;
|
||||
case 11: c.folder_active = true; break;
|
||||
case 12: c._folder_source = &c; break;
|
||||
case 13: c.folder_dest = &c; break;
|
||||
}
|
||||
assert(!c.canReleaseRakSdForEthernet());
|
||||
assert(!c.prepareRakSdForEthernet() && !c.releasedRakSdEthernet());
|
||||
assert(c.sd_cache.resets == 0 && c.card.releases == 0);
|
||||
assert(c.manager.removals == 0 && c.manager.clears == 0 && c.manager.interest);
|
||||
}
|
||||
for (bool disconnected : {false, true}) {
|
||||
Context c;
|
||||
if (disconnected) c.disconnected_folder_dest = &c;
|
||||
else c.fetch_to_folder = true;
|
||||
assert(!c.prepareRakSdForEthernet() && c.card.releases == 0);
|
||||
}
|
||||
Context c;
|
||||
c.serve_self_leaves = static_cast<uint8_t*>(malloc(8));
|
||||
c.serve_self_proof = static_cast<uint8_t*>(malloc(8));
|
||||
assert(c.canReleaseRakSdForEthernet() && c.prepareRakSdForEthernet());
|
||||
assert(c.card.releases == 1 && c.sd_cache.resets == 1);
|
||||
assert(!c._sd_cache_source_attached && c.manager.removals == 1);
|
||||
assert(!c.serving && !c.serve_self_leaves && !c.serve_self_proof && c.buffer_releases == 1);
|
||||
assert(!c.manager.interest && c._sd_cache_init_retry_at == 0);
|
||||
#ifdef OTA_SEEDER_ONLY
|
||||
assert(c.manager.accept_full && c.fetch_store.selections == 0);
|
||||
#else
|
||||
assert(!c.manager.accept_full && c.fetch_store.selections == 1);
|
||||
#endif
|
||||
assert(c.releasedRakSdEthernet() && c.manager.interest);
|
||||
assert(c.card.releases == 1); // release callback never probes or remounts
|
||||
}
|
||||
'''
|
||||
|
||||
|
||||
class RakSdContextHandoffTests(unittest.TestCase):
|
||||
def test_actual_idle_handoff_and_every_pinned_work_guard(self):
|
||||
source = (ROOT / 'src/helpers/ota/OtaContext.h').read_text()
|
||||
methods = '\n'.join(extract_braced(source, signature) for signature in (
|
||||
'bool canReleaseRakSdForEthernet() const', 'bool prepareRakSdForEthernet()',
|
||||
'bool releasedRakSdEthernet()'))
|
||||
harness = HARNESS.replace('@METHODS@', methods)
|
||||
for seeder in (False, True):
|
||||
with self.subTest(seeder=seeder):
|
||||
build(self, harness, ['OTA_RAK15002_SD_STORE',
|
||||
*(['OTA_SEEDER_ONLY'] if seeder else [])])
|
||||
|
||||
def test_companion_physical_ethernet_and_sd_commands_keep_failed_admission_off(self):
|
||||
source = (ROOT / 'examples/companion_radio/main.cpp').read_text()
|
||||
method = extract_braced(source, 'bool handleRak4631PeripheralCommand(')
|
||||
fixture = r'''
|
||||
#include <cassert>
|
||||
#include <cstdint>
|
||||
#include <cstdio>
|
||||
#include <cstring>
|
||||
struct Board {} board;
|
||||
struct Ethernet {
|
||||
bool active = false, allowed = true;
|
||||
unsigned enables = 0, disables = 0;
|
||||
bool tryEnable() { ++enables; if (!allowed) return false; active = true; return true; }
|
||||
bool tryDisable() { ++disables; if (!allowed) return false; active = false; return true; }
|
||||
bool hardwareActive() const { return active; }
|
||||
const char* last_error() const { return "storage busy"; }
|
||||
} ethernet_interface;
|
||||
namespace mesh { namespace ota {
|
||||
bool admission = true;
|
||||
unsigned acquires = 0;
|
||||
struct Card {
|
||||
unsigned queries = 0;
|
||||
bool okay = true;
|
||||
bool getSpace(Board&, uint64_t& used, uint64_t& free) {
|
||||
++queries; used = 42; free = 4096; return okay;
|
||||
}
|
||||
const char* last_error() const { return "no SD card"; }
|
||||
};
|
||||
struct Context { Card card; Card& sdStagingStore() { return card; } } context;
|
||||
Context& ota_ctx() { assert(admission); return context; }
|
||||
bool ota_acquire_context(char* reply, size_t cap) {
|
||||
++acquires; if (!admission) snprintf(reply, cap, "Error: queue busy"); return admission;
|
||||
}
|
||||
} }
|
||||
@METHOD@
|
||||
int main() {
|
||||
char reply[160];
|
||||
assert(handleRak4631PeripheralCommand("get eth", reply, sizeof reply));
|
||||
assert(strstr(reply, "eth=off") && !ethernet_interface.enables);
|
||||
assert(!handleRak4631PeripheralCommand("get unrelated", reply, sizeof reply));
|
||||
assert(handleRak4631PeripheralCommand("set eth both", reply, sizeof reply));
|
||||
assert(strstr(reply, "Error") && !ethernet_interface.enables);
|
||||
ethernet_interface.allowed = false;
|
||||
assert(handleRak4631PeripheralCommand("set eth on", reply, sizeof reply));
|
||||
assert(strstr(reply, "storage busy") && !ethernet_interface.active);
|
||||
ethernet_interface.allowed = true;
|
||||
assert(handleRak4631PeripheralCommand("set eth on", reply, sizeof reply));
|
||||
assert(strstr(reply, "eth=on") && ethernet_interface.active);
|
||||
assert(handleRak4631PeripheralCommand("get sdcard free", reply, sizeof reply));
|
||||
assert(strstr(reply, "Ethernet off") && !mesh::ota::acquires);
|
||||
assert(handleRak4631PeripheralCommand("set eth off", reply, sizeof reply));
|
||||
assert(!ethernet_interface.active);
|
||||
mesh::ota::admission = false;
|
||||
assert(handleRak4631PeripheralCommand("get sdcard", reply, sizeof reply));
|
||||
assert(strstr(reply, "queue busy") && !mesh::ota::context.card.queries);
|
||||
mesh::ota::admission = true;
|
||||
assert(handleRak4631PeripheralCommand("get sdcard free", reply, sizeof reply));
|
||||
assert(strstr(reply, "used=42 free=4096 bytes") && mesh::ota::context.card.queries == 1);
|
||||
mesh::ota::context.card.okay = false;
|
||||
assert(handleRak4631PeripheralCommand("get sdcard", reply, sizeof reply));
|
||||
assert(strstr(reply, "no SD card"));
|
||||
}
|
||||
'''.replace('@METHOD@', method)
|
||||
build(self, fixture)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -0,0 +1,535 @@
|
||||
"""Native regression for the exact RAK15002 receiver capability and selector."""
|
||||
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
|
||||
def method(source, signature):
|
||||
start = source.index(signature)
|
||||
opening = source.index("{", start)
|
||||
depth, end = 1, opening + 1
|
||||
while depth:
|
||||
depth += (source[end] == "{") - (source[end] == "}")
|
||||
end += 1
|
||||
return source[start:end]
|
||||
|
||||
|
||||
def run_cpp(source, defines=()):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
path = Path(directory)
|
||||
cpp = path / "test.cpp"
|
||||
cpp.write_text(source, encoding="utf-8")
|
||||
exe = path / "test"
|
||||
subprocess.run([
|
||||
"c++", "-std=c++17", "-Wall", "-Wextra", "-Werror",
|
||||
*("-D" + value for value in defines), "-I", str(ROOT),
|
||||
str(cpp), "-o", str(exe),
|
||||
], check=True, timeout=30)
|
||||
subprocess.run([str(exe)], check=True, timeout=10)
|
||||
|
||||
|
||||
class RakSdReceiverTest(unittest.TestCase):
|
||||
def test_actual_apply_requires_exact_sd_loader_and_selects_runtime_handoff(self):
|
||||
apply = (ROOT / "src/helpers/ota/OtaApply.cpp").read_text()
|
||||
functions = "\n".join(method(apply, signature) for signature in (
|
||||
"bool ota_apply_mota_nrf52(OtaStoreSdNrf52& store",
|
||||
"bool ota_apply_mota_nrf52(OtaStoreQspiNrf52& store",
|
||||
"void ota_reboot_to_apply() { // public:",
|
||||
))
|
||||
run_cpp(r'''
|
||||
#include <cassert>
|
||||
#include <cstring>
|
||||
#include "src/helpers/ota/OtaBlInfo.h"
|
||||
#include "src/helpers/ota/OtaFlashLayout_nrf52.h"
|
||||
using namespace mesh::ota;
|
||||
namespace mesh { namespace ota {
|
||||
static bool identity_ok = true, verified = true, clear_ok = true;
|
||||
static const char* identity_name = "4631_DFU";
|
||||
static unsigned verification_calls = 0, resets = 0;
|
||||
static uint8_t request = 0, source = 0;
|
||||
static OtaBlCaps caps;
|
||||
struct OtaBootloaderIdentity { bool crc_ok = true; const char* device_name; };
|
||||
bool ota_installed_bootloader_identity(OtaBootloaderIdentity& out) {
|
||||
out.device_name = identity_name; return identity_ok;
|
||||
}
|
||||
OtaBlCaps appCaps() { return caps; }
|
||||
#define ota_bootloader_app_caps appCaps
|
||||
struct OtaStoreFlashNrf52 {
|
||||
bool hybrid = false;
|
||||
bool is_hybrid() const { return hybrid; }
|
||||
bool publish_hybrid_handoff() { return true; }
|
||||
};
|
||||
static OtaStoreFlashNrf52* g_nrf52_apply_store = nullptr;
|
||||
static uint8_t g_nrf52_external_handoff = GPREGRET2_OTA_STAGE_QSPI;
|
||||
struct OtaStoreSdNrf52 {};
|
||||
struct OtaStoreQspiNrf52 {
|
||||
static bool headerW25Detected() { return false; }
|
||||
uint32_t jedec_id() const { return 0xef4015; }
|
||||
};
|
||||
struct SignerAllowlist {};
|
||||
struct ApplyState {};
|
||||
template<class Store> bool ota_apply_mota_nrf52_external(
|
||||
Store&, const SignerAllowlist&, uint8_t flag, const char*, bool trusted,
|
||||
ApplyState&, char*) {
|
||||
++verification_calls;
|
||||
assert(trusted == (flag == OTA_BL_STORAGE_SD));
|
||||
return verified;
|
||||
}
|
||||
void ota_nrf52_set_reset_handoff(uint8_t req, uint8_t src) { request = req; source = src; }
|
||||
bool ota_nrf52_clear_reset_reasons() { return clear_ok; }
|
||||
void NVIC_SystemReset() { ++resets; }
|
||||
uint32_t simulatedCeiling() { return MOTA_NRF52_STAGE_CEILING_EXPANDED; }
|
||||
#define ota_nrf52_effective_stage_ceiling simulatedCeiling
|
||||
''' + functions + r'''
|
||||
} }
|
||||
int main() {
|
||||
OtaStoreSdNrf52 sd; OtaStoreQspiNrf52 nor; OtaStoreFlashNrf52 internal;
|
||||
SignerAllowlist allow; ApplyState state; char msg[160];
|
||||
caps.optional_sd_app_storage = 0;
|
||||
g_nrf52_apply_store = &internal;
|
||||
assert(!ota_apply_mota_nrf52(sd, allow, state, msg));
|
||||
assert(!g_nrf52_apply_store && verification_calls == 0);
|
||||
caps.optional_sd_app_storage = OTA_BL_STORAGE_SD; identity_ok = false;
|
||||
assert(!ota_apply_mota_nrf52(sd, allow, state, msg) && verification_calls == 0);
|
||||
identity_ok = true; identity_name = "3401_DFU";
|
||||
assert(!ota_apply_mota_nrf52(sd, allow, state, msg) && verification_calls == 0);
|
||||
identity_name = "4631_DFU"; verified = false;
|
||||
assert(!ota_apply_mota_nrf52(sd, allow, state, msg));
|
||||
assert(verification_calls == 1 && g_nrf52_external_handoff == GPREGRET2_OTA_STAGE_QSPI);
|
||||
verified = true;
|
||||
assert(ota_apply_mota_nrf52(sd, allow, state, msg));
|
||||
ota_reboot_to_apply();
|
||||
assert(request == GPREGRET_OTA_APPLY && source == GPREGRET2_OTA_STAGE_SD && resets == 1);
|
||||
assert(ota_apply_mota_nrf52(nor, allow, state, msg));
|
||||
ota_reboot_to_apply();
|
||||
assert(request == GPREGRET_OTA_APPLY && source == GPREGRET2_OTA_STAGE_QSPI && resets == 2);
|
||||
g_nrf52_apply_store = &internal;
|
||||
ota_reboot_to_apply();
|
||||
assert(source == GPREGRET2_OTA_STAGE_EXPANDED && resets == 3);
|
||||
internal.hybrid = true;
|
||||
ota_reboot_to_apply();
|
||||
assert(source == GPREGRET2_OTA_STAGE_HYBRID && resets == 4);
|
||||
clear_ok = false;
|
||||
ota_reboot_to_apply();
|
||||
assert(request == 0 && source == 0xbd && resets == 5);
|
||||
}
|
||||
''', ("OTA_RAK15002_SD_STORE", "OTA_RAK_AUTO_STORE", "OTA_SD_STORE", "OTA_QSPI_STORE"))
|
||||
|
||||
def test_actual_pull_gate_distinguishes_sd_nor_and_internal_boot_packages(self):
|
||||
cli = (ROOT / "src/helpers/ota/OtaCli.cpp").read_text()
|
||||
error = cli.index("ERR bootloader cannot apply from detected QSPI storage")
|
||||
start = cli.rfind("#elif defined(OTA_RAK_AUTO_STORE)", 0, error)
|
||||
end = cli.index("#elif defined(OTA_QSPI_STORE)", error)
|
||||
gate = cli[cli.index("\n", start) + 1:end]
|
||||
run_cpp(r'''
|
||||
#include <cassert>
|
||||
#include <cstdio>
|
||||
#include <cstring>
|
||||
#include <cstdint>
|
||||
constexpr uint8_t OTA_BL_STORAGE_QSPI = 4, OTA_BL_STORAGE_SD = 1, MFLAG_SIGNED = 2;
|
||||
struct Store {
|
||||
int mode;
|
||||
bool usesExternal() const { return mode == 1 || mode == 2; }
|
||||
bool usesQspi() const { return mode == 1; }
|
||||
bool usesSd() const { return mode == 2; }
|
||||
bool usesInternal() const { return mode == 0; }
|
||||
const char* selectionReason() const { return "not prepared"; }
|
||||
};
|
||||
bool rejects(bool selboot, int medium, uint8_t flags, uint8_t sd, uint8_t selflags = 2) {
|
||||
struct { Store fetch_store; } c = {{medium}};
|
||||
struct { uint8_t storage_flags, optional_sd_app_storage; } bl = {flags, sd};
|
||||
char reply[160];
|
||||
''' + gate + r'''
|
||||
return false;
|
||||
}
|
||||
int main() {
|
||||
assert(!rejects(false, 2, 0x1f, 1));
|
||||
assert(rejects(false, 2, 0x1e, 0));
|
||||
assert(rejects(false, 2, 0x1f, 1, 0));
|
||||
assert(!rejects(false, 1, 0x1e, 0));
|
||||
assert(rejects(false, 1, 0x0a, 0));
|
||||
assert(!rejects(false, 0, 0x0a, 0));
|
||||
assert(rejects(false, 3, 0x1f, 1));
|
||||
for (int medium = 0; medium < 4; ++medium)
|
||||
assert(!rejects(true, medium, 0x0a, 0, 0)); // privileged path is internal, never SD
|
||||
}
|
||||
''', ("OTA_RAK15002_SD_STORE",))
|
||||
|
||||
def test_layout_and_status_dispatch_do_not_prepare_optional_storage(self):
|
||||
layout = (ROOT / "src/helpers/StorageLayout.cpp").read_text()
|
||||
start = layout.index("#if defined(ENABLE_OTA) && defined(OTA_RAK15002_SD_STORE)")
|
||||
end = layout.index("#elif defined(ENABLE_OTA) && defined(OTA_RAK_AUTO_STORE)", start)
|
||||
body = layout[layout.index("\n", start) + 1:end]
|
||||
self.assertNotIn("getSpace(", body)
|
||||
self.assertNotIn("ota_acquire_context", body)
|
||||
self.assertNotIn("ota_ctx()", body)
|
||||
run_cpp(r'''
|
||||
#include <cassert>
|
||||
#include <cstdarg>
|
||||
#include <cstdio>
|
||||
#include <cstring>
|
||||
namespace mesh { namespace ota {
|
||||
struct Store {
|
||||
int medium = -1;
|
||||
const char* selectionReason() const { return medium < 0 ? "not probed" : "selected"; }
|
||||
bool usesSd() const { return medium == 2; }
|
||||
bool usesQspi() const { return medium == 1; }
|
||||
bool usesInternal() const { return medium == 0; }
|
||||
unsigned staged_size() const { return 0; }
|
||||
};
|
||||
struct Context { Store fetch_store; };
|
||||
static Context* active = nullptr;
|
||||
const Context* ota_context_if_active() { return active; }
|
||||
} }
|
||||
void appendStorageLayout(char* reply, size_t reply_size, const char* format, ...) {
|
||||
va_list args; va_start(args, format);
|
||||
vsnprintf(reply + strlen(reply), reply_size - strlen(reply), format, args);
|
||||
va_end(args);
|
||||
}
|
||||
void status(char* reply) {
|
||||
int board = 0; size_t reply_size = 160;
|
||||
reply[0] = 0;
|
||||
''' + body + r'''
|
||||
}
|
||||
int main() {
|
||||
char reply[160];
|
||||
status(reply); assert(strstr(reply, "not-prepared"));
|
||||
mesh::ota::Context context; mesh::ota::active = &context;
|
||||
status(reply);
|
||||
#if defined(OTA_SEEDER_ONLY)
|
||||
assert(strstr(reply, "source-only"));
|
||||
#else
|
||||
assert(strstr(reply, "not-prepared"));
|
||||
for (int medium = 0; medium < 3; ++medium) {
|
||||
context.fetch_store.medium = medium;
|
||||
status(reply);
|
||||
assert(strstr(reply, medium == 0 ? "medium=internal" : medium == 1 ? "medium=qspi" : "medium=sd"));
|
||||
}
|
||||
#endif
|
||||
}
|
||||
''')
|
||||
# Full Companion uses the separate stable SD archive, not an adaptive
|
||||
# receiver. Compile the same production status block in both roles.
|
||||
run_cpp(r'''
|
||||
#include <cassert>
|
||||
#include <cstdarg>
|
||||
#include <cstdio>
|
||||
#include <cstring>
|
||||
namespace mesh { namespace ota {
|
||||
struct Context {};
|
||||
const Context* ota_context_if_active() { static Context c; return &c; }
|
||||
} }
|
||||
void appendStorageLayout(char* reply, size_t reply_size, const char* format, ...) {
|
||||
va_list args; va_start(args, format);
|
||||
vsnprintf(reply + strlen(reply), reply_size - strlen(reply), format, args);
|
||||
va_end(args);
|
||||
}
|
||||
int main() {
|
||||
int board = 0; size_t reply_size = 160; char reply[160] = {};
|
||||
''' + body + r'''
|
||||
assert(strstr(reply, "source-only"));
|
||||
}
|
||||
''', ("OTA_SEEDER_ONLY",))
|
||||
cli = (ROOT / "src/helpers/ota/OtaCli.cpp").read_text()
|
||||
dispatcher = method(cli, "bool handle_ota_command(")
|
||||
self.assertLess(dispatcher.index("return handle_status(reply, c)"),
|
||||
dispatcher.index("c.prepareRakStorage()"))
|
||||
self.assertIn('"qspi|storage|self|id|stats|help|?|h"', dispatcher)
|
||||
|
||||
def test_existing_local_combined_loader_advertises_the_exact_new_capability(self):
|
||||
candidate = (ROOT.parent / "Adafruit_nRF52_Bootloader_OTAFIX" /
|
||||
"_build-rak4631-all-in-one-test-20261015/cmake/bootloader.bin")
|
||||
if not candidate.is_file():
|
||||
self.skipTest("Optional previously-built OTAFIX qualification candidate is absent")
|
||||
# Read-only admission of an existing ARM binary, not a bootloader build
|
||||
# or firmware upload. CI need not have the separate loader repository.
|
||||
image = candidate.read_bytes()
|
||||
self.assertEqual(len(image), 0xA000)
|
||||
source = r'''
|
||||
#include <cassert>
|
||||
#include "src/helpers/ota/OtaBlInfo.h"
|
||||
using namespace mesh::ota;
|
||||
static const uint8_t image[] = { @IMAGE@ };
|
||||
int main() {
|
||||
const auto app = ota_bl_app_caps_scan(image, sizeof(image));
|
||||
const auto boot = ota_bl_update_caps_scan_aligned(image, sizeof(image), 0x0a);
|
||||
assert(app.present && app.storage_flags == 0x1f && app.optional_app_storage == 0x14);
|
||||
assert(app.optional_sd_app_storage == 1);
|
||||
assert(boot.present && boot.storage_flags == 0x0a && boot.optional_sd_app_storage == 1);
|
||||
}
|
||||
'''.replace("@IMAGE@", ",".join(str(value) for value in image))
|
||||
run_cpp(source)
|
||||
|
||||
def test_memory_mapped_capabilities_are_scoped_to_rak4631(self):
|
||||
source = r'''
|
||||
#include <cassert>
|
||||
#include <cstring>
|
||||
#include <sys/mman.h>
|
||||
#include "src/helpers/ota/OtaBlInfo.h"
|
||||
using namespace mesh::ota;
|
||||
int main() {
|
||||
auto* flash = static_cast<uint8_t*>(mmap(reinterpret_cast<void*>(uintptr_t(0xf4000)),
|
||||
0xa000, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS | MAP_FIXED, -1, 0));
|
||||
assert(flash == reinterpret_cast<uint8_t*>(uintptr_t(0xf4000)));
|
||||
memset(flash, 0, 0xa000);
|
||||
const uint8_t records[] = {
|
||||
'M','O','T','A','B','L','D','R',3,0,5,0,0x0a,0,0,0,
|
||||
'M','O','T','A','R','A','M','A',1,0,72,0,0,0,1,0,
|
||||
'M','O','T','A','S','T','O','R',1,0,16,0,0x14,0,0,0,
|
||||
'M','O','T','A','S','D','R','A',1,0,72,0,1,0x53,0,0,
|
||||
};
|
||||
memcpy(flash, records, sizeof(records));
|
||||
auto app = ota_bootloader_app_caps();
|
||||
auto boot = ota_bootloader_update_caps();
|
||||
#if defined(RAK_4631) || defined(OTA_RAK15002_SD_STORE)
|
||||
assert(app.storage_flags == 0x1f && app.optional_sd_app_storage == 1);
|
||||
assert(boot.storage_flags == 0x0a && boot.optional_sd_app_storage == 1);
|
||||
#else
|
||||
assert(app.storage_flags == 0x1e && app.optional_sd_app_storage == 0);
|
||||
assert(boot.storage_flags == 0x0a && boot.optional_sd_app_storage == 0);
|
||||
#endif
|
||||
assert(ota_bootloader_self_update_caps_valid(boot));
|
||||
assert(munmap(flash, 0xa000) == 0);
|
||||
}
|
||||
'''
|
||||
for scope in ((), ("RAK_4631",), ("OTA_RAK15002_SD_STORE",)):
|
||||
with self.subTest(scope=scope):
|
||||
run_cpp(source, ("NRF52_PLATFORM", "OTA_RAK_AUTO_STORE",
|
||||
"OTA_INTERNAL_BOOTLOADER_UPDATE", *scope))
|
||||
|
||||
def test_exact_supplement_preserves_legacy_application_and_update_views(self):
|
||||
run_cpp(r'''
|
||||
#include <cassert>
|
||||
#include <cstring>
|
||||
#include <initializer_list>
|
||||
#include "src/helpers/ota/OtaBlInfo.h"
|
||||
using namespace mesh::ota;
|
||||
static const uint8_t boot[16] = {
|
||||
'M','O','T','A','B','L','D','R',3,0,5,0,0x0a,0,0,0,
|
||||
};
|
||||
static const uint8_t bundle[32] = {
|
||||
'M','O','T','A','R','A','M','A',1,0,72,0,0,0,1,0,
|
||||
'M','O','T','A','S','T','O','R',1,0,16,0,0x14,0,0,0,
|
||||
};
|
||||
static const uint8_t sd[16] = {
|
||||
'M','O','T','A','S','D','R','A',1,0,72,0,1,0x53,0,0,
|
||||
};
|
||||
static void make(uint8_t* image) {
|
||||
memset(image, 0, 256);
|
||||
memcpy(image, boot, sizeof(boot));
|
||||
memcpy(image + 32, bundle, sizeof(bundle));
|
||||
memcpy(image + 80, sd, sizeof(sd));
|
||||
}
|
||||
static void check(const uint8_t* image, uint8_t expected) {
|
||||
auto app = ota_bl_app_caps_scan(image, 256);
|
||||
auto update = ota_bl_update_caps_scan_aligned(image, 256, 0x0a);
|
||||
assert(app.present && app.storage_flags == (0x1e | expected));
|
||||
assert(app.optional_app_storage == 0x14 && app.optional_sd_app_storage == expected);
|
||||
assert(update.present && update.storage_flags == 0x0a);
|
||||
assert(update.optional_app_storage == 0x14 && update.optional_sd_app_storage == expected);
|
||||
}
|
||||
int main() {
|
||||
uint8_t image[256];
|
||||
make(image); check(image, 1);
|
||||
// Old loaders retain exactly their old internal/NOR contract.
|
||||
memset(image + 80, 0, sizeof(sd)); check(image, 0);
|
||||
for (unsigned i = 0; i < sizeof(sd); ++i) {
|
||||
make(image); image[80 + i] ^= 1; check(image, 0);
|
||||
}
|
||||
make(image); memcpy(image + 112, sd, sizeof(sd)); check(image, 0);
|
||||
make(image); memcpy(image + 112, sd, sizeof(sd)); image[125] ^= 1; check(image, 0);
|
||||
make(image); memcpy(image + 248, sd, 8); check(image, 0); // truncated duplicate
|
||||
make(image); memmove(image + 81, image + 80, sizeof(sd)); image[80] = 0; check(image, 0);
|
||||
make(image);
|
||||
assert(ota_bl_rak_sd_app_storage(image, 95, 0x0a, 0x14) == 0);
|
||||
assert(ota_bl_rak_sd_app_storage(nullptr, 256, 0x0a, 0x14) == 0);
|
||||
for (uint8_t primary : {uint8_t(9), uint8_t(0x0e), uint8_t(0x1e)}) {
|
||||
make(image); image[12] = primary;
|
||||
auto app = ota_bl_app_caps_scan(image, sizeof(image));
|
||||
assert(app.optional_sd_app_storage == 0 && app.storage_flags == primary);
|
||||
}
|
||||
// Missing/corrupt retained-RAM/NOR metadata cannot borrow the SD marker.
|
||||
for (unsigned i = 0; i < sizeof(bundle); ++i) {
|
||||
make(image); image[32 + i] ^= 1;
|
||||
assert(ota_bl_app_caps_scan(image, sizeof(image)).optional_sd_app_storage == 0);
|
||||
}
|
||||
make(image); memcpy(image + 128, boot, sizeof(boot));
|
||||
assert(!ota_bl_update_caps_scan_aligned(image, sizeof(image), 0x0a).present);
|
||||
}
|
||||
''')
|
||||
|
||||
def test_actual_adaptive_store_defers_io_and_preserves_each_owner(self):
|
||||
header = (ROOT / "src/helpers/ota/OtaStoreAdaptiveNrf52.h").read_text()
|
||||
body = header[header.index("class OtaStoreAdaptiveNrf52"):
|
||||
header.index("} // namespace ota")]
|
||||
source = r'''
|
||||
#include <cassert>
|
||||
#include <cstring>
|
||||
#include <new>
|
||||
#include "src/helpers/ota/OtaStore.h"
|
||||
#include "src/helpers/ota/OtaBlInfo.h"
|
||||
#include "src/helpers/ota/OtaRakStoragePolicy.h"
|
||||
#include "src/helpers/nrf52/Rak4631SharedSpi.h"
|
||||
using namespace mesh::ota;
|
||||
namespace mesh { namespace ota {
|
||||
static unsigned probes = 0, card_checks = 0, sd_probes = 0;
|
||||
static unsigned internal_live = 0, nor_live = 0;
|
||||
static bool card = true, media = true, sd_cap = true, valid_identity = true;
|
||||
static bool release_ok = true;
|
||||
static uint8_t detected = 2;
|
||||
static const char* device_name = "4631_DFU";
|
||||
struct OtaBootloaderIdentity { bool crc_ok = true; const char* device_name; };
|
||||
bool ota_installed_bootloader_identity(OtaBootloaderIdentity& out) {
|
||||
out.device_name = device_name; return valid_identity;
|
||||
}
|
||||
struct MotaManifest { bool is_bootloader() const { return false; } };
|
||||
bool mota_parse_manifest(const uint8_t*, size_t, MotaManifest&) { return false; }
|
||||
OtaBlCaps appCaps() {
|
||||
OtaBlCaps c; c.present = true; c.apply_abi = 3; c.codec_mask = 5;
|
||||
c.storage_flags = sd_cap ? 0x1f : 0x1e; c.optional_app_storage = 0x14;
|
||||
c.optional_sd_app_storage = sd_cap ? 1 : 0; return c;
|
||||
}
|
||||
OtaBlCaps bootCaps() { auto c = appCaps(); c.storage_flags = 0x0a; return c; }
|
||||
bool ota_rak15002_sd_card_present() {
|
||||
assert(!rak4631_ethernet_owns_spi()); ++card_checks; return card;
|
||||
}
|
||||
class Store : public OtaStore {
|
||||
public:
|
||||
uint32_t size = 0;
|
||||
bool begin(uint32_t n) override { size = n; return true; }
|
||||
bool write(uint32_t, const uint8_t*, uint32_t) override { return true; }
|
||||
bool read(uint32_t, uint8_t*, uint32_t) const override { return true; }
|
||||
uint32_t capacity() const override { return 65536; }
|
||||
uint32_t staged_size() const override { return size; }
|
||||
void clear() override { size = 0; }
|
||||
};
|
||||
class OtaStoreFlashNrf52 : public Store {
|
||||
public:
|
||||
OtaStoreFlashNrf52() { ++internal_live; assert(nor_live == 0); }
|
||||
~OtaStoreFlashNrf52() { --internal_live; }
|
||||
bool plan_layout(bool full, uint32_t, uint32_t, uint32_t, bool bootloader) override {
|
||||
return !full || bootloader;
|
||||
}
|
||||
};
|
||||
class OtaStoreQspiNrf52 : public Store {
|
||||
public:
|
||||
OtaStoreQspiNrf52() { ++nor_live; assert(internal_live == 0); }
|
||||
~OtaStoreQspiNrf52() { --nor_live; }
|
||||
static uint8_t autoDetect() { assert(!card && !rak4631_ethernet_owns_spi()); ++probes; return detected; }
|
||||
static bool headerW25Detected() { return false; }
|
||||
uint32_t jedec_id() const { return 0xef4015; }
|
||||
uint8_t status1() const { return 0; }
|
||||
const char* last_stage() const { return "idle"; }
|
||||
const char* last_error() const { return ""; }
|
||||
};
|
||||
class OtaStoreSdNrf52 : public Store {
|
||||
public:
|
||||
bool probeMedia() { assert(card && !rak4631_ethernet_owns_spi()); ++sd_probes; return media; }
|
||||
bool releaseBus() { return release_ok && size == 0; }
|
||||
const char* last_error() const { return "SD test"; }
|
||||
};
|
||||
#define ota_bootloader_app_caps appCaps
|
||||
#define ota_bootloader_update_caps bootCaps
|
||||
''' + body + r'''
|
||||
} }
|
||||
int main() {
|
||||
{
|
||||
OtaStoreAdaptiveNrf52 store;
|
||||
auto* stable_sd = &store.sdStore();
|
||||
assert(!store.usesSd() && !store.usesQspi() && !store.usesInternal());
|
||||
assert(!store.usesExternal() && store.capacity() == 0 && store.staged_size() == 0);
|
||||
assert(card_checks == 0 && sd_probes == 0 && probes == 0);
|
||||
assert(store.selectStorage() && store.usesSd() && store.usesExternal());
|
||||
assert(sd_probes == 1 && probes == 0);
|
||||
assert(store.plan_layout(true, 100000, 365, 100000, false));
|
||||
assert(store.begin(100370));
|
||||
assert(!store.resetSelection());
|
||||
// A malformed caller cannot grant a previously selected external store
|
||||
// access after Ethernet took ownership, with or without the old combined
|
||||
// recipe flag. Its staged data must remain untouched.
|
||||
rak4631_set_ethernet_spi_owner(true);
|
||||
uint8_t byte = 0;
|
||||
assert(!store.begin(200) && !store.write(0, &byte, 1) && !store.read(0, &byte, 1));
|
||||
assert(store.capacity() == 0);
|
||||
store.clear();
|
||||
assert(stable_sd->staged_size() == 100370);
|
||||
rak4631_set_ethernet_spi_owner(false);
|
||||
assert(!store.plan_layout(true, 40960, 365, 40960, true)); // never abandon SD data
|
||||
assert(store.usesSd() && stable_sd->staged_size() == 100370);
|
||||
store.clear();
|
||||
assert(store.plan_layout(true, 40960, 365, 40960, true));
|
||||
assert(store.usesInternal() && &store.sdStore() == stable_sd);
|
||||
assert(store.begin(41330) && !store.resetSelection());
|
||||
store.clear();
|
||||
assert(store.plan_layout(true, 100000, 365, 100000, false) && store.usesSd());
|
||||
release_ok = false;
|
||||
assert(!store.resetSelection() && store.usesSd());
|
||||
release_ok = true;
|
||||
assert(store.resetSelection() && !store.usesSd());
|
||||
assert(&store.sdStore() == stable_sd);
|
||||
}
|
||||
assert(internal_live == 0 && nor_live == 0);
|
||||
probes = sd_probes = card_checks = 0; card = false;
|
||||
{
|
||||
OtaStoreAdaptiveNrf52 store;
|
||||
assert(store.selectStorage() && store.usesQspi() && !store.usesSd());
|
||||
assert(probes == 1 && sd_probes == 0);
|
||||
assert(store.begin(100));
|
||||
assert(!store.resetSelection());
|
||||
store.clear(); release_ok = false;
|
||||
assert(!store.resetSelection() && store.usesQspi() && nor_live == 1);
|
||||
release_ok = true; assert(store.resetSelection() && nor_live == 0);
|
||||
}
|
||||
card = true; sd_cap = false; probes = sd_probes = 0;
|
||||
{
|
||||
OtaStoreAdaptiveNrf52 store;
|
||||
assert(store.selectStorage() && store.usesInternal());
|
||||
assert(strstr(store.selectionReason(), "newer OTAFIX"));
|
||||
assert(probes == 0 && sd_probes == 0); // don't clock SD or same-CS NOR
|
||||
assert(!store.plan_layout(true, 100000, 365, 100000, false));
|
||||
}
|
||||
sd_cap = true; media = false;
|
||||
{
|
||||
OtaStoreAdaptiveNrf52 store;
|
||||
assert(store.selectStorage() && store.usesInternal());
|
||||
assert(probes == 0 && sd_probes == 1);
|
||||
}
|
||||
media = true; valid_identity = false; sd_probes = 0;
|
||||
{
|
||||
OtaStoreAdaptiveNrf52 store;
|
||||
assert(!store.selectStorage());
|
||||
assert(!store.usesExternal() && !store.usesInternal());
|
||||
assert(sd_probes == 0 && probes == 0);
|
||||
}
|
||||
valid_identity = true; device_name = "3401_DFU";
|
||||
{
|
||||
OtaStoreAdaptiveNrf52 store;
|
||||
assert(!store.selectStorage() && sd_probes == 0 && probes == 0);
|
||||
}
|
||||
device_name = "4631_DFU"; card_checks = 0;
|
||||
rak4631_set_ethernet_spi_owner(true);
|
||||
{
|
||||
OtaStoreAdaptiveNrf52 store;
|
||||
assert(store.selectStorage() && store.usesInternal());
|
||||
assert(card_checks == 0 && sd_probes == 0 && probes == 0);
|
||||
}
|
||||
rak4631_set_ethernet_spi_owner(false);
|
||||
assert(internal_live == 0 && nor_live == 0);
|
||||
}
|
||||
'''
|
||||
for legacy_combined in (False, True):
|
||||
with self.subTest(legacy_combined=legacy_combined):
|
||||
defines = ["OTA_RAK15002_SD_STORE", "OTA_INTERNAL_BOOTLOADER_UPDATE",
|
||||
"OTA_RAK_AUTO_STORE"]
|
||||
if legacy_combined:
|
||||
defines.append("RAK4631_COMBINED_ETHERNET")
|
||||
run_cpp(source, defines)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,211 @@
|
||||
"""Optional-SD room Ethernet must use the guarded, opt-in runtime."""
|
||||
|
||||
from pathlib import Path
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
from test_rak4631_ethernet_runtime import FIXTURE, SCENARIOS
|
||||
from test_rak_sd_receiver import method, run_cpp
|
||||
from test_serial_wifi_sessions import SANITIZER_FLAGS
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
|
||||
class RakSdRoomEthernetTest(unittest.TestCase):
|
||||
def test_actual_role_callbacks_reserve_only_after_storage_release(self):
|
||||
source = (ROOT / "examples/simple_room_server/main.cpp").read_text()
|
||||
callbacks = "\n".join(method(source, signature) for signature in (
|
||||
"static bool prepareSdEthernet(",
|
||||
"static bool canReleaseSdEthernet(",
|
||||
"static bool releaseSdEthernet(",
|
||||
))
|
||||
harness = r'''
|
||||
#include <cassert>
|
||||
#include <cstdio>
|
||||
#include <cstring>
|
||||
#include "src/helpers/nrf52/Rak4631SharedSpi.h"
|
||||
namespace mesh { namespace ota {
|
||||
static bool fitted = false, prepare_ok = true, release_ok = true;
|
||||
static unsigned prepares = 0, releases = 0;
|
||||
struct Context {
|
||||
bool prepareRakSdForEthernet() {
|
||||
++prepares;
|
||||
assert(!rak4631_ethernet_owns_spi());
|
||||
return prepare_ok;
|
||||
}
|
||||
bool canReleaseRakSdForEthernet() { return release_ok; }
|
||||
bool releasedRakSdEthernet() {
|
||||
++releases;
|
||||
assert(rak4631_ethernet_owns_spi()); // Keep reservation until callback completes.
|
||||
return release_ok;
|
||||
}
|
||||
};
|
||||
static Context context;
|
||||
static Context* active = nullptr;
|
||||
Context* ota_context_if_active() { return active; }
|
||||
bool ota_rak15002_sd_card_present() { return fitted; }
|
||||
} }
|
||||
''' + callbacks + r'''
|
||||
int main() {
|
||||
using namespace mesh::ota;
|
||||
char error[160];
|
||||
assert(!prepareSdEthernet(nullptr, error, sizeof(error)));
|
||||
assert(strstr(error, "not ready") && prepares == 0);
|
||||
active = &context;
|
||||
fitted = true;
|
||||
assert(!prepareSdEthernet(nullptr, error, sizeof(error)));
|
||||
assert(strstr(error, "SD card fitted") && prepares == 0);
|
||||
fitted = false; prepare_ok = false;
|
||||
assert(!prepareSdEthernet(nullptr, error, sizeof(error)));
|
||||
assert(rak4631_spi_owner() == RakSharedSpiOwner::None);
|
||||
prepare_ok = true;
|
||||
assert(rak4631_try_acquire_spi(RakSharedSpiOwner::Nor));
|
||||
assert(!prepareSdEthernet(nullptr, error, sizeof(error)));
|
||||
assert(rak4631_spi_owner() == RakSharedSpiOwner::Nor); // Never steal uncertain NOR.
|
||||
rak4631_release_spi(RakSharedSpiOwner::Nor);
|
||||
assert(prepareSdEthernet(nullptr, error, sizeof(error)));
|
||||
assert(rak4631_ethernet_owns_spi());
|
||||
release_ok = false;
|
||||
assert(!canReleaseSdEthernet(nullptr));
|
||||
assert(!releaseSdEthernet(nullptr) && rak4631_ethernet_owns_spi());
|
||||
active = nullptr;
|
||||
assert(!releaseSdEthernet(nullptr) && rak4631_ethernet_owns_spi());
|
||||
active = &context; release_ok = true;
|
||||
assert(canReleaseSdEthernet(nullptr) && releaseSdEthernet(nullptr));
|
||||
assert(rak4631_spi_owner() == RakSharedSpiOwner::None && releases == 2);
|
||||
}
|
||||
'''
|
||||
run_cpp(harness, ("OTA_RAK15002_SD_STORE",))
|
||||
repeater = (ROOT / "examples/simple_repeater/main.cpp").read_text()
|
||||
combined = "\n".join(method(repeater, signature) for signature in (
|
||||
"static bool prepareCombinedEthernet(",
|
||||
"static bool canReleaseCombinedEthernet(",
|
||||
"static bool releaseCombinedEthernet(",
|
||||
))
|
||||
for original, replacement in (
|
||||
("prepareCombinedEthernet", "prepareSdEthernet"),
|
||||
("canReleaseCombinedEthernet", "canReleaseSdEthernet"),
|
||||
("releaseCombinedEthernet", "releaseSdEthernet"),
|
||||
):
|
||||
combined = combined.replace(original, replacement)
|
||||
# No legacy NOR hooks exist in this mock. A wrong branch selected
|
||||
# under both flags fails compilation before the ownership assertions.
|
||||
run_cpp(harness.replace(callbacks, combined),
|
||||
("OTA_RAK15002_SD_STORE", "RAK4631_COMBINED_ETHERNET"))
|
||||
|
||||
def test_actual_runtime_without_legacy_combined_flag(self):
|
||||
compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++")
|
||||
if compiler is None:
|
||||
self.skipTest("a host C++17 compiler is required")
|
||||
with tempfile.TemporaryDirectory(prefix="rak-sd-room-ethernet-") as directory:
|
||||
binary = Path(directory) / "runtime"
|
||||
fixture = (FIXTURE / "test.cpp").read_text()
|
||||
fixture = fixture.replace("#include <helpers/nrf52/EthernetCLI.h>",
|
||||
"static std::atomic<bool> mock_sd_fitted{false};\n"
|
||||
"#include <helpers/nrf52/EthernetCLI.h>")
|
||||
fixture = fixture.replace(" hooks.context = &context;",
|
||||
" hooks.context = &context;\n"
|
||||
" hooks.storage_conflict = [](void*) { return mock_sd_fitted.load(); };")
|
||||
cases = r'''
|
||||
case 35: { // Card insertion gates main loop and retains owner until explicit OFF.
|
||||
start_online(); accept_connection();
|
||||
EthernetSpiAccess expected = EthernetSpiAccess::Idle;
|
||||
wait_until([&] { expected = EthernetSpiAccess::Idle;
|
||||
return ethernet_spi_access.compare_exchange_strong(expected, EthernetSpiAccess::Client); });
|
||||
const int before = hardware_operations.load();
|
||||
mock_sd_fitted.store(true);
|
||||
ethernet_loop_maintain();
|
||||
ethernet_spi_access.store(EthernetSpiAccess::Idle);
|
||||
wait_until([] { return ethernet_worker_stopped.load(); });
|
||||
assert(ethernet_state.load() == EthernetCliState::Faulted);
|
||||
assert(!spi_active.load() && bus_owner.load() && ethernet_get_enabled());
|
||||
assert(hardware_operations.load() == before + 1); // SPI.end only: no socket CLOSE/SEND.
|
||||
ethernet_loop_maintain();
|
||||
assert(bus_owner.load() && context.releases == 0);
|
||||
assert(command("eth on").find("off first") != std::string::npos);
|
||||
context.off_allowed = false;
|
||||
assert(command("eth off").find("locked") != std::string::npos);
|
||||
assert(bus_owner.load());
|
||||
context.off_allowed = true;
|
||||
finish_off();
|
||||
break;
|
||||
}
|
||||
case 36: { // An entered vendor DHCP call completes; no further vendor IO follows insertion.
|
||||
Ethernet.begin_blocked.store(true);
|
||||
assert(command("eth on").find("starting") != std::string::npos);
|
||||
wait_until([] { return Ethernet.begin_count.load() == 1; });
|
||||
const int before = hardware_operations.load();
|
||||
mock_sd_fitted.store(true);
|
||||
unblock(Ethernet.begin_blocked);
|
||||
wait_until([] { return ethernet_worker_stopped.load(); });
|
||||
assert(hardware_operations.load() == before + 1);
|
||||
assert(!spi_active.load() && bus_owner.load());
|
||||
finish_off();
|
||||
break;
|
||||
}
|
||||
case 37: { // Retained Stream admission detects insertion before any controller register access.
|
||||
start_online(); accept_connection();
|
||||
EthernetSpiAccess expected = EthernetSpiAccess::Idle;
|
||||
wait_until([&] { expected = EthernetSpiAccess::Idle;
|
||||
return ethernet_spi_access.compare_exchange_strong(expected, EthernetSpiAccess::Client); });
|
||||
const int before = hardware_operations.load();
|
||||
mock_sd_fitted.store(true);
|
||||
assert_busy_facade();
|
||||
ethernet_spi_access.store(EthernetSpiAccess::Idle);
|
||||
wait_until([] { return ethernet_worker_stopped.load(); });
|
||||
assert(hardware_operations.load() == before + 1 && bus_owner.load());
|
||||
finish_off();
|
||||
break;
|
||||
}
|
||||
case 38: { // Card inserted after reservation but before worker scheduling: no hardware init.
|
||||
defer_task_start.store(true);
|
||||
assert(command("eth on").find("starting") != std::string::npos);
|
||||
mock_sd_fitted.store(true);
|
||||
unblock(defer_task_start);
|
||||
wait_until([] { return ethernet_worker_stopped.load(); });
|
||||
assert(hardware_operations.load() == 0 && !spi_active.load() && bus_owner.load());
|
||||
finish_off();
|
||||
break;
|
||||
}
|
||||
'''
|
||||
fixture = fixture.replace(" default: assert(false);", cases + " default: assert(false);")
|
||||
fixture_path = Path(directory) / "fixture.cpp"
|
||||
fixture_path.write_text(fixture, encoding="utf-8")
|
||||
compiled = subprocess.run([
|
||||
compiler, "-std=c++17", "-pthread", "-Wall", "-Wextra", "-Werror",
|
||||
*SANITIZER_FLAGS, "-DETHERNET_ENABLED=1", "-DOTA_RAK15002_SD_STORE=1",
|
||||
f"-I{FIXTURE / 'mocks'}", f"-I{ROOT / 'src'}",
|
||||
str(fixture_path), "-o", str(binary),
|
||||
], capture_output=True, text=True, timeout=60)
|
||||
self.assertEqual(compiled.returncode, 0, compiled.stdout + compiled.stderr)
|
||||
descriptions = (*SCENARIOS, "online SD insertion", "SD insertion during DHCP",
|
||||
"retained Stream SD insertion", "SD insertion before worker start")
|
||||
for scenario, description in enumerate(descriptions):
|
||||
with self.subTest(scenario=description):
|
||||
checked = subprocess.run([str(binary), str(scenario)],
|
||||
capture_output=True, text=True, timeout=10)
|
||||
self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr)
|
||||
self.assertIn(f"PASS: RAK4631 Ethernet runtime scenario {scenario}", checked.stdout)
|
||||
|
||||
def test_room_wires_default_off_after_mesh_begin_and_keeps_renewal_stream(self):
|
||||
source = (ROOT / "examples/simple_room_server/main.cpp").read_text()
|
||||
setup = method(source, "void setup()")
|
||||
configure = setup.index("ethernet_configure(hooks)")
|
||||
self.assertLess(setup.index("the_mesh.begin(fs)"), configure)
|
||||
self.assertLess(configure, setup.index("ethernet_start_task()"))
|
||||
self.assertNotIn("ethernet_configure(hooks, true)", setup)
|
||||
for assignment in ("hooks.prepare = prepareSdEthernet",
|
||||
"hooks.can_release = canReleaseSdEthernet",
|
||||
"hooks.released = releaseSdEthernet",
|
||||
"hooks.storage_conflict ="):
|
||||
self.assertIn(assignment, setup)
|
||||
self.assertIn("#if defined(OTA_RAK15002_SD_STORE)\n"
|
||||
" if (ethernet_take_session_reset() || !ethernet_get_enabled())", source)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user