From b7c2636aa96eb3411c6f6c542ca486ad39ed4cd9 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Fri, 16 Oct 2026 18:44:16 -0700 Subject: [PATCH] Retry nonblocking Ethernet operations in runtime regression fixture --- .../rak4631_ethernet_runtime/test.cpp | 28 ++++++--- test/test_rak4631_ethernet_runtime.py | 62 ++++++++++++++++++- 2 files changed, 82 insertions(+), 8 deletions(-) diff --git a/test/fixtures/rak4631_ethernet_runtime/test.cpp b/test/fixtures/rak4631_ethernet_runtime/test.cpp index 66046b88b..d810a52a9 100644 --- a/test/fixtures/rak4631_ethernet_runtime/test.cpp +++ b/test/fixtures/rak4631_ethernet_runtime/test.cpp @@ -60,6 +60,16 @@ static void wait_until(const std::function& condition) { std::this_thread::sleep_for(std::chrono::milliseconds(1)); } } +static void write_all(const uint8_t* data, size_t size) { + size_t accepted = 0; + wait_until([&] { + const size_t remaining = size - accepted; + const size_t written = ethernet_client.write(data + accepted, remaining); + assert(written <= remaining); + accepted += written; + return accepted == size; + }); +} void mock_bus_touch() { assert(bus_owner.load() && spi_active.load()); assert(ethernet_spi_access.load() != EthernetSpiAccess::Idle); @@ -392,23 +402,27 @@ int main(int argc, char** argv) { start_online(); assert(command("get eth") == "> on"); assert(command("eth on").find("already on") != std::string::npos); - const int before_status = hardware_operations.load(); + const int before_status = main_hardware_operations.load(); assert(command("eth.status") == "ETH: 192.168.1.42:23"); - assert(hardware_operations.load() == before_status); + assert(main_hardware_operations.load() == before_status); accept_connection(); assert(ethernet_take_session_reset() && !ethernet_take_session_reset()); - assert(ethernet_client.connected()); + wait_until([] { return ethernet_client.connected(); }); mock_ethernet_input = "eth.status\r\n"; char line[32] = {}; - assert(ethernet_read_line(line, sizeof(line)) && std::string(line) == "eth.status"); - wait_until([] { return ethernet_is_running(); }); - assert(ethernet_client.write(reinterpret_cast("abcde"), 5) == 5); - wait_until([] { return output().find("abcde") != std::string::npos && ethernet_is_running(); }); + // Online is only a snapshot, not a reservation against the SPI worker. + // Keep parser state and retry only bytes the Stream has not accepted. + wait_until([&] { return ethernet_read_line(line, sizeof(line)); }); + assert(std::string(line) == "eth.status"); + write_all(reinterpret_cast("abcde"), 5); + const std::string expected = std::string(ETHERNET_CLI_BANNER) + "\r\n\r\nabcde"; + wait_until([&] { return output() == expected; }); mock_socket_connected[0].store(false); wait_until([] { return ethernet_session_reset.load() && !ethernet_client_attached.load(); }); wait_until([] { return ethernet_is_running(); }); assert(ethernet_take_session_reset() && !ethernet_client.connected()); finish_off(); + assert(output() == expected); // No delayed duplicate writes after drain. break; } case 8: { // Renewal owns SPI; every retained Stream call yields without raw IO. diff --git a/test/test_rak4631_ethernet_runtime.py b/test/test_rak4631_ethernet_runtime.py index 09816f831..6709787b2 100644 --- a/test/test_rak4631_ethernet_runtime.py +++ b/test/test_rak4631_ethernet_runtime.py @@ -53,6 +53,64 @@ SCENARIOS = ( class Rak4631EthernetRuntimeTests(unittest.TestCase): + def test_fixture_retries_busy_and_short_writes_without_duplicates(self): + compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++") + if compiler is None: + self.skipTest("a host C++17 compiler is required") + fixture = (FIXTURE / "test.cpp").read_text(encoding="ascii") + waiter = method(fixture, "static void wait_until(") + writer = method(fixture, "static void write_all(") + harness = r''' +#include +#include +#include +#include +#include +#include +struct Client { + unsigned calls = 0; + std::string output; + size_t write(const uint8_t* data, size_t size) { + const size_t capacity[] = {0, 2, 0, 1, 1, 1}; + assert(calls < sizeof(capacity) / sizeof(capacity[0])); + const size_t offered = capacity[calls++]; + const size_t accepted = size < offered ? size : offered; + output.append(reinterpret_cast(data), accepted); + return accepted; + } +} ethernet_client; +@WAITER@ +@WRITER@ +int main() { + const uint8_t payload[] = {'a', 'b', 'c', 'd', 'e'}; + write_all(payload, sizeof(payload)); + assert(ethernet_client.output == "abcde" && "busy/short retries must preserve exact bytes"); + assert(ethernet_client.calls == 6); +} +''' + with tempfile.TemporaryDirectory(prefix="meshcore-ethernet-fixture-write-") as directory: + for name, implementation in ( + ("retry-unsent", writer), + ("single-attempt", "static void write_all(const uint8_t* data, size_t size) {\n" + "wait_until([] { return true; });\n" + "assert(ethernet_client.write(data, size) == size);\n}"), + ("duplicate-prefix", writer.replace("data + accepted", "data")), + ): + with self.subTest(implementation=name): + binary = Path(directory) / name + compiled = subprocess.run( + [compiler, "-std=c++17", "-pthread", "-Wall", "-Wextra", "-Werror", + *SANITIZER_FLAGS, "-x", "c++", "-", "-o", str(binary)], + input=harness.replace("@WAITER@", waiter).replace("@WRITER@", implementation), + capture_output=True, text=True, timeout=60) + self.assertEqual(compiled.returncode, 0, compiled.stdout + compiled.stderr) + checked = subprocess.run([str(binary)], capture_output=True, text=True, timeout=10) + if name == "retry-unsent": + self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr) + else: + self.assertNotEqual(checked.returncode, 0) + self.assertIn("Assertion", checked.stderr) + def test_unblock_holds_predicate_mutex_before_notifying(self): compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++") if compiler is None: @@ -141,7 +199,9 @@ int main() { capture_output=True, text=True, timeout=60, ) self.assertEqual(compiled.returncode, 0, compiled.stdout + compiled.stderr) - for scenario, description in enumerate(SCENARIOS): + cases = list(enumerate(SCENARIOS)) + cases.extend((7, f"online ownership race repeat {index}") for index in range(32)) + for scenario, description in cases: with self.subTest(scenario=description): checked = subprocess.run([str(binary), str(scenario)], capture_output=True, text=True, timeout=10)