Handle correlated C++ guards in host test extraction

test/cpp_source.py

@@ -1,8 +1,15 @@
 """Extract original C++ compound bodies without choosing a build configuration.
 
-This is a lexical test helper, not a C++ parser or preprocessor. Conditional
-alternatives must agree on brace depth at their join. All alternatives and
+This is a lexical test helper, not a C++ parser or preprocessor. Repeated,
+normalized conditional expressions share symbolic choices, allowing a guarded
+opening brace to be closed by a later identical guard. Other expressions are
+opaque atoms, not evaluated C preprocessor arithmetic. All alternatives and
 directives remain in the returned source for the real compiler to select.
+Reusing a retained choice after a potentially invalidating macro change fails
+closed; balanced local default definitions can join and forget their choices.
+Expansion dependencies of opaque expressions are deliberately not simulated.
+Conditions must be stable feature expressions, not aliases for volatile/location-dependent
+macro expansions. Direct __LINE__/__COUNTER__ expressions are rejected too.
 """
 
 import re
@@ -10,6 +17,7 @@ import re
 
 _RAW = re.compile(r'(?:u8|u|U|L)?R"([^\s()\\]{0,16})\(')
 _DIRECTIVE = re.compile(r"#\s*([A-Za-z_]\w*)")
+MAX_CONDITIONAL_PATHS = 64
 
 
 def _quoted_end(text, start):
@@ -135,6 +143,173 @@ def _events(text):
         line_start = False
 
 
