Files
MeshCore/test/test_cli_settings_contract.py
T
mikecarper 4f76345b67 Apply infrastructure radio changes after a two-second reply handoff
Persist set freq and set radio atomically, then apply through the existing safe radio scheduler without rebooting. Retain the old tuple until the delay and reply handoff finish, including queued room/sensor replies and TX-complete callbacks. End temporary sessions when accepting a permanent change.

Display applied RF settings even when radio2 is off. Share radio validation across immediate and scheduled setters to preserve STM32 Mini flash capacity. Update command documentation, regression coverage, and CI.

Validated production timing, persistence, display, CLI, fleet/reply, RX, sleep and Morse regressions. W12 repeater, room-server and sensor builds pass. Canonical STM32 Mini Repeater passes flash and RAM admission with 448 bytes of flash remaining. Physical W12 RF testing was not performed.
2026-10-26 00:38:13 -07:00

351 lines
21 KiB
Python

"""Audit CLI get/set coverage; execute migrated radio settings through real branches.
The inventory checks parser entry points, not documentation/whitelist mentions.
Read-only queries and settings written through another command have explicit
exceptions. Hardware callbacks are mocked in the native round-trip test.
"""
from pathlib import Path
import re
import shutil
import subprocess
import sys
import tempfile
import unittest
from test_replay_reset_integration import extract_braced
ROOT = Path(__file__).resolve().parents[1]
COMMON = ROOT / 'src/helpers/CommonCLI.cpp'
OBSERVER = ROOT / 'src/helpers/CommonCLI_Observer.cpp'
RADIO = ROOT / 'src/helpers/CommonRadioPrefs.cpp'
TOKEN = re.compile(r'//[^\n]*|/\*[\s\S]*?\*/|"(?:\\.|[^"\\])*"|\'(?:\\.|[^\'\\])*\'|[{}]')
def literal_keys(method, variable='config', top_level=True):
"""Extract literal key comparisons, excluding nested comparisons of values."""
matches = list(re.finditer(
r'(?:memcmp|strncmp|strcmp|configKeyEquals)\(' + variable +
r',\s*"([a-z][\w.]*)(?: [^"\n]*)?"', method))
depth, cursor, result = 0, 0, set()
tokens = list(TOKEN.finditer(method))
for match in matches:
while cursor < len(tokens) and tokens[cursor].start() < match.start():
depth += (tokens[cursor].group() == '{') - (tokens[cursor].group() == '}')
cursor += 1
if not top_level or depth == 1:
result.add(match[1])
return result
def methods(text, prefix):
return tuple(extract_braced(text, f'{prefix}{verb}Cmd(') for verb in ('Set', 'Get'))
def full_command_keys(text):
comparisons = re.findall(
r'(?:strcmp|strncmp|memcmp)\((?:command|cmd|text),\s*"(get|set) ([a-z][\w.]*)[ "]', text)
return tuple({key for kind, key in comparisons if kind == verb} for verb in ('set', 'get'))
COMMON_QUERY_EXCEPTIONS = {
'bootloader.ver': 'Installed bootloader identity',
'bridge.running': 'Live bridge state; set bridge.enabled controls intent',
'espnow.running': 'Live ESP-NOW bridge state; set espnow.enabled controls intent',
'rs232.running': 'Live UART bridge state; set rs232.enabled controls intent',
'rs232.status': 'Live heartbeat, negotiated standby and UART owner diagnostics',
'bridge.type': 'Build-selected bridge transport',
'password': 'Written by the password command, not set password',
'public.key': 'Derived from identity; written through set prv.key',
'pwrmgt.bootmv': 'Boot-time voltage measurement',
'pwrmgt.bootreason': 'Boot/reset cause',
'pwrmgt.source': 'Measured power source',
'pwrmgt.support': 'Hardware power-management capabilities',
'radio.rxps.config': 'Detailed view of set radio.rxps',
'radio.rxps.status': 'Live radio status for set radio.rxps',
'role': 'Build-selected firmware role',
'rxps.wd': 'Runtime RX power-saving recovery counters',
'tx.status': 'HIL-only applied-power verification for guarded RF tests',
'wifi.pwd': 'Setter is in the observer or standalone-WiFi delegate',
}
OBSERVER_QUERY_EXCEPTIONS = {
'link.dns': 'Live resolver list, gateway, and retained per-medium lease diagnostics',
'mqtt.config.valid': 'Configuration validation result',
'mqtt.enabled': 'Stored in common NodePrefs; set mqtt.enabled is handled by CommonCLI',
'mqtt.ntp.diag': 'NTP diagnostics',
'mqtt.presets': 'List of available presets',
'mqtt.running': 'Live state; set mqtt.enabled controls intent',
'mqtt.stopping': 'Pending worker-owned shutdown; set mqtt.enabled off requests it',
'mqtt.stats': 'Runtime MQTT statistics',
'wifi.status': 'Live connection status',
}
class CLISettingsContractTest(unittest.TestCase):
def test_shared_primary_parser_can_be_skipped_without_changing_other_commands(self):
harness = r'''
#include <cassert>
#include <cstdio>
#include <cstring>
#include "examples/companion_radio/NodePrefs.h"
int main() {
CompanionNodePrefs prefs;
prefs.freq = 915; prefs.bw = 125; prefs.sf = 7; prefs.cr = 5;
auto* radio = prefs.getRadioPrefs(); radio->clearDirty();
char reply[160] = "unchanged";
for (const char* text : {"set radio 916,125,8,6", "set radio invalid"}) {
assert(!radio->handleCommand(text, 0, reply, false));
assert(!strcmp(reply, "unchanged") && !radio->isDirty());
assert(prefs.freq == 915 && prefs.bw == 125 && prefs.sf == 7 && prefs.cr == 5);
}
assert(radio->handleCommand("get radio", 0, reply, false));
float f = 0, bw = 0; unsigned sf = 0, cr = 0;
assert(sscanf(reply, "> %f,%f,%u,%u", &f, &bw, &sf, &cr) == 4);
assert(f == 915 && bw == 125 && sf == 7 && cr == 5);
assert(radio->handleCommand("set af 2", 0, reply, false));
assert(!strcmp(reply, "OK") && prefs.airtime_factor == 2 && radio->isDirty());
radio->clearDirty();
// The three-argument/default path keeps the complete legacy parser.
assert(radio->handleCommand("set radio invalid", 0, reply));
assert(strstr(reply, "invalid radio") && !radio->isDirty());
assert(radio->handleCommand("set radio 916,125,8,6", 0, reply));
assert(!strncmp(reply, "OK", 2) && radio->isDirty());
assert(prefs.freq == 916 && prefs.bw == 125 && prefs.sf == 8 && prefs.cr == 6);
}
'''
with tempfile.TemporaryDirectory(prefix='common-radio-dispatch-') as directory:
work = Path(directory)
(work / 'test.cpp').write_text('#include <initializer_list>\n' + harness)
(work / 'Utils.h').write_text('''#pragma once
#include <Arduino.h>
#include <cassert>
namespace mesh { struct Utils {
static void printHex(Stream&,const uint8_t*,size_t){assert(false);}
static void fromHex(uint8_t*,size_t,const char*){assert(false);}
}; }
''')
sources = ['ConfigSerializer.cpp', 'DynamicConfigSerializer.cpp',
'CommonRadioPrefs.cpp', 'radiolib/RXPowerSaving.cpp',
'radiolib/RXPowerSavingCLI.cpp', 'TxtDataHelpers.cpp']
compiled = subprocess.run(['g++', '-std=c++17', '-DNRF52_PLATFORM=1',
'-fsanitize=address,undefined', '-fno-sanitize-recover=all', '-fno-pie', '-no-pie',
'-I', str(work), '-I', str(ROOT / 'test/fixtures/radio_profiles/mocks'),
'-I', str(ROOT / 'test/mocks'), '-I', str(ROOT / 'src'), '-I', str(ROOT),
str(work / 'test.cpp'), *(str(ROOT / 'src/helpers' / name) for name in sources),
'-o', str(work / 'test')], text=True, capture_output=True)
self.assertEqual(compiled.returncode, 0, compiled.stdout + compiled.stderr)
checked = subprocess.run([str(work / 'test')], text=True, capture_output=True)
self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr)
def test_feature_guarded_query_coverage(self):
compiler = shutil.which('g++') or shutil.which('clang++')
self.assertIsNotNone(compiler, 'Host C++ compiler required')
profiles = {
'minimal': [],
'hil': ['MESH_HIL_TX_FREQ_MHZ'],
'nrf52_gps_sd': ['NRF52_PLATFORM', 'ENV_INCLUDE_GPS', 'OTA_SD_STORE', 'NRF52_POWER_MANAGEMENT'],
'esp32_web': ['ESP_PLATFORM', 'ESP32_PLATFORM', 'ADMIN_PASSWORD', 'MESH_USB_LOGGING_AVAILABLE'],
'esp32_g3': ['ESP_PLATFORM', 'ESP32_PLATFORM', 'STATION_G3_ESP32'],
'esp32_mqtt': ['ESP_PLATFORM', 'ESP32_PLATFORM', 'WITH_MQTT_BRIDGE', 'WITH_BRIDGE',
'WITH_MQTT_NEIGHBORS', 'MESH_USB_LOGGING_AVAILABLE'],
'esp32_mqtt_espnow': ['ESP_PLATFORM', 'ESP32_PLATFORM', 'WITH_MQTT_BRIDGE',
'WITH_ESPNOW_BRIDGE', 'WITH_BRIDGE'],
'rs232_gps': ['WITH_BRIDGE', 'WITH_RS232_BRIDGE', 'ENV_INCLUDE_GPS'],
'rs232_espnow': ['ESP_PLATFORM', 'ESP32_PLATFORM', 'WITH_BRIDGE',
'WITH_RS232_BRIDGE', 'WITH_ESPNOW_BRIDGE'],
'mqtt_rs232_espnow': ['ESP_PLATFORM', 'ESP32_PLATFORM', 'WITH_BRIDGE',
'WITH_MQTT_BRIDGE', 'WITH_RS232_BRIDGE', 'WITH_ESPNOW_BRIDGE'],
'mqtt_rs232': ['ESP_PLATFORM', 'ESP32_PLATFORM', 'WITH_BRIDGE',
'WITH_MQTT_BRIDGE', 'WITH_RS232_BRIDGE'],
'espnow': ['ESP_PLATFORM', 'ESP32_PLATFORM', 'WITH_BRIDGE', 'WITH_ESPNOW_BRIDGE', 'MESH_PRIMARY_ESPNOW'],
'lr2021': ['USE_LR2021'],
}
for path, prefix, exceptions in (
(COMMON, 'void CommonCLI::handle', COMMON_QUERY_EXCEPTIONS),
(OBSERVER, 'bool CommonCLI::handleObserver', OBSERVER_QUERY_EXCEPTIONS),
):
original = '\n'.join(methods(path.read_text(encoding='utf-8'), prefix))
for profile, macros in profiles.items():
with self.subTest(parser=path.name, profile=profile):
# Preprocess the production bodies so a setter behind a
# different feature guard cannot satisfy a visible getter.
result = subprocess.run(
[compiler, '-E', '-P', '-x', 'c++', *[f'-D{key}=1' for key in macros], '-'],
input=original, text=True, capture_output=True, check=True)
setter, getter = methods(result.stdout, prefix)
sets, gets = literal_keys(setter), literal_keys(getter)
gets = {'flood.retry.bucket' if k == 'flood.retry.bucket.' else k for k in gets}
self.assertFalse(gets - sets - set(exceptions))
if path == COMMON:
self.assertIn('path.hash.mode', sets)
def test_common_and_observer_query_coverage(self):
for path, prefix, exceptions in (
(COMMON, 'void CommonCLI::handle', COMMON_QUERY_EXCEPTIONS),
(OBSERVER, 'bool CommonCLI::handleObserver', OBSERVER_QUERY_EXCEPTIONS),
):
with self.subTest(parser=path.name):
setter, getter = methods(path.read_text(encoding='utf-8'), prefix)
sets, gets = literal_keys(setter), literal_keys(getter)
# The query embeds the bucket index in the key; the setter
# accepts it as a separate argument.
gets = {'flood.retry.bucket' if k == 'flood.retry.bucket.' else k for k in gets}
self.assertEqual(gets - sets, set(exceptions),
f'{path.name}: missing setter or undocumented read-only query')
def test_mqtt_slot_query_coverage(self):
setter, getter = methods(OBSERVER.read_text(encoding='utf-8'), 'bool CommonCLI::handleObserver')
sets = literal_keys(setter, 'subcmd', False)
gets = literal_keys(getter, 'subcmd', False)
self.assertGreater(len(sets), 5, 'MQTT slot parser moved; update the audit')
self.assertEqual(gets - sets, {'diag'})
def test_every_migrated_radio_setter_is_reachable_in_infrastructure(self):
common_set, common_get = methods(COMMON.read_text(encoding='utf-8'), 'void CommonCLI::handle')
shared = extract_braced(RADIO.read_text(encoding='utf-8'), 'bool CommonRadioPrefs::handleCommand(')
migrated = set(re.findall(r'startsWith\(command, "set ([\w.]+) ', shared))
self.assertEqual(len(migrated), 13, 'Review changes to the shared radio command surface')
self.assertFalse(migrated - literal_keys(common_set))
self.assertFalse(migrated - literal_keys(common_get))
def test_split_parser_delegates_are_wired(self):
common = COMMON.read_text(encoding='utf-8')
setter, getter = methods(common, 'void CommonCLI::handle')
self.assertIn('handleObserverSetCmd(sender_timestamp, config, reply)', setter)
self.assertIn('handleObserverGetCmd(sender_timestamp, config, reply)', getter)
for role, filename in (('simple_repeater', 'MyMesh.cpp'),
('simple_room_server', 'MyMesh.cpp'),
('simple_sensor', 'SensorMesh.cpp')):
text = (ROOT/'examples'/role/filename).read_text(encoding='utf-8')
self.assertIn('_cli.handleCommand(sender_timestamp, command, reply)', text)
for verb in ('get', 'set'):
block = extract_braced(common, f'if (memcmp(command, "{verb} ", 4) == 0)')
self.assertIn(f'handle{verb.title()}Cmd(sender_timestamp, command, reply)', block)
def test_companion_allowlist_is_backed_by_shared_parser(self):
text = (ROOT/'examples/companion_radio/MyMesh.cpp').read_text(encoding='utf-8')
allowlist = extract_braced(text, 'static bool isCompanionRadioPrefsCommand(')
shared = extract_braced(RADIO.read_text(encoding='utf-8'), 'bool CommonRadioPrefs::handleCommand(')
for verb, key in re.findall(r'"(get|set) ([\w.]+) ?"', allowlist):
self.assertIn(f'"{verb} {key}' + (' ' if verb == 'set' else '') + '"', shared)
handler = extract_braced(text, 'bool MyMesh::handleCommand(')
self.assertIn('_prefs.getRadioPrefs()->handleCommand(', handler)
self.assertIn('_prefs.getRadioPrefs()->isDirty()', handler)
self.assertIn('savePrefs()', handler)
def test_role_query_coverage(self):
common_set, _ = methods(COMMON.read_text(encoding='utf-8'), 'void CommonCLI::handle')
common_keys = literal_keys(common_set)
for role, filename, exceptions in (
('simple_repeater', 'MyMesh.cpp', {
'acl', # ACL list, changed with ACL commands
'battery.alert.region', # set battery.alert on <region>
'clock.sync', # clock-sync delegate
'host', # Runtime host configuration view
'recent.repeaters', # Discovery/cache query
}),
('simple_room_server', 'MyMesh.cpp', {'acl'}),
('simple_sensor', 'SensorMesh.cpp', {
'acl',
'tracker', # Aggregate status; set tracker.<key> changes settings
}),
):
text = (ROOT/'examples'/role/filename).read_text(encoding='utf-8')
sets, gets = full_command_keys(text)
with self.subTest(role=role):
self.assertEqual(gets - sets - common_keys, exceptions)
companion = (ROOT/'examples/companion_radio/MyMesh.cpp').read_text(encoding='utf-8')
sets, gets = full_command_keys(companion)
terminal = extract_braced(companion, 'void MyMesh::handleTerminalCommand(')
terminal_set = extract_braced(terminal, 'if (strncmp(command, "set ", 4) == 0)')
sets |= literal_keys(terminal_set)
shared = extract_braced(companion, 'static bool isCompanionRadioPrefsCommand(')
sets |= {k for verb, k in re.findall(r'"(get|set) ([\w.]+) ?"', shared) if verb == 'set'}
self.assertEqual(gets - sets, {
'bootloader.ver', # Installed bootloader identity
'contact.cache', 'contact.cache.timing', # Contact-storage diagnostics
'display.wifi', # Connection/display status
'wifi.tcp', # Read-only app listener/traffic diagnostic
'dm.held', # Read-only pending-DM count
'flash.health', # Read-only flash health diagnostics
'key.backup.transport', # Identity-backup transport capability
'mqtt', 'mqtt.running', # MQTT connection status
'password', 'prv.key', # Local secret/identity reads
'public.key', # Read-only public identity
'pwrmgt.bootreason', 'role', # Boot/build facts
'radio.rxps.config', # Detail view of radio.rxps
'tx.status', # HIL-only applied-power verification
'wifi.status', 'wifi.ip', # Live WiFi link state
})
def test_literal_key_comparison_lengths(self):
for path in (COMMON, OBSERVER):
text = path.read_text(encoding='utf-8')
for match in re.finditer(r'(?:memcmp|strncmp)\((?:config|subcmd),\s*"([^"\n]*)",\s*(\d+)\)', text):
key, length = match[1], int(match[2])
with self.subTest(file=path.name, key=key):
# Some exact getters compare the terminating NUL too.
self.assertIn(length, (len(key), len(key) + 1))
if key.endswith(' '):
self.assertEqual(length, len(key), 'Setter must not compare the value against NUL')
def test_native_radio_setting_round_trips(self):
compiler = shutil.which('g++') or shutil.which('clang++')
self.assertIsNotNone(compiler, 'Host C++ compiler required')
source = COMMON.read_text(encoding='utf-8')
original = '\n'.join(methods(source, 'void CommonCLI::handle'))
keys = ['radio', 'freq', 'af', 'dutycycle', 'int.thresh', 'cad', 'radio.rxgain',
'tx', 'rxdelay', 'agc.reset.interval', 'multi.acks', 'txdelay', 'direct.txdelay',
'flood.max', 'flood.max.advert', 'flood.max.unscoped']
dispatch = ' else '.join(extract_braced(source, f'if (memcmp(command, "{verb} ", 4) == 0)')
for verb in ('get', 'set'))
template = (ROOT/'test/fixtures/cli_settings/main.cpp').read_text(encoding='utf-8')
with tempfile.TemporaryDirectory() as directory:
cpp, exe = Path(directory)/'test.cpp', Path(directory)/'test.exe'
for macros in ([], ['-DUSE_LR2021=1'], ['-DSTATION_G3_ESP32=1'],
['-DSTATION_G3_ESP32=1', '-DUSE_LR2021=1']):
with self.subTest(macros=macros):
# Resolve complete, balanced production methods first. A
# following board-only else-if can own the closing brace
# of extra.sf (or any selected arm). Extracting that arm
# before preprocessing would retain its unmatched #if.
result = subprocess.run(
[compiler, '-E', '-P', '-x', 'c++', *macros, '-'],
input=original, text=True, capture_output=True, check=True)
setter, getter = methods(result.stdout, 'void CommonCLI::handle')
# Keep the actual active conditions and bodies. Only
# unrelated settings needing hardware dependencies are
# omitted; board delegation has its own real G3 suite.
set_blocks = [extract_braced(setter, 'if (strncmp(config, "path.hash.mode", 14)')]
get_blocks = [extract_braced(getter, 'if (configKeyEquals(config, "path.hash.mode"))')]
set_blocks.append(extract_braced(setter, 'if (strcmp(config, "extra.sf") == 0 ||'))
get_blocks.append(extract_braced(getter, 'if (strcmp(config, "extra.sf") == 0)'))
for key in keys:
set_blocks.append(extract_braced(setter, f'if (memcmp(config, "{key} ", {len(key)+1}) == 0)'))
get_blocks.append(extract_braced(getter, f'if (configKeyEquals(config, "{key}"))'))
fixture = template.replace('@SET@', ' else '.join(set_blocks))
fixture = fixture.replace('@GET@', ' else '.join(get_blocks))
fixture = fixture.replace('@DISPATCH@', dispatch)
fixture = fixture.replace('@PRIMARY_VALIDATE@', extract_braced(
source, 'bool CommonCLI::acceptsPrimaryRadioParams(').replace('CommonCLI::', ''))
fixture = fixture.replace('@KEY_EQUALS@', extract_braced(source, 'static bool configKeyEquals('))
fixture = fixture.replace('@TX_METHODS@', '\n'.join(
extract_braced(source, signature).replace('CommonCLI::', 'CLI::')
for signature in ('void CommonCLI::applyAndSaveTxPower(',
'void CommonCLI::servicePendingTxPower(')))
fixture = fixture.replace('@RX_METHODS@', '\n'.join(
extract_braced(source, signature).replace('CommonCLI::', 'CLI::')
for signature in ('mesh::RadioParamApplyResult CommonCLI::applyRxSetting(',
'bool CommonCLI::saveRxSetting(',
'void CommonCLI::restoreRxSetting(')))
cpp.write_text(fixture, encoding='utf-8')
sanitizers = (['-fsanitize=address,undefined', '-fno-sanitize-recover=all',
'-fno-pie', '-no-pie'] if sys.platform.startswith('linux') else [])
subprocess.run([compiler, '-std=c++17', '-Wall', '-Wextra', '-Werror',
*sanitizers, *macros, '-I', str(ROOT/'src'), str(cpp),
'-o', str(exe)], check=True, timeout=60)
subprocess.run([str(exe)], check=True, timeout=15)
if __name__ == '__main__':
unittest.main()