"""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 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', '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', 'role': 'Build-selected firmware role', 'rxps.wd': 'Runtime RX power-saving recovery counters', 'wifi.pwd': 'Setter is in the observer or standalone-WiFi delegate', } OBSERVER_QUERY_EXCEPTIONS = { 'mqtt.config.valid': 'Configuration validation result', 'mqtt.ntp.diag': 'NTP diagnostics', 'mqtt.presets': 'List of available presets', 'mqtt.running': 'Live state; set mqtt.enabled controls intent', 'mqtt.stats': 'Runtime MQTT statistics', 'wifi.status': 'Live connection status', } class CLISettingsContractTest(unittest.TestCase): def test_feature_guarded_query_coverage(self): compiler = shutil.which('g++') or shutil.which('clang++') self.assertIsNotNone(compiler, 'Host C++ compiler required') profiles = { 'minimal': [], '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_mqtt': ['ESP_PLATFORM', 'ESP32_PLATFORM', 'WITH_MQTT_BRIDGE', 'WITH_BRIDGE', 'WITH_MQTT_NEIGHBORS', 'MESH_USB_LOGGING_AVAILABLE'], 'rs232_gps': ['WITH_BRIDGE', 'WITH_RS232_BRIDGE', 'ENV_INCLUDE_GPS'], '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 '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'}), ): 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, { 'contact.cache', 'contact.cache.timing', # Contact-storage diagnostics 'display.wifi', # Connection/display status 'mqtt', 'mqtt.running', # MQTT connection status 'password', 'prv.key', # Local secret/identity reads 'pwrmgt.bootreason', 'role', # Boot/build facts 'radio.rxps.config', # Detail view of radio.rxps 'wifi.pwd', # Saved transport credentials }) 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') setter, getter = 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'] # Preserve the actual conditions and bodies. The only excluded branches # are unrelated settings needing board-specific dependencies. 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: block = extract_braced(setter, f'if (memcmp(config, "{key} ", {len(key)+1}) == 0)') # A feature guard immediately before this closing brace controls # the following else-if arm, which is not part of this fixture. block = re.sub(r'\n#if[^\n]*\n\s*}$', '\n}', block) set_blocks.append(block) get_blocks.append(extract_braced(getter, f'if (configKeyEquals(config, "{key}"))')) dispatch = ' else '.join(extract_braced(source, f'if (memcmp(command, "{verb} ", 4) == 0)') for verb in ('get', 'set')) fixture = (ROOT/'test/fixtures/cli_settings/main.cpp').read_text(encoding='utf-8') fixture = fixture.replace('@SET@', ' else '.join(set_blocks)) fixture = fixture.replace('@GET@', ' else '.join(get_blocks)) fixture = fixture.replace('@DISPATCH@', dispatch) fixture = fixture.replace('@KEY_EQUALS@', extract_braced(source, 'static bool configKeyEquals(')) with tempfile.TemporaryDirectory() as directory: cpp, exe = Path(directory)/'test.cpp', Path(directory)/'test.exe' cpp.write_text(fixture, encoding='utf-8') for macros in ([], ['-DUSE_LR2021=1']): with self.subTest(macros=macros): subprocess.run([compiler, '-std=c++17', '-Wall', '-Wextra', '-Werror', *macros, '-I', str(ROOT/'src'), str(cpp), '-o', str(exe)], check=True) subprocess.run([str(exe)], check=True) if __name__ == '__main__': unittest.main()