Share constrained STM32 code while preserving firmware features

Omit unbound forwarding-region callbacks only for exclusive Companion source roles. Reuse the compact neighbor sorter in repeaters for every RPC order and the CLI list, removing duplicate sort implementations while preserving storage and size gates.

test/test_client_acl_role.py

@@ -19,6 +19,33 @@ FIXTURE = ROOT / "test/fixtures/client_acl_spiffs"
 
 
 class ClientAclRoleTests(unittest.TestCase):
+    def test_complete_exclusive_companion_role_uses_one_global_definition(self):
+        definition = POLICY.COMPANION_DEFINITION
+        for role in ("+<../examples/companion_radio>",
+                     "+<../examples/companion_radio/*.cpp>",
+                     ["+<*.cpp>", "+<..\\examples\\companion_radio\\*.cpp>"]):
+            self.assertEqual(POLICY.companion_role_policy(role), {definition: 1})
+        for source in ("", "+<*>", "+<../examples/companion_radio/main.cpp>",
+                       "+<../examples/companion_radio> -<../examples/companion_radio/MyMesh.cpp>",
+                       "+<../examples/companion_radio> +<../examples/custom_role>",
+                       "+<../examples/simple_repeater>"):
+            self.assertEqual(POLICY.companion_role_policy(source), {})
+        for other in ("simple_repeater", *POLICY.OTHER_ROLES):
+            if other == "companion_radio":
+                continue
+            self.assertEqual(POLICY.companion_role_policy(
+                "+<../examples/companion_radio> +<../examples/" + other + ">"), {})
+        role = "+<../examples/companion_radio>"
+        for flag in ("-D" + definition + "=0", "-D" + definition + "=1", "-U" + definition):
+            self.assertEqual(POLICY.companion_role_policy(role, flag), {})
+        self.assertEqual(POLICY.companion_role_policy(role, cppdefines=[(definition, 1)]), {})
+        for value in ("true", "2", "-1"):
+            with self.assertRaises(ValueError):
+                POLICY.companion_role_policy(role, "-D" + definition + "=" + value)
+        with self.assertRaises(ValueError):
+            POLICY.companion_role_policy("+<../examples/simple_repeater>",
+                                         "-D" + definition + "=1")
+
     def test_only_complete_repeater_role_qualifies(self):
         for value in ("+<../examples/simple_repeater>", "+<../examples/simple_repeater/*.cpp>",
                       ["+<*.cpp>", "+<..\\examples\\simple_repeater\\*.cpp>"]):
@@ -57,6 +84,9 @@ class ClientAclRoleTests(unittest.TestCase):
         del env["appended"]
         POLICY.install(env)
         self.assertNotIn("appended", env)
+        env["SRC_FILTER"] = "+<../examples/companion_radio/*.cpp>"
+        POLICY.install(env)
+        self.assertEqual(env["appended"], {"CPPDEFINES": [(POLICY.COMPANION_DEFINITION, 1)]})
 
     def test_every_platform_base_installs_role_hook(self):
         ini = (ROOT / "platformio.ini").read_text()

test/test_radio_profile_cli.py

@@ -9,7 +9,7 @@ import unittest
 ROOT = Path(__file__).resolve().parents[1]
 
 class RadioProfileCLITest(unittest.TestCase):
