diff --git a/test/test_companion_power_diagnostics.py b/test/test_companion_power_diagnostics.py index 1b3c5321e..4f412e888 100644 --- a/test/test_companion_power_diagnostics.py +++ b/test/test_companion_power_diagnostics.py @@ -7,24 +7,30 @@ import subprocess import tempfile import unittest +from cpp_source import body + ROOT = Path(__file__).resolve().parents[1] class CompanionPowerDiagnosticsTests(unittest.TestCase): def test_native_usb_wait_counts_only_the_sdk_call_and_preserves_fallback(self): source = (ROOT / "examples/companion_radio/main.cpp").read_text(encoding="ascii") - start = source.index(" bool sleep_failed = false;") - block = source[start:source.index("#elif COMPANION_IDF_PM_AVAILABLE", start)] + block = body(source, "if (can_sleep && the_mesh.canIdleDualRadio())") code = r''' +#define ESP32_PLATFORM 1 +#define ENABLE_USB_INTERFACE 1 +#define ARDUINO_USB_CDC_ON_BOOT 1 #include #include #include +#include using mesh::powerdiag::Reason; using esp_err_t = int; constexpr int ESP_OK = 0; static uint32_t clock_us = 100000; static uint32_t millis() { return clock_us / 1000; } static uint32_t micros() { return clock_us; } +int64_t esp_timer_get_time() { return clock_us; } static bool ready = true; static int timer_error = 0, sleep_error = 0; static unsigned ready_calls = 0, timer_calls = 0, wait_calls = 0, delay_calls = 0; @@ -35,8 +41,20 @@ static int esp_sleep_enable_timer_wakeup(uint64_t us) { static int esp_light_sleep_start() { ++wait_calls; clock_us += 500; return sleep_error; } static unsigned pdMS_TO_TICKS(unsigned ms) { return ms; } static void vTaskDelay(unsigned ms) { assert(ms == 10); ++delay_calls; clock_us += 10000; } +struct Mesh { + bool dual=false,profile_ready=false; + unsigned profile_queries=0; + bool canIdleDualRadio()const{return true;} + bool isDualRadioActive()const{return dual;} + bool getDualRadioIdleDelay(uint32_t& ms){++profile_queries;ms=3;return profile_ready;} +} the_mesh; +struct Board { + unsigned profile_calls=0;uint32_t deadline=0; + void sleepUntilMicroseconds(uint32_t value){++profile_calls;deadline=value;} +} board; static void attempt() { - const uint32_t idle_slice_ms = 10; + const bool can_sleep=true,noise_floor_wait=false; + const uint32_t noise_floor_delay_ms=0; @BLOCK@ } int main() { @@ -65,17 +83,40 @@ int main() { if (mode >= 2) assert(value.detail == -static_cast(mode)); } } + // Scanner rejection retains polling and must never take the native + // single-profile SDK wait or fallback RTOS delay. Exercise the actual + // production deadline helper, rather than substituting a helper result. + for (unsigned mode=0;mode<3;++mode) { + clock_us=100000;setEnabled(diagnostics,millis());reset(millis()); + decision(0,millis());the_mesh.dual=true;the_mesh.profile_ready=mode==2; + the_mesh.profile_queries=board.profile_calls=0; + ready=mode!=0;timer_error=sleep_error=0; + ready_calls=timer_calls=wait_calls=delay_calls=0; + attempt(); + assert(ready_calls==1 && timer_calls==0 && wait_calls==0 && delay_calls==0); + assert(the_mesh.profile_queries==(mode!=0) && board.profile_calls==(mode==2)); + if(mode==2)assert(board.deadline==103000); + const auto value=snapshot(millis()); + assert(value.calls==0 && value.returns==0 && value.errors==0); + if(diagnostics) { + assert(value.board_vetoes==(mode==0)); + assert(value.current==(mode==0?mask(Reason::Board):mode==1?mask(Reason::DualRadio):0)); + } + } + the_mesh.dual=false; } } '''.replace("@BLOCK@", block) with tempfile.TemporaryDirectory(prefix="meshcore-companion-power-") as directory: cpp = Path(directory) / "test.cpp" binary = Path(directory) / "test" + (Path(directory) / "esp_timer.h").write_text( + "#pragma once\n#include \nint64_t esp_timer_get_time();\n", encoding="ascii") cpp.write_text(code, encoding="ascii") built = subprocess.run([ os.environ.get("CXX", "c++"), "-std=c++17", "-Wall", "-Wextra", "-Werror", "-fsanitize=address,undefined", "-fno-sanitize-recover=all", "-fno-pie", "-no-pie", - "-I", str(ROOT / "src"), str(cpp), "-o", str(binary), + "-I", str(directory), "-I", str(ROOT / "src"), str(cpp), "-o", str(binary), ], capture_output=True, text=True, timeout=60) self.assertEqual(built.returncode, 0, built.stdout + built.stderr) result = subprocess.run([str(binary)], capture_output=True, text=True, timeout=20)