mikecarper
4f114e914b
Extend the cooperative contact transaction to uncached ESP32 storage, including the PSRAM policy used by the XIAO and V4. Preserve owned contact copies, the 152-byte disk format, revision cancellation and synchronous durability operations. Keep contact capacities unchanged. Avoid SDK error logs when optional blob or transaction cleanup files are absent, while preserving metadata and deletion failure handling. Validate cached ESP/nRF and uncached PSRAM paths under ASan/UBSan, including bounded writes, mutation cancellation, faults, recovery and reboot readback. Update the existing uncached production fixture.
85 lines
3.9 KiB
Python
85 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Run the uncached Companion persistence paths against faulting filesystems."""
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
from test_replay_reset_integration import extract_braced
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
class CompanionUncachedStorageTest(unittest.TestCase):
|
|
def test_esp32_uncached_storage(self):
|
|
self.run_platform("ESP32_PLATFORM")
|
|
|
|
def test_stm32_uncached_storage(self):
|
|
self.run_platform("STM32_PLATFORM")
|
|
|
|
def test_rp2040_uncached_storage(self):
|
|
self.run_platform("RP2040_PLATFORM")
|
|
|
|
def run_platform(self, platform):
|
|
source = (ROOT / "examples/companion_radio/DataStore.cpp").read_text()
|
|
signatures = [
|
|
"static bool companionPathPresence(",
|
|
"bool DataStore::loadMainIdentity(",
|
|
"bool DataStore::canCreateMainIdentity() const",
|
|
"bool DataStore::saveMainIdentity(",
|
|
"static bool deserializeContactRecord(",
|
|
"void DataStore::loadContacts(",
|
|
"bool DataStore::saveContacts(",
|
|
"bool DataStore::hasIncompleteContactLoad() const",
|
|
]
|
|
if platform == "ESP32_PLATFORM":
|
|
signatures.extend((
|
|
"static bool serializeContactRecord(",
|
|
"DataStore::~DataStore()",
|
|
"void DataStore::cancelContactWrite(",
|
|
"bool DataStore::serviceContactWrite(",
|
|
"bool DataStore::markContactDirty(",
|
|
"bool DataStore::releaseContact(",
|
|
"bool DataStore::serviceContactWrites(",
|
|
"bool DataStore::flushContactWrites(",
|
|
"bool DataStore::hasPendingContactWrites() const",
|
|
))
|
|
# The serializer must be declared before save/service uses it.
|
|
signatures.insert(0, signatures.pop(signatures.index(
|
|
"static bool serializeContactRecord(")))
|
|
packet = (ROOT / "src/Packet.cpp").read_text()
|
|
generated = "namespace mesh {\n" + extract_braced(
|
|
packet, "bool Packet::isValidPathLen(") + "\n}\n"
|
|
generated += "\n".join(extract_braced(source, s) for s in signatures)
|
|
with tempfile.TemporaryDirectory(prefix="mesh-uncached-store-") as temp:
|
|
temp = Path(temp)
|
|
(temp / "store_under_test.h").write_text(generated)
|
|
header = (ROOT / "examples/companion_radio/DataStore.h").read_text()
|
|
start = header.index(" mesh::ContactFileTransaction* _contact_write")
|
|
end = header.index("\n#endif", start)
|
|
(temp / "contact_write_state_under_test.h").write_text(header[start:end])
|
|
# IdentityStore.h only needs the platform filesystem declaration;
|
|
# the concrete mock below implements the same file API.
|
|
(temp / "FS.h").write_text(
|
|
"#pragma once\nnamespace fs { using FS = FakeFilesystem; }\n"
|
|
)
|
|
binary = temp / "test"
|
|
compiled = subprocess.run([
|
|
"c++", "-std=c++17", "-O1", "-g", "-Wall", "-Wextra",
|
|
"-Werror", "-Wno-unused-parameter", "-fsanitize=address,undefined", "-fno-pie", "-no-pie",
|
|
"-DMESH_CONTACT_CACHE=0", "-D" + platform + "=1",
|
|
"-I", str(temp),
|
|
"-I", str(ROOT / "test/fixtures/contact_cache/mocks"),
|
|
"-I", str(ROOT / "src"),
|
|
"-I", str(ROOT / "lib/ed25519"),
|
|
str(ROOT / "test/fixtures/companion_uncached_storage/test.cpp"),
|
|
"-o", str(binary),
|
|
], capture_output=True, text=True, timeout=60)
|
|
self.assertEqual(compiled.returncode, 0, compiled.stdout + compiled.stderr)
|
|
result = subprocess.run([str(binary)], capture_output=True, text=True, timeout=30)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|