Simon
4 months ago
parent
f23a9ade26
commit
426feb0db0
@ -1,23 +1,26 @@
|
||||
|
||||
from twisted.internet import reactor
|
||||
from twisted.web.xmlrpc import Proxy
|
||||
|
||||
|
||||
def printValue(value):
|
||||
print(repr(value))
|
||||
reactor.stop()
|
||||
|
||||
|
||||
def printError(error):
|
||||
print("error", error)
|
||||
reactor.stop()
|
||||
|
||||
|
||||
def capitalize(value):
|
||||
print(value)
|
||||
|
||||
|
||||
proxy = Proxy(b"http://localhost:7080/xmlrpc")
|
||||
# The callRemote method accepts a method name and an argument list.
|
||||
proxy.callRemote("FD_API.reset", '2', '55555').addCallbacks(capitalize, printError)
|
||||
reactor.run()
|
||||
import json
|
||||
import sys
|
||||
from urllib import request
|
||||
|
||||
|
||||
def post(path, payload):
|
||||
body = json.dumps(payload).encode("utf-8")
|
||||
req = request.Request(
|
||||
"http://127.0.0.1:8000" + path,
|
||||
data=body,
|
||||
headers={"content-type": "application/json"},
|
||||
method="POST",
|
||||
)
|
||||
with request.urlopen(req, timeout=3) as response:
|
||||
return json.loads(response.read().decode("utf-8"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) != 4:
|
||||
print("usage: api_client.py <dmrid> <key> <options>")
|
||||
raise SystemExit(2)
|
||||
print(post("/api/v1/options/set", {
|
||||
"dmrid": int(sys.argv[1]),
|
||||
"key": sys.argv[2],
|
||||
"options": sys.argv[3],
|
||||
}))
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,113 @@
|
||||
# FreeDMR API
|
||||
|
||||
FreeDMR includes an experimental HTTP/JSON API for small live control-plane
|
||||
actions. It is intended for local administration and automation, not for public
|
||||
internet exposure.
|
||||
|
||||
Enable it with:
|
||||
|
||||
```ini
|
||||
[GLOBAL]
|
||||
ENABLE_API: True
|
||||
```
|
||||
|
||||
When enabled, the API listens on TCP port `8000`.
|
||||
|
||||
## Safety Notes
|
||||
|
||||
FreeDMR is a live voice routing process. API requests are deliberately limited
|
||||
to small in-memory operations so they do not delay DMR voice packet handling.
|
||||
Request bodies larger than 8192 bytes are rejected.
|
||||
|
||||
Bind or firewall port `8000` appropriately. Do not expose it publicly without a
|
||||
trusted reverse proxy and access controls.
|
||||
|
||||
## Authentication
|
||||
|
||||
User-level endpoints require:
|
||||
|
||||
- `dmrid`: the connected HBP peer/repeater DMR ID
|
||||
- `key`: the session options key for that peer
|
||||
|
||||
System-level endpoints require:
|
||||
|
||||
- `systemkey`: the FreeDMR system API key
|
||||
|
||||
## Endpoints
|
||||
|
||||
### Health
|
||||
|
||||
```bash
|
||||
curl http://127.0.0.1:8000/api/v1/health
|
||||
```
|
||||
|
||||
### Version
|
||||
|
||||
```bash
|
||||
curl http://127.0.0.1:8000/api/v1/version
|
||||
```
|
||||
|
||||
### Get Options
|
||||
|
||||
```bash
|
||||
curl -X POST http://127.0.0.1:8000/api/v1/options/get \
|
||||
-H 'content-type: application/json' \
|
||||
-d '{"dmrid":1234567,"key":"secret"}'
|
||||
```
|
||||
|
||||
If no live options are present, the response is:
|
||||
|
||||
```json
|
||||
{"ok":true,"connected":true,"has_options":false,"options":""}
|
||||
```
|
||||
|
||||
### Set Options
|
||||
|
||||
```bash
|
||||
curl -X POST http://127.0.0.1:8000/api/v1/options/set \
|
||||
-H 'content-type: application/json' \
|
||||
-d '{"dmrid":1234567,"key":"secret","options":"KEY=secret;TS1=91;DIAL=2350"}'
|
||||
```
|
||||
|
||||
The `options` value must be the complete FreeDMR `OPTIONS` string. The API does
|
||||
not add or preserve `KEY=...` automatically.
|
||||
|
||||
### Reset Peer Session
|
||||
|
||||
```bash
|
||||
curl -X POST http://127.0.0.1:8000/api/v1/reset \
|
||||
-H 'content-type: application/json' \
|
||||
-d '{"dmrid":1234567,"key":"secret"}'
|
||||
```
|
||||
|
||||
FreeDMR expects one HBP peer per master instance, so this resets the master
|
||||
instance that owns the authenticated peer session.
|
||||
|
||||
### Reset All Connections
|
||||
|
||||
```bash
|
||||
curl -X POST http://127.0.0.1:8000/api/v1/system/resetall \
|
||||
-H 'content-type: application/json' \
|
||||
-d '{"systemkey":"system-secret"}'
|
||||
```
|
||||
|
||||
### Stop FreeDMR
|
||||
|
||||
```bash
|
||||
curl -X POST http://127.0.0.1:8000/api/v1/system/kill \
|
||||
-H 'content-type: application/json' \
|
||||
-d '{"systemkey":"system-secret"}'
|
||||
```
|
||||
|
||||
## Responses
|
||||
|
||||
Successful responses include `"ok": true`. Failed responses include
|
||||
`"ok": false` and an `error` string.
|
||||
|
||||
Common errors:
|
||||
|
||||
- `invalid_credentials`
|
||||
- `invalid_json`
|
||||
- `missing_options`
|
||||
- `request_too_large`
|
||||
- `not_found`
|
||||
@ -1,3 +0,0 @@
|
||||
home = /usr/bin
|
||||
include-system-site-packages = false
|
||||
version = 3.10.12
|
||||
@ -0,0 +1 @@
|
||||
"""FreeDMR test package."""
|
||||
@ -0,0 +1 @@
|
||||
"""Test harness helpers for FreeDMR."""
|
||||
@ -0,0 +1,491 @@
|
||||
"""In-process deterministic packet harness for bridge_master tests.
|
||||
|
||||
This module is test-only. It avoids UDP sockets and replaces production
|
||||
network sends with capture functions while leaving production modules unchanged.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from types import SimpleNamespace
|
||||
import copy
|
||||
import importlib
|
||||
import unittest
|
||||
|
||||
|
||||
DMRD = b"DMRD"
|
||||
HBPF_VOICE = 0x0
|
||||
HBPF_VOICE_SYNC = 0x1
|
||||
HBPF_DATA_SYNC = 0x2
|
||||
HBPF_SLT_VHEAD = 0x1
|
||||
HBPF_SLT_VTERM = 0x2
|
||||
ID_MAX = 16776415
|
||||
PEER_MAX = 4294967295
|
||||
|
||||
|
||||
def require_bridge_master():
|
||||
"""Import bridge_master or skip tests when runtime deps are unavailable."""
|
||||
|
||||
try:
|
||||
return importlib.import_module("bridge_master")
|
||||
except ModuleNotFoundError as exc:
|
||||
raise unittest.SkipTest(
|
||||
f"bridge_master runtime dependency is not installed: {exc.name}"
|
||||
) from exc
|
||||
|
||||
|
||||
def bytes_3(value: int | bytes) -> bytes:
|
||||
if isinstance(value, bytes):
|
||||
if len(value) != 3:
|
||||
raise ValueError("expected exactly 3 bytes")
|
||||
return value
|
||||
return int(value).to_bytes(3, "big")
|
||||
|
||||
|
||||
def bytes_4(value: int | bytes) -> bytes:
|
||||
if isinstance(value, bytes):
|
||||
if len(value) != 4:
|
||||
raise ValueError("expected exactly 4 bytes")
|
||||
return value
|
||||
return int(value).to_bytes(4, "big")
|
||||
|
||||
|
||||
def int_id(value: int | bytes) -> int:
|
||||
if isinstance(value, int):
|
||||
return value
|
||||
return int.from_bytes(value, "big")
|
||||
|
||||
|
||||
def acl_permit_all(max_id: int = ID_MAX) -> tuple[bool, list[tuple[int, int]]]:
|
||||
return True, [(1, max_id)]
|
||||
|
||||
|
||||
def hbp_bits(slot: int, call_type: str, frame_type: int, dtype_vseq: int) -> int:
|
||||
bits = ((frame_type & 0x3) << 4) | (dtype_vseq & 0xF)
|
||||
if slot == 2:
|
||||
bits |= 0x80
|
||||
if call_type == "unit":
|
||||
bits |= 0x40
|
||||
return bits
|
||||
|
||||
|
||||
def parse_dmr_fields(packet: bytes) -> dict[str, object]:
|
||||
if len(packet) < 20 or packet[:4] != DMRD:
|
||||
return {"raw": packet}
|
||||
|
||||
bits = packet[15]
|
||||
if bits & 0x40:
|
||||
call_type = "unit"
|
||||
elif (bits & 0x23) == 0x23:
|
||||
call_type = "vcsbk"
|
||||
else:
|
||||
call_type = "group"
|
||||
|
||||
return {
|
||||
"opcode": packet[:4],
|
||||
"seq": packet[4],
|
||||
"rf_src": packet[5:8],
|
||||
"dst_id": packet[8:11],
|
||||
"peer_id": packet[11:15],
|
||||
"bits": bits,
|
||||
"slot": 2 if bits & 0x80 else 1,
|
||||
"call_type": call_type,
|
||||
"frame_type": (bits & 0x30) >> 4,
|
||||
"dtype_vseq": bits & 0xF,
|
||||
"stream_id": packet[16:20],
|
||||
"dmr_payload": packet[20:53],
|
||||
"ber": packet[53:54],
|
||||
"rssi": packet[54:55],
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PacketSpec:
|
||||
peer_id: int | bytes = 1001
|
||||
rf_src: int | bytes = 3120001
|
||||
dst_id: int | bytes = 91
|
||||
slot: int = 2
|
||||
stream_id: int | bytes = 0x01020304
|
||||
seq: int = 0
|
||||
call_type: str = "group"
|
||||
frame_type: int = HBPF_VOICE
|
||||
dtype_vseq: int = 0
|
||||
payload: bytes = b"\x00" * 33
|
||||
ber: bytes = b"\x00"
|
||||
rssi: bytes = b"\x00"
|
||||
delay: float = 0.0
|
||||
|
||||
def data(self) -> bytes:
|
||||
if len(self.payload) != 33:
|
||||
raise ValueError("DMR payload must be exactly 33 bytes")
|
||||
return b"".join(
|
||||
[
|
||||
DMRD,
|
||||
bytes([self.seq & 0xFF]),
|
||||
bytes_3(self.rf_src),
|
||||
bytes_3(self.dst_id),
|
||||
bytes_4(self.peer_id),
|
||||
bytes([hbp_bits(self.slot, self.call_type, self.frame_type, self.dtype_vseq)]),
|
||||
bytes_4(self.stream_id),
|
||||
self.payload,
|
||||
self.ber,
|
||||
self.rssi,
|
||||
]
|
||||
)
|
||||
|
||||
def decoded_args(self) -> tuple[bytes, bytes, bytes, int, int, str, int, int, bytes, bytes]:
|
||||
return (
|
||||
bytes_4(self.peer_id),
|
||||
bytes_3(self.rf_src),
|
||||
bytes_3(self.dst_id),
|
||||
self.seq & 0xFF,
|
||||
self.slot,
|
||||
self.call_type,
|
||||
self.frame_type,
|
||||
self.dtype_vseq,
|
||||
bytes_4(self.stream_id),
|
||||
self.data(),
|
||||
)
|
||||
|
||||
def decoded_obp_args(
|
||||
self,
|
||||
packet_hash: bytes = b"",
|
||||
hops: bytes = b"",
|
||||
source_server: int | bytes = 9990,
|
||||
source_rptr: int | bytes = 0,
|
||||
) -> tuple[bytes, bytes, bytes, int, int, str, int, int, bytes, bytes, bytes, bytes, bytes, bytes, bytes, bytes]:
|
||||
return (
|
||||
bytes_4(self.peer_id),
|
||||
bytes_3(self.rf_src),
|
||||
bytes_3(self.dst_id),
|
||||
self.seq & 0xFF,
|
||||
self.slot,
|
||||
self.call_type,
|
||||
self.frame_type,
|
||||
self.dtype_vseq,
|
||||
bytes_4(self.stream_id),
|
||||
self.data(),
|
||||
packet_hash,
|
||||
hops,
|
||||
bytes_4(source_server),
|
||||
self.ber,
|
||||
self.rssi,
|
||||
bytes_4(source_rptr),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class CapturedPacket:
|
||||
target_system: str
|
||||
packet: bytes
|
||||
hops: bytes | None = None
|
||||
ber: bytes = b"\x00"
|
||||
rssi: bytes = b"\x00"
|
||||
source_server: bytes = b"\x00\x00\x00\x00"
|
||||
source_rptr: bytes = b"\x00\x00\x00\x00"
|
||||
fields: dict[str, object] = field(init=False)
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
self.fields = parse_dmr_fields(self.packet)
|
||||
|
||||
|
||||
class PacketCapture:
|
||||
def __init__(self) -> None:
|
||||
self.packets: list[CapturedPacket] = []
|
||||
|
||||
def recorder(self, target_system: str):
|
||||
def record(
|
||||
packet: bytes,
|
||||
hops: bytes | None = b"",
|
||||
ber: bytes = b"\x00",
|
||||
rssi: bytes = b"\x00",
|
||||
source_server: bytes = b"\x00\x00\x00\x00",
|
||||
source_rptr: bytes = b"\x00\x00\x00\x00",
|
||||
) -> None:
|
||||
self.packets.append(
|
||||
CapturedPacket(
|
||||
target_system=target_system,
|
||||
packet=packet,
|
||||
hops=hops,
|
||||
ber=ber,
|
||||
rssi=rssi,
|
||||
source_server=source_server,
|
||||
source_rptr=source_rptr,
|
||||
)
|
||||
)
|
||||
|
||||
return record
|
||||
|
||||
def for_system(self, system: str) -> list[CapturedPacket]:
|
||||
return [packet for packet in self.packets if packet.target_system == system]
|
||||
|
||||
|
||||
class ReportCapture:
|
||||
def __init__(self) -> None:
|
||||
self.events: list[bytes] = []
|
||||
|
||||
def send_bridgeEvent(self, data: bytes) -> None:
|
||||
self.events.append(data)
|
||||
|
||||
|
||||
class FakeClock:
|
||||
def __init__(self, start: float = 1_700_000_000.0) -> None:
|
||||
self.now = float(start)
|
||||
|
||||
def time(self) -> float:
|
||||
return self.now
|
||||
|
||||
def advance(self, seconds: float) -> float:
|
||||
self.now += seconds
|
||||
return self.now
|
||||
|
||||
|
||||
class FakeReactor:
|
||||
def __init__(self) -> None:
|
||||
self.later: list[tuple[float, object, tuple, dict]] = []
|
||||
self.thread_calls: list[tuple[object, tuple, dict]] = []
|
||||
|
||||
def callLater(self, delay, func, *args, **kwargs):
|
||||
self.later.append((delay, func, args, kwargs))
|
||||
return SimpleNamespace(cancel=lambda: None, active=lambda: True)
|
||||
|
||||
def callInThread(self, func, *args, **kwargs):
|
||||
self.thread_calls.append((func, args, kwargs))
|
||||
|
||||
def callFromThread(self, func, *args, **kwargs):
|
||||
return func(*args, **kwargs)
|
||||
|
||||
|
||||
class FakeTransport:
|
||||
def __init__(self) -> None:
|
||||
self.writes: list[tuple[bytes, tuple[str, int] | None]] = []
|
||||
|
||||
def write(self, packet: bytes, sockaddr=None) -> None:
|
||||
self.writes.append((packet, sockaddr))
|
||||
|
||||
|
||||
def minimal_config(system_names: tuple[str, ...] = ("MASTER-A", "MASTER-B")) -> dict:
|
||||
config = {
|
||||
"GLOBAL": {
|
||||
"SERVER_ID": bytes_4(9990),
|
||||
"USE_ACL": False,
|
||||
"TG1_ACL": acl_permit_all(),
|
||||
"TG2_ACL": acl_permit_all(),
|
||||
"SUB_ACL": acl_permit_all(),
|
||||
"GEN_STAT_BRIDGES": False,
|
||||
"DATA_GATEWAY": False,
|
||||
"VALIDATE_SERVER_IDS": False,
|
||||
},
|
||||
"REPORTS": {"REPORT": False},
|
||||
"ALIASES": {"PATH": "./", "SUB_MAP_FILE": ""},
|
||||
"ALLSTAR": {"ENABLED": False},
|
||||
"SYSTEMS": {},
|
||||
"_SUB_IDS": {},
|
||||
"_PEER_IDS": {},
|
||||
"_LOCAL_SUBSCRIBER_IDS": {},
|
||||
"_SERVER_IDS": {},
|
||||
"CHECKSUMS": {},
|
||||
}
|
||||
for name in system_names:
|
||||
config["SYSTEMS"][name] = {
|
||||
"MODE": "MASTER",
|
||||
"ENABLED": True,
|
||||
"REPEAT": True,
|
||||
"MAX_PEERS": 1,
|
||||
"IP": "127.0.0.1",
|
||||
"PORT": 0,
|
||||
"PASSPHRASE": b"",
|
||||
"GROUP_HANGTIME": 0,
|
||||
"USE_ACL": False,
|
||||
"REG_ACL": acl_permit_all(PEER_MAX),
|
||||
"SUB_ACL": acl_permit_all(),
|
||||
"TG1_ACL": acl_permit_all(),
|
||||
"TG2_ACL": acl_permit_all(),
|
||||
"DEFAULT_UA_TIMER": 1,
|
||||
"SINGLE_MODE": True,
|
||||
"VOICE_IDENT": False,
|
||||
"TS1_STATIC": "",
|
||||
"TS2_STATIC": "",
|
||||
"DEFAULT_REFLECTOR": 0,
|
||||
"GENERATOR": 0,
|
||||
"ANNOUNCEMENT_LANGUAGE": "en_GB",
|
||||
"ALLOW_UNREG_ID": True,
|
||||
"PROXY_CONTROL": False,
|
||||
"OVERRIDE_IDENT_TG": False,
|
||||
"PEERS": {},
|
||||
}
|
||||
return config
|
||||
|
||||
|
||||
def add_openbridge_system(config: dict, name: str = "OBP-1", network_id: int = 1) -> dict:
|
||||
config["SYSTEMS"][name] = {
|
||||
"MODE": "OPENBRIDGE",
|
||||
"ENABLED": True,
|
||||
"NETWORK_ID": bytes_4(network_id),
|
||||
"IP": "127.0.0.1",
|
||||
"PORT": 0,
|
||||
"PASSPHRASE": b"test-passphrase\x00\x00\x00\x00\x00\x00",
|
||||
"TARGET_IP": "127.0.0.1",
|
||||
"TARGET_PORT": 0,
|
||||
"TARGET_SOCK": ("127.0.0.1", 0),
|
||||
"USE_ACL": False,
|
||||
"SUB_ACL": acl_permit_all(),
|
||||
"TG1_ACL": acl_permit_all(),
|
||||
"TG2_ACL": acl_permit_all(),
|
||||
"RELAX_CHECKS": True,
|
||||
"ENHANCED_OBP": False,
|
||||
"VER": 5,
|
||||
}
|
||||
return config
|
||||
|
||||
|
||||
def active_bridge(
|
||||
name: str,
|
||||
tg_id: int,
|
||||
entries: tuple[tuple[str, int], ...],
|
||||
timeout_minutes: int = 1,
|
||||
) -> dict[str, list[dict]]:
|
||||
tg_bytes = bytes_3(tg_id)
|
||||
return {
|
||||
name: [
|
||||
{
|
||||
"SYSTEM": system,
|
||||
"TS": slot,
|
||||
"TGID": tg_bytes,
|
||||
"ACTIVE": True,
|
||||
"TIMEOUT": timeout_minutes * 60,
|
||||
"TO_TYPE": "ON",
|
||||
"OFF": [],
|
||||
"ON": [tg_bytes],
|
||||
"RESET": [],
|
||||
"TIMER": 0,
|
||||
}
|
||||
for system, slot in entries
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
class DeterministicScenario:
|
||||
def __init__(self, config: dict | None = None, bridges: dict | None = None) -> None:
|
||||
self.config = config or minimal_config()
|
||||
self.bridges = bridges or {}
|
||||
self.clock = FakeClock()
|
||||
self.capture = PacketCapture()
|
||||
self.reports: dict[str, ReportCapture] = {}
|
||||
self.transports: dict[str, FakeTransport] = {}
|
||||
self.reactor = FakeReactor()
|
||||
self.bm = None
|
||||
self._saved_attrs: dict[str, object] = {}
|
||||
self._saved_systems: dict | None = None
|
||||
|
||||
def __enter__(self):
|
||||
self.bm = require_bridge_master()
|
||||
self._saved_systems = dict(self.bm.systems)
|
||||
|
||||
for attr in (
|
||||
"CONFIG",
|
||||
"BRIDGES",
|
||||
"SUB_MAP",
|
||||
"peer_ids",
|
||||
"subscriber_ids",
|
||||
"talkgroup_ids",
|
||||
"local_subscriber_ids",
|
||||
"server_ids",
|
||||
"checksums",
|
||||
"reactor",
|
||||
"time",
|
||||
"words",
|
||||
):
|
||||
if hasattr(self.bm, attr):
|
||||
self._saved_attrs[attr] = getattr(self.bm, attr)
|
||||
|
||||
self.bm.CONFIG = self.config
|
||||
self.bm.BRIDGES = copy.deepcopy(self.bridges)
|
||||
self.bm.SUB_MAP = {}
|
||||
self.bm.peer_ids = {}
|
||||
self.bm.subscriber_ids = {}
|
||||
self.bm.talkgroup_ids = {}
|
||||
self.bm.local_subscriber_ids = {}
|
||||
self.bm.server_ids = {}
|
||||
self.bm.checksums = {}
|
||||
self.bm.words = {"en_GB": {"silence": b"", "busy": b"", "notlinked": b"", "linkedto": b"", "to": b""}}
|
||||
self.bm.reactor = self.reactor
|
||||
self.bm.time = self.clock.time
|
||||
|
||||
self.bm.systems.clear()
|
||||
for system_name, system_config in self.config["SYSTEMS"].items():
|
||||
report = ReportCapture()
|
||||
self.reports[system_name] = report
|
||||
if system_config["MODE"] == "MASTER":
|
||||
system = self.bm.routerHBP(system_name, self.config, report)
|
||||
elif system_config["MODE"] == "OPENBRIDGE":
|
||||
system = self.bm.routerOBP(system_name, self.config, report)
|
||||
else:
|
||||
continue
|
||||
system.send_system = self.capture.recorder(system_name)
|
||||
transport = FakeTransport()
|
||||
system.transport = transport
|
||||
self.transports[system_name] = transport
|
||||
self.bm.systems[system_name] = system
|
||||
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc, tb) -> None:
|
||||
if self.bm is None:
|
||||
return
|
||||
self.bm.systems.clear()
|
||||
if self._saved_systems is not None:
|
||||
self.bm.systems.update(self._saved_systems)
|
||||
|
||||
for attr in (
|
||||
"CONFIG",
|
||||
"BRIDGES",
|
||||
"SUB_MAP",
|
||||
"peer_ids",
|
||||
"subscriber_ids",
|
||||
"talkgroup_ids",
|
||||
"local_subscriber_ids",
|
||||
"server_ids",
|
||||
"checksums",
|
||||
"reactor",
|
||||
"time",
|
||||
"words",
|
||||
):
|
||||
if attr in self._saved_attrs:
|
||||
setattr(self.bm, attr, self._saved_attrs[attr])
|
||||
elif hasattr(self.bm, attr):
|
||||
delattr(self.bm, attr)
|
||||
|
||||
@property
|
||||
def systems(self):
|
||||
return self.bm.systems
|
||||
|
||||
@property
|
||||
def bridge_state(self):
|
||||
return self.bm.BRIDGES
|
||||
|
||||
def inject_hbp(self, system_name: str, packet: PacketSpec) -> None:
|
||||
self.systems[system_name].dmrd_received(*packet.decoded_args())
|
||||
|
||||
def inject_obp(self, system_name: str, packet: PacketSpec) -> None:
|
||||
self.systems[system_name].dmrd_received(*packet.decoded_obp_args())
|
||||
|
||||
def inject_datagram(self, system_name: str, packet: bytes, sockaddr=("127.0.0.1", 50000)) -> None:
|
||||
self.systems[system_name].datagramReceived(packet, sockaddr)
|
||||
|
||||
def register_peer(
|
||||
self,
|
||||
system_name: str,
|
||||
peer_id: int | bytes = 1001,
|
||||
sockaddr=("127.0.0.1", 50000),
|
||||
callsign: bytes = b"TEST ",
|
||||
) -> bytes:
|
||||
peer = bytes_4(peer_id)
|
||||
self.config["SYSTEMS"][system_name]["PEERS"][peer] = {
|
||||
"CONNECTION": "YES",
|
||||
"SOCKADDR": sockaddr,
|
||||
"CALLSIGN": callsign,
|
||||
"RADIO_ID": peer,
|
||||
"LAST_PING": self.clock.time(),
|
||||
}
|
||||
return peer
|
||||
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,162 @@
|
||||
import io
|
||||
import json
|
||||
import sys
|
||||
import types
|
||||
import unittest
|
||||
|
||||
|
||||
def install_dmr_utils_stub():
|
||||
if "dmr_utils3.utils" in sys.modules:
|
||||
return None
|
||||
dmr_utils3 = types.ModuleType("dmr_utils3")
|
||||
utils = types.ModuleType("dmr_utils3.utils")
|
||||
|
||||
def bytes_4(value):
|
||||
return int(value).to_bytes(4, "big")
|
||||
|
||||
utils.bytes_4 = bytes_4
|
||||
sys.modules["dmr_utils3"] = dmr_utils3
|
||||
sys.modules["dmr_utils3.utils"] = utils
|
||||
return ("dmr_utils3", "dmr_utils3.utils")
|
||||
|
||||
|
||||
class FakeRequest:
|
||||
def __init__(self, path, payload=None):
|
||||
self.postpath = [part.encode("utf-8") for part in path.strip("/").split("/") if part]
|
||||
self.content = io.BytesIO(
|
||||
b"" if payload is None else json.dumps(payload).encode("utf-8")
|
||||
)
|
||||
self.code = None
|
||||
self.headers = {}
|
||||
|
||||
def setResponseCode(self, code):
|
||||
self.code = code
|
||||
|
||||
def setHeader(self, name, value):
|
||||
self.headers[name] = value
|
||||
|
||||
def getHeader(self, name):
|
||||
if name == "content-length":
|
||||
return str(len(self.content.getvalue()))
|
||||
return None
|
||||
|
||||
|
||||
class APITest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
try:
|
||||
import twisted.web.resource # noqa: F401
|
||||
except ModuleNotFoundError as exc:
|
||||
self.skipTest(f"Twisted is not installed: {exc}")
|
||||
self.stubbed_modules = install_dmr_utils_stub()
|
||||
import API
|
||||
|
||||
self.api = API
|
||||
self.peer_id = (1234567).to_bytes(4, "big")
|
||||
self.config = {
|
||||
"GLOBAL": {"SYSTEM_API_KEY": "system-secret", "_KILL_SERVER": False},
|
||||
"SYSTEMS": {
|
||||
"MASTER-A": {
|
||||
"MODE": "MASTER",
|
||||
"PEERS": {self.peer_id: {}},
|
||||
"_opt_key": "peer-secret",
|
||||
},
|
||||
"OBP-A": {
|
||||
"MODE": "OPENBRIDGE",
|
||||
"PEERS": {},
|
||||
},
|
||||
},
|
||||
}
|
||||
self.bridges = {}
|
||||
self.controller = API.FD_APIController(self.config, self.bridges)
|
||||
|
||||
def tearDown(self):
|
||||
if self.stubbed_modules:
|
||||
for module in self.stubbed_modules:
|
||||
sys.modules.pop(module, None)
|
||||
|
||||
def test_getoptions_returns_clear_no_options_response(self):
|
||||
result = self.controller.getoptions("MASTER-A")
|
||||
|
||||
self.assertEqual(
|
||||
result,
|
||||
{"connected": True, "has_options": False, "options": ""},
|
||||
)
|
||||
|
||||
def test_getoptions_decodes_byte_options_for_json(self):
|
||||
self.config["SYSTEMS"]["MASTER-A"]["OPTIONS"] = b"KEY=peer-secret;TS1=91"
|
||||
|
||||
result = self.controller.getoptions("MASTER-A")
|
||||
|
||||
self.assertEqual(result["options"], "KEY=peer-secret;TS1=91")
|
||||
self.assertTrue(result["has_options"])
|
||||
|
||||
def test_setoptions_stores_full_options_string_unchanged(self):
|
||||
options = "KEY=peer-secret;TS1=91;DIAL=2350"
|
||||
|
||||
self.controller.options("MASTER-A", options)
|
||||
|
||||
self.assertEqual(self.config["SYSTEMS"]["MASTER-A"]["OPTIONS"], options)
|
||||
|
||||
def test_user_reset_is_allowed_only_for_matching_peer_key(self):
|
||||
system = self.controller.validateKey(1234567, "peer-secret")
|
||||
|
||||
self.assertEqual(system, "MASTER-A")
|
||||
self.controller.reset(system)
|
||||
self.assertTrue(self.config["SYSTEMS"]["MASTER-A"]["_reset"])
|
||||
self.assertFalse(self.controller.validateKey(1234567, "wrong"))
|
||||
|
||||
def test_system_kill_sets_existing_control_flag(self):
|
||||
self.assertTrue(self.controller.validateSystemKey("system-secret"))
|
||||
|
||||
self.controller.killserver()
|
||||
|
||||
self.assertTrue(self.config["GLOBAL"]["_KILL_SERVER"])
|
||||
|
||||
def test_options_get_endpoint_returns_json(self):
|
||||
resource = self.api.make_api_resource(self.config, self.bridges)
|
||||
request = FakeRequest(
|
||||
"/api/v1/options/get",
|
||||
{"dmrid": 1234567, "key": "peer-secret"},
|
||||
)
|
||||
|
||||
body = resource.render_POST(request)
|
||||
|
||||
self.assertEqual(request.code, 200)
|
||||
self.assertEqual(
|
||||
json.loads(body.decode("utf-8")),
|
||||
{"ok": True, "connected": True, "has_options": False, "options": ""},
|
||||
)
|
||||
|
||||
def test_options_get_endpoint_rejects_bad_key(self):
|
||||
resource = self.api.make_api_resource(self.config, self.bridges)
|
||||
request = FakeRequest(
|
||||
"/api/v1/options/get",
|
||||
{"dmrid": 1234567, "key": "wrong"},
|
||||
)
|
||||
|
||||
body = resource.render_POST(request)
|
||||
|
||||
self.assertEqual(request.code, 401)
|
||||
self.assertEqual(
|
||||
json.loads(body.decode("utf-8")),
|
||||
{"ok": False, "error": "invalid_credentials"},
|
||||
)
|
||||
|
||||
def test_endpoint_rejects_large_request_body(self):
|
||||
resource = self.api.make_api_resource(self.config, self.bridges)
|
||||
request = FakeRequest(
|
||||
"/api/v1/options/set",
|
||||
{"dmrid": 1234567, "key": "peer-secret", "options": "A" * 9000},
|
||||
)
|
||||
|
||||
body = resource.render_POST(request)
|
||||
|
||||
self.assertEqual(request.code, 413)
|
||||
self.assertEqual(
|
||||
json.loads(body.decode("utf-8")),
|
||||
{"ok": False, "error": "request_too_large"},
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@ -0,0 +1,165 @@
|
||||
import importlib
|
||||
import io
|
||||
import sys
|
||||
import types
|
||||
import unittest
|
||||
from contextlib import redirect_stdout
|
||||
|
||||
|
||||
class AuxiliaryToolTests(unittest.TestCase):
|
||||
def test_report_receiver_bool_flag(self):
|
||||
import report_receiver
|
||||
|
||||
self.assertTrue(report_receiver.bool_flag("1"))
|
||||
self.assertTrue(report_receiver.bool_flag("true"))
|
||||
self.assertTrue(report_receiver.bool_flag("yes"))
|
||||
self.assertFalse(report_receiver.bool_flag("0"))
|
||||
self.assertFalse(report_receiver.bool_flag(""))
|
||||
self.assertFalse(report_receiver.bool_flag(None))
|
||||
|
||||
def test_ami_factory_builds_protocol_with_instance_state(self):
|
||||
try:
|
||||
import AMI
|
||||
except ModuleNotFoundError as exc:
|
||||
self.skipTest(str(exc))
|
||||
|
||||
factory = AMI.AMI.AMIClientFactory(
|
||||
AMI.AMI.AMIClient,
|
||||
b"user",
|
||||
b"secret",
|
||||
b"1234",
|
||||
b"ilink 3 2350",
|
||||
)
|
||||
protocol = factory.buildProtocol(None)
|
||||
|
||||
self.assertEqual(protocol.username, b"user")
|
||||
self.assertEqual(protocol.secret, b"secret")
|
||||
self.assertEqual(protocol.nodenum, b"1234")
|
||||
self.assertEqual(protocol.command, b"ilink 3 2350")
|
||||
|
||||
def test_report_sql_uses_factory_db_and_parameterized_insert(self):
|
||||
self._install_mysql_stub()
|
||||
try:
|
||||
import report_sql
|
||||
report_sql = importlib.reload(report_sql)
|
||||
except ModuleNotFoundError as exc:
|
||||
self.skipTest(str(exc))
|
||||
|
||||
fake_db = _FakeDB()
|
||||
fake_reactor = object()
|
||||
factory = report_sql.reportClientFactory(report_sql.reportClient, fake_db, fake_reactor)
|
||||
with redirect_stdout(io.StringIO()):
|
||||
client = factory.buildProtocol(None)
|
||||
|
||||
self.assertIs(client.db, fake_db)
|
||||
self.assertIs(client.reactor, fake_reactor)
|
||||
|
||||
event = {
|
||||
"type": "GROUP VOICE",
|
||||
"event": "START",
|
||||
"trx": "RX",
|
||||
"system": "SYSTEM",
|
||||
"streamid": "1234",
|
||||
"peerid": "5678",
|
||||
"subid": "9012",
|
||||
"slot": "2",
|
||||
"dstid": "2350",
|
||||
"duration": "0",
|
||||
}
|
||||
with redirect_stdout(io.StringIO()):
|
||||
client.send_mysql(event)
|
||||
|
||||
statement, params = fake_db.cursor_obj.executed
|
||||
self.assertIn("%s", statement)
|
||||
self.assertEqual(params[0], "GROUP VOICE")
|
||||
self.assertEqual(params[8], "2350")
|
||||
self.assertTrue(fake_db.committed)
|
||||
self.assertTrue(fake_db.cursor_obj.closed)
|
||||
|
||||
def test_proxy_environment_bool_parser(self):
|
||||
saved_modules = self._install_proxy_stubs()
|
||||
try:
|
||||
import hotspot_proxy_v2
|
||||
hotspot_proxy_v2 = importlib.reload(hotspot_proxy_v2)
|
||||
|
||||
self.assertTrue(hotspot_proxy_v2.bool_from_env("1"))
|
||||
self.assertTrue(hotspot_proxy_v2.bool_from_env("true"))
|
||||
self.assertTrue(hotspot_proxy_v2.bool_from_env("yes"))
|
||||
self.assertFalse(hotspot_proxy_v2.bool_from_env("0"))
|
||||
self.assertFalse(hotspot_proxy_v2.bool_from_env(""))
|
||||
self.assertFalse(hotspot_proxy_v2.bool_from_env(None))
|
||||
finally:
|
||||
self._restore_modules(saved_modules)
|
||||
|
||||
def _install_mysql_stub(self):
|
||||
mysql_module = types.ModuleType("mysql")
|
||||
connector_module = types.ModuleType("mysql.connector")
|
||||
|
||||
class ConnectorError(Exception):
|
||||
pass
|
||||
|
||||
connector_module.Error = ConnectorError
|
||||
connector_module.errorcode = types.SimpleNamespace(
|
||||
ER_ACCESS_DENIED_ERROR=1045,
|
||||
ER_BAD_DB_ERROR=1049,
|
||||
)
|
||||
mysql_module.connector = connector_module
|
||||
sys.modules["mysql"] = mysql_module
|
||||
sys.modules["mysql.connector"] = connector_module
|
||||
|
||||
def _install_proxy_stubs(self):
|
||||
stubbed = ["dmr_utils3", "dmr_utils3.utils", "Pyro5", "Pyro5.api"]
|
||||
saved_modules = {name: sys.modules.get(name) for name in stubbed + ["hotspot_proxy_v2"]}
|
||||
|
||||
dmr_utils3_module = types.ModuleType("dmr_utils3")
|
||||
dmr_utils3_utils_module = types.ModuleType("dmr_utils3.utils")
|
||||
dmr_utils3_utils_module.int_id = lambda value: int.from_bytes(value, "big")
|
||||
dmr_utils3_module.utils = dmr_utils3_utils_module
|
||||
pyro5_module = types.ModuleType("Pyro5")
|
||||
pyro5_api_module = types.ModuleType("Pyro5.api")
|
||||
pyro5_api_module.Proxy = object
|
||||
pyro5_module.api = pyro5_api_module
|
||||
sys.modules["dmr_utils3"] = dmr_utils3_module
|
||||
sys.modules["dmr_utils3.utils"] = dmr_utils3_utils_module
|
||||
sys.modules["Pyro5"] = pyro5_module
|
||||
sys.modules["Pyro5.api"] = pyro5_api_module
|
||||
sys.modules.pop("hotspot_proxy_v2", None)
|
||||
return saved_modules
|
||||
|
||||
def _restore_modules(self, saved_modules):
|
||||
for name, module in saved_modules.items():
|
||||
if module is None:
|
||||
sys.modules.pop(name, None)
|
||||
else:
|
||||
sys.modules[name] = module
|
||||
|
||||
|
||||
class _FakeCursor:
|
||||
def __init__(self):
|
||||
self.executed = None
|
||||
self.closed = False
|
||||
|
||||
def execute(self, statement, params):
|
||||
self.executed = (statement, params)
|
||||
|
||||
def close(self):
|
||||
self.closed = True
|
||||
|
||||
|
||||
class _FakeDB:
|
||||
def __init__(self):
|
||||
self.cursor_obj = _FakeCursor()
|
||||
self.committed = False
|
||||
|
||||
def is_connected(self):
|
||||
return True
|
||||
|
||||
def cursor(self):
|
||||
return self.cursor_obj
|
||||
|
||||
def commit(self):
|
||||
self.committed = True
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@ -0,0 +1,32 @@
|
||||
import ast
|
||||
import pathlib
|
||||
import unittest
|
||||
|
||||
|
||||
ROOT = pathlib.Path(__file__).resolve().parents[1]
|
||||
|
||||
|
||||
def load_bridge_helper(name):
|
||||
source = (ROOT / "bridge.py").read_text()
|
||||
module = ast.parse(source)
|
||||
for node in module.body:
|
||||
if isinstance(node, ast.FunctionDef) and node.name == name:
|
||||
namespace = {}
|
||||
exec(compile(ast.Module([node], []), "bridge.py", "exec"), namespace)
|
||||
return namespace[name]
|
||||
raise AssertionError(f"bridge.py helper not found: {name}")
|
||||
|
||||
|
||||
class BridgeBackportTests(unittest.TestCase):
|
||||
def test_dmrd_seq_delta_is_modulo_256(self):
|
||||
dmrd_seq_delta = load_bridge_helper("dmrd_seq_delta")
|
||||
|
||||
self.assertIsNone(dmrd_seq_delta(1, False))
|
||||
self.assertEqual(dmrd_seq_delta(2, 1), 1)
|
||||
self.assertEqual(dmrd_seq_delta(0, 255), 1)
|
||||
self.assertEqual(dmrd_seq_delta(2, 255), 3)
|
||||
self.assertEqual(dmrd_seq_delta(250, 2), 248)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
Loading…
Reference in new issue