+def _uncommented(text):
+    """Remove comments without changing literal contents or joining words."""
+    cursor, pieces = 0, []
+    for kind, begin, finish, _ in _events(text):
+        if kind == "ignored":
+            pieces.extend((text[cursor:begin], " "))
+            cursor = finish
+    pieces.append(text[cursor:])
+    return "".join(pieces)
+
+
+def _directive_expression(text, start, end):
+    logical = re.sub(r"\\\r?\n", "", text[start:end])
+    # Scan after '#' so the lexer can see comments before the keyword too.
+    logical = "#" + _uncommented(logical[1:])
+    match = _DIRECTIVE.match(logical)
+    if not match:
+        raise AssertionError("invalid C++ conditional directive")
+    return logical[match.end():].strip()
+
+
+def _outer_operand(expression):
+    if expression.startswith("("):
+        depth = 0
+        for kind, begin, _, value in _events(expression):
+            if kind == "code" and value in "()":
+                depth += 1 if value == "(" else -1
+                if depth == 0:
+                    return expression[1:-1].strip() if begin == len(expression) - 1 else None
+    return None
+
+
+def _simple_condition(expression):
+    match = re.fullmatch(r"defined\s*(?:\(\s*([A-Za-z_]\w*)\s*\)|([A-Za-z_]\w*))", expression)
+    if match:
+        return "defined(" + (match[1] or match[2]) + ")"
+    if re.fullmatch(r"[A-Za-z_]\w*|(?:0[xX][0-9a-fA-F]+|[0-9]+)[uUlL]*", expression):
+        return expression
+    return None
+
+
+def _condition_identifiers(expression):
+    pieces, cursor = [], 0
+    for kind, begin, end, _ in _events(expression):
+        if kind == "literal":
+            pieces.extend((expression[cursor:begin], " "))
+            cursor = end
+    pieces.append(expression[cursor:])
+    return set(re.findall(r"\b[A-Za-z_]\w*", "".join(pieces)))
+
+
+def _condition(text, start, end, directive):
+    expression = _directive_expression(text, start, end)
+    if directive in ("ifdef", "ifndef"):
+        if not re.fullmatch(r"[A-Za-z_]\w*", expression):
+            raise AssertionError("invalid C++ conditional macro name")
+        expression = "defined(" + expression + ")"
+    inverted = directive == "ifndef"
+    while expression:
+        outer = _outer_operand(expression)
+        if outer is not None:
+            expression = outer
+            continue
+        if expression.startswith("!") and not expression.startswith("!="):
+            # '!A && B' is not the complement of 'A && B'. Peel only a
+            # whole unary operand; arbitrary expressions stay opaque.
+            operand = expression[1:].lstrip()
+            probe = operand.lstrip("!").lstrip()
+            if _simple_condition(probe) is not None or _outer_operand(probe) is not None:
+                inverted = not inverted
+                expression = operand
+                continue
+        simple = _simple_condition(expression)
+        if simple is not None and simple.startswith("defined("):
+            return simple, inverted
+        if {"__LINE__", "__COUNTER__"} & _condition_identifiers(expression):
+            raise AssertionError("volatile C++ conditional expression is unsupported")
+        return simple or expression, inverted
+    raise AssertionError("empty C++ conditional expression")
+
+
+class _Path:
+    def __init__(self, depth=1, choices=None, branches=()):
+        self.depth = depth
+        self.choices = {} if choices is None else choices
+        # Each frame is (parent active, previous alternative taken, active).
+        self.branches = branches
+
+    def active(self):
+        return not self.branches or self.branches[-1][2]
+
+
+def _conditional_paths(paths, condition, inverted, invalidated, alternative=False):
+    result = []
+    for path in paths:
+        parent, taken = (path.branches[-1][:2] if alternative else (path.active(), False))
+        prefix = path.branches[:-1] if alternative else path.branches
+        if not parent or taken:
+            result.append(_Path(path.depth, path.choices, prefix + ((parent, taken, False),)))
+            continue
+        if condition in path.choices and condition in invalidated:
+            raise AssertionError("C++ macro mutation or directive invalidates a retained conditional choice")
+        values = (path.choices[condition],) if condition in path.choices else (False, True)
+        for value in values:
+            choices = dict(path.choices)
+            choices[condition] = value
+            active = value != inverted
+            result.append(_Path(path.depth, choices, prefix + ((True, active, active),)))
+            if len(result) > MAX_CONDITIONAL_PATHS:
+                raise AssertionError("C++ conditional path complexity exceeds 64")
+    if len(result) > MAX_CONDITIONAL_PATHS:
+        raise AssertionError("C++ conditional path complexity exceeds 64")
+    return result
+
+
+def _merge_paths(paths):
+    # Forget irrelevant conditions as soon as both possibilities leave the
+    # same brace/conditional state. Only unresolved brace effects retain them.
+    # Pair and emit cubes in a stable order so conservative complexity limits
+    # do not depend on Python's randomized string/set iteration order.
+    def choice_order(choice):
+        return len(choice), tuple(sorted(choice))
+
+    grouped = {}
+    for path in paths:
+        grouped.setdefault((path.depth, path.branches), set()).add(frozenset(path.choices.items()))
+    result = []
+    for (depth, branches), choices in sorted(grouped.items()):
+        changed = True
+        while changed:
+            changed = False
+            items = sorted(choices, key=choice_order)
+            for index, first in enumerate(items):
+                for second in items[index + 1:]:
+                    if first <= second or second <= first:
+                        choices.discard(second if first <= second else first)
+                    elif (len(first ^ second) == 2 and
+                          {key for key, _ in first} == {key for key, _ in second}):
+                        choices.difference_update((first, second))
+                        choices.add(first & second)
+                    else:
+                        continue
+                    changed = True
+                    break
+                if changed:
+                    break
+        result.extend(_Path(depth, dict(sorted(choice)), branches)
+                      for choice in sorted(choices, key=choice_order))
+    return result
+
+
+def _invalidate_macro_choices(paths, invalidated, text, start, end):
+    name = re.match(r"([A-Za-z_]\w*)", _directive_expression(text, start, end))
+    if not name:
+        raise AssertionError("invalid C++ macro mutation")
+    for path in paths:
+        if not path.active():
+            continue
+        for condition in path.choices:
+            defined = re.fullmatch(r"defined\(([A-Za-z_]\w*)\)", condition)
+            # A numeric/opaque expression may expand through another macro
+            # supplied outside this body. Pure defined(NAME) does not expand
+            # NAME, so unrelated changes cannot alter that particular atom.
+            if defined is None or defined[1] == name[1]:
+                invalidated.add(condition)
+
+
 def body(text, signature):
     """Return the original signature and complete compound body.
 
@@ -142,16 +317,19 @@ def body(text, signature):
     AssertionError for incomplete or ambiguous bodies. Signatures in comments,
     literals and macro definitions do not select a body.
     """
-    if not signature:
+    if not signature.strip():
         raise ValueError("empty C++ signature")
     start = text.find(signature)
+    leading = len(signature) - len(signature.lstrip())
+    anchor = start + leading if start >= 0 else -1
     events = iter(_events(text))
     for kind, begin, end, value in events:
         if start < 0:
             raise ValueError("C++ signature not found: " + signature)
-        while begin <= start < end and kind not in ("code", "brace"):
+        while begin <= anchor < end and kind not in ("code", "brace"):
             start = text.find(signature, end)
-        if begin <= start < end:
+            anchor = start + leading if start >= 0 else -1
+        if begin <= anchor < end:
             break
     else:
         raise ValueError("C++ signature not found: " + signature)
@@ -162,37 +340,51 @@ def body(text, signature):
     else:
         raise AssertionError("missing C++ body opening brace")
 
-    depth = 1
+    paths = [_Path()]
     branches = []
+    invalidated = set()
     for kind, begin, end, value in events:
         if kind == "directive":
             if value in ("if", "ifdef", "ifndef"):
-                branches.append({"entry": depth, "ends": [], "else": False})
+                paths = _conditional_paths(paths, *_condition(text, begin, end, value), invalidated)
+                branches.append(False)
             elif value in ("elif", "else"):
-                if not branches or branches[-1]["else"]:
+                if not branches or branches[-1]:
                     raise AssertionError("unexpected C++ conditional alternative")
-                frame = branches[-1]
-                frame["ends"].append(depth)
-                frame["else"] = value == "else"
-                depth = frame["entry"]
+                if value == "elif":
+                    paths = _conditional_paths(paths, *_condition(text, begin, end, value), invalidated,
+                                               alternative=True)
+                else:
+                    branches[-1] = True
+                    for path in paths:
+                        parent, taken, _ = path.branches[-1]
+                        path.branches = path.branches[:-1] + ((parent, True, parent and not taken),)
             elif value == "endif":
                 if not branches:
                     raise AssertionError("unexpected C++ conditional endif")
-                frame = branches.pop()
-                frame["ends"].append(depth)
-                if not frame["else"]:
-                    frame["ends"].append(frame["entry"])
-                if len(set(frame["ends"])) != 1:
-                    raise AssertionError("C++ conditional alternatives have different brace depths")
-                depth = frame["ends"][0]
-                if depth == 0 and not branches:
-                    return text[start:end]
-        elif depth == 0 and kind != "ignored":
-            raise AssertionError("code after a conditional C++ body closing brace")
-        elif kind == "brace":
-            depth += 1 if value == "{" else -1
-            if depth < 0:
-                raise AssertionError("unexpected C++ closing brace")
-            if depth == 0 and not branches:
-                return text[start:end]
+                branches.pop()
+                for path in paths:
+                    path.branches = path.branches[:-1]
+                paths = _merge_paths(paths)
+                # A balanced local #ifndef/#define needs no later correlation.
+                # Keep taint only while some outstanding brace/branch effect
+                # still depends on that choice; never reuse an invalidated one.
+                invalidated.intersection_update(condition for path in paths for condition in path.choices)
+            elif value in ("define", "undef"):
+                _invalidate_macro_choices(paths, invalidated, text, begin, end)
+            elif value in ("include", "include_next", "line") or (value == "pragma" and
+                    re.match(r"(?:push_macro|pop_macro)\b", _directive_expression(text, begin, end))):
+                invalidated.update(condition for path in paths if path.active() for condition in path.choices)
+        elif kind != "ignored":
+            for path in paths:
+                if not path.active():
+                    continue
+                if path.depth == 0:
+                    raise AssertionError("code after a conditional C++ body closing brace")
+                if kind == "brace":
+                    path.depth += 1 if value == "{" else -1
+                    if path.depth < 0:
+                        raise AssertionError("unexpected C++ closing brace")
+        if not branches and all(path.depth == 0 for path in paths):
+            return text[start:end]
     raise AssertionError("unclosed C++ body or conditional")

test/test_cpp_source.py

@@ -2,11 +2,14 @@
 """Qualify the shared C++ extractor against real and conditional source."""
 
 from pathlib import Path
+import os
 import subprocess
+import sys
 import tempfile
 import unittest
 
-from cpp_source import body
+from cpp_source import MAX_CONDITIONAL_PATHS, body
+from test_radio_receive_contract import method
 
 
 ROOT = Path(__file__).resolve().parents[1]
@@ -17,6 +20,24 @@ class CppSourceTests(unittest.TestCase):
         function = "void target() { if (true) { call(); } }"
         self.assertEqual(body(function + "\nvoid next() {}", "void target()"), function)
 
+    def test_leading_whitespace_signature_is_exact_and_preserved(self):
+        function = "\n  void target() { work(); }"
+        source = "void before() {}" + function + "\nvoid next() {}"
+        for extract in (body, method):
+            with self.subTest(extractor=extract.__name__):
+                self.assertEqual(extract(source, "\n  void target()"), function)
+                with self.assertRaises(ValueError):
+                    extract("void target() {}", "\nvoid target()")
+                with self.assertRaises(ValueError):
+                    extract(source, " \n\t")
+
+    def test_leading_whitespace_signature_in_a_comment_cannot_select_a_body(self):
+        source = "/*\nvoid target() { wrong(); }\n*/\nvoid target() { correct(); }"
+        expected = "\nvoid target() { correct(); }"
+        for extract in (body, method):
+            with self.subTest(extractor=extract.__name__):
+                self.assertEqual(extract(source, "\nvoid target()"), expected)
+
     def test_preserves_legacy_if_and_class_extraction(self):
         source = 'if (strcmp(config, "powersaving on") == 0) { set(true); } else { set(false); }'
         self.assertEqual(body(source, 'if (strcmp(config, "powersaving on") == 0'), source.split(" else")[0])
@@ -105,6 +126,116 @@ const char* example = "void target() { }";
         function = "void target() {\n#if A\n  if (first) { work(); }\n#endif\n}"
         self.assertEqual(body(function + "\nvoid next() {}", "void target()"), function)
 
+    def test_separated_feature_guards_rejoin_without_exponential_history(self):
+        function = "void target() {\n"
+        # Sequential balanced guards must not retain every historical flag.
+        for index in range(100):
+            function += f"#if FEATURE_{index}\n  if (ready) {{ work(); }}\n#endif\n"
+        # Six independently guarded opens require exactly 64 live paths.
+        for index in range(6):
+            function += f"#ifdef FEATURE_{index}\n  if (ready) {{\n#endif\n"
+        function += "    work();\n"
+        for index in reversed(range(6)):
+            function += f"#if defined ( FEATURE_{index} )\n  }}\n#endif\n"
+        function += "}"
+        self.assertEqual(body(function + "\nvoid next() {}", "void target()"), function)
+        self.assertEqual(MAX_CONDITIONAL_PATHS, 64)
+        excessive = function.replace("    work();\n", "#if EXTRA\nif (ready) {\n#endif\n    work();\n", 1)
+        with self.assertRaisesRegex(AssertionError, "complexity exceeds 64"):
+            body(excessive, "void target()")
+
+    def test_opaque_identity_and_unary_precedence_cannot_create_false_correlations(self):
+        pairs = (
+            ("!A && B", "!(A && B)"),
+            ("A B", "AB"),
+            ("'a b' == VALUE", "'ab' == VALUE"),
+            ("'defined A' == VALUE", "'defined(A)' == VALUE"),
+            ("FLAG", "defined(FLAG)"),
+        )
+        for opening, closing in pairs:
+            source = ("void target() {\n#if " + opening + "\nif (ready) {\n#endif\nwork();\n#if " +
+                      closing + "\n}\n#endif\n}\nvoid next() {}")
+            with self.subTest(opening=opening, closing=closing), self.assertRaises(AssertionError):
+                body(source, "void target()")
+
+    def test_symbolic_merge_and_body_are_reproducible_across_hash_seeds(self):
+        script = r'''
+from itertools import product
+from cpp_source import _Path, _merge_paths, body
+
+flags = ("FEATURE_A", "FEATURE_B", "FEATURE_C")
+paths = [_Path(choices=dict(zip(flags, values)))
+         for values in product((False, True), repeat=len(flags))]
+merged = _merge_paths(paths)
+assert len(merged) == 1 and merged[0].choices == {}, merged
+source = "void target() {\n"
+for flag in flags:
+    source += "#ifdef " + flag + "\nif (ready) {\n#endif\n"
+source += "work();\n"
+for flag in reversed(flags):
+    source += "#if defined(" + flag + ")\n}\n#endif\n"
+source += "}"
+assert body(source + "\nvoid next() { missing_symbol(); }", "void target()") == source
+print([(path.depth, sorted(path.choices.items()), path.branches) for path in merged])
+'''
+        outputs = []
+        for seed in (0, 1, 2, 3, 7, 11, 42, 123):
+            with self.subTest(seed=seed):
+                checked = subprocess.run([sys.executable, "-B", "-c", script], cwd=ROOT / "test",
+                                         env=dict(os.environ, PYTHONHASHSEED=str(seed)),
+                                         capture_output=True, text=True, timeout=10)
+                self.assertEqual(checked.returncode, 0, checked.stderr)
+                outputs.append(checked.stdout)
+        self.assertEqual(set(outputs), {"[(1, [], ())]\n"})
+
+    def test_macro_mutations_cannot_reuse_a_retained_guard(self):
+        mutations = ("#undef A", "#define A 0", "#include <unknown.h>",
+                     "#include_next <unknown.h>", "#line 20",
+                     '#pragma push_macro("A")', '#pragma pop_macro("A")')
+        for mutation in mutations:
+            source = ("void target() {\n#ifdef A\nif (ready) {\n#endif\n" + mutation +
+                      "\n#ifdef A\n}\n#endif\n}")
+            with self.subTest(mutation=mutation), self.assertRaisesRegex(AssertionError, "invalidate"):
+                body(source, "void target()")
+        # Opaque A may expand through B outside this body. Do not pretend a
+        # token-only dependency scan can prove an unrelated mutation safe.
+        alias = ("#define A B\nvoid target() {\n#if A\nif (ready) {\n#endif\n#undef B\n"
+                 "#define B 0\n#if A\n}\n#endif\n}")
+        with self.assertRaisesRegex(AssertionError, "macro mutation"):
+            body(alias, "void target()")
+        # Defining B cannot alter whether A is defined. A mutation in the
+        # known-inactive arm of the retained A guard is ignored as in C++.
+        safe = ("void target() {\n#ifdef A\nif (ready) {\n#endif\n#define B 1\n"
+                "#ifdef A\n#ifndef A\n#undef A\n#endif\n}\n#endif\n}")
+        self.assertEqual(body(safe, "void target()"), safe)
+
+    def test_volatile_condition_values_fail_closed_but_definedness_is_stable(self):
+        for value in ("__LINE__", "(__COUNTER__ > 1)", "!__LINE__"):
+            source = "void target() {\n#if " + value + "\nwork();\n#endif\n}"
+            with self.subTest(value=value), self.assertRaisesRegex(AssertionError, "volatile"):
+                body(source, "void target()")
+        source = "void target() {\n#ifdef __LINE__\nif (ready) {\n#endif\nwork();\n#if defined(__LINE__)\n}\n#endif\n}"
+        self.assertEqual(body(source, "void target()"), source)
+
+    def test_balanced_local_macro_defaults_can_join_before_future_guards(self):
+        source = '''void target() {
+#if defined(RXEN) || defined(TXEN)
+#ifndef RXEN
+#define RXEN -1
+#endif
+#ifndef TXEN
+#define TXEN -1
+#endif
+  use_pins(RXEN, TXEN);
+#endif
+#if defined(RXEN) || defined(TXEN)
+  use_pins(RXEN, TXEN);
+#endif
+}'''
+        for extract in (body, method):
+            with self.subTest(extractor=extract.__name__):
+                self.assertEqual(extract(source + "\nvoid next() {}", "void target()"), source)
+
     def test_malformed_or_ambiguous_sources_fail_explicitly(self):
         malformed = (
             "void target() {",
@@ -146,6 +277,129 @@ const char* example = "void target() { }";
         self.assertIn("!ethernet_client.connected()", extracted)
         self.assertNotIn("usbLoggingRecoverySafe", extracted)
 
+    def test_legacy_receive_extractor_retains_the_complete_actual_room_loop(self):
+        source = (ROOT / "examples/simple_room_server/main.cpp").read_text()
+        signature = "\nvoid loop() {"
+        expected = source[source.index(signature):].rstrip()
+        # The raw-brace counter previously ran off this source because both
+        # preprocessor alternatives open an if, but share one closing brace.
+        source += "\nvoid must_not_be_extracted() { missing_symbol(); }\n"
+        for extract in (body, method):
+            with self.subTest(extractor=extract.__name__):
+                extracted = extract(source, signature)
+                self.assertEqual(extracted, expected)
+                self.assertIn("!ethernet_get_enabled()", extracted)
+                self.assertIn("!ethernet_client.connected()", extracted)
+                self.assertIn("mesh::serviceUsbLoggingWatchdog", extracted)
+                self.assertNotIn("must_not_be_extracted", extracted)
+
+    def test_actual_setup_begin_and_radio_init_keep_correlated_guard_boundaries(self):
+        cases = (
+            ("examples/companion_radio/main.cpp", "\nvoid setup()", "\nstatic void serviceUsbLoggingOnlySession"),
+            ("examples/companion_radio/MyMesh.cpp", "void MyMesh::begin(", "\nvoid MyMesh::configureRadioFromPrefs("),
+            ("src/helpers/radiolib/CustomSX1262.h", "bool std_init(SPIClass* spi = NULL)",
+             "\n    int16_t startReceiveDutyCycle("),
+        )
+        for filename, signature, following in cases:
+            source = (ROOT / filename).read_text()
+            start = source.index(signature)
+            expected = source[start:source.index(following, start)].rstrip()
+            isolated = expected + "\nvoid must_not_be_extracted() { missing_symbol(); }"
+            for extract in (body, method):
+                with self.subTest(filename=filename, extractor=extract.__name__):
+                    self.assertEqual(extract(source, signature), expected)
+                    self.assertEqual(extract(isolated, signature), expected)
+                    self.assertEqual(extract(expected, signature), expected)
+
+    def test_separated_nested_inverted_and_elif_guards_compile_in_every_configuration(self):
+        source = '''int target(bool condition) {
+  int value = 0;
+  (void)condition;
+#ifdef EXCLUSIVE
+  if (condition) {
+#endif
+    value += 1;
+#if (defined ( EXCLUSIVE ))
+  }
+#endif
+#ifndef SECOND
+  if (condition) {
+#endif
+    value += 2;
+#if !defined SECOND
+  }
+#endif
+#ifdef OUTER
+#ifndef INNER
+  if (condition) {
+#endif
+#endif
+    value += 4;
+#if defined(OUTER)
+#if !(defined(INNER))
+  }
+#endif
+#endif
+#ifdef EXCLUSIVE
+  if (condition) {
+#elif defined(SECOND)
+  if (!condition) {
+#endif
+    value += 8;
+#if defined EXCLUSIVE
+  }
+#elif defined (SECOND)
+  }
+#endif
+  return value;
+}'''
+        main = r'''
+int main() {
+  for (int c = 0; c < 2; ++c) {
+    int expected = 0;
+#ifdef EXCLUSIVE
+    if (c) expected += 1;
+#else
+    expected += 1;
+#endif
+#ifndef SECOND
+    if (c) expected += 2;
+#else
+    expected += 2;
+#endif
+#if defined(OUTER) && !defined(INNER)
+    if (c) expected += 4;
+#else
+    expected += 4;
+#endif
+#ifdef EXCLUSIVE
+    if (c) expected += 8;
+#elif defined(SECOND)
+    if (!c) expected += 8;
+#else
+    expected += 8;
+#endif
+    assert(target(c != 0) == expected);
+  }
+}
+'''
+        with tempfile.TemporaryDirectory() as directory:
+            cpp, binary = Path(directory) / "split.cpp", Path(directory) / "split"
+            for extract in (body, method):
+                extracted = extract(source + "\nvoid next() { missing_symbol(); }", "int target(")
+                self.assertEqual(extracted, source)
+                cpp.write_text("#include <cassert>\n" + extracted + main)
+                for mask in range(16):
+                    defines = ["-D" + name for bit, name in enumerate(("EXCLUSIVE", "SECOND", "OUTER", "INNER"))
+                               if mask & (1 << bit)]
+                    with self.subTest(extractor=extract.__name__, defines=defines):
+                        compiled = subprocess.run(["g++", "-std=c++17", "-Wall", "-Wextra", "-Werror", *defines,
+                                                   str(cpp), "-o", str(binary)],
+                                                  capture_output=True, text=True, timeout=30)
+                        self.assertEqual(compiled.returncode, 0, compiled.stderr)
+                        checked = subprocess.run([str(binary)], capture_output=True, text=True, timeout=10)
+                        self.assertEqual(checked.returncode, 0, checked.stderr)
+
     def test_both_real_conditional_forms_compile_without_firmware_changes(self):
         source = '''int target(bool condition) {
   int value = 0;
@@ -163,30 +417,33 @@ const char* example = "void target() { }";
 }
 void must_not_be_extracted() { missing_symbol(); }
 '''
-        extracted = body(source, "int target(")
         with tempfile.TemporaryDirectory() as directory:
             cpp = Path(directory) / "conditional.cpp"
-            cpp.write_text("#include <cassert>\n" + extracted + r'''
+            for conditional in ("RAK4631_COMBINED_ETHERNET", "OTA_RAK15002_SD_STORE"):
+                actual = source.replace("RAK4631_COMBINED_ETHERNET", conditional)
+                for extract in (body, method):
+                    extracted = extract(actual, "int target(")
+                    cpp.write_text("#include <cassert>\n" + extracted + r'''
 int main() {
 #if !defined(ETHERNET_ENABLED)
   assert(target(false) == 0 && target(true) == 0);
-#elif defined(RAK4631_COMBINED_ETHERNET)
+#elif defined(TEST_COMBINED_BRANCH)
   assert(target(false) == 0 && target(true) == 3);
 #else
   assert(target(false) == 3 && target(true) == 0);
 #endif
 }
 ''')
-            for defines in ([], ["-DETHERNET_ENABLED"],
-                            ["-DETHERNET_ENABLED", "-DRAK4631_COMBINED_ETHERNET"]):
-                with self.subTest(defines=defines):
-                    binary = Path(directory) / "conditional"
-                    compile_result = subprocess.run(
-                        ["g++", "-std=c++17", "-Wall", "-Wextra", "-Werror", *defines,
-                         str(cpp), "-o", str(binary)], capture_output=True, text=True, timeout=30)
-                    self.assertEqual(compile_result.returncode, 0, compile_result.stderr)
-                    run = subprocess.run([str(binary)], capture_output=True, text=True, timeout=10)
-                    self.assertEqual(run.returncode, 0, run.stderr)
+                    for defines in ([], ["-DETHERNET_ENABLED"],
+                                    ["-DETHERNET_ENABLED", f"-D{conditional}", "-DTEST_COMBINED_BRANCH"]):
+                        with self.subTest(conditional=conditional, extractor=extract.__name__, defines=defines):
+                            binary = Path(directory) / "conditional"
+                            compile_result = subprocess.run(
+                                ["g++", "-std=c++17", "-Wall", "-Wextra", "-Werror", *defines,
+                                 str(cpp), "-o", str(binary)], capture_output=True, text=True, timeout=30)
+                            self.assertEqual(compile_result.returncode, 0, compile_result.stderr)
+                            run = subprocess.run([str(binary)], capture_output=True, text=True, timeout=10)
+                            self.assertEqual(run.returncode, 0, run.stderr)
 
 
 if __name__ == "__main__":

test/test_radio_receive_contract.py

@@ -8,16 +8,13 @@ import sys
 import tempfile
 import unittest
 
+from cpp_source import body
+
 ROOT = Path(__file__).resolve().parents[1]
 
 def method(text, signature):
-    start = text.index(signature)
-    end = text.index('{', start) + 1
-    depth = 1
-    while depth:
-        depth += (text[end] == '{') - (text[end] == '}')
-        end += 1
-    return text[start:end]
+    """Keep the shared legacy API without double-counting C++ alternatives."""
+    return body(text, signature)
 
 HARNESS = r'''
 #include <cassert>