mirror of
https://github.com/mikecarper/MeshCore.git
synced 2026-11-11 21:17:38 +00:00
mikecarper
cddc0b7350
Update reviewed build inventories, hosted documentation links and the CLI token fixture for the recent M33S, wardriving and clock changes. Keep advert signature validation compact and preserve all bounds checks. Use equivalent integer battery scaling and a scoped LTO loop policy to fit Tiny Relay Companion without reducing features, storage or capacities. Validated all five STM32 smoke images with flash and runtime RAM gates, 188 native routing/advert cases, sanitizer-based receive checks, exhaustive battery scaling, and the failing host check suites.
157 lines
8.5 KiB
Python
157 lines
8.5 KiB
Python
"""Keep constrained CLI prose separate from commands and protocol replies."""
|
|
from pathlib import Path
|
|
import configparser
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
FIXTURE = ROOT / 'test/fixtures/cli_reply_text.json'
|
|
SELECTION = re.compile(
|
|
r'MESH_CLI_REPLY_TEXT\(("(?:\\.|[^"\\])*"),\s*("(?:\\.|[^"\\])*")\)')
|
|
TOKENS = re.compile(r'"(?:\\.|[^"\\])*"|\'(?:\\.|[^\'\\])*\'|'
|
|
r'//[^\n]*|/\*.*?\*/|[A-Za-z_][A-Za-z_0-9]*|[^\s]', re.S)
|
|
|
|
|
|
def original_tokens(source):
|
|
source = re.sub(r'^#include [<"](?:helpers/)?CLIReplyText\.h[>"]\s*\n',
|
|
'', source, flags=re.M)
|
|
source = SELECTION.sub(lambda match: match.group(1), source)
|
|
return '\n'.join(token.group() for token in TOKENS.finditer(source)
|
|
if not token.group().startswith(('//', '/*'))).encode()
|
|
|
|
|
|
class CliReplyTextTests(unittest.TestCase):
|
|
def test_only_reviewed_human_literals_change_and_control_flow_stays_exact(self):
|
|
fixture = json.loads(FIXTURE.read_text())
|
|
self.assertEqual(fixture['schema'], 1)
|
|
expected = {}
|
|
for row in fixture['messages']:
|
|
expected.setdefault(row['file'], set()).add((row['verbose'], row['concise']))
|
|
old, new = row['verbose'], row['concise']
|
|
self.assertLess(len(new), len(old))
|
|
formats = r'%(?:[-+ #0]*[0-9.*]*[a-zA-Z]+|%)'
|
|
self.assertEqual(re.findall(formats, old), re.findall(formats, new))
|
|
category = r'^[ \t]*(Error|ERROR|ERR|Err|OK)(?=[: ,\-])'
|
|
self.assertEqual(re.findall(category, old), re.findall(category, new))
|
|
self.assertEqual(re.findall(r'[\r\n]', old), re.findall(r'[\r\n]', new))
|
|
self.assertEqual(re.findall(r'\d+(?:\.\d+)?', old),
|
|
re.findall(r'\d+(?:\.\d+)?', new))
|
|
self.assertFalse(old.startswith(('> ', '{', 'KEYBACKUP ', 'ACK')))
|
|
protected = old.lower().replace('[reboot]', '')
|
|
if (row['file'] == 'src/helpers/ota/OtaCli.cpp' and
|
|
old == 'ERR self serve requires ESP32 or external-storage nRF52; host sources use ota folder'):
|
|
# This describes source capabilities, not host data or its retention.
|
|
protected = protected.replace('host sources use ota folder', 'ota folder')
|
|
self.assertNotRegex(protected,
|
|
r'host|replay|password|login|save unconfirmed|reboot|'
|
|
r'retained|warning|applying in|restore scheduled')
|
|
for protected in ('[reboot]', 'default'):
|
|
if protected in old:
|
|
self.assertIn(protected, new)
|
|
if 'unchanged' in old:
|
|
self.assertIn('sleep policy unchanged', old)
|
|
self.assertIn('sleep policy unchanged', new)
|
|
for file, messages in expected.items():
|
|
source = (ROOT / file).read_text()
|
|
actual = {(json.loads(match.group(1)), json.loads(match.group(2)))
|
|
for match in SELECTION.finditer(source)}
|
|
self.assertEqual(actual, messages, file)
|
|
# Substituting each original literal recovers every original C++
|
|
# token, including command tests, branches, calls and arguments.
|
|
# Review unrelated behavior changes before refreshing a fixture
|
|
# hash; keep their tokens covered by this guard as well.
|
|
self.assertEqual(hashlib.sha256(original_tokens(source)).hexdigest(),
|
|
fixture['noncompact_tokens_sha256'][file], file)
|
|
|
|
def test_ota_prose_keeps_discovery_and_retry_parser_keywords(self):
|
|
messages = {row['verbose']: row['concise']
|
|
for row in json.loads(FIXTURE.read_text())['messages']
|
|
if row['file'] == 'src/helpers/ota/OtaCli.cpp'}
|
|
self.assertEqual(len(messages), 15)
|
|
self.assertGreaterEqual(sum(len(old) - len(new)
|
|
for old, new in messages.items()), 240)
|
|
for prefix in ('No updates seen yet', 'ERR no such update',
|
|
'no trusted signer keys'):
|
|
rows = [(old, new) for old, new in messages.items() if old.startswith(prefix)]
|
|
self.assertEqual(len(rows), 1, prefix)
|
|
self.assertTrue(rows[0][1].startswith(prefix), prefix)
|
|
for old, new in messages.items():
|
|
self.assertFalse(old.startswith(('OK', 'usage:', 'ERR usage:')))
|
|
for token in re.findall(r'`([^`]+)`', old):
|
|
self.assertIn(token, new)
|
|
if 'disabled' in old:
|
|
self.assertIn('disabled', new)
|
|
if 'cannot install' in old:
|
|
self.assertIn('cannot install', new)
|
|
if 'cannot update its bootloader' in old:
|
|
self.assertIn('cannot update its bootloader', new)
|
|
|
|
def test_profile_activation_is_explicit_and_default_is_off(self):
|
|
fixture = json.loads(FIXTURE.read_text())
|
|
enabled = []
|
|
for path in [ROOT / 'platformio.ini', *sorted((ROOT / 'variants').rglob('*.ini'))]:
|
|
config = configparser.ConfigParser(interpolation=None, strict=False)
|
|
config.read(path)
|
|
for section in config.sections():
|
|
if re.search(r'-D\s*MESH_CLI_COMPACT_REPLIES_DEFAULT=1\b',
|
|
config[section].get('build_flags', '')):
|
|
self.assertTrue(section.startswith('env:'), str(path))
|
|
enabled.append(section[4:])
|
|
self.assertEqual(sorted(enabled), sorted(fixture['profiles']))
|
|
header = (ROOT / 'src/helpers/CLIReplyText.h').read_text()
|
|
self.assertIn('#ifndef MESH_CLI_COMPACT_REPLIES\n', header)
|
|
self.assertIn('#define MESH_CLI_COMPACT_REPLIES 0', header)
|
|
|
|
def test_actual_header_selects_exact_text_without_a_runtime_lookup(self):
|
|
compiler = os.environ.get('CXX') or shutil.which('g++') or shutil.which('clang++')
|
|
if not compiler:
|
|
self.skipTest('host C++ compiler required')
|
|
messages = json.loads(FIXTURE.read_text())['messages']
|
|
selections = ',\n'.join('MESH_CLI_REPLY_TEXT(%s, %s)' %
|
|
(json.dumps(row['verbose']), json.dumps(row['concise']))
|
|
for row in messages)
|
|
source = ('#include <CLIReplyText.h>\n#include <cstdio>\n'
|
|
'#ifdef _WIN32\n#include <fcntl.h>\n#include <io.h>\n#endif\n'
|
|
'int main(){\n'
|
|
# Compare the selected literals without Windows stdio's
|
|
# newline conversion changing the executable's output.
|
|
'#ifdef _WIN32\n_setmode(_fileno(stdout),_O_BINARY);\n#endif\n'
|
|
'const char* text[]={'+selections+'};'
|
|
'for(const char* line:text)puts(line);}\n')
|
|
with tempfile.TemporaryDirectory(prefix='cli-reply-text-') as folder:
|
|
work = Path(folder)
|
|
executable = work / ('test.exe' if os.name == 'nt' else 'test')
|
|
(work / 'test.cpp').write_text(source)
|
|
for flags, key in (([], 'verbose'),
|
|
(['-DMESH_CLI_COMPACT_REPLIES=0'], 'verbose'),
|
|
(['-DMESH_CLI_COMPACT_REPLIES=1'], 'concise'),
|
|
(['-DMESH_CLI_COMPACT_REPLIES_DEFAULT=1'], 'concise'),
|
|
(['-DMESH_CLI_COMPACT_REPLIES_DEFAULT=1',
|
|
'-DMESH_CLI_COMPACT_REPLIES=0'], 'verbose'),
|
|
(['-DMESH_CLI_COMPACT_REPLIES_DEFAULT=0',
|
|
'-DMESH_CLI_COMPACT_REPLIES=1'], 'concise')):
|
|
built = subprocess.run([compiler, '-std=c++11', '-Os', '-Wall', '-Wextra',
|
|
'-Werror', *flags, '-I'+str(ROOT / 'src/helpers'),
|
|
str(work / 'test.cpp'), '-o', str(executable)], capture_output=True)
|
|
self.assertEqual(built.returncode, 0, built.stderr.decode(errors='replace'))
|
|
result = subprocess.run([str(executable)], capture_output=True)
|
|
self.assertEqual(result.returncode, 0, result.stderr.decode(errors='replace'))
|
|
self.assertEqual(result.stdout,
|
|
''.join(row[key] + '\n' for row in messages).encode())
|
|
if key == 'concise':
|
|
image = executable.read_bytes()
|
|
selected = {row['concise'] for row in messages}
|
|
for row in messages:
|
|
if row['verbose'] not in selected:
|
|
self.assertNotIn(row['verbose'].encode()+b'\0', image)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|