"""Compile production secondary CLI and Dispatcher with injected transport/storage faults.""" from pathlib import Path import os import shutil import subprocess import tempfile import unittest 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] LIFECYCLE = r''' #include #include #include #include #include static uint32_t now_ms = 1000; uint32_t millis() { return now_ms; } namespace mesh { struct Packet {}; struct Mesh { virtual void onSendComplete(Packet*) {} virtual void onSendFail(Packet*) {} virtual void onRadioProfileCopyQueued(Packet*, const Packet*, uint8_t) {} }; } struct Profiles { bool pending = false, accepted = false; unsigned completions = 0; bool finishReplyMutation(bool delivered) { ++completions; accepted = delivered; pending = delivered; return true; } }; struct CLI { Profiles profiles; struct NoMorseReply { void complete(const void*) {} void fail(const void*) {} void copy(const void*, const void*) {} } morse; NoMorseReply& morseReply() { return morse; } Profiles& radioProfiles() { return profiles; } }; struct Manager { std::vector queue; int getOutboundTotal() const { return queue.size(); } mesh::Packet* getOutboundByIdx(int i) { return queue[i]; } mesh::Packet* removeOutboundByIdx(int i) { auto* packet = queue[i]; queue.erase(queue.begin()+i); return packet; } }; struct MyMesh : mesh::Mesh { CLI _cli; Manager manager; Manager* _mgr = &manager; mesh::TempRadioReplyBarrier temp_radio_reply_barrier; bool radio_reply_secondary = false, primary_radio_mutation_starts_temp = false; uint32_t radio_reply_deadline = 0; mesh::Packet* pending_battery_alert_packet = nullptr; bool battery_alert_sent = false; uint64_t last_battery_alert_sent = 0, uptime_millis = 0; uint32_t last_millis = 0; mesh::Packet* outbound = nullptr; unsigned released = 0, normal_schedules = 0, cancellations = 0; const mesh::Packet* getOutboundInFlight() const { return outbound; } void cancelOutboundRadioRetry(const mesh::Packet* p) { assert(p == outbound); ++cancellations; } void releasePacket(mesh::Packet*) { ++released; } bool scheduleNormalRadio() { ++normal_schedules; return true; } void onSendComplete(mesh::Packet*) override; void onSendFail(mesh::Packet*) override; void onRadioProfileCopyQueued(mesh::Packet*, const mesh::Packet*, uint8_t) override; void finishRadioReply(bool); void serviceRadioReplyDeadline(); void arm(mesh::Packet* a, mesh::Packet* b = nullptr) { temp_radio_reply_barrier.prepare(a); if (b) onRadioProfileCopyQueued(b, a, 0); temp_radio_reply_barrier.arm(a); radio_reply_deadline = now_ms + 300000; } }; @METHODS@ int main() { unsigned scenarios = 0; for (bool secondary : {false, true}) for (unsigned success = 0; success < 4; ++success) { MyMesh m; mesh::Packet a, b, unrelated; m.radio_reply_secondary = secondary; m.primary_radio_mutation_starts_temp = true; m._cli.profiles.pending = secondary; m.arm(&a, &b); m.onSendComplete(&unrelated); assert(m.temp_radio_reply_barrier.waiting()); if (success & 1) m.onSendComplete(&a); else m.onSendFail(&a); assert(m.temp_radio_reply_barrier.waiting()); assert(m._cli.profiles.completions == 0 && m.normal_schedules == 0); if (success & 2) m.onSendComplete(&b); else m.onSendFail(&b); assert(!m.temp_radio_reply_barrier.waiting()); assert(m.radio_reply_deadline == 0); if (secondary) { assert(m._cli.profiles.completions == 1 && m._cli.profiles.accepted == (success != 0)); assert(m.normal_schedules == 0); } else assert(m.normal_schedules == (success == 0 ? 1U : 0U)); ++scenarios; } { // Deleting an active schedule may restore saved radio, but must not erase other slots on ACK failure. MyMesh m; mesh::Packet a; m.primary_radio_mutation_starts_temp = false; m.arm(&a); m.onSendFail(&a); assert(!m.temp_radio_reply_barrier.waiting() && m.normal_schedules == 0); ++scenarios; } for (bool rollover : {false, true}) { now_ms = rollover ? UINT32_MAX - 10 : 1000; MyMesh m; mesh::Packet a, b, ordinary; m.radio_reply_secondary = true; m.arm(&a, &b); m.manager.queue = {&a, &ordinary, &b}; now_ms += 299999; m.serviceRadioReplyDeadline(); assert(m.temp_radio_reply_barrier.waiting() && m.released == 0); ++now_ms; m.serviceRadioReplyDeadline(); assert(!m.temp_radio_reply_barrier.waiting() && m.released == 2); assert(m.manager.queue.size() == 1 && m.manager.queue[0] == &ordinary); assert(!m._cli.profiles.accepted && m._cli.profiles.completions == 1); ++scenarios; } { // A physically transmitting copy retains ownership until its actual completion/watchdog. MyMesh m; mesh::Packet a, b; m.radio_reply_secondary = true; m.arm(&a, &b); m.outbound = &a; m.manager.queue = {&b}; now_ms += 300000; m.serviceRadioReplyDeadline(); assert(m.cancellations == 1 && m.released == 1); assert(m.temp_radio_reply_barrier.waiting() && m._cli.profiles.completions == 0); m.onSendComplete(&a); assert(!m.temp_radio_reply_barrier.waiting() && m._cli.profiles.accepted); ++scenarios; } { // Once one copy succeeded, a timeout can retire the other without canceling the mutation. MyMesh m; mesh::Packet a, b; m.radio_reply_secondary = true; m.arm(&a, &b); m.onSendComplete(&a); m.manager.queue = {&b}; now_ms += 300000; m.serviceRadioReplyDeadline(); assert(m.released == 1 && m._cli.profiles.accepted); assert(!m.temp_radio_reply_barrier.waiting()); ++scenarios; } assert(scenarios == 13); } ''' class RadioReplyMutationTest(unittest.TestCase): @staticmethod def compiler_flags(): return ['-std=c++17', '-Wall', '-Wextra'] + ([] if os.name == 'nt' else ['-fsanitize=address,undefined', '-fno-omit-frame-pointer', '-fno-pie', '-no-pie']) def test_radio_transactions_and_coding_rate_restore(self): compiler = shutil.which('g++') or shutil.which('clang++') self.assertIsNotNone(compiler) with tempfile.TemporaryDirectory() as work: exe = Path(work) / 'radio_replies.exe' result = subprocess.run([compiler, *self.compiler_flags(), '-I', str(ROOT/'test/fixtures/radio_profiles/mocks'), '-I', str(ROOT/'test/mocks'), '-I', str(ROOT/'src'), str(ROOT/'src/helpers/RadioProfileCLI.cpp'), str(ROOT/'src/Dispatcher.cpp'), str(ROOT/'src/Packet.cpp'), str(ROOT/'src/helpers/StaticPoolPacketManager.cpp'), str(ROOT/'test/fixtures/radio_profiles/reply_mutation_test.cpp'), '-o', str(exe)], capture_output=True, text=True) self.assertEqual(result.returncode, 0, result.stderr[-14000:]) env = os.environ.copy() # The production pool is a boot-lifetime singleton without a # destructor; this harness constructs several independent pools. # Keep bounds/UAF/UB sanitizers, but don't count that intentional # lifetime allocation as a host-process teardown leak. if os.name != 'nt': env['ASAN_OPTIONS'] = env.get('ASAN_OPTIONS', '') + ':detect_leaks=0' subprocess.run([str(exe)], check=True, env=env) def test_production_repeater_ack_callbacks_and_deadline(self): source = (ROOT/'examples/simple_repeater/MyMesh.cpp').read_text(encoding='utf-8') methods = '\n'.join(method(source, signature) for signature in ( 'void MyMesh::onSendComplete(', 'void MyMesh::onSendFail(', 'void MyMesh::onRadioProfileCopyQueued(', 'void MyMesh::finishRadioReply(', 'void MyMesh::serviceRadioReplyDeadline(')) compiler = shutil.which('g++') or shutil.which('clang++') self.assertIsNotNone(compiler) with tempfile.TemporaryDirectory() as work: cpp = Path(work)/'lifecycle.cpp'; exe = Path(work)/'lifecycle.exe' cpp.write_text(LIFECYCLE.replace('@METHODS@', methods), encoding='utf-8') result = subprocess.run([compiler, *self.compiler_flags(), '-I', str(ROOT/'src'), str(cpp), '-o', str(exe)], capture_output=True, text=True) self.assertEqual(result.returncode, 0, result.stderr[-14000:]) subprocess.run([str(exe)], check=True) if __name__ == '__main__': unittest.main()