-    def test_full_companion_keeps_rf_presets_without_unused_region_code(self):
+    def test_all_companions_keep_rf_presets_without_unused_region_code(self):
         compiler = shutil.which('g++') or shutil.which('clang++')
         self.assertIsNotNone(compiler)
         code = r'''
@@ -23,7 +23,7 @@ int main() {
   f.cmd("set radio2 mediamesh");
   f.advance(2000);
   assert(f.radio.p.enabled() && fabsf(f.radio.p.secondary.params.freq - 911.150f) < 0.0001f);
-#if defined(COMPANION_RADIO_FULL)
+#if defined(COMPANION_RADIO_FULL) || MESH_ROLE_COMPANION
   assert(!scope.commits && !scope.checks);
 #else
   assert(scope.enabled && scope.durable && scope.commits);
@@ -37,9 +37,11 @@ int main() {
         with tempfile.TemporaryDirectory(prefix='profile-region-role-') as work:
             source = Path(work)/'role.cpp'
             source.write_text(code, encoding='ascii')
-            for full in (False, True):
-                binary = Path(work)/('companion' if full else 'infrastructure')
-                flags = ['-DCOMPANION_RADIO_FULL=1'] if full else []
+            roles = {'infrastructure': [], 'full': ['-DCOMPANION_RADIO_FULL=1'],
+                     'ordinary': ['-DMESH_ROLE_COMPANION=1']}
+            for role, flags in roles.items():
+                companion = role != 'infrastructure'
+                binary = Path(work)/role
                 compiled = subprocess.run([
                     compiler, '-std=c++17', '-Os', '-ffunction-sections',
                     '-fdata-sections', '-Wl,--gc-sections',
@@ -53,11 +55,12 @@ int main() {
                 self.assertEqual(compiled.returncode, 0, compiled.stderr)
                 run = subprocess.run([str(binary)], capture_output=True, text=True, timeout=10)
                 self.assertEqual(run.returncode, 0, run.stdout+run.stderr)
-                sizes[full] = int(run.stdout)
+                sizes[role] = int(run.stdout)
                 image = binary.read_bytes()
                 for marker in (b'MediaMesh scope save pending', b'MediaMesh region unavailable or full'):
-                    self.assertEqual(marker in image, not full)
-        self.assertLess(sizes[True], sizes[False])
+                    self.assertEqual(marker in image, not companion)
+        for role in ('full', 'ordinary'):
+            self.assertLess(sizes[role], sizes['infrastructure'])
 
     def test_commands_persistence_and_timers(self):
         compiler = shutil.which('g++') or shutil.which('clang++')

test/test_small_table_sort.py

@@ -8,6 +8,8 @@ import sys
 import tempfile
 import unittest
 
+from cpp_source import body
+
 ROOT = Path(__file__).resolve().parents[1]
 
 HARNESS = r'''
