You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
60 lines
3.0 KiB
60 lines
3.0 KiB
"""Low-ID unit data stays local without suppressing Gateway observation."""
|
|
import unittest
|
|
|
|
from tests.harness.deterministic import (
|
|
DeterministicScenario, PacketSpec, minimal_config, add_openbridge_system,
|
|
bytes_3, HBPF_DATA_SYNC,
|
|
)
|
|
|
|
|
|
class UnitDataMeshContainmentTests(unittest.TestCase):
|
|
def test_mesh_requires_subscriber_source_and_destination(self):
|
|
pairs = (
|
|
(2341, 2342690, False),
|
|
(5000, 2342690, False),
|
|
(5005, 2342690, False),
|
|
(900999, 2342690, False),
|
|
(999999, 2342690, False),
|
|
(2342690, 5005, False),
|
|
(2342690, 999999, False),
|
|
(1000000, 2342690, True),
|
|
(2342690, 1000000, True),
|
|
(2342690, 2342691, True),
|
|
)
|
|
for ingress in ("MASTER-A", "OBP-IN", "DATA-GATEWAY"):
|
|
for source, destination, mesh_allowed in pairs:
|
|
for dtype in (3, 6, 7, 8):
|
|
with self.subTest(ingress=ingress, source=source,
|
|
destination=destination, dtype=dtype):
|
|
config = minimal_config(("MASTER-A", "MASTER-B"))
|
|
config["GLOBAL"]["DATA_GATEWAY"] = True
|
|
for name, network_id in (("OBP-IN", 3001), ("OBP-OUT", 3002),
|
|
("DATA-GATEWAY", 9991)):
|
|
add_openbridge_system(config, name, network_id)
|
|
with DeterministicScenario(config=config) as scenario:
|
|
scenario.clock.advance(10)
|
|
scenario.bm.SUB_MAP[bytes_3(destination)] = (
|
|
"MASTER-B", 2, scenario.clock.time())
|
|
packet = PacketSpec(
|
|
rf_src=source, dst_id=destination, slot=2,
|
|
call_type="unit", frame_type=HBPF_DATA_SYNC,
|
|
dtype_vseq=dtype, payload=bytes([dtype]) * 33,
|
|
)
|
|
inject = (scenario.inject_hbp if ingress == "MASTER-A"
|
|
else scenario.inject_obp)
|
|
inject(ingress, packet)
|
|
mesh = scenario.capture.for_system("OBP-OUT")
|
|
self.assertEqual(len(mesh), int(mesh_allowed))
|
|
local = scenario.capture.for_system("MASTER-B")
|
|
self.assertEqual(len(local), 1)
|
|
gateway = scenario.capture.for_system("DATA-GATEWAY")
|
|
self.assertEqual(len(gateway), int(ingress != "DATA-GATEWAY"))
|
|
for forwarded in mesh + local + gateway:
|
|
self.assertEqual(forwarded.fields["rf_src"], bytes_3(source))
|
|
self.assertEqual(forwarded.fields["dst_id"], bytes_3(destination))
|
|
self.assertEqual(forwarded.fields["dmr_payload"], packet.payload)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|