Update openapi and fix the test script

This commit is contained in:
Joshua Mesilane
2026-04-07 16:28:49 +10:00
parent 38e1fbe3f9
commit 7370cdc688
2 changed files with 19 additions and 20 deletions
+16 -17
View File
@@ -46,7 +46,7 @@ def _make_config(**overrides) -> dict:
"node_name": "test-node",
},
"mesh": {
"global_flood_allow": True,
"unscoped_flood_allow": True,
"loop_detect": "off",
},
"delays": {
@@ -228,11 +228,10 @@ class TestFloodForward:
assert result is None
assert "do not retransmit" in pkt.drop_reason.lower()
def test_global_flood_deny_plain_flood(self, handler):
handler.config["mesh"]["global_flood_allow"] = False
def test_unscoped_flood_deny_plain_flood(self, handler):
handler.config["mesh"]["unscoped_flood_allow"] = False
pkt = _make_flood_packet()
# When global_flood_allow=False, flood_forward calls _check_transport_codes
# which will fail because there are no transport codes on a plain flood
# When unscoped_flood_allow=False, flood_forward should fail on a packet type without a transport code defined
result = handler.flood_forward(pkt)
assert result is None
@@ -656,26 +655,26 @@ class TestHashStabilityThroughForwarding:
# ===================================================================
# 9. Global flood policy
# 9. unscoped flood policy
# ===================================================================
class TestGlobalFloodPolicy:
"""global_flood_allow=False blocks plain flood, transport checked."""
class TestUnscopedFloodPolicy:
"""unscoped_flood_allow=False blocks plain flood, transport checked."""
def test_flood_blocked_by_policy(self, handler):
handler.config["mesh"]["global_flood_allow"] = False
handler.config["mesh"]["unscoped_flood_allow"] = False
pkt = _make_flood_packet()
result = handler.flood_forward(pkt)
assert result is None
def test_direct_unaffected_by_flood_policy(self, handler):
handler.config["mesh"]["global_flood_allow"] = False
handler.config["mesh"]["unscoped_flood_allow"] = False
pkt = _make_direct_packet()
result = handler.direct_forward(pkt)
assert result is not None # direct is not blocked by flood policy
def test_transport_flood_checked_when_policy_off(self, handler):
handler.config["mesh"]["global_flood_allow"] = False
handler.config["mesh"]["unscoped_flood_allow"] = False
pkt = _make_transport_flood_packet()
# Will call _check_transport_codes which will fail (no storage keys)
result = handler.flood_forward(pkt)
@@ -812,7 +811,7 @@ class TestGetDropReason:
assert "Path too long" in reason
def test_flood_policy_reason(self, handler):
handler.config["mesh"]["global_flood_allow"] = False
handler.config["mesh"]["unscoped_flood_allow"] = False
pkt = _make_flood_packet()
reason = handler._get_drop_reason(pkt)
assert "flood" in reason.lower()
@@ -1202,7 +1201,7 @@ BAD_PACKETS = [
"no path"),
("bad_flood_policy_off",
"Plain flood when global_flood_allow=False (needs config override)",
"Plain flood when unscoped_flood_allow=False (needs config override)",
lambda: _make_flood_packet(payload=b"\x01\x02"),
"transport codes"),
@@ -1323,9 +1322,9 @@ class TestBadPacketArray:
BAD_PACKETS, ids=_bad_ids,
)
def test_process_packet_drops(self, handler, name, desc, builder, expected_reason):
# Two entries need global_flood_allow=False
# Two entries need unscoped_flood_allow=False
if "policy_off" in name:
handler.config["mesh"]["global_flood_allow"] = False
handler.config["mesh"]["unscoped_flood_allow"] = False
pkt = builder()
result = handler.process_packet(pkt, snr=5.0)
@@ -1337,7 +1336,7 @@ class TestBadPacketArray:
)
def test_drop_reason_set(self, handler, name, desc, builder, expected_reason):
if "policy_off" in name:
handler.config["mesh"]["global_flood_allow"] = False
handler.config["mesh"]["unscoped_flood_allow"] = False
pkt = builder()
handler.process_packet(pkt, snr=5.0)
@@ -1353,7 +1352,7 @@ class TestBadPacketArray:
def test_bad_packet_not_marked_seen(self, handler, name, desc, builder, expected_reason):
"""Dropped packets must NOT pollute the seen cache."""
if "policy_off" in name:
handler.config["mesh"]["global_flood_allow"] = False
handler.config["mesh"]["unscoped_flood_allow"] = False
pkt = builder()
handler.process_packet(pkt, snr=5.0)