Fit Mini Repeater with approved optional service cuts
This commit is contained in:
mikecarper committed
1 parent
7a9d099039
commit
55f67f9062
30 files changed
+1259
-34
No files matched your search
@@ -9,6 +9,7 @@ from test_radio_receive_contract import method
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
HARNESS = r'''
|
||||
#include <Arduino.h>
|
||||
#include <helpers/CarrierTools.h>
|
||||
#include <helpers/CarrierWaveCLI.h>
|
||||
#include <cassert>
|
||||
#include <helpers/radiolib/RadioInterruptEvents.h>
|
||||
@@ -175,7 +176,7 @@ struct RadioLibWrapper : mesh::Radio {
|
||||
bool stopCarrierWave();
|
||||
bool serviceCarrierWave();
|
||||
bool serviceMorseTx();
|
||||
bool serviceMorseRx();
|
||||
@RX_SERVICE_DECLARATION@
|
||||
bool supportsMorseTx() const override;
|
||||
mesh::MorseTxResult startMorseTx(float,uint8_t,const char*) override;
|
||||
mesh::MorseTxResult stopMorseTx() override;
|
||||
@@ -476,11 +477,15 @@ def production_harness():
|
||||
if signature in ('mesh::MorseRxResult RadioLibWrapper::startMorseRx(',
|
||||
'mesh::MorseRxResult RadioLibWrapper::stopMorseRx(',
|
||||
'size_t RadioLibWrapper::readMorseCapture(',
|
||||
'void RadioLibWrapper::clearMorseCapture('):
|
||||
'void RadioLibWrapper::clearMorseCapture(',
|
||||
'bool RadioLibWrapper::serviceMorseRx('):
|
||||
return '#if MESH_ENABLE_MORSE_RX\n'+method(source,signature)+'\n#endif'
|
||||
return method(source,signature)
|
||||
code = HARNESS.replace('@METHODS@', '\n'.join(extract_transition(s) for s in signatures))
|
||||
header = (ROOT/'src/helpers/radiolib/RadioLibWrappers.h').read_text()
|
||||
service_start = header.rfind('#if MESH_ENABLE_MORSE_RX\n',0,header.index(' bool serviceMorseRx();'))
|
||||
service_end = header.index('\n#endif',service_start) + len('\n#endif')
|
||||
code = code.replace('@RX_SERVICE_DECLARATION@',header[service_start:service_end])
|
||||
fields_start = header.rfind('#if MESH_ENABLE_MORSE_RX',0,header.index('mesh::MorseRx* _morse_rx'))
|
||||
fields_end = header.index(' // End Morse RX state.',fields_start)
|
||||
code = code.replace('@MORSE_RX_FIELDS@',header[fields_start:fields_end])
|
||||
|
||||
@@ -222,7 +222,7 @@ int main(void) {
|
||||
str(harness), str(ED25519 / "fe.c"), "-o", str(executable)], check=True)
|
||||
subprocess.run([str(executable)], check=True)
|
||||
|
||||
def test_only_size_constrained_stm32_companions_opt_in(self):
|
||||
def test_only_approved_size_constrained_stm32_profiles_opt_in(self):
|
||||
profiles = []
|
||||
for ini in (ROOT / "variants").glob("*/platformio.ini"):
|
||||
section = ""
|
||||
@@ -236,6 +236,7 @@ int main(void) {
|
||||
("tiny_relay", "[env:Tiny_Relay_companion_radio_usb]"),
|
||||
("wio-e5-dev", "[env:wio-e5_companion_radio_usb]"),
|
||||
("wio-e5-mini", "[env:wio-e5-mini_companion_radio_usb]"),
|
||||
("wio-e5-mini", "[env:wio-e5-mini_repeater]"),
|
||||
])
|
||||
|
||||
|
||||
|
||||
@@ -128,7 +128,7 @@ int main(void) {
|
||||
str(harness), str(ED25519 / "sc.c"), "-o", str(executable)], check=True)
|
||||
subprocess.run([str(executable)], check=True)
|
||||
|
||||
def test_only_size_constrained_stm32_companions_enable_compact_scalars(self):
|
||||
def test_only_approved_size_constrained_stm32_profiles_enable_compact_scalars(self):
|
||||
profiles = []
|
||||
for ini in (ROOT / "variants").glob("*/platformio.ini"):
|
||||
section = ""
|
||||
@@ -142,6 +142,7 @@ int main(void) {
|
||||
("tiny_relay", "[env:Tiny_Relay_companion_radio_usb]"),
|
||||
("wio-e5-dev", "[env:wio-e5_companion_radio_usb]"),
|
||||
("wio-e5-mini", "[env:wio-e5-mini_companion_radio_usb]"),
|
||||
("wio-e5-mini", "[env:wio-e5-mini_repeater]"),
|
||||
])
|
||||
|
||||
|
||||
|
||||
@@ -8,6 +8,8 @@ import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
from cpp_source import body
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
@@ -23,6 +25,7 @@ HARNESS = r'''
|
||||
#include <vector>
|
||||
#include <cstdint>
|
||||
#include <helpers/FileRead.h>
|
||||
#include <helpers/PacketFileLoggingConfig.h>
|
||||
@USB_CONFIG@
|
||||
#define MESH_ESP32_USB_CONSOLE_COOPERATIVE 1
|
||||
#define MESH_USB_CONSOLE_COOPERATIVE 1
|
||||
@@ -397,6 +400,23 @@ RECENT_TEST = r'''
|
||||
|
||||
|
||||
class CooperativeOutputTest(unittest.TestCase):
|
||||
def role_methods(self, text, role):
|
||||
dump_start = text.index("void MyMesh::dumpLogFile()")
|
||||
guard = "#if MESH_ENABLE_PACKET_FILE_LOGGING\n"
|
||||
if text[:dump_start].endswith(guard):
|
||||
dump_start -= len(guard)
|
||||
dump_end = text.index("\n#if MESH_USB_CONSOLE_COOPERATIVE\nbool MyMesh::hasPendingSerialOutput", dump_start)
|
||||
methods = text[dump_start:dump_end] + "\n" + "\n".join(body(text, signature) for signature in (
|
||||
"bool MyMesh::hasPendingSerialOutput()", "void MyMesh::cancelPendingSerialOutput()",
|
||||
"void MyMesh::servicePendingSerialOutput()"))
|
||||
if role == "simple_repeater":
|
||||
format_start = text.index("static void formatLocalSnrX4(")
|
||||
format_end = text.index("\nvoid MyMesh::formatRecentRepeatersReply", format_start)
|
||||
recent_start = text.index("void MyMesh::printRecentRepeatersSerial()")
|
||||
recent_end = text.index("\nbool MyMesh::setRecentRepeater(", recent_start)
|
||||
methods = text[format_start:format_end] + "\n" + text[recent_start:recent_end] + "\n" + methods
|
||||
return methods
|
||||
|
||||
def test_usb_acl_commands_route_to_the_cooperative_pump(self):
|
||||
for role in ("simple_repeater", "simple_room_server"):
|
||||
with self.subTest(role=role):
|
||||
@@ -415,17 +435,7 @@ class CooperativeOutputTest(unittest.TestCase):
|
||||
for config in ("MESH_ESP32_TINYUSB_NONBLOCKING", "MESH_ESP32_HWCDC_SESSION_GUARD")):
|
||||
with self.subTest(role=role, transport=config), tempfile.TemporaryDirectory() as temporary:
|
||||
text = (ROOT / f"examples/{role}/MyMesh.cpp").read_text()
|
||||
dump_start = text.index("void MyMesh::dumpLogFile()")
|
||||
dump_end = text.index("\n#if MESH_USB_CONSOLE_COOPERATIVE\nbool MyMesh::hasPendingSerialOutput", dump_start)
|
||||
pump_start = text.index("bool MyMesh::hasPendingSerialOutput()", dump_end)
|
||||
pump_end = text.index("\n#endif", pump_start)
|
||||
methods = text[dump_start:dump_end] + "\n" + text[pump_start:pump_end]
|
||||
if role == "simple_repeater":
|
||||
format_start = text.index("static void formatLocalSnrX4(")
|
||||
format_end = text.index("\nvoid MyMesh::formatRecentRepeatersReply", format_start)
|
||||
recent_start = text.index("void MyMesh::printRecentRepeatersSerial()")
|
||||
recent_end = text.index("\nbool MyMesh::setRecentRepeater(", recent_start)
|
||||
methods = text[format_start:format_end] + "\n" + text[recent_start:recent_end] + "\n" + methods
|
||||
methods = self.role_methods(text, role)
|
||||
program = HARNESS.replace("@METHODS@", methods).replace(
|
||||
"@RECENT_TEST@", RECENT_TEST if role == "simple_repeater" else ""
|
||||
).replace("@USB_CONFIG@", f"#define {config} 1")
|
||||
@@ -441,6 +451,59 @@ class CooperativeOutputTest(unittest.TestCase):
|
||||
)
|
||||
self.assertEqual(run.returncode, 0, run.stdout + run.stderr)
|
||||
|
||||
def test_file_off_control_preserves_shared_acl_and_recent_output(self):
|
||||
compiler = shutil.which("g++") or shutil.which("clang++")
|
||||
self.assertIsNotNone(compiler, "A host C++ compiler is required")
|
||||
text = (ROOT / "examples/simple_repeater/MyMesh.cpp").read_text()
|
||||
program = HARNESS[:HARNESS.index("int main() {")].replace(
|
||||
"@METHODS@", self.role_methods(text, "simple_repeater")).replace(
|
||||
"@USB_CONFIG@", "#define MESH_ESP32_TINYUSB_NONBLOCKING 1")
|
||||
program += r'''
|
||||
int main() {
|
||||
setupFile("must not be read\n");
|
||||
MyMesh radio;
|
||||
radio.dumpLogFile();
|
||||
require(!radio.hasPendingSerialOutput() && console.output.empty(), "file-off dump is active");
|
||||
require(fs.data->reads == 0, "file-off dump read persistent storage");
|
||||
std::string expected = "ACL:\r\n";
|
||||
for (int i = 0; i < 32; ++i) {
|
||||
FakeAcl::Client client;
|
||||
std::fill(std::begin(client.id.pub_key), std::end(client.id.pub_key), i);
|
||||
radio.acl.clients.push_back(client);
|
||||
char key[PUB_KEY_SIZE * 2 + 1], line[80];
|
||||
mesh::Utils::toHex(key, client.id.pub_key, PUB_KEY_SIZE);
|
||||
snprintf(line, sizeof(line), "%02X %s\n", client.permissions, key);
|
||||
expected += line;
|
||||
}
|
||||
radio.printAclSerial();
|
||||
require(radio.hasPendingSerialOutput(), "file-off ACL was disabled");
|
||||
console.capacity = 0;
|
||||
radio.servicePendingSerialOutput();
|
||||
require(console.output.empty(), "file-off ACL ignored backpressure");
|
||||
console.short_limit = 3;
|
||||
drain(radio, 73);
|
||||
require(console.output == expected, "file-off ACL lost retained suffix");
|
||||
require(fs.data->reads == 0, "shared ACL output read persistent log");
|
||||
radio.printAclSerial();
|
||||
radio.servicePendingSerialOutput();
|
||||
radio.cancelPendingSerialOutput();
|
||||
require(!radio.hasPendingSerialOutput() && radio.serial_log_pending_size == 0,
|
||||
"file-off cancellation retained ACL suffix");
|
||||
@RECENT_TEST@
|
||||
std::cout << "file-off shared output checks passed\n";
|
||||
}
|
||||
'''
|
||||
program = program.replace("@RECENT_TEST@", RECENT_TEST)
|
||||
with tempfile.TemporaryDirectory(prefix="mini-file-off-output-") as directory:
|
||||
executable = Path(directory) / "pump"
|
||||
build = subprocess.run(
|
||||
[compiler, "-std=c++17", "-O0", "-DMESH_ENABLE_PACKET_FILE_LOGGING=0",
|
||||
"-I", str(ROOT / "src"), "-x", "c++", "-", "-o", str(executable)],
|
||||
input=program, text=True, capture_output=True, timeout=60)
|
||||
self.assertEqual(build.returncode, 0, build.stdout + build.stderr)
|
||||
run = subprocess.run([str(executable)], text=True, capture_output=True, timeout=30)
|
||||
self.assertEqual(run.returncode, 0, run.stdout + run.stderr)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,197 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Compile the complete disabled management implementation at its FS boundary."""
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
HEADER = r'''
|
||||
#pragma once
|
||||
#include <cassert>
|
||||
#include <cstddef>
|
||||
#include <cstdint>
|
||||
#include <cstdio>
|
||||
#include <cstring>
|
||||
#define OUT_PATH_UNKNOWN 255
|
||||
struct TransportKey { unsigned marker = 17; };
|
||||
struct Filesystem {
|
||||
unsigned reads = 0;
|
||||
bool exists(const char* path) {
|
||||
// No management/data.tx probe, recovery or write is permitted when off.
|
||||
assert(!strcmp(path, "/fleet_channel"));
|
||||
++reads;
|
||||
return true;
|
||||
}
|
||||
};
|
||||
using FILESYSTEM = Filesystem;
|
||||
struct Callbacks { bool supportsFleetControl() const { return true; } };
|
||||
namespace mesh {
|
||||
struct Mesh {};
|
||||
struct RegionMap {};
|
||||
struct FleetChannel {
|
||||
static inline unsigned live = 0;
|
||||
FleetChannel(FILESYSTEM*, Callbacks*, RegionMap*) { ++live; }
|
||||
~FleetChannel() { --live; }
|
||||
bool handleConfig(const char*, char* reply, size_t size) {
|
||||
snprintf(reply, size, "fleet remains available"); return true;
|
||||
}
|
||||
};
|
||||
}
|
||||
struct CommonCLI {
|
||||
mesh::Mesh* _management_mesh = nullptr;
|
||||
FILESYSTEM* _management_fs = nullptr;
|
||||
mesh::RegionMap region;
|
||||
mesh::RegionMap* _region_map = ®ion;
|
||||
Callbacks callbacks;
|
||||
Callbacks* _callbacks = &callbacks;
|
||||
mesh::FleetChannel* _fleet_channel = nullptr;
|
||||
unsigned beacon_starts = 0;
|
||||
~CommonCLI() { delete _fleet_channel; }
|
||||
void beginBeacons() { ++beacon_starts; }
|
||||
void beginManagement(mesh::Mesh&, FILESYSTEM*);
|
||||
void loopManagement();
|
||||
bool handleFleetCommand(const char*, char*);
|
||||
bool getDataTxPath(const uint8_t*&, uint8_t&) const;
|
||||
bool resolveDataTxScope(TransportKey&, char*, size_t, bool*) const;
|
||||
bool adoptLegacyDataTxPath(const uint8_t*, uint8_t);
|
||||
bool setDataTxPath(const char*, char*, size_t);
|
||||
bool handleDataTxCommand(char*, char*);
|
||||
bool handleManagementCommand(char*, char*);
|
||||
};
|
||||
'''
|
||||
|
||||
HARNESS = r'''
|
||||
#include "CommonCLI.h"
|
||||
#include <string>
|
||||
#include <vector>
|
||||
int main() {
|
||||
Filesystem fs;
|
||||
mesh::Mesh mesh;
|
||||
{
|
||||
CommonCLI cli;
|
||||
cli.beginManagement(mesh, &fs);
|
||||
assert(cli._management_mesh == &mesh && cli._management_fs == &fs);
|
||||
assert(cli.beacon_starts == MESH_ENABLE_TX_BEACONS);
|
||||
assert(mesh::FleetChannel::live == MESH_ENABLE_FLEET_CONTROL);
|
||||
assert(fs.reads == MESH_ENABLE_FLEET_CONTROL);
|
||||
cli.loopManagement();
|
||||
assert(fs.reads == MESH_ENABLE_FLEET_CONTROL);
|
||||
#if MESH_ENABLE_FLEET_CONTROL
|
||||
char fleet_reply[160] = {};
|
||||
assert(cli.handleFleetCommand("get fleet.stats", fleet_reply));
|
||||
assert(!strcmp(fleet_reply, "fleet remains available"));
|
||||
#endif
|
||||
const uint8_t byte = 42;
|
||||
const uint8_t* path = &byte;
|
||||
uint8_t length = 3;
|
||||
assert(!cli.getDataTxPath(path, length));
|
||||
assert(path == nullptr && length == OUT_PATH_UNKNOWN);
|
||||
assert(!cli.adoptLegacyDataTxPath(&byte, 1));
|
||||
TransportKey key;
|
||||
char name[8] = "old";
|
||||
bool ambiguous = true;
|
||||
assert(!cli.resolveDataTxScope(key, name, sizeof(name), &ambiguous));
|
||||
assert(!*name && !ambiguous && key.marker == 17);
|
||||
assert(!cli.resolveDataTxScope(key, nullptr, 0, nullptr));
|
||||
for (const char* command : {"ver", "board", "stats-core", "stats-radio",
|
||||
"stats-packets", "get usb.logging", "get radio2",
|
||||
"get password", "host cpu-temp", "get data.txExtra"}) {
|
||||
char input[160], reply[160] = "not handled";
|
||||
strcpy(input, command);
|
||||
assert(!cli.handleManagementCommand(input, reply));
|
||||
assert(!strcmp(input, command) && !strcmp(reply, "not handled"));
|
||||
}
|
||||
for (const char* command : {"get mgmt", "get mgmt.password", "set mgmt.enabled on",
|
||||
"get mgmt.path", "set mgmt.path direct",
|
||||
"get data.tx", "get data.tx.path", "set data.tx region none"}) {
|
||||
char input[160], reply[160] = {};
|
||||
strcpy(input, command);
|
||||
assert(cli.handleManagementCommand(input, reply));
|
||||
assert(!strcmp(input, command));
|
||||
assert(!strcmp(reply, !strncmp(command + 4, "data.tx", 7)
|
||||
? "Err - data.tx unavailable" : "ERR: management unavailable"));
|
||||
}
|
||||
for (size_t length : {size_t(0), size_t(5), size_t(96), size_t(200)}) {
|
||||
const std::string original = "set mgmt.password " + std::string(length, 's');
|
||||
std::vector<char> guarded(original.size() + 9, '\x5A');
|
||||
char* command = guarded.data() + 4;
|
||||
memcpy(command, original.c_str(), original.size() + 1);
|
||||
char reply[160] = {};
|
||||
assert(cli.handleManagementCommand(command, reply));
|
||||
assert(!strcmp(reply, "ERR: management unavailable"));
|
||||
assert(!memcmp(command, "set mgmt.password ", 18));
|
||||
for (size_t i = 18; i <= original.size(); ++i) assert(command[i] == 0);
|
||||
for (size_t i = 0; i < 4; ++i) assert(guarded[i] == '\x5A');
|
||||
for (size_t i = original.size() + 5; i < guarded.size(); ++i) assert(guarded[i] == '\x5A');
|
||||
}
|
||||
char guarded_reply[12]; memset(guarded_reply, '\x5A', sizeof(guarded_reply));
|
||||
assert(!cli.setDataTxPath("direct", guarded_reply + 2, 6));
|
||||
assert(guarded_reply[0] == '\x5A' && guarded_reply[1] == '\x5A');
|
||||
assert(guarded_reply[7] == 0 && guarded_reply[8] == '\x5A');
|
||||
assert(fs.reads == MESH_ENABLE_FLEET_CONTROL);
|
||||
}
|
||||
assert(mesh::FleetChannel::live == 0);
|
||||
puts("PASS: disabled management fails closed, wipes passwords and retains Fleet/Beacon context");
|
||||
}
|
||||
'''
|
||||
|
||||
|
||||
class ManagementFeatureGateTest(unittest.TestCase):
|
||||
def test_default_is_enabled_and_explicit_values_are_boolean(self):
|
||||
compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++")
|
||||
self.assertIsNotNone(compiler, "a host C++17 compiler is required")
|
||||
header = ROOT / "src/helpers/ManagementFeatureConfig.h"
|
||||
for platform in (None, "STM32_PLATFORM", "NRF52_PLATFORM", "ESP32_PLATFORM", "RP2040_PLATFORM"):
|
||||
for value in (None, 0, 1, -1, 2):
|
||||
with self.subTest(platform=platform, value=value):
|
||||
command = [compiler, "-E", "-P", "-x", "c++", "-include", str(header)]
|
||||
if platform:
|
||||
command.append("-D" + platform + "=1")
|
||||
if value is not None:
|
||||
command.append("-DMESH_ENABLE_MANAGEMENT_REPORTING=" + str(value))
|
||||
command.append("-")
|
||||
result = subprocess.run(command, input="MESH_ENABLE_MANAGEMENT_REPORTING\n",
|
||||
capture_output=True, text=True, timeout=10)
|
||||
if value in (-1, 2):
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn("must be 0 or 1", result.stderr)
|
||||
else:
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
self.assertEqual(result.stdout.strip(), str(1 if value is None else value))
|
||||
|
||||
def test_whole_disabled_translation_units_have_no_crypto_or_storage_dependency(self):
|
||||
compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++")
|
||||
self.assertIsNotNone(compiler, "a host C++17 compiler is required")
|
||||
with tempfile.TemporaryDirectory(prefix="management-off-") as directory:
|
||||
work = Path(directory)
|
||||
(work / "CommonCLI.h").write_text(HEADER, encoding="ascii")
|
||||
(work / "test.cpp").write_text(HARNESS, encoding="ascii")
|
||||
for name in ("CommonCLI_Management.cpp", "ManagementReporter.cpp"):
|
||||
shutil.copyfile(ROOT / "src/helpers" / name, work / name)
|
||||
sanitizers = (["-fsanitize=address,undefined", "-fno-sanitize-recover=all",
|
||||
"-fno-pie", "-no-pie"] if sys.platform.startswith("linux") else [])
|
||||
for beacons, fleet in ((0, 0), (0, 1), (1, 0), (1, 1)):
|
||||
with self.subTest(beacons=beacons, fleet=fleet):
|
||||
executable = work / ("test-%u-%u" % (beacons, fleet))
|
||||
result = subprocess.run([
|
||||
compiler, "-std=c++17", "-O1", "-Wall", "-Wextra", "-Werror", *sanitizers,
|
||||
"-DMESH_ENABLE_MANAGEMENT_REPORTING=0", "-DMESH_ENABLE_TX_BEACONS=" + str(beacons),
|
||||
"-DMESH_ENABLE_FLEET_CONTROL=" + str(fleet), "-I", str(work),
|
||||
"-I", str(ROOT / "src/helpers"), str(work / "CommonCLI_Management.cpp"),
|
||||
str(work / "ManagementReporter.cpp"), str(work / "test.cpp"), "-o", str(executable),
|
||||
], capture_output=True, text=True, timeout=60)
|
||||
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
||||
result = subprocess.run([str(executable)], capture_output=True, text=True, timeout=15)
|
||||
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
||||
self.assertIn("PASS: disabled management fails closed", result.stdout)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,416 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Keep constrained Mini Repeater builds compatible with the MQTT packet log.
|
||||
|
||||
Compile the production repeater RX method, Dispatcher TX method, packet wire
|
||||
codec and bounded serial writer under all eight feature-gate combinations.
|
||||
Hardware transport/radio are stand-ins; this does not qualify physical USB/RF.
|
||||
"""
|
||||
|
||||
import configparser
|
||||
import itertools
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
from cpp_source import body
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
HARNESS = r'''
|
||||
#include <Arduino.h>
|
||||
inline uint32_t micros() { return millis() * 1000U; }
|
||||
#include <Packet.h>
|
||||
#include <helpers/CarrierTools.h>
|
||||
#include <helpers/ManagementFeatureConfig.h>
|
||||
#include <helpers/PacketFileLoggingConfig.h>
|
||||
#include <helpers/SerialPacketLog.h>
|
||||
#include <algorithm>
|
||||
#include <cassert>
|
||||
#include <cstdarg>
|
||||
#include <cstdio>
|
||||
#include <cstring>
|
||||
#include <limits>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
static_assert(MESH_PACKET_LOGGING == 1, "packet logging must remain compiled");
|
||||
static_assert(MESH_PACKET_LOGGING_COMPACT == 1, "retain compact serial protocol");
|
||||
static_assert(MESH_ENABLE_CARRIER_TOOLS == EXPECT_CARRIER, "exercise actual flag");
|
||||
static_assert(MESH_ENABLE_MANAGEMENT_REPORTING == EXPECT_MANAGEMENT, "exercise actual flag");
|
||||
static_assert(MESH_ENABLE_PACKET_FILE_LOGGING == EXPECT_FILE_LOGGING, "exercise actual flag");
|
||||
|
||||
struct Port : Stream {
|
||||
std::string output;
|
||||
bool stalled = false;
|
||||
size_t chunk = 7, wedge_after = std::numeric_limits<size_t>::max();
|
||||
int availableForWrite() override {
|
||||
return stalled || output.size() >= wedge_after ? 0 : int(chunk);
|
||||
}
|
||||
size_t write(const uint8_t* data, size_t size) override {
|
||||
if (!availableForWrite()) return 0;
|
||||
size = std::min(size, chunk);
|
||||
output.append(reinterpret_cast<const char*>(data), size);
|
||||
return size;
|
||||
}
|
||||
} log_port;
|
||||
|
||||
namespace mesh {
|
||||
bool logging_enabled = true;
|
||||
bool isUsbLoggingEnabled() { return logging_enabled; }
|
||||
Stream& usbLoggingPort() { return log_port; }
|
||||
|
||||
@PACKET_METHODS@
|
||||
|
||||
enum class RadioParamApplyResult { APPLIED, BUSY };
|
||||
struct Profile { uint8_t cr = 5; };
|
||||
struct Profiles {
|
||||
Profile value;
|
||||
const Profile& params(uint8_t) const { return value; }
|
||||
};
|
||||
struct Radio {
|
||||
bool send_ok = true;
|
||||
unsigned sends = 0;
|
||||
std::vector<uint8_t> raw;
|
||||
Profiles settings;
|
||||
RadioParamApplyResult prepareTransmitProfile(uint8_t, bool) {
|
||||
return RadioParamApplyResult::APPLIED;
|
||||
}
|
||||
Profiles* profiles() { return &settings; }
|
||||
uint32_t getEstAirtimeFor(int) { return 10; }
|
||||
bool setCodingRate(uint8_t) { return true; }
|
||||
bool startSendRaw(const uint8_t* bytes, int length) {
|
||||
++sends;
|
||||
if (send_ok) raw.assign(bytes, bytes + length);
|
||||
return send_ok;
|
||||
}
|
||||
float getLastSNR() const { return 2.5f; }
|
||||
float getLastRSSI() const { return -92; }
|
||||
};
|
||||
struct Clock { uint32_t getMillis() const { return millis(); } };
|
||||
struct Dispatcher {
|
||||
Packet* outbound = nullptr;
|
||||
Radio radio;
|
||||
Radio* _radio = &radio;
|
||||
Clock clock;
|
||||
Clock* _ms = &clock;
|
||||
uint8_t outbound_restore_cr = 0;
|
||||
uint32_t outbound_start = 0, outbound_expiry = 0;
|
||||
bool restoreOutboundTxOverrides() { return true; }
|
||||
bool isPacketRadioCurrent(const Packet*) const { return true; }
|
||||
uint8_t getDefaultTxCodingRate() const { return 5; }
|
||||
uint32_t futureMillis(uint32_t wait) const { return millis() + wait; }
|
||||
const char* getLogDateTime() const { return "unused compact timestamp"; }
|
||||
bool startOutboundTransmit();
|
||||
};
|
||||
|
||||
@TX_METHOD@
|
||||
} // namespace mesh
|
||||
|
||||
#define PACKET_LOG_FILE "/packet_log"
|
||||
struct FakeFS {
|
||||
unsigned opens = 0, writes = 0, removes = 0;
|
||||
std::string output;
|
||||
void remove(const char* name) {
|
||||
assert(!std::strcmp(name, PACKET_LOG_FILE));
|
||||
++removes;
|
||||
}
|
||||
};
|
||||
struct File {
|
||||
FakeFS* fs;
|
||||
explicit operator bool() const { return true; }
|
||||
void print(const char* text) { ++fs->writes; fs->output += text; }
|
||||
void printf(const char* format, ...) {
|
||||
char text[1024];
|
||||
va_list args;
|
||||
va_start(args, format);
|
||||
const int length = std::vsnprintf(text, sizeof(text), format, args);
|
||||
va_end(args);
|
||||
assert(length >= 0 && size_t(length) < sizeof(text));
|
||||
++fs->writes;
|
||||
fs->output += text;
|
||||
}
|
||||
void close() {}
|
||||
};
|
||||
struct Callbacks {
|
||||
virtual void setLoggingOn(bool enable) = 0;
|
||||
virtual void eraseLogFile() = 0;
|
||||
};
|
||||
inline uint8_t getRetryLogCodingRate(const mesh::Packet*, uint8_t fallback) { return fallback; }
|
||||
inline uint16_t rxPowerSavingPreambleForParams(uint8_t, float) { return 16; }
|
||||
struct MyMesh : Callbacks {
|
||||
bool _logging = false;
|
||||
FakeFS fs;
|
||||
FakeFS* _fs = &fs;
|
||||
mesh::Radio radio;
|
||||
mesh::Radio* _radio = &radio;
|
||||
unsigned refreshed = 0;
|
||||
uint8_t active_sf = 7;
|
||||
float active_bw = 125;
|
||||
File openAppend(const char* name) {
|
||||
assert(!std::strcmp(name, PACKET_LOG_FILE));
|
||||
++fs.opens;
|
||||
return File{&fs};
|
||||
}
|
||||
const char* getLogDateTime() const { return "timestamp"; }
|
||||
uint8_t getDefaultTxCodingRate() const { return 5; }
|
||||
void formatFloodRetryPath(char* output, size_t size, const mesh::Packet*) {
|
||||
std::snprintf(output, size, "01");
|
||||
}
|
||||
bool formatFloodRetryHeard(char* output, size_t size, const mesh::Packet*) {
|
||||
std::snprintf(output, size, "02");
|
||||
return true;
|
||||
}
|
||||
void refreshFloodRetryReachability(const mesh::Packet*) { ++refreshed; }
|
||||
@INLINE_CALLBACKS@
|
||||
void logRxRaw(float snr, float rssi, const uint8_t raw[], int len);
|
||||
void logRx(mesh::Packet* pkt, int len, float score);
|
||||
void logTx(mesh::Packet* pkt, int len);
|
||||
void logTxFail(mesh::Packet* pkt, int len);
|
||||
void onFloodRetryEvent(const char* event, const mesh::Packet* packet,
|
||||
uint32_t delay_millis, uint8_t retry_attempt);
|
||||
};
|
||||
@RX_METHOD@
|
||||
@FILE_METHODS@
|
||||
|
||||
static void reset() {
|
||||
log_port = Port{};
|
||||
mesh::logging_enabled = true;
|
||||
mesh::serialLogDroppedCount() = 0;
|
||||
mesh::serialLogPortSeen() = false;
|
||||
g_mock_millis = 100;
|
||||
}
|
||||
static std::string hex(const uint8_t* bytes, size_t size) {
|
||||
const char* digits = "0123456789ABCDEF";
|
||||
std::string output;
|
||||
for (size_t i = 0; i < size; ++i) {
|
||||
output += digits[bytes[i] >> 4];
|
||||
output += digits[bytes[i] & 15];
|
||||
}
|
||||
return output;
|
||||
}
|
||||
static mesh::Packet makePacket(bool maximum = false) {
|
||||
mesh::Packet packet;
|
||||
packet.header = (PAYLOAD_TYPE_GRP_DATA << PH_TYPE_SHIFT) | ROUTE_TYPE_TRANSPORT_DIRECT;
|
||||
packet.transport_codes[0] = 0x1234;
|
||||
packet.transport_codes[1] = 0xABCD;
|
||||
packet.setPathHashSizeAndCount(2, maximum ? 32 : 2);
|
||||
for (unsigned i = 0; i < packet.getPathByteLen(); ++i) packet.path[i] = uint8_t(i);
|
||||
packet.payload_len = maximum ? MAX_PACKET_PAYLOAD : 3;
|
||||
for (unsigned i = 0; i < packet.payload_len; ++i) packet.payload[i] = uint8_t(i ^ 0xAB);
|
||||
return packet;
|
||||
}
|
||||
|
||||
int main() {
|
||||
MyMesh repeater;
|
||||
// Actual inline callbacks and persistent packet emitters remain independent
|
||||
// of live USB logging. The file-off control cannot be re-enabled at runtime.
|
||||
assert(!repeater._logging);
|
||||
repeater.setLoggingOn(true);
|
||||
assert(repeater._logging == bool(EXPECT_FILE_LOGGING));
|
||||
mesh::Packet stored = makePacket();
|
||||
repeater.logRx(&stored, 42, 0.25f);
|
||||
repeater.logTx(&stored, 42);
|
||||
repeater.logTxFail(&stored, 42);
|
||||
repeater.onFloodRetryEvent("good", &stored, 12, 1);
|
||||
assert(repeater.refreshed == 1); // Reachability is not a logging side effect.
|
||||
repeater.onFloodRetryEvent("good", nullptr, 12, 1);
|
||||
repeater.onFloodRetryEvent("failure", &stored, 12, 1);
|
||||
assert(repeater.refreshed == 1);
|
||||
repeater.eraseLogFile();
|
||||
if (EXPECT_FILE_LOGGING) {
|
||||
assert(repeater.fs.opens == 5 && repeater.fs.writes > 0 && repeater.fs.removes == 1);
|
||||
for (const char* marker : {": RX,", ": TX,", ": TX FAIL!", ": FLOOD RETRY good"})
|
||||
assert(repeater.fs.output.find(marker) != std::string::npos);
|
||||
} else {
|
||||
assert(repeater.fs.opens == 0 && repeater.fs.writes == 0 && repeater.fs.removes == 0);
|
||||
assert(repeater.fs.output.empty());
|
||||
}
|
||||
repeater.setLoggingOn(false);
|
||||
assert(!repeater._logging);
|
||||
const unsigned old_opens = repeater.fs.opens;
|
||||
repeater.logRx(&stored, 42, 0.25f);
|
||||
repeater.logTx(&stored, 42);
|
||||
repeater.logTxFail(&stored, 42);
|
||||
repeater.onFloodRetryEvent("good", &stored, 12, 1);
|
||||
assert(repeater.refreshed == 2 && repeater.fs.opens == old_opens);
|
||||
const uint8_t raw[] = {0x00, 0xAB, 0xFF};
|
||||
reset();
|
||||
repeater.logRxRaw(2.5f, -92, raw, sizeof(raw));
|
||||
assert(log_port.output == "R00ABFF\n");
|
||||
assert(g_mock_millis == 100 && mesh::serialLogDroppedCount() == 0);
|
||||
|
||||
uint8_t maximum_rx[MAX_TRANS_UNIT];
|
||||
for (unsigned i = 0; i < sizeof(maximum_rx); ++i) maximum_rx[i] = uint8_t(i);
|
||||
reset();
|
||||
repeater.logRxRaw(0, -100, maximum_rx, sizeof(maximum_rx));
|
||||
assert(log_port.output == "R" + hex(maximum_rx, sizeof(maximum_rx)) + "\n");
|
||||
assert(log_port.output.size() == 2 + MAX_TRANS_UNIT * 2);
|
||||
|
||||
for (const bool maximum : {false, true}) {
|
||||
mesh::Packet packet = makePacket(maximum);
|
||||
uint8_t wire[MAX_TRANS_UNIT];
|
||||
const size_t length = packet.writeTo(wire);
|
||||
assert(length == size_t(packet.getRawLength()));
|
||||
if (maximum) assert(length == 254);
|
||||
// Exercise real decoding/bounds too: feature gates must not alter packet
|
||||
// framing, transport codes, path bytes, payload or accepted wire lengths.
|
||||
mesh::Packet decoded;
|
||||
assert(decoded.readFrom(wire, uint8_t(length)));
|
||||
assert(decoded.header == packet.header && decoded.path_len == packet.path_len);
|
||||
assert(decoded.payload_len == packet.payload_len);
|
||||
assert(!std::memcmp(decoded.payload, packet.payload, packet.payload_len));
|
||||
assert(!decoded.readFrom(wire, 1));
|
||||
auto invalid = std::vector<uint8_t>(wire, wire + length);
|
||||
invalid[5] = 0xC1; // reserved four-byte path mode
|
||||
assert(!decoded.readFrom(invalid.data(), uint8_t(invalid.size())));
|
||||
|
||||
reset();
|
||||
mesh::Dispatcher dispatcher;
|
||||
dispatcher.outbound = &packet;
|
||||
assert(dispatcher.startOutboundTransmit());
|
||||
assert(dispatcher.radio.raw == std::vector<uint8_t>(wire, wire + length));
|
||||
assert(log_port.output == "T" + hex(wire, length) + "\n");
|
||||
assert(mesh::serialLogDroppedCount() == 0);
|
||||
|
||||
reset();
|
||||
mesh::logging_enabled = false;
|
||||
repeater.logRxRaw(0, -100, maximum_rx, sizeof(maximum_rx));
|
||||
assert(dispatcher.startOutboundTransmit());
|
||||
assert(log_port.output.empty());
|
||||
assert(dispatcher.radio.raw == std::vector<uint8_t>(wire, wire + length));
|
||||
|
||||
reset();
|
||||
dispatcher.radio.send_ok = false;
|
||||
assert(!dispatcher.startOutboundTransmit());
|
||||
assert(log_port.output.empty()); // Never claim a TX which did not start.
|
||||
}
|
||||
|
||||
// An initially undrained host must cost neither time nor a false drop.
|
||||
reset();
|
||||
log_port.stalled = true;
|
||||
repeater.logRxRaw(0, -100, raw, sizeof(raw));
|
||||
assert(log_port.output.empty() && g_mock_millis == 100);
|
||||
assert(mesh::serialLogDroppedCount() == 0);
|
||||
|
||||
// Once a host was observed, refused RX and TX records count as lost. RF
|
||||
// transmission still proceeds; recovery exposes the existing DROP marker.
|
||||
reset();
|
||||
repeater.logRxRaw(0, -100, raw, sizeof(raw));
|
||||
log_port.output.clear();
|
||||
log_port.stalled = true;
|
||||
repeater.logRxRaw(0, -100, raw, sizeof(raw));
|
||||
mesh::Packet packet = makePacket();
|
||||
mesh::Dispatcher dispatcher;
|
||||
dispatcher.outbound = &packet;
|
||||
assert(dispatcher.startOutboundTransmit() && dispatcher.radio.sends == 1);
|
||||
assert(log_port.output.empty() && g_mock_millis == 100);
|
||||
assert(mesh::serialLogDroppedCount() == 2);
|
||||
log_port.stalled = false;
|
||||
repeater.logRxRaw(0, -100, raw, sizeof(raw));
|
||||
assert(log_port.output == "DROP:2\r\nR00ABFF\n");
|
||||
|
||||
// Keep the bounded writer's existing mid-record stall behavior explicit.
|
||||
reset();
|
||||
mesh::serialLogPortSeen() = true;
|
||||
log_port.wedge_after = 14;
|
||||
repeater.logRxRaw(0, -100, maximum_rx, sizeof(maximum_rx));
|
||||
assert(log_port.output.size() == 14);
|
||||
assert(g_mock_millis - 100 <= SERIAL_LOG_WRITE_BUDGET_MS + 1);
|
||||
assert(mesh::serialLogDroppedCount() == 1);
|
||||
std::puts("PASS: production compact repeater RX/TX logging survives all feature gates");
|
||||
}
|
||||
'''
|
||||
|
||||
|
||||
class MiniRepeaterLoggingTest(unittest.TestCase):
|
||||
def test_actual_raw_rx_tx_packet_codec_and_backpressure_with_all_gate_combinations(self):
|
||||
compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++")
|
||||
self.assertIsNotNone(compiler, "a host C++17 compiler is required")
|
||||
packet = (ROOT / "src/Packet.cpp").read_text()
|
||||
packet_methods = "\n".join(body(packet, signature) for signature in (
|
||||
"Packet::Packet()", "bool Packet::isValidPathLen(",
|
||||
"size_t Packet::writePath(", "int Packet::getRawLength() const",
|
||||
"uint8_t Packet::writeTo(", "bool Packet::readFrom("))
|
||||
dispatcher = (ROOT / "src/Dispatcher.cpp").read_text()
|
||||
repeater = (ROOT / "examples/simple_repeater/MyMesh.cpp").read_text()
|
||||
program = HARNESS.replace("@PACKET_METHODS@", packet_methods).replace(
|
||||
"@TX_METHOD@", body(dispatcher, "bool Dispatcher::startOutboundTransmit()"))
|
||||
program = program.replace("@RX_METHOD@", body(repeater, "void MyMesh::logRxRaw("))
|
||||
header = (ROOT / "examples/simple_repeater/MyMesh.h").read_text()
|
||||
program = program.replace("@INLINE_CALLBACKS@", "\n".join(body(header, signature) for signature in (
|
||||
"void setLoggingOn(", "void eraseLogFile()")))
|
||||
program = program.replace("@FILE_METHODS@", "\n".join(body(repeater, signature) for signature in (
|
||||
"void MyMesh::logRx(", "void MyMesh::logTx(", "void MyMesh::logTxFail(",
|
||||
"void MyMesh::onFloodRetryEvent(")))
|
||||
sanitizers = (["-fsanitize=address,undefined", "-fno-sanitize-recover=all",
|
||||
"-fno-pie", "-no-pie"] if sys.platform.startswith("linux") else [])
|
||||
with tempfile.TemporaryDirectory(prefix="mini-experiment-logging-") as directory:
|
||||
source, executable = Path(directory) / "test.cpp", Path(directory) / "test"
|
||||
source.write_text(program, encoding="ascii")
|
||||
for carrier, management, file_logging in itertools.product((0, 1), repeat=3):
|
||||
with self.subTest(carrier_tools=carrier, management_reporting=management,
|
||||
packet_file_logging=file_logging):
|
||||
command = [compiler, "-std=c++17", "-O1", "-Wall", "-Wextra", "-Werror",
|
||||
"-Wno-unused-parameter", "-Wno-unused-variable",
|
||||
"-Wno-unused-but-set-variable", "-Wno-sign-compare", *sanitizers,
|
||||
"-DARDUINO=1", "-DSTM32_PLATFORM=1", "-DMESH_DEBUG=0",
|
||||
"-DMESH_PACKET_LOGGING=1", "-DMESH_PACKET_LOGGING_COMPACT=1",
|
||||
f"-DMESH_ENABLE_CARRIER_TOOLS={carrier}",
|
||||
f"-DMESH_ENABLE_MANAGEMENT_REPORTING={management}",
|
||||
f"-DMESH_ENABLE_PACKET_FILE_LOGGING={file_logging}",
|
||||
f"-DEXPECT_CARRIER={carrier}", f"-DEXPECT_MANAGEMENT={management}",
|
||||
f"-DEXPECT_FILE_LOGGING={file_logging}",
|
||||
"-I", str(ROOT / "test/mocks"), "-I", str(ROOT / "src"),
|
||||
str(source), "-o", str(executable)]
|
||||
built = subprocess.run(command, capture_output=True, text=True, timeout=60)
|
||||
self.assertEqual(built.returncode, 0, built.stdout + built.stderr)
|
||||
tested = subprocess.run([str(executable)], capture_output=True, text=True, timeout=15)
|
||||
self.assertEqual(tested.returncode, 0, tested.stdout + tested.stderr)
|
||||
self.assertIn("PASS: production compact repeater RX/TX", tested.stdout)
|
||||
|
||||
def test_baseline_storage_logging_host_cli_and_authentication_contracts_remain(self):
|
||||
config = configparser.ConfigParser(interpolation=None, inline_comment_prefixes=(";",))
|
||||
config.read(ROOT / "variants/wio-e5-mini/platformio.ini")
|
||||
variant = config["env:wio-e5-mini_repeater"]
|
||||
self.assertEqual(variant.getint("board_upload.maximum_size"), 245760)
|
||||
flags = variant["build_flags"]
|
||||
self.assertRegex(flags, r"-D\s+LFS_FLASH_TOTAL_SIZE=16384\b")
|
||||
self.assertRegex(flags, r"-D\s+MESH_PACKET_LOGGING_COMPACT=1\b")
|
||||
self.assertNotRegex(flags, r"-D\s+MESH_ENABLE_HOST_CLI=0\b")
|
||||
self.assertNotRegex(flags, r"-D\s+MESH_PACKET_LOGGING=0\b")
|
||||
header = (ROOT / "examples/simple_repeater/MyMesh.h").read_text()
|
||||
self.assertIn("#ifndef MESH_ENABLE_HOST_CLI\n#define MESH_ENABLE_HOST_CLI 1\n#endif", header)
|
||||
file_config = (ROOT / "src/helpers/PacketFileLoggingConfig.h").read_text()
|
||||
self.assertRegex(file_config, r"#ifndef MESH_ENABLE_PACKET_FILE_LOGGING\s+"
|
||||
r"#define MESH_ENABLE_PACKET_FILE_LOGGING 1\s+#endif")
|
||||
source = (ROOT / "examples/simple_repeater/MyMesh.cpp").read_text()
|
||||
dispatch = body(source, "void MyMesh::handleCommand(uint32_t sender_timestamp, ClientInfo* sender,")
|
||||
self.assertLess(dispatch.index("handleReplayResetCommand(sender, command, reply, usb_origin)"),
|
||||
dispatch.index("_cli.handleCommand("))
|
||||
self.assertIn("if (sender != nullptr && sender_timestamp == 0) sender_timestamp = 1;", dispatch)
|
||||
receive = source[source.index("else if (type == PAYLOAD_TYPE_TXT_MSG && len > 5"):]
|
||||
self.assertIn("client->isAdmin() || client->isRegionMgr() || client->isFilterMgr()", receive)
|
||||
self.assertIn("sender_timestamp < client->last_timestamp && !cached_retry", receive)
|
||||
host_complete = body(source, "bool MyMesh::completeHostCliRequest(")
|
||||
self.assertIn("client_index < 0 || !acl.getClientByIdx(client_index)->isAdmin()", host_complete)
|
||||
host_reply = body(source, "bool MyMesh::handleHostCliSerialReply(")
|
||||
self.assertIn("parsed.request_id != deferred_cli_command.request_id", host_reply)
|
||||
self.assertIn("parsed.request_nonce != host_cli_nonce", host_reply)
|
||||
self.assertIn("else if (!host_cli_claimed)", host_reply)
|
||||
deferred = body(source, "void __attribute__((noinline)) MyMesh::processDeferredCliCommand()")
|
||||
self.assertIn("if (!client->isAdmin())", deferred)
|
||||
self.assertIn("self_id.sign(signature, (const uint8_t*)signed_content,", deferred)
|
||||
common = body((ROOT / "src/helpers/CommonCLI.cpp").read_text(), "void CommonCLI::handleCommand(")
|
||||
self.assertIn("mesh::handleUsbLoggingWatchdogCommand(", common)
|
||||
self.assertIn("_radio_profiles.handle(", common)
|
||||
self.assertIn('#endif\n } else if (sender_timestamp == 0 && memcmp(command, "stats-packets",', common)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -176,12 +176,15 @@ template<typename T,typename=void> struct HasReceiveRead : std::false_type {};
|
||||
template<typename T> struct HasReceiveRead<T,std::void_t<decltype(&T::readMorseRssi)>> : std::true_type {};
|
||||
template<typename T,typename=void> struct HasReceiveBandwidth : std::false_type {};
|
||||
template<typename T> struct HasReceiveBandwidth<T,std::void_t<decltype(&T::morseReceiverBandwidthKHz)>> : std::true_type {};
|
||||
template<typename T,typename=void> struct HasReceiveService : std::false_type {};
|
||||
template<typename T> struct HasReceiveService<T,std::void_t<decltype(&T::serviceMorseRx)>> : std::true_type {};
|
||||
int main() {
|
||||
static_assert(MESH_ENABLE_MORSE_RX==EXPECTED_RX,"STM32 RX opt-in must be explicit");
|
||||
static_assert(HasCapturePointer<STMReceive>::value==bool(EXPECTED_RX),"disabled RX must exclude capture metadata");
|
||||
static_assert(HasCaptureAllocator<STMReceive>::value==bool(EXPECTED_RX),"disabled RX must exclude allocation machinery");
|
||||
static_assert(HasCaptureOwner<STMReceive>::value==bool(EXPECTED_RX),"disabled RX must exclude ownership state");
|
||||
static_assert(HasCaptureReadErrors<STMReceive>::value==bool(EXPECTED_RX),"disabled RX must exclude read-error state");
|
||||
static_assert(HasReceiveService<STMReceive>::value==bool(EXPECTED_RX),"disabled RX must exclude its service implementation");
|
||||
#if MESH_ENABLE_MORSE_RX || !defined(STM32_PLATFORM)
|
||||
constexpr bool expected_native_hooks=true;
|
||||
#else
|
||||
@@ -203,7 +206,6 @@ int main() {
|
||||
assert(!w.supportsMorseRx());
|
||||
assert(w.startMorseRx(902.1f,25,60000)==mesh::MorseRxResult::Unsupported);
|
||||
assert(!w.chip.standby_calls && !w.board.awake);
|
||||
assert(!w.serviceMorseRx());
|
||||
assert(w.stopMorseRx()==mesh::MorseRxResult::Unsupported);
|
||||
mesh::MorseRxDiagnostics stats;
|
||||
assert(!w.morseRxDiagnostics(stats) && !w.morseCaptureGeneration() && !w.morseCaptureLength());
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
"""Run the production shared profile command parser and persistence with a memory FS."""
|
||||
from pathlib import Path
|
||||
import itertools
|
||||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
@@ -14,10 +15,11 @@ class RadioProfileCLITest(unittest.TestCase):
|
||||
with tempfile.TemporaryDirectory() as work:
|
||||
# Compact STM32 images omit fleet staging. Both paths must preserve
|
||||
# local schedules, persistence retries and temporary lease bounds.
|
||||
for fleet_enabled in (False, True):
|
||||
exe = Path(work) / f'profiles-{int(fleet_enabled)}.exe'
|
||||
for fleet_enabled, carrier_enabled in itertools.product((False, True), repeat=2):
|
||||
exe = Path(work) / f'profiles-{int(fleet_enabled)}-{int(carrier_enabled)}.exe'
|
||||
subprocess.run([compiler, '-std=c++17', '-Wall', '-Wextra',
|
||||
f'-DMESH_ENABLE_FLEET_CONTROL={int(fleet_enabled)}',
|
||||
f'-DMESH_ENABLE_CARRIER_TOOLS={int(carrier_enabled)}',
|
||||
'-I', str(ROOT/'test/fixtures/radio_profiles/mocks'),
|
||||
'-I', str(ROOT/'test/mocks'), '-I', str(ROOT/'src'),
|
||||
str(ROOT/'src/helpers/RadioProfileCLI.cpp'),
|
||||
@@ -26,7 +28,8 @@ class RadioProfileCLITest(unittest.TestCase):
|
||||
for scenario in (None, 'permanent_order', 'temporary_boundary', 'lease_creation',
|
||||
'save_failure', 'schedule_permutations', 'storage_isolation',
|
||||
'relative_schedule'):
|
||||
with self.subTest(fleet=fleet_enabled, scenario=scenario or 'commands_persistence_and_timers'):
|
||||
with self.subTest(fleet=fleet_enabled, carrier=carrier_enabled,
|
||||
scenario=scenario or 'commands_persistence_and_timers'):
|
||||
subprocess.run([str(exe)] + ([scenario] if scenario else []), check=True)
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
@@ -21,6 +21,7 @@ class STM32CompanionProfileTest(unittest.TestCase):
|
||||
variant = "rak3x72"
|
||||
target = "RAK_3x72_companion_radio_usb"
|
||||
base = "rak3x72"
|
||||
extra_timer_targets = ()
|
||||
|
||||
def setUp(self):
|
||||
self.config = configparser.ConfigParser(
|
||||
@@ -28,12 +29,16 @@ class STM32CompanionProfileTest(unittest.TestCase):
|
||||
self.config.read(ROOT / "variants" / self.variant / "platformio.ini")
|
||||
self.companion = self.config["env:" + self.target]
|
||||
|
||||
def test_unused_timer_wrappers_are_removed_only_from_headless_companion(self):
|
||||
def test_unused_timer_wrappers_are_removed_only_from_approved_headless_roles(self):
|
||||
for flag in ("RADIOLIB_TONE_UNSUPPORTED", "HAL_TIM_MODULE_ONLY"):
|
||||
self.assertRegex(self.companion["build_flags"], r"-D\s+" + flag + r"\b")
|
||||
for section in self.config.sections():
|
||||
if section != self.companion.name:
|
||||
self.assertNotIn(flag, self.config[section].get("build_flags", ""))
|
||||
flags = self.config[section].get("build_flags", "")
|
||||
if section in self.extra_timer_targets:
|
||||
self.assertRegex(flags, r"-D\s+" + flag + r"\b")
|
||||
else:
|
||||
self.assertNotIn(flag, flags)
|
||||
|
||||
def test_flash_and_filesystem_boundaries_do_not_move(self):
|
||||
self.assertNotIn("board_upload.maximum_size", self.companion)
|
||||
@@ -93,6 +98,7 @@ class WioSTM32CompanionProfileTest(STM32CompanionProfileTest):
|
||||
variant = "wio-e5-mini"
|
||||
target = "wio-e5-mini_companion_radio_usb"
|
||||
base = "lora_e5_mini"
|
||||
extra_timer_targets = ("env:wio-e5-mini_repeater",)
|
||||
|
||||
def test_headless_button_controls_are_retained(self):
|
||||
self.assertRegex(self.companion["build_flags"],
|
||||
|
||||
@@ -0,0 +1,188 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Guard the constrained Mini Repeater recipe without bypassing size/storage gates."""
|
||||
|
||||
import configparser
|
||||
import re
|
||||
import runpy
|
||||
import shlex
|
||||
import subprocess
|
||||
import textwrap
|
||||
from pathlib import Path
|
||||
import unittest
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
TARGET = "wio-e5-mini_repeater"
|
||||
SECTION = "env:" + TARGET
|
||||
LINK_SIZE_FLAGS = (
|
||||
"-fno-caller-saves",
|
||||
"-fno-tree-dominator-opts",
|
||||
"-fno-reorder-blocks",
|
||||
"-fno-reorder-functions",
|
||||
"-fno-tree-switch-conversion",
|
||||
)
|
||||
OPTIONAL_CUTS = (
|
||||
"MESH_ENABLE_CARRIER_TOOLS",
|
||||
"MESH_ENABLE_MANAGEMENT_REPORTING",
|
||||
"MESH_ENABLE_PACKET_FILE_LOGGING",
|
||||
)
|
||||
|
||||
|
||||
def read_config(path):
|
||||
config = configparser.ConfigParser(
|
||||
interpolation=None, inline_comment_prefixes=(";",))
|
||||
config.read(path)
|
||||
return config
|
||||
|
||||
|
||||
class FakeEnvironment(dict):
|
||||
def Append(self, **values):
|
||||
for name, items in values.items():
|
||||
self.setdefault(name, []).extend(items)
|
||||
|
||||
def AppendUnique(self, **values):
|
||||
for name, items in values.items():
|
||||
current = self.setdefault(name, [])
|
||||
current.extend(item for item in items if item not in current)
|
||||
|
||||
|
||||
class MiniRepeaterProfileTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.config = read_config(ROOT / "variants/wio-e5-mini/platformio.ini")
|
||||
self.repeater = self.config[SECTION]
|
||||
self.flags = self.repeater["build_flags"]
|
||||
|
||||
def test_existing_flash_filesystem_and_neighbour_contracts_are_retained(self):
|
||||
self.assertEqual(self.repeater.getint("board_upload.maximum_size"), 245760)
|
||||
self.assertRegex(self.flags, r"-D\s+LFS_FLASH_TOTAL_SIZE=16384\b")
|
||||
self.assertRegex(self.flags, r"-D\s+MAX_NEIGHBOURS=50\b")
|
||||
self.assertEqual(245760 + 16384, 256 * 1024)
|
||||
self.assertEqual(self.repeater["platform_packages"].split(), [
|
||||
"platformio/toolchain-gccarmnoneeabi@^1.140201.0"])
|
||||
self.assertEqual(self.repeater["build_unflags"].split(), ["-Os"])
|
||||
self.assertIn("+<../examples/simple_repeater/*.cpp>",
|
||||
self.repeater["build_src_filter"])
|
||||
self.assertNotIn("MAX_CLIENTS", self.flags)
|
||||
self.assertNotIn("--no-check", self.repeater.get("extra_scripts", ""))
|
||||
|
||||
def test_three_opt_outs_apply_only_to_exact_mini_repeater(self):
|
||||
for macro in OPTIONAL_CUTS:
|
||||
self.assertRegex(self.flags, r"-D\s+" + macro + r"=0\b")
|
||||
for path in [ROOT / "platformio.ini", *sorted((ROOT / "variants").rglob("platformio.ini"))]:
|
||||
config = read_config(path)
|
||||
for name in config.sections():
|
||||
flags = config[name].get("build_flags", "")
|
||||
for macro in OPTIONAL_CUTS:
|
||||
with self.subTest(path=path.relative_to(ROOT), section=name, macro=macro):
|
||||
if path == ROOT / "variants/wio-e5-mini/platformio.ini" and name == SECTION:
|
||||
continue
|
||||
self.assertNotRegex(flags, r"-D\s*" + macro + r"=0\b")
|
||||
# Retain the profile's pre-existing trims, rather than adding unrelated cuts.
|
||||
disabled = set(re.findall(r"-D\s*(MESH_ENABLE_\w+)=0\b", self.flags))
|
||||
self.assertEqual(disabled, {*OPTIONAL_CUTS, "MESH_ENABLE_CLOCK_SYNC",
|
||||
"MESH_ENABLE_FLOOD_GROUP_MODERATION"})
|
||||
|
||||
def test_host_cli_live_logging_and_existing_core_features_remain(self):
|
||||
self.assertNotIn("MESH_ENABLE_HOST_CLI", self.flags)
|
||||
header = (ROOT / "examples/simple_repeater/MyMesh.h").read_text()
|
||||
self.assertRegex(header, r"#ifndef MESH_ENABLE_HOST_CLI\s+"
|
||||
r"#define MESH_ENABLE_HOST_CLI 1\b")
|
||||
self.assertNotRegex(self.flags, r"-D\s*(?:MESH_PACKET_LOGGING|MESH_USB_LOGGING_MERGED)=0\b")
|
||||
self.assertRegex(self.flags, r"-D\s+MESH_PACKET_LOGGING_COMPACT=1\b")
|
||||
self.assertRegex(self.flags, r"-D\s+MESH_ENABLE_RECENT_REPEATERS=1\b")
|
||||
# Numeric shortcut GET is deliberately not part of this size profile.
|
||||
self.assertNotIn("NUMERIC", self.flags)
|
||||
|
||||
def test_compact_crypto_and_unused_timer_wrappers_are_explicit(self):
|
||||
for macro in ("ED25519_COMPACT_SHA512=1", "ED25519_COMPACT_SCALAR=1",
|
||||
"ED25519_COMPACT_FIELD=1", "RADIOLIB_TONE_UNSUPPORTED",
|
||||
"HAL_TIM_MODULE_ONLY"):
|
||||
self.assertRegex(self.flags, r"-D\s+" + macro + r"\b")
|
||||
common = read_config(ROOT / "platformio.ini")
|
||||
self.assertRegex(common["stm32_base"]["build_flags"],
|
||||
r"-D\s+ED25519_COMPACT_BASE=1\b")
|
||||
self.assertIn("post:scripts/check_firmware_ram.py",
|
||||
common["stm32_base"]["extra_scripts"])
|
||||
|
||||
def test_measured_compiler_flags_reach_both_compile_and_lto_link(self):
|
||||
tokens = shlex.split(self.flags)
|
||||
for flag in ("-Oz", "-flto", *LINK_SIZE_FLAGS):
|
||||
self.assertIn(flag, tokens)
|
||||
scripts = self.repeater["extra_scripts"].split()
|
||||
self.assertIn("${stm32_base.extra_scripts}", scripts)
|
||||
self.assertIn("post:scripts/stm32_mini_size.py", scripts)
|
||||
helper = ROOT / "scripts/stm32_mini_size.py"
|
||||
for target in (TARGET, "wio-e5-mini_sensor", "wio-e5-mini_companion_radio_usb",
|
||||
"RAK_3x72_companion_radio_usb", "wio-e5_repeater"):
|
||||
with self.subTest(target=target):
|
||||
environment = FakeEnvironment(PIOENV=target, LINKFLAGS=["keep-existing-flag"])
|
||||
runpy.run_path(str(helper), init_globals={
|
||||
"env": environment,
|
||||
"Import": lambda name: self.assertEqual(name, "env"),
|
||||
})
|
||||
expected = ["keep-existing-flag", *(LINK_SIZE_FLAGS if target == TARGET else ())]
|
||||
self.assertEqual(environment["LINKFLAGS"], expected)
|
||||
self.assertEqual(set(environment), {"PIOENV", "LINKFLAGS"})
|
||||
|
||||
def test_capability_contract_declares_only_exact_mini_repeater_omissions(self):
|
||||
command = r'''
|
||||
set -euo pipefail
|
||||
source build.sh
|
||||
pio() { echo 'FORBIDDEN PlatformIO call in host capability test' >&2; return 99; }
|
||||
target="$1"
|
||||
BUILD_PROFILE_FOR_TARGET="$2"
|
||||
PIO_CONFIG_JSON='[]'
|
||||
PIO_ENV_PLATFORM_BY_NAME["$target"]=STM32_PLATFORM
|
||||
PIO_ENV_BOARD_BY_NAME["$target"]=lora_e5_mini
|
||||
BUILD_CAPABILITIES=()
|
||||
BUILD_EXPECTATIONS=()
|
||||
BUILD_APPLICATION_EXPECTATIONS=()
|
||||
BUILD_REDUCTIONS=(caller.record)
|
||||
PLATFORMIO_BUILD_FLAGS=-DKEEP_CALLER_FLAGS=1
|
||||
PLATFORMIO_BUILD_UNFLAGS=-DKEEP_CALLER_UNFLAGS=1
|
||||
declare_build_capability_contract "$target" STM32_PLATFORM
|
||||
[ "$PLATFORMIO_BUILD_FLAGS" = -DKEEP_CALLER_FLAGS=1 ]
|
||||
[ "$PLATFORMIO_BUILD_UNFLAGS" = -DKEEP_CALLER_UNFLAGS=1 ]
|
||||
printf '%s\n' "${BUILD_REDUCTIONS[@]}"
|
||||
'''
|
||||
omissions = [
|
||||
"radio.carrier_tools omitted to preserve the Mini Repeater app boundary",
|
||||
"management.reporting and data.tx omitted to preserve the Mini Repeater app boundary",
|
||||
"logging.file omitted; live USB/MQTT packet logging retained",
|
||||
]
|
||||
for target in (TARGET, "wio-e5-mini_companion_radio_usb", "wio-e5-mini_sensor",
|
||||
"wio-e5_repeater", "RAK_3x72_repeater", TARGET + "_custom"):
|
||||
for profile in ("standard", "full"):
|
||||
with self.subTest(target=target, profile=profile):
|
||||
result = subprocess.run(
|
||||
["bash", "-c", command, "mini-capability-contract", target, profile],
|
||||
cwd=ROOT, capture_output=True, text=True, check=True, timeout=10)
|
||||
self.assertEqual(result.stderr, "")
|
||||
self.assertEqual(result.stdout.splitlines(),
|
||||
["caller.record", *(omissions if target == TARGET else ())])
|
||||
|
||||
def test_ci_links_exact_cable_only_profile_with_release_logging_and_gates(self):
|
||||
workflow = (ROOT / ".github/workflows/run-unit-tests.yml").read_text()
|
||||
job = re.search(r"^ stm32-mini-repeater-smoke:\n(.*?)(?=^ [\w-]+:|\Z)",
|
||||
workflow, re.MULTILINE | re.DOTALL)
|
||||
self.assertIsNotNone(job)
|
||||
script = job[1].replace("\\\n", "")
|
||||
self.assertIn("set -euo pipefail", script)
|
||||
self.assertIn("bash build.sh build-firmware " + TARGET, script)
|
||||
self.assertIn("--firmware-version v1.18.0.1-halo-keymind-cascade-dev", script)
|
||||
self.assertIn("--radio-preset usa-cascade-fixed --profile cascade", script)
|
||||
self.assertIn("--allow-no-ota", script)
|
||||
self.assertIn("DISABLE_DEBUG: '0'", script)
|
||||
self.assertNotIn("--require-ota", script)
|
||||
self.assertNotIn("pio run", script)
|
||||
self.assertNotIn(" &", script)
|
||||
self.assertNotRegex(script, r"continue-on-error|maximum_size|LFS_FLASH_TOTAL_SIZE|MAX_NEIGHBOURS")
|
||||
for test in ("test_stm32_mini_repeater_profile.py", "test_mini_repeater_logging.py",
|
||||
"test_management_feature_gate.py", "test_carrier_wave.py"):
|
||||
self.assertIn("python3 -B test/" + test + " -v", script)
|
||||
command = textwrap.dedent(job[1].rsplit(" run: |\n", 1)[1])
|
||||
subprocess.run(["bash", "-n"], input=command, text=True, check=True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user