@@ -140,8 +142,125 @@ void sortElsewhere(void* items, size_t count, size_t width,
 }
 '''
 
+NEIGHBOUR_HARNESS = r'''
+#include <algorithm>
+#include <cassert>
+#include <cstdint>
+#include <cstdio>
+#include <cstring>
+#include <numeric>
+#include <random>
+#include <vector>
+#include <helpers/SmallTableSort.h>
+struct NeighbourInfo {
+  uint32_t heard_timestamp;
+  int8_t snr;
+  uint16_t identity;
+};
+@COMPARATORS@
+static void rpcSort(NeighbourInfo** sorted_neighbours, int16_t neighbours_count,
+                    uint8_t order_by) {
+  @RPC_SORT@
+}
+static void cliSort(NeighbourInfo** sorted_neighbours, int16_t neighbours_count) {
+  @CLI_SORT@
+}
+static bool before(const NeighbourInfo* a, const NeighbourInfo* b, unsigned order) {
+  if (order == 0) return a->heard_timestamp > b->heard_timestamp;
+  if (order == 1) return a->heard_timestamp < b->heard_timestamp;
+  if (order == 2) return a->snr > b->snr;
+  return a->snr < b->snr;
+}
+int main() {
+  std::mt19937 random(0x18180002);
+  const uint32_t timestamps[] = {1, 42, 42, UINT32_MAX, 1, UINT32_MAX - 1};
+  const int8_t snrs[] = {INT8_MIN, -16, -1, 0, 0, 7, 7, INT8_MAX};
+  int (*const compare[])(const void*, const void*) = {
+    cmp_neighbours_newest_to_oldest, cmp_neighbours_oldest_to_newest,
+    cmp_neighbours_strongest_to_weakest, cmp_neighbours_weakest_to_strongest
+  };
+  for (size_t count : {0U, 1U, 2U, 16U, 50U, 100U}) {
+    std::vector<NeighbourInfo> records(count);
+    for (size_t i = 0; i < count; ++i) {
+      records[i].heard_timestamp = timestamps[i % 6];
+      records[i].snr = snrs[i % 8];
+      records[i].identity = static_cast<uint16_t>(i);
+    }
+    const auto original = records;
+    for (unsigned order = 0; order < 4; ++order) {
+      std::vector<NeighbourInfo*> actual(count), expected(count);
+      for (size_t i = 0; i < count; ++i) actual[i] = &records[i];
+      std::shuffle(actual.begin(), actual.end(), random);
+      expected = actual;
+      std::sort(expected.begin(), expected.end(), [order](const NeighbourInfo* a,
+                                                         const NeighbourInfo* b) {
+        return before(a, b, order);
+      });
+      rpcSort(actual.data(), static_cast<int16_t>(count), order);
+      std::vector<bool> seen(count);
+      for (size_t i = 0; i < count; ++i) {
+        const auto index = actual[i]->identity;
+        assert(index < count && !seen[index] && actual[i] == &records[index]);
+        seen[index] = true;
+        assert(!before(actual[i], expected[i], order));
+        assert(!before(expected[i], actual[i], order));
+        assert(compare[order](&actual[i], &expected[i]) == 0);
+        if (i) assert(!before(actual[i], actual[i - 1], order));
+      }
+      // Unsupported order values retain the caller's original pointer order.
+      const auto unchanged = actual;
+      rpcSort(actual.data(), static_cast<int16_t>(count), 4);
+      assert(actual == unchanged);
+      cliSort(actual.data(), static_cast<int16_t>(count));
+      for (size_t i = 1; i < count; ++i) {
+        assert(actual[i - 1]->heard_timestamp >= actual[i]->heard_timestamp);
+      }
+      for (size_t i = 0; i < count; ++i) {
+        assert(records[i].heard_timestamp == original[i].heard_timestamp);
+        assert(records[i].snr == original[i].snr);
+        assert(records[i].identity == original[i].identity);
+      }
+    }
+  }
+  puts("PASS: real neighbour RPC orders and CLI, duplicate keys, negative SNR and pointer records");
+}
+'''
+
 
 class SmallTableSortTests(unittest.TestCase):
+    def test_real_neighbour_rpc_and_cli_orderings(self):
+        compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++")
+        self.assertIsNotNone(compiler, "a host C++17 compiler is required")
+        production = (ROOT / "examples/simple_repeater/MyMesh.cpp").read_text()
+        names = ("newest_to_oldest", "oldest_to_newest",
+                 "strongest_to_weakest", "weakest_to_strongest")
+        comparators = "\n".join(body(production, "static int cmp_neighbours_" + name + "(")
+                                for name in names)
+        start = production.index("if (order_by == 0) {")
+        rpc_sort = production[start:production.index("// build results buffer", start)]
+        formatter = body(production, "void MyMesh::formatNeighborsReply(")
+        start = formatter.index("// sort neighbours newest to oldest")
+        cli_sort = formatter[start:formatter.index("for (int i = 0;", start)]
+        source = NEIGHBOUR_HARNESS.replace("@COMPARATORS@", comparators)
+        source = source.replace("@RPC_SORT@", rpc_sort).replace("@CLI_SORT@", cli_sort)
+        for platform in ("STM32_PLATFORM", "NRF52_PLATFORM", "ESP32_PLATFORM", "RP2040_PLATFORM"):
+            with self.subTest(platform=platform), tempfile.TemporaryDirectory(
+                    prefix="neighbour-sort-") as directory:
+                work = Path(directory)
+                (work / "test.cpp").write_text(source)
+                binary = work / "sort"
+                command = [compiler, "-std=c++17", "-O2", "-Wall", "-Wextra", "-Werror",
+                           "-D" + platform + "=1", "-I" + str(ROOT / "src"),
+                           str(work / "test.cpp"), "-o", str(binary)]
+                if sys.platform.startswith("linux"):
+                    command[1:1] = ["-fsanitize=address,undefined", "-fno-sanitize-recover=all",
+                                    "-fno-pie", "-no-pie"]
+                built = subprocess.run(command, capture_output=True, text=True, timeout=60)
+                self.assertEqual(built.returncode, 0, built.stdout + built.stderr)
+                tested = subprocess.run([str(binary)], capture_output=True, text=True, timeout=10)
+                self.assertEqual(tested.returncode, 0, tested.stdout + tested.stderr)
+                self.assertIn("negative SNR and pointer records", tested.stdout)
+
     def test_sort_on_all_platforms(self):
         compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++")
         self.assertIsNotNone(compiler, "a host C++17 compiler is required")