mirror of
https://github.com/pyMC-dev/pyMC_Repeater.git
synced 2026-06-11 16:54:44 +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.
215 lines
8.5 KiB
Python
215 lines
8.5 KiB
Python
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
|
|
from repeater.handler_helpers.mesh_cli import MeshCLI
|
|
|
|
|
|
def _base_config():
|
|
return {
|
|
"version": "3.2.1",
|
|
"repeater": {
|
|
"name": "node-a",
|
|
"mode": "forward",
|
|
"latitude": 1.2,
|
|
"longitude": 3.4,
|
|
"airtime_factor": 1.1,
|
|
"advert_interval_minutes": 120,
|
|
"flood_advert_interval_hours": 24,
|
|
"max_flood_hops": 20,
|
|
"rx_delay_base": 0.2,
|
|
"tx_delay_factor": 1.3,
|
|
"direct_tx_delay_factor": 0.6,
|
|
"multi_acks": 2,
|
|
"interference_threshold": -115,
|
|
"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 _cfg_mgr(save_ok=True, err=None):
|
|
return SimpleNamespace(
|
|
save_to_file=MagicMock(return_value=(save_ok, err)),
|
|
live_update_daemon=MagicMock(),
|
|
)
|
|
|
|
|
|
def test_handle_command_admin_and_prefix_behavior():
|
|
cli = MeshCLI("/tmp/cfg.yaml", _base_config(), _cfg_mgr())
|
|
|
|
assert cli.handle_command(b"a", "help", is_admin=False) == "Error: Admin permission required"
|
|
assert cli.handle_command(b"a", "12|help set", is_admin=True).startswith("12|")
|
|
|
|
|
|
def test_help_routing_and_basic_unknown_paths():
|
|
cli = MeshCLI("/tmp/cfg.yaml", _base_config(), _cfg_mgr(), enable_regions=False)
|
|
|
|
assert "pyMC CLI Commands" in cli._route_command("help")
|
|
assert "No detailed help" in cli._route_command("help nope")
|
|
assert cli._route_command("start ota").startswith("Error:")
|
|
assert cli._route_command("sensor read").startswith("Error:")
|
|
assert cli._route_command("gps on").startswith("Error:")
|
|
assert cli._route_command("stats-foo").startswith("Error:")
|
|
assert cli._route_command("region load x").startswith("Error: Region commands not available")
|
|
assert cli._route_command("unknown") == "Unknown command"
|
|
|
|
|
|
def test_cmd_advert_branches_and_success_schedule():
|
|
cli = MeshCLI("/tmp/cfg.yaml", _base_config(), _cfg_mgr(), send_advert_callback=MagicMock())
|
|
|
|
# No callback configured.
|
|
cli_no_cb = MeshCLI("/tmp/cfg.yaml", _base_config(), _cfg_mgr(), send_advert_callback=None)
|
|
assert cli_no_cb._cmd_advert().startswith("Error: Advert functionality")
|
|
|
|
# Callback present but no event loop.
|
|
cli._event_loop = None
|
|
assert cli._cmd_advert() == "Error: Event loop not available"
|
|
|
|
# Event loop available/running and schedule succeeds.
|
|
fake_loop = SimpleNamespace(is_running=lambda: True)
|
|
cli._event_loop = fake_loop
|
|
|
|
with patch(
|
|
"asyncio.run_coroutine_threadsafe", side_effect=lambda coro, _loop: coro.close()
|
|
) as run_ts:
|
|
out = cli._cmd_advert()
|
|
|
|
assert out == "OK - Advert sent"
|
|
run_ts.assert_called_once()
|
|
|
|
|
|
def test_cmd_password_save_success_failure_and_exception():
|
|
cfg = _base_config()
|
|
ok_mgr = _cfg_mgr(save_ok=True)
|
|
cli_ok = MeshCLI("/tmp/cfg.yaml", cfg, ok_mgr)
|
|
|
|
assert cli_ok._cmd_password("password ") == "Error: Password cannot be empty"
|
|
assert cli_ok._cmd_password("password newpw") == "password now: newpw"
|
|
ok_mgr.live_update_daemon.assert_called_once_with(["security"])
|
|
|
|
bad_mgr = _cfg_mgr(save_ok=False, err="disk")
|
|
cli_bad = MeshCLI("/tmp/cfg.yaml", _base_config(), bad_mgr)
|
|
assert "Failed to save config" in cli_bad._cmd_password("password x")
|
|
|
|
ex_mgr = SimpleNamespace(
|
|
save_to_file=MagicMock(side_effect=RuntimeError("boom")),
|
|
live_update_daemon=MagicMock(),
|
|
)
|
|
cli_ex = MeshCLI("/tmp/cfg.yaml", _base_config(), ex_mgr)
|
|
assert cli_ex._cmd_password("password x") == "Error: Failed to save password"
|
|
|
|
|
|
def test_cmd_get_public_key_and_neighbor_branches():
|
|
cli = MeshCLI("/tmp/cfg.yaml", _base_config(), _cfg_mgr())
|
|
|
|
assert cli._cmd_get("public.key") == "Error: Identity not available"
|
|
|
|
cli.identity = SimpleNamespace(get_public_key=lambda: b"\x01" * 32)
|
|
assert cli._cmd_get("public.key") == "> " + (b"\x01" * 32).hex()
|
|
|
|
cli.identity = SimpleNamespace(get_public_key=MagicMock(side_effect=RuntimeError("bad")))
|
|
assert cli._cmd_get("public.key").startswith("Error:")
|
|
|
|
# neighbors: no storage
|
|
assert cli._cmd_neighbors() == "Error: Storage not available"
|
|
|
|
# neighbors: empty, filtered empty, then formatted output
|
|
storage = SimpleNamespace(get_neighbors=lambda: {})
|
|
cli.storage_handler = storage
|
|
assert cli._cmd_neighbors() == "No neighbors discovered yet"
|
|
|
|
storage.get_neighbors = lambda: {
|
|
"aa": {"is_repeater": False, "zero_hop": False, "last_seen": 1}
|
|
}
|
|
assert "No repeaters or zero hop" in cli._cmd_neighbors()
|
|
|
|
storage.get_neighbors = lambda: {
|
|
"abcdef12feed": {"is_repeater": True, "zero_hop": False, "last_seen": 10, "snr": 4.9},
|
|
"11223344aabb": {"is_repeater": False, "zero_hop": True, "last_seen": 20, "snr": 1.2},
|
|
}
|
|
with patch("time.time", return_value=30):
|
|
out = cli._cmd_neighbors()
|
|
|
|
assert "abcdef12:20:4" in out
|
|
assert "11223344:10:1" in out
|
|
|
|
cli.storage_handler = SimpleNamespace(
|
|
get_neighbors=MagicMock(side_effect=RuntimeError("db fail"))
|
|
)
|
|
assert cli._cmd_neighbors().startswith("Error:")
|
|
|
|
|
|
def test_cmd_set_updates_and_validation_errors():
|
|
cfg = _base_config()
|
|
mgr = _cfg_mgr()
|
|
cli = MeshCLI("/tmp/cfg.yaml", cfg, mgr)
|
|
|
|
assert cli._cmd_set("af 2.5") == "OK"
|
|
assert cfg["repeater"]["airtime_factor"] == 2.5
|
|
|
|
assert cli._cmd_set("name node-z") == "OK"
|
|
assert cfg["repeater"]["node_name"] == "node-z"
|
|
|
|
assert cli._cmd_set("repeat off").endswith("OFF")
|
|
assert cfg["repeater"]["mode"] == "monitor"
|
|
|
|
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 g") == "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("flood.advert.interval 2").startswith("Error: interval range")
|
|
assert cli._cmd_set("flood.max 100") == "Error: max 64"
|
|
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("agc.reset.interval 10") == "OK - interval rounded to 8"
|
|
assert cli._cmd_set("bad") == "Error: Missing value"
|
|
assert cli._cmd_set("tx nope").startswith("Error: invalid value")
|
|
assert cli._cmd_set("unknown.key 1") == "unknown config: unknown.key"
|
|
|
|
|
|
def test_misc_commands_and_routes():
|
|
cli = MeshCLI("/tmp/cfg.yaml", _base_config(), _cfg_mgr(), enable_regions=True)
|
|
|
|
assert cli._cmd_region("region").startswith("Error:")
|
|
assert cli._cmd_region("region load us").startswith("Error:")
|
|
assert cli._cmd_region("region save").startswith("Error:")
|
|
assert cli._cmd_region("region remove x").startswith("Error:")
|
|
assert cli._cmd_region("region unknown").startswith("Err -")
|
|
|
|
assert cli._cmd_setperm("setperm") == "Err - bad params"
|
|
assert cli._cmd_setperm("setperm abc zz") == "Err - invalid permissions"
|
|
assert cli._cmd_setperm("setperm abc 2").startswith("Error:")
|
|
|
|
assert cli._cmd_tempradio("tempradio 1 2 3").startswith("Error: Expected")
|
|
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 nope") == "Error, invalid params"
|
|
assert cli._cmd_tempradio("tempradio 915 125 7 5 10").startswith("Error:")
|
|
|
|
assert cli._cmd_neighbor_remove("neighbor.remove ") == "ERR: Missing pubkey"
|
|
assert cli._cmd_neighbor_remove("neighbor.remove abc").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 whatever") == "Unknown log command"
|