diff --git a/test/fixtures/official_web_console/firmware.cpp b/test/fixtures/official_web_console/firmware.cpp index 705637818..e1c92640e 100644 --- a/test/fixtures/official_web_console/firmware.cpp +++ b/test/fixtures/official_web_console/firmware.cpp @@ -43,6 +43,7 @@ struct MyMesh { Prefs* _prefs = &prefs; Sensors* _sensors = &sensors; unsigned saves = 0; + bool dual_radio_active = false; void savePrefs() { ++saves; } void set(const char* config, char* reply) { @SETTER@ } void get(const char* config, char* reply) { @GETTER@ } @@ -53,6 +54,7 @@ struct MyMesh { } void cancelPendingSerialOutput() {} bool hasPendingSerialOutput() { return false; } + bool isDualRadioActive() { return dual_radio_active; } Prefs* getNodePrefs() { return _prefs; } unsigned getPowerSaveSleepSeconds(unsigned limit) { return limit; } bool millisHasNowPassed(unsigned deadline) { return millis() > deadline; } diff --git a/test/test_official_web_console.py b/test/test_official_web_console.py index 1c4b04aeb..44974fc20 100644 --- a/test/test_official_web_console.py +++ b/test/test_official_web_console.py @@ -44,7 +44,7 @@ def official_console(): def firmware_harness(*, lose_sleep_guard=False, lose_crlf=False, - lose_raw_uart_guard=False): + lose_raw_uart_guard=False, lose_dual_radio_guard=False): # Retain the existing host peripheral boundary; use the production methods # and a virtual clock to exercise minutes/hours without slowing CI. source = HARNESS.split("int main()", 1)[0] @@ -78,6 +78,10 @@ def firmware_harness(*, lose_sleep_guard=False, lose_crlf=False, morse_declaration = " const bool morse_timing = mesh::morseTimingActive(&radio_driver);" assert morse_declaration in main[:start], "Missing live Morse ownership snapshot" power_loop = morse_declaration + "\n" + main[start:end] + if lose_dual_radio_guard: + guard = " && !the_mesh.isDualRadioActive()" + assert power_loop.count(guard) == 1, "Missing radio2 idle-delay negative-control boundary" + power_loop = power_loop.replace(guard, "") morse_helper = body((ROOT / "src/helpers/MorseLoopPolicy.h").read_text(), "inline bool morseTimingActive(const Radio* radio)") # Model only RF ownership at the peripheral boundary; use the real helper @@ -157,6 +161,48 @@ class OfficialWebConsoleTests(unittest.TestCase): self.assertNotEqual(result.returncode, 0, "Negative control did not detect buffered replies") self.assertIn("power-saving setter", result.stderr) + def run_dual_radio_trace(self, *, lose_dual_radio_guard=False): + source = firmware_harness(lose_dual_radio_guard=lose_dual_radio_guard) + source = source.split("int main() {", 1)[0] + r''' +int main() { + attachHost(true); Serial.terminal_open = true; + prefs.powersaving_enabled = 1; mock_millis = 300000; + const unsigned before = yield_ms; + the_mesh.dual_radio_active = true; + power_loop(); + require(sleep_calls == 0 && yield_ms == before, "Active radio2 inserted an idle delay"); + the_mesh.dual_radio_active = false; + power_loop(); + require(sleep_calls == 0 && yield_ms == before + 1, "Idle delay did not resume after radio2"); + prefs.powersaving_enabled = 0; + power_loop(); + require(sleep_calls == 0 && yield_ms == before + 1, "PS-off idle behavior changed"); +} +''' + with tempfile.TemporaryDirectory(prefix="official-console-radio2-ownership-") as temporary: + cpp = Path(temporary) / "ownership.cpp" + binary = Path(temporary) / "ownership" + cpp.write_text(source, encoding="ascii") + sanitizers = (["-fsanitize=address,undefined", "-fno-sanitize-recover=all", + "-fno-pie", "-no-pie"] if sys.platform.startswith("linux") else []) + result = subprocess.run([self.compiler, "-std=c++17", "-Wall", "-Wextra", + "-DARDUINO_USB_MODE=1", "-DARDUINO_USB_CDC_ON_BOOT=1", "-DESP32_PLATFORM=1", + "-DMESH_ESP32_USB_CONSOLE_COOPERATIVE=1", "-DMESH_USB_CONSOLE_COOPERATIVE=1", + "-DMESH_USB_LOGGING_AVAILABLE=1", "-DCONFIG_TINYUSB_ENABLED=0", + "-I", str(ROOT / "src"), *sanitizers, str(cpp), "-o", str(binary)], + text=True, capture_output=True, timeout=60) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + return subprocess.run([str(binary)], text=True, capture_output=True, timeout=20) + + def test_actual_radio2_ownership_skips_idle_delay_then_resumes(self): + result = self.run_dual_radio_trace() + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + + def test_missing_radio2_idle_guard_is_detected(self): + result = self.run_dual_radio_trace(lose_dual_radio_guard=True) + self.assertNotEqual(result.returncode, 0, "Negative control did not detect radio2 idle delay") + self.assertIn("Active radio2 inserted an idle delay", result.stderr) + def test_actual_morse_ownership_blocks_sleep_and_idle_delay_then_resumes(self): source = firmware_harness().split("int main() {", 1)[0] + r''' int main() {