361 lines
16 KiB
Python
361 lines
16 KiB
Python
"""Exercise delayed OTA response classification and infrastructure reply wiring."""
|
|
from pathlib import Path
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
from test_radio_receive_contract import method
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
class ReplyTxIntegrationTest(unittest.TestCase):
|
|
def test_deferred_responses_keep_the_request_profile(self):
|
|
mesh_source = (ROOT / 'src/Mesh.cpp').read_text()
|
|
dispatcher = (ROOT / 'src/Dispatcher.cpp').read_text()
|
|
harness = r'''
|
|
#include <cassert>
|
|
#include <vector>
|
|
#include <RadioProfiles.h>
|
|
#include <Packet.h>
|
|
#include <helpers/ota/OtaManager.h>
|
|
#include <helpers/ota/OtaProtocol.h>
|
|
#include "mota_vectors.h"
|
|
namespace mesh {
|
|
namespace ota {
|
|
struct OtaContext { OtaManager& manager; };
|
|
static OtaContext* active;
|
|
OtaContext* ota_context_if_active() { return active; }
|
|
OtaContext& ota_ctx() { return *active; }
|
|
}
|
|
struct Manager {
|
|
int free=40;
|
|
int getFreeCount() { return free; }
|
|
};
|
|
static int queuedPacedOtaResponses(Manager*) { return 0; }
|
|
static constexpr int OTA_EGRESS_QUEUE_CREDIT=4, OTA_EGRESS_MIN_FREE=4;
|
|
@PACED@
|
|
struct Radio {
|
|
RadioProfiles config;
|
|
const RadioProfiles* profiles() const { return &config; }
|
|
};
|
|
struct Mesh {
|
|
Manager manager; Manager* _mgr=&manager;
|
|
Radio radio; Radio* _radio=&radio;
|
|
ota::OtaManager ota_manager;
|
|
ota::OtaContext context{ota_manager};
|
|
Packet packet;
|
|
std::vector<uint8_t> masks;
|
|
uint8_t receive_profile=0;
|
|
bool receiving=false;
|
|
Mesh() { ota::active=&context; }
|
|
bool isAnyTempRadioActive() { return true; }
|
|
Packet* createOtaPacket(const uint8_t*,uint16_t) {
|
|
packet=Packet(); packet.header=PAYLOAD_TYPE_OTA << PH_TYPE_SHIFT;
|
|
packet.radio_profile=receive_profile; packet.radio_local=!receiving;
|
|
packet.radio_generation=receiving ? radio.config.generation[receive_profile] : 0;
|
|
return &packet;
|
|
}
|
|
void releasePacket(Packet*) {}
|
|
bool sendOtaFlood(Packet* p) { masks.push_back(getTransmitProfileMask(p)); return true; }
|
|
uint8_t getTransmitProfileMask(const Packet*) const;
|
|
static bool otaSendAdapter(void*,const uint8_t*,uint16_t,bool);
|
|
void service() { @SERVICE@ }
|
|
};
|
|
@MASK@
|
|
@ADAPTER@
|
|
}
|
|
int main() {
|
|
mesh::Mesh node;
|
|
node.radio.config.primary_temporary=true;
|
|
node.radio.config.secondary_temporary=true;
|
|
node.radio.config.secondary.mode=mesh::RadioProfileMode::RxTx;
|
|
node.radio.config.cross=mesh::RadioCrossMode::Off;
|
|
node.ota_manager.begin(0,node.otaSendAdapter,&node);
|
|
assert(node.ota_manager.serve(MOTA_VEC,sizeof(MOTA_VEC)));
|
|
mesh::ota::ReqMsg request{};
|
|
memcpy(request.manifest_id,EXP_MERKLE_ROOT,4);
|
|
request.block_idx=0; request.want_mask=1;
|
|
uint8_t wire[MAX_PACKET_PAYLOAD];
|
|
const auto length=mesh::ota::encode_req(wire,sizeof(wire),request);
|
|
node.receiving=true; node.receive_profile=1;
|
|
mesh::ota::OtaReplyRoute route;
|
|
route.profile=1; route.generation=node.radio.config.generation[1];
|
|
assert(node.ota_manager.on_message(wire,length,route));
|
|
assert(node.masks.empty()); // DATA is deliberately deferred until later
|
|
node.receiving=false; node.receive_profile=0;
|
|
node.manager.free=4; node.service(); // backpressure must retain origin
|
|
assert(node.ota_manager.pendingServeJobs()==1 && node.masks.empty());
|
|
node.manager.free=40;
|
|
node.service();
|
|
assert(node.masks.size()==1 && node.masks[0]==2);
|
|
assert(node.packet.radio_profile==1 && !node.packet.radio_local);
|
|
assert(node.packet.radio_generation==route.generation);
|
|
assert(node.ota_manager.replyRoute().profile==0xFF);
|
|
|
|
// Infrastructure reply-both still overrides AUTO origin selection, including
|
|
// a receive-only secondary when the configured reply policy forces it.
|
|
node.ota_manager.clearPendingEgress(); node.masks.clear();
|
|
node.radio.config.reply_tx=mesh::RADIO_TX_BOTH;
|
|
node.radio.config.reply_force=true;
|
|
node.radio.config.secondary.mode=mesh::RadioProfileMode::Rx;
|
|
assert(node.ota_manager.on_message(wire,length,route));
|
|
node.service();
|
|
assert(node.masks.size()==1 && node.masks[0]==3);
|
|
node.radio.config.reply_tx=mesh::RADIO_TX_AUTO;
|
|
node.radio.config.reply_force=false;
|
|
node.radio.config.secondary.mode=mesh::RadioProfileMode::RxTx;
|
|
|
|
// Identical block requests heard on separate radio domains must not merge.
|
|
node.ota_manager.clearPendingEgress(); node.masks.clear();
|
|
assert(node.ota_manager.on_message(wire,length,route));
|
|
auto primary=route; primary.profile=0;
|
|
assert(node.ota_manager.on_message(wire,length,primary));
|
|
assert(node.ota_manager.pendingServeJobs()==2);
|
|
node.service();
|
|
mesh::ota::ReqProofMsg proof{};
|
|
memcpy(proof.manifest_id,EXP_MERKLE_ROOT,4); proof.block_idx=0;
|
|
uint8_t proof_wire[MAX_PACKET_PAYLOAD];
|
|
const auto proof_len=mesh::ota::encode_req_proof(proof_wire,sizeof(proof_wire),proof);
|
|
assert(node.ota_manager.on_message(proof_wire,proof_len,primary));
|
|
assert(node.ota_manager.pendingServeJobs()==2); // primary proof merges only with primary DATA
|
|
assert(node.ota_manager.on_message(proof_wire,proof_len,route));
|
|
node.service(); // secondary proof
|
|
node.service(); // primary DATA
|
|
node.service(); // primary proof
|
|
assert(node.ota_manager.pendingServeJobs()==0);
|
|
assert((node.masks==std::vector<uint8_t>{2,2,1,1}));
|
|
|
|
// Reconfiguration retires the stale job even when it was waiting on proof.
|
|
node.masks.clear();
|
|
assert(node.ota_manager.on_message(wire,length,route));
|
|
node.service();
|
|
const auto sent=node.ota_manager.packetsSent();
|
|
++node.radio.config.generation[1];
|
|
auto replacement=route; ++replacement.generation;
|
|
assert(node.ota_manager.on_message(wire,length,replacement));
|
|
assert(node.ota_manager.pendingServeJobs()==2);
|
|
node.service();
|
|
assert(node.ota_manager.pendingServeJobs()==1);
|
|
assert(node.ota_manager.packetsSent()==sent); // discarded is not reported as sent
|
|
node.service();
|
|
assert(node.masks.size()==2 && node.packet.radio_generation==replacement.generation);
|
|
node.radio.config.secondary.mode=mesh::RadioProfileMode::Off;
|
|
node.service();
|
|
assert(node.ota_manager.pendingServeJobs()==0);
|
|
node.radio.config.secondary.mode=mesh::RadioProfileMode::RxTx;
|
|
|
|
// The paced manifest queue also separates origins and expires stale sessions.
|
|
node.masks.clear();
|
|
mesh::ota::GetManifestMsg manifest{};
|
|
memcpy(manifest.manifest_id,EXP_MERKLE_ROOT,4); manifest.want_mask=1;
|
|
uint8_t manifest_wire[MAX_PACKET_PAYLOAD];
|
|
const auto manifest_len=mesh::ota::encode_get_manifest(manifest_wire,sizeof(manifest_wire),manifest);
|
|
assert(node.ota_manager.on_message(manifest_wire,manifest_len,route));
|
|
assert(node.ota_manager.on_message(manifest_wire,manifest_len,replacement));
|
|
assert(node.ota_manager.on_message(manifest_wire,manifest_len,primary));
|
|
assert(node.ota_manager.pendingManifestJobs()==3);
|
|
node.service(); // stale route is discarded even before its pacing deadline
|
|
assert(node.ota_manager.pendingManifestJobs()==2 && node.masks.empty());
|
|
node.ota_manager.set_clock(1000000);
|
|
node.service(); node.service();
|
|
assert((node.masks==std::vector<uint8_t>{2,1}));
|
|
assert(node.ota_manager.pendingManifestJobs()==0);
|
|
|
|
// Changing primary also retires its old request before any fragment is sent.
|
|
node.masks.clear();
|
|
assert(node.ota_manager.on_message(wire,length,primary));
|
|
++node.radio.config.generation[0];
|
|
++primary.generation;
|
|
assert(node.ota_manager.on_message(wire,length,primary));
|
|
node.service();
|
|
assert(node.masks.empty() && node.ota_manager.pendingServeJobs()==1);
|
|
node.service();
|
|
assert(node.masks.size()==1 && node.masks[0]==1);
|
|
assert(node.packet.radio_generation==primary.generation);
|
|
node.ota_manager.clearPendingEgress();
|
|
|
|
// LEAVES is paced and retains its request scope after the receive stack
|
|
// unwinds. Backpressure must not consume the pending descriptor.
|
|
node.masks.clear();
|
|
mesh::ota::GetLeavesMsg leaves{};
|
|
memcpy(leaves.manifest_id,EXP_MERKLE_ROOT,4); leaves.want_mask=1;
|
|
uint8_t leaves_wire[MAX_PACKET_PAYLOAD];
|
|
const auto leaves_len=mesh::ota::encode_get_leaves(leaves_wire,sizeof(leaves_wire),leaves);
|
|
node.receiving=true; node.receive_profile=1;
|
|
assert(node.ota_manager.on_message(leaves_wire,leaves_len,replacement));
|
|
assert(node.masks.empty() && node.ota_manager.pendingManifestJobs()==1);
|
|
node.receiving=false; node.receive_profile=0;
|
|
node.manager.free=4;
|
|
node.ota_manager.set_clock(2000000);
|
|
node.service();
|
|
assert(node.masks.empty() && node.ota_manager.pendingManifestJobs()==1);
|
|
node.manager.free=40;
|
|
node.service();
|
|
assert(node.masks.size()==1 && node.masks[0]==2);
|
|
assert(node.packet.radio_reply && !node.packet.radio_local);
|
|
assert(node.ota_manager.pendingManifestJobs()==0);
|
|
// Deferred affinity cannot bleed into a later local send.
|
|
const uint8_t data=mesh::ota::OTA_DATA;
|
|
assert(node.otaSendAdapter(&node,&data,1,false));
|
|
assert(node.masks.size()==2 && node.masks[1]==1 && node.packet.radio_local);
|
|
|
|
// A regular primary and temporary secondary stay isolated unless cross is on.
|
|
node.radio.config.primary_temporary=false;
|
|
node.masks.clear();
|
|
assert(node.ota_manager.on_message(wire,length,replacement));
|
|
node.service();
|
|
assert(node.masks.size()==1 && node.masks[0]==2);
|
|
node.ota_manager.clearPendingEgress();
|
|
node.radio.config.cross=mesh::RadioCrossMode::On;
|
|
assert(node.ota_manager.on_message(wire,length,replacement));
|
|
node.service();
|
|
assert(node.masks.size()==2 && node.masks[1]==3);
|
|
}
|
|
'''
|
|
harness = harness.replace('@PACED@', method(mesh_source, 'static bool isPacedOtaResponse('))
|
|
harness = harness.replace('@ADAPTER@', method(mesh_source, 'bool Mesh::otaSendAdapter('))
|
|
harness = harness.replace('@MASK@', method(dispatcher, 'uint8_t Dispatcher::getTransmitProfileMask(')
|
|
.replace('Dispatcher::', 'Mesh::'))
|
|
start = mesh_source.index('ota::ota_ctx().manager.serviceEgress(')
|
|
service = mesh_source[start:mesh_source.index('}, this);', start) + len('}, this);')]
|
|
harness = harness.replace('@SERVICE@', service)
|
|
self.assertIn('context->manager.on_message(pkt->payload, pkt->payload_len, route)', mesh_source)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory)
|
|
source = path / 'routing.cpp'; source.write_text(harness)
|
|
binary = path / 'routing'
|
|
sources = ['OtaManager.cpp', 'OtaProtocol.cpp', 'MotaContainer.cpp', 'MerkleTree.cpp']
|
|
result = subprocess.run([os.environ.get('CXX', 'g++'), '-std=c++17',
|
|
'-I', str(ROOT / 'src'), '-I', str(ROOT / 'test/mocks'),
|
|
'-I', str(ROOT / 'test/test_ota'), str(source),
|
|
*[str(ROOT / 'src/helpers/ota' / name) for name in sources],
|
|
str(ROOT / 'src/Utils.cpp'), str(ROOT / 'src/Packet.cpp'), '-o', str(binary)],
|
|
capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
result = subprocess.run([str(binary)], capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
def test_async_ota_adapter_marks_only_response_types_and_retains_backpressure(self):
|
|
source = (ROOT / 'src/Mesh.cpp').read_text()
|
|
adapter = method(source, 'bool Mesh::otaSendAdapter(')
|
|
paced = method(source, 'static bool isPacedOtaResponse(')
|
|
harness = r'''
|
|
#include <cassert>
|
|
#include <cstdint>
|
|
#include <initializer_list>
|
|
#include <helpers/ota/OtaFormat.h>
|
|
namespace mesh {
|
|
struct Packet {
|
|
bool radio_reply=false, radio_local=true;
|
|
uint8_t radio_profile=0, radio_origin=0;
|
|
uint32_t radio_generation=0, radio_origin_generation=0;
|
|
};
|
|
namespace ota {
|
|
struct Route { uint32_t generation=0; uint8_t profile=0xFF; };
|
|
struct Context { struct Manager { Route replyRoute() const { return {}; } } manager; };
|
|
Context* ota_context_if_active() { return nullptr; }
|
|
}
|
|
struct Manager { int free=10; int getFreeCount() { return free; } };
|
|
static int queued;
|
|
static int queuedPacedOtaResponses(Manager*) { return queued; }
|
|
static constexpr int OTA_EGRESS_QUEUE_CREDIT=@CREDIT@, OTA_EGRESS_MIN_FREE=@RESERVE@;
|
|
@PACED@
|
|
struct Mesh {
|
|
Manager manager; Manager* _mgr=&manager;
|
|
Packet packet;
|
|
bool temporary=true, allocation=true, admit=true, was_reply=false;
|
|
uint8_t mask=1;
|
|
int allocations=0, sends=0;
|
|
bool isAnyTempRadioActive() { return temporary; }
|
|
Packet* createOtaPacket(const uint8_t*,uint16_t) {
|
|
++allocations;
|
|
// Deliberately opposite to expected for half the types: classification
|
|
// must not accidentally depend on the receive call stack or a prior send.
|
|
packet.radio_reply=!packet.radio_reply;
|
|
if (!allocation || !manager.free) return nullptr;
|
|
--manager.free;
|
|
return &packet;
|
|
}
|
|
uint8_t getTransmitProfileMask(const Packet*) { return mask; }
|
|
void releasePacket(Packet*) { ++manager.free; }
|
|
bool sendOtaFlood(Packet* p) { ++sends; was_reply=p->radio_reply; releasePacket(p); return admit; }
|
|
static bool otaSendAdapter(void*,const uint8_t*,uint16_t,bool);
|
|
};
|
|
@ADAPTER@
|
|
}
|
|
int main() {
|
|
mesh::Mesh node;
|
|
for (unsigned type=0;type<256;++type) {
|
|
const uint8_t msg=type;
|
|
bool reply=type==0x03 || type==0x05 || type==0x07 || type==0x09 || type==0x0b;
|
|
assert(node.otaSendAdapter(&node,&msg,1,true));
|
|
assert(node.was_reply==reply);
|
|
}
|
|
const uint8_t data=mesh::ota::OTA_DATA, proof=mesh::ota::OTA_PROOF;
|
|
const uint8_t manifest=mesh::ota::OTA_MANIFEST, leaves=mesh::ota::OTA_LEAVES;
|
|
for (const uint8_t type : {data,proof,manifest,leaves}) {
|
|
for (uint8_t mask : {1,2,3}) {
|
|
for (int queued=0;queued<=mesh::OTA_EGRESS_QUEUE_CREDIT+1;++queued) {
|
|
for (int free=0;free<=mesh::OTA_EGRESS_MIN_FREE+3;++free) {
|
|
node.mask=mask; mesh::queued=queued; node.manager.free=free;
|
|
const int copies=mask==3 ? 2 : 1;
|
|
const bool allowed=queued+copies<=mesh::OTA_EGRESS_QUEUE_CREDIT
|
|
&& free>=mesh::OTA_EGRESS_MIN_FREE+copies;
|
|
const int allocations=node.allocations;
|
|
assert(node.otaSendAdapter(&node,&type,1,true)==allowed);
|
|
assert(node.manager.free==free); // failure releases, simulated successful TX drains
|
|
if (queued>=mesh::OTA_EGRESS_QUEUE_CREDIT || free<=mesh::OTA_EGRESS_MIN_FREE)
|
|
assert(node.allocations==allocations); // normal backpressure is not pool exhaustion
|
|
}
|
|
}
|
|
}
|
|
}
|
|
node.mask=0; node.manager.free=10; mesh::queued=0;
|
|
assert(!node.otaSendAdapter(&node,&data,1,true));
|
|
assert(node.manager.free==10);
|
|
node.mask=3;
|
|
int allocated=node.allocations;
|
|
node.manager.free=10; node.temporary=false;
|
|
assert(!node.otaSendAdapter(&node,&data,1,true));
|
|
assert(node.allocations==allocated);
|
|
node.temporary=true; node.allocation=false;
|
|
int sends=node.sends;
|
|
assert(!node.otaSendAdapter(&node,&data,1,true)); assert(node.sends==sends);
|
|
node.allocation=true; node.admit=false;
|
|
assert(!node.otaSendAdapter(&node,&data,1,true));
|
|
}
|
|
'''
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
cpp = Path(folder) / 'reply.cpp'
|
|
exe = Path(folder) / 'reply'
|
|
header = (ROOT / 'src/Mesh.h').read_text()
|
|
credit = re.search(r'#define OTA_EGRESS_QUEUE_CREDIT (\d+)', header).group(1)
|
|
reserve = re.search(r'#define OTA_EGRESS_MIN_FREE (\d+)', header).group(1)
|
|
cpp.write_text(harness.replace('@ADAPTER@', adapter).replace('@PACED@', paced)
|
|
.replace('@CREDIT@', credit).replace('@RESERVE@', reserve))
|
|
result = subprocess.run([os.environ.get('CXX', 'g++'), '-std=c++17',
|
|
'-Wall', '-Wextra', '-I', str(ROOT / 'src'), str(cpp), '-o', str(exe)],
|
|
capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
subprocess.run([str(exe)], check=True)
|
|
|
|
def test_all_infrastructure_roles_enable_replies_and_mark_delayed_completions(self):
|
|
common = (ROOT / 'src/helpers/CommonCLI.cpp').read_text()
|
|
self.assertIn('_radio_profiles.begin(fs, _callbacks->getProfileRadio(), _rtc, true);', common)
|
|
for path in ['examples/simple_repeater/MyMesh.cpp',
|
|
'examples/simple_room_server/MyMesh.cpp',
|
|
'examples/simple_sensor/SensorMesh.cpp']:
|
|
with self.subTest(path=path):
|
|
text = (ROOT / path).read_text()
|
|
self.assertIn('_cli.loadPrefs(', text)
|
|
self.assertIn('packet->radio_reply = true;', text)
|
|
room = (ROOT / 'examples/simple_room_server/MyMesh.cpp').read_text()
|
|
self.assertIn('reply->radio_reply = true; // asynchronous response to the room subscription', room)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|