mirror of
https://github.com/pyMC-dev/pyMC_Repeater.git
synced 2026-06-11 00:34:46 +02:00
Lloyd
45a44eb47b
- Updated byte representations in tests to use lowercase hex format for consistency. - Reformatted code for better readability, including line breaks and indentation adjustments. - Consolidated multiple lines into single lines where appropriate to enhance clarity. - Ensured that all test cases maintain consistent formatting and style across the test suite.
268 lines
10 KiB
Python
268 lines
10 KiB
Python
from unittest.mock import MagicMock, patch
|
|
|
|
|
|
from repeater.handler_helpers.repeater_cli import MeshCLI, RepeaterCLI
|
|
from repeater.identity_manager import IdentityManager
|
|
|
|
|
|
class _FakeIdentity:
|
|
def __init__(self, pubkey: bytes, addr: bytes = b"\xaa\xbb"):
|
|
self._pubkey = pubkey
|
|
self._addr = addr
|
|
|
|
def get_public_key(self):
|
|
return self._pubkey
|
|
|
|
def get_address_bytes(self):
|
|
return self._addr
|
|
|
|
|
|
def _base_config():
|
|
return {
|
|
"version": "9.9.9",
|
|
"repeater": {
|
|
"name": "node-1",
|
|
"mode": "forward",
|
|
"latitude": 12.3,
|
|
"longitude": 45.6,
|
|
"airtime_factor": 1.1,
|
|
"advert_interval_minutes": 120,
|
|
"flood_advert_interval_hours": 24,
|
|
"max_flood_hops": 32,
|
|
"rx_delay_base": 0.4,
|
|
"tx_delay_factor": 1.2,
|
|
"direct_tx_delay_factor": 0.7,
|
|
"multi_acks": 2,
|
|
"interference_threshold": -111,
|
|
"agc_reset_interval": 8,
|
|
},
|
|
"radio": {
|
|
"frequency": 915000000,
|
|
"bandwidth": 125000,
|
|
"spreading_factor": 7,
|
|
"coding_rate": 5,
|
|
"tx_power": 22,
|
|
},
|
|
"security": {"guest_password": "guest", "allow_read_only": True},
|
|
}
|
|
|
|
|
|
def test_identity_manager_register_lookup_and_collision_paths():
|
|
mgr = IdentityManager(config={})
|
|
id_a = _FakeIdentity(bytes([0x11]) + b"A" * 31, addr=b"\x01\x02")
|
|
id_b_collision = _FakeIdentity(bytes([0x11]) + b"B" * 31, addr=b"\x03\x04")
|
|
|
|
assert mgr.register_identity("alpha", id_a, {"k": 1}, "repeater") is True
|
|
assert mgr.has_identity(0x11) is True
|
|
assert mgr.get_identity_by_hash(0x11)[0] is id_a
|
|
assert mgr.get_identity_by_name("alpha")[0] is id_a
|
|
|
|
# Collision on first pubkey byte should be rejected.
|
|
assert mgr.register_identity("beta", id_b_collision, {"k": 2}, "room_server") is False
|
|
|
|
|
|
def test_identity_manager_list_and_type_filtering():
|
|
mgr = IdentityManager(config={})
|
|
id_a = _FakeIdentity(bytes([0x22]) + b"A" * 31)
|
|
id_b = _FakeIdentity(bytes([0x33]) + b"B" * 31)
|
|
|
|
mgr.register_identity("rep-main", id_a, {"x": 1}, "repeater")
|
|
mgr.register_identity("room-a", id_b, {"y": 2}, "room_server")
|
|
|
|
listed = mgr.list_identities()
|
|
assert len(listed) == 2
|
|
assert any(item["hash"] == "0x22" and item["name"] == "repeater:rep-main" for item in listed)
|
|
assert any(item["hash"] == "0x33" and item["type"] == "room_server" for item in listed)
|
|
|
|
assert mgr.has_identity_type("repeater") is True
|
|
assert mgr.has_identity_type("room_server") is True
|
|
assert mgr.has_identity_type("unknown") is False
|
|
|
|
by_type = mgr.get_identities_by_type("room_server")
|
|
assert len(by_type) == 1
|
|
assert by_type[0][0] == "room-a"
|
|
|
|
|
|
def test_identity_manager_list_handles_none_identity_fields():
|
|
mgr = IdentityManager(config={})
|
|
mgr.identities[0x44] = (None, {}, "repeater")
|
|
mgr.registered_hashes[0x44] = "repeater:ghost"
|
|
|
|
listed = mgr.list_identities()
|
|
assert listed[0]["address"] == "N/A"
|
|
assert listed[0]["public_key"] is None
|
|
|
|
|
|
def test_repeater_cli_alias_points_to_mesh_cli():
|
|
assert RepeaterCLI is MeshCLI
|
|
|
|
|
|
def test_cli_non_admin_and_prefix_passthrough():
|
|
cfg = _base_config()
|
|
save = MagicMock()
|
|
cli = MeshCLI("/tmp/config.yaml", cfg, save)
|
|
|
|
assert cli.handle_command(b"x", "help", is_admin=False) == "Error: Admin permission required"
|
|
assert cli.handle_command(b"x", "01|help set", is_admin=True).startswith("01|")
|
|
|
|
|
|
def test_cli_help_and_route_unknown_commands():
|
|
cli = MeshCLI("/tmp/config.yaml", _base_config(), MagicMock())
|
|
|
|
help_text = cli._route_command("help")
|
|
assert "pyMC CLI Commands" in help_text
|
|
|
|
assert "No detailed help" in cli._route_command("help not-a-topic")
|
|
assert cli._route_command("start ota").startswith("Error:")
|
|
assert cli._route_command("gps now").startswith("Error:")
|
|
assert cli._route_command("stats-air").startswith("Error:")
|
|
assert cli._route_command("totally-unknown") == "Unknown command"
|
|
|
|
|
|
def test_cli_reboot_uses_service_utils_result():
|
|
cli = MeshCLI("/tmp/config.yaml", _base_config(), MagicMock())
|
|
|
|
with patch("repeater.service_utils.restart_service", return_value=(True, "restarted")):
|
|
assert cli._cmd_reboot() == "OK - restarted"
|
|
|
|
with patch("repeater.service_utils.restart_service", return_value=(False, "denied")):
|
|
assert cli._cmd_reboot() == "Error: denied"
|
|
|
|
|
|
def test_cli_clock_time_password_and_version_commands():
|
|
cfg = _base_config()
|
|
save = MagicMock()
|
|
cli = MeshCLI("/tmp/config.yaml", cfg, save, identity_type="room_server")
|
|
|
|
assert "UTC" in cli._cmd_clock("clock")
|
|
assert "not needed" in cli._cmd_clock("clock sync")
|
|
assert cli._cmd_clock("clock bad") == "Unknown clock command"
|
|
assert cli._cmd_time("time 1 2").startswith("Error:")
|
|
|
|
assert cli._cmd_password("password ") == "Error: Password cannot be empty"
|
|
assert cli._cmd_password("password newpass") == "password now: newpass"
|
|
assert cfg["security"]["password"] == "newpass"
|
|
save.assert_called()
|
|
|
|
assert cli._cmd_version() == "pyMC_room_server v9.9.9"
|
|
|
|
|
|
def test_cli_get_commands_cover_expected_fields():
|
|
cli = MeshCLI("/tmp/config.yaml", _base_config(), MagicMock())
|
|
|
|
assert cli._cmd_get("af") == "> 1.1"
|
|
assert cli._cmd_get("name") == "> node-1"
|
|
assert cli._cmd_get("repeat") == "> on"
|
|
assert cli._cmd_get("lat") == "> 12.3"
|
|
assert cli._cmd_get("lon") == "> 45.6"
|
|
assert cli._cmd_get("radio") == "> 915.0,125.0,7,5"
|
|
assert cli._cmd_get("freq") == "> 915.0"
|
|
assert cli._cmd_get("tx") == "> 22"
|
|
assert cli._cmd_get("role") == "> repeater"
|
|
assert cli._cmd_get("guest.password") == "> guest"
|
|
assert cli._cmd_get("allow.read.only") == "> on"
|
|
assert cli._cmd_get("advert.interval") == "> 120"
|
|
assert cli._cmd_get("flood.advert.interval") == "> 24"
|
|
assert cli._cmd_get("flood.max") == "> 32"
|
|
assert cli._cmd_get("rxdelay") == "> 0.4"
|
|
assert cli._cmd_get("txdelay") == "> 1.2"
|
|
assert cli._cmd_get("direct.txdelay") == "> 0.7"
|
|
assert cli._cmd_get("multi.acks") == "> 2"
|
|
assert cli._cmd_get("int.thresh") == "> -111"
|
|
assert cli._cmd_get("agc.reset.interval") == "> 8"
|
|
assert cli._cmd_get("public.key").startswith("Error:")
|
|
assert cli._cmd_get("missing") == "??: missing"
|
|
|
|
|
|
def test_cli_set_commands_apply_and_validate_ranges():
|
|
cfg = _base_config()
|
|
save = MagicMock()
|
|
cli = MeshCLI("/tmp/config.yaml", cfg, save)
|
|
|
|
assert cli._cmd_set("af 2.5") == "OK"
|
|
assert cfg["repeater"]["airtime_factor"] == 2.5
|
|
|
|
assert cli._cmd_set("name repeater-z") == "OK"
|
|
assert cfg["repeater"]["name"] == "repeater-z"
|
|
|
|
assert cli._cmd_set("repeat off").endswith("OFF")
|
|
assert cfg["repeater"]["mode"] == "monitor"
|
|
|
|
assert cli._cmd_set("lat 1.25") == "OK"
|
|
assert cli._cmd_set("lon 2.5") == "OK"
|
|
|
|
assert cli._cmd_set("radio 900000000 250000 9 6").startswith("OK")
|
|
assert cfg["radio"]["frequency"] == 900000000.0
|
|
|
|
assert cli._cmd_set("freq 868000000").startswith("OK")
|
|
assert cli._cmd_set("tx 17") == "OK"
|
|
assert cli._cmd_set("guest.password gpw") == "OK"
|
|
assert cli._cmd_set("allow.read.only off") == "OK"
|
|
|
|
assert cli._cmd_set("advert.interval 59").startswith("Error: interval range")
|
|
assert cli._cmd_set("advert.interval 60") == "OK"
|
|
|
|
assert cli._cmd_set("flood.advert.interval 2").startswith("Error: interval range")
|
|
assert cli._cmd_set("flood.advert.interval 48") == "OK"
|
|
|
|
assert cli._cmd_set("flood.max 65") == "Error: max 64"
|
|
assert cli._cmd_set("flood.max 64") == "OK"
|
|
|
|
assert cli._cmd_set("rxdelay -1") == "Error: cannot be negative"
|
|
assert cli._cmd_set("txdelay -1") == "Error: cannot be negative"
|
|
assert cli._cmd_set("direct.txdelay -1") == "Error: cannot be negative"
|
|
|
|
assert cli._cmd_set("multi.acks 5") == "OK"
|
|
assert cli._cmd_set("int.thresh -120") == "OK"
|
|
assert cli._cmd_set("agc.reset.interval 10") == "OK - interval rounded to 8"
|
|
|
|
|
|
def test_cli_set_command_error_paths():
|
|
cfg = _base_config()
|
|
save = MagicMock()
|
|
cli = MeshCLI("/tmp/config.yaml", cfg, save)
|
|
|
|
assert cli._cmd_set("af") == "Error: Missing value"
|
|
assert cli._cmd_set("radio 1 2 3") == "Error: Expected freq bw sf cr"
|
|
assert cli._cmd_set("unknown.key 1") == "unknown config: unknown.key"
|
|
assert cli._cmd_set("tx not-int").startswith("Error: invalid value")
|
|
|
|
cli.save_config = MagicMock(side_effect=RuntimeError("disk full"))
|
|
assert cli._cmd_set("name x").startswith("Error:")
|
|
|
|
|
|
def test_cli_setperm_region_neighbor_tempradio_log_paths():
|
|
cli = MeshCLI("/tmp/config.yaml", _base_config(), MagicMock(), enable_regions=False)
|
|
|
|
assert cli._cmd_setperm("setperm") == "Err - bad params"
|
|
assert cli._cmd_setperm("setperm deadbeef zz") == "Err - invalid permissions"
|
|
assert cli._cmd_setperm("setperm deadbeef 2").startswith("Error:")
|
|
|
|
assert "not available" in cli._route_command("region load us")
|
|
|
|
cli_regions = MeshCLI("/tmp/config.yaml", _base_config(), MagicMock(), enable_regions=True)
|
|
assert cli_regions._cmd_region("region").startswith("Error:")
|
|
assert cli_regions._cmd_region("region load x").startswith("Error:")
|
|
assert cli_regions._cmd_region("region save").startswith("Error:")
|
|
assert cli_regions._cmd_region("region allowf").startswith("Error:")
|
|
assert cli_regions._cmd_region("region what").startswith("Err -")
|
|
|
|
assert cli._cmd_neighbors().startswith("Error:")
|
|
assert cli._cmd_neighbor_remove("neighbor.remove ") == "ERR: Missing pubkey"
|
|
assert cli._cmd_neighbor_remove("neighbor.remove 001122").startswith("Error:")
|
|
|
|
assert cli._cmd_tempradio("tempradio 1 2 3").startswith("Error:")
|
|
assert cli._cmd_tempradio("tempradio 299 125 7 5 10") == "Error: invalid frequency"
|
|
assert cli._cmd_tempradio("tempradio 915 6 7 5 10") == "Error: invalid bandwidth"
|
|
assert cli._cmd_tempradio("tempradio 915 125 4 5 10") == "Error: invalid spreading factor"
|
|
assert cli._cmd_tempradio("tempradio 915 125 7 9 10") == "Error: invalid coding rate"
|
|
assert cli._cmd_tempradio("tempradio 915 125 7 5 0") == "Error: invalid timeout"
|
|
assert cli._cmd_tempradio("tempradio 915 125 7 5 x") == "Error, invalid params"
|
|
assert cli._cmd_tempradio("tempradio 915 125 7 5 10").startswith("Error:")
|
|
|
|
assert cli._cmd_log("log start").startswith("Error:")
|
|
assert cli._cmd_log("log stop").startswith("Error:")
|
|
assert cli._cmd_log("log erase").startswith("Error:")
|
|
assert cli._cmd_log("log") == "Error: Use journalctl to view logs"
|
|
assert cli._cmd_log("log weird") == "Unknown log command"
|