mirror of
https://github.com/pyMC-dev/pyMC_Repeater.git
synced 2026-10-31 04:01:59 +01:00
agessaman
2aafedae02
MeshCLI._auto_add_discovery_result reached for storage_handler.record_advert and returned early when it was absent. record_advert lives on StorageCollector, but MeshCLI is constructed with the SQLiteHandler (main.py -> TextHelper -> MeshCLI), which exposes only store_advert. The guard therefore always failed in production: `discover.neighbors` discovered nodes and recorded none of them, so `neighbors` listed whatever adverts happened to arrive rather than what discovery found. Tests missed it because they injected a hand-rolled double that did have record_advert. Extract the shared persistence into discovery.persist_discovery_result, which accepts either storage object its callers actually hold: record_advert when present (store plus the MQTT/Glass advert publish), falling back to store_advert. neighbors_publisher._enrich_discovery_result, which had its own working copy against the StorageCollector, now calls it too, so the two paths build an identical advert row from one implementation. The regression test specs its mock off the real SQLiteHandler class rather than hand-rolling the surface, so a double can no longer claim a method the wired-in object does not have. It fails against the previous code with KeyError 'auto_added'.
663 lines
26 KiB
Python
663 lines
26 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",
|
|
"owner_info": "owner-a",
|
|
"latitude": 1.2,
|
|
"longitude": 3.4,
|
|
"airtime_factor": 1.1,
|
|
"advert_interval_minutes": 120,
|
|
# Stale orphan key from the old CLI; nothing must read it.
|
|
"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,
|
|
"security": {
|
|
"admin_password": "adminpw",
|
|
"guest_password": "guest",
|
|
"allow_read_only": True,
|
|
"max_clients": 5,
|
|
},
|
|
},
|
|
"radio": {
|
|
"frequency": 915000000,
|
|
"bandwidth": 125000,
|
|
"spreading_factor": 7,
|
|
"coding_rate": 5,
|
|
"tx_power": 22,
|
|
},
|
|
"mesh": {"path_hash_mode": 0, "loop_detect": "minimal"},
|
|
# Decoy: the legacy top-level section the CLI must no longer touch.
|
|
"security": {"guest_password": "stale-top-level", "allow_read_only": False},
|
|
}
|
|
|
|
|
|
def _cfg_mgr(save_ok=True):
|
|
return SimpleNamespace(
|
|
save_to_file=MagicMock(return_value=save_ok),
|
|
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 "openHop 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("discover.neighbors").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"
|
|
# The password must land where LoginHelper reads it, not the legacy top-level section.
|
|
assert cfg["repeater"]["security"]["admin_password"] == "newpw"
|
|
assert "password" not in cfg["security"]
|
|
ok_mgr.live_update_daemon.assert_called_once_with(["repeater"])
|
|
|
|
bad_mgr = _cfg_mgr(save_ok=False)
|
|
cli_bad = MeshCLI("/tmp/cfg.yaml", _base_config(), bad_mgr)
|
|
assert "Failed to save config" in cli_bad._cmd_password("password x")
|
|
bad_mgr.live_update_daemon.assert_not_called()
|
|
|
|
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 zero hop repeaters" 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},
|
|
"99aabbccddeeff00": {
|
|
"is_repeater": True,
|
|
"zero_hop": True,
|
|
"last_seen": 21,
|
|
"snr": 6.3,
|
|
},
|
|
}
|
|
with patch("time.time", return_value=30):
|
|
out = cli._cmd_neighbors()
|
|
|
|
assert "99aabbcc:9:6" in out
|
|
assert "abcdef12:20:4" not in out
|
|
assert "11223344:10:1" not in out
|
|
|
|
cli.storage_handler = SimpleNamespace(
|
|
get_neighbors=MagicMock(side_effect=RuntimeError("db fail"))
|
|
)
|
|
assert cli._cmd_neighbors().startswith("Error:")
|
|
|
|
|
|
def test_cmd_get_owner_info_path_hash_mode_and_loop_detect():
|
|
cli = MeshCLI("/tmp/cfg.yaml", _base_config(), _cfg_mgr())
|
|
|
|
assert cli._cmd_get("owner.info") == "> owner-a"
|
|
assert cli._cmd_get("path.hash.mode") == "> 0"
|
|
assert cli._cmd_get("loop.detect") == "> minimal"
|
|
|
|
|
|
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"
|
|
|
|
# CLI input is MHz/kHz (firmware parity); the config stores Hz.
|
|
assert cli._cmd_set("radio 900 250 9 6").startswith("OK")
|
|
assert cfg["radio"]["frequency"] == 900000000
|
|
assert cfg["radio"]["bandwidth"] == 250000
|
|
assert cfg["radio"]["spreading_factor"] == 9
|
|
assert cfg["radio"]["coding_rate"] == 6
|
|
assert cli._cmd_set("radio 100 250 9 6") == "Error, invalid radio params"
|
|
assert cli._cmd_set("radio 900 250 4 6") == "Error, invalid radio params"
|
|
|
|
assert cli._cmd_set("freq 868.5").startswith("OK")
|
|
assert cfg["radio"]["frequency"] == 868500000
|
|
assert cli._cmd_set("tx 17") == "OK - restart repeater to apply"
|
|
assert cfg["radio"]["tx_power"] == 17
|
|
assert cli._cmd_set("guest.password g") == "OK"
|
|
assert cfg["repeater"]["security"]["guest_password"] == "g"
|
|
assert cfg["security"]["guest_password"] == "stale-top-level"
|
|
assert cli._cmd_set("owner.info Alice|Ops") == "OK"
|
|
assert cfg["repeater"]["owner_info"] == "Alice\nOps"
|
|
assert cli._cmd_set("allow.read.only off") == "OK"
|
|
assert cfg["repeater"]["security"]["allow_read_only"] is False
|
|
assert cli._cmd_set("path.hash.mode 2") == "OK"
|
|
assert cfg["mesh"]["path_hash_mode"] == 2
|
|
assert cli._cmd_set("path.hash.mode 3") == "Error: path.hash.mode must be 0, 1, or 2"
|
|
assert cli._cmd_set("loop.detect strict") == "OK"
|
|
assert cfg["mesh"]["loop_detect"] == "strict"
|
|
assert (
|
|
cli._cmd_set("loop.detect weird")
|
|
== "Error: loop.detect must be off, minimal, moderate, or strict"
|
|
)
|
|
|
|
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.advert.interval 12") == "OK"
|
|
# The engine's timer key is updated; the stale orphan key is left alone.
|
|
assert cfg["repeater"]["send_advert_interval_hours"] == 12
|
|
assert cfg["repeater"]["flood_advert_interval_hours"] == 24
|
|
assert cli._cmd_set("flood.max 100") == "Error: max 64"
|
|
assert cli._cmd_set("rxdelay -1") == "Error, must be 0-20"
|
|
assert cli._cmd_set("rxdelay 20.5") == "Error, must be 0-20"
|
|
assert cli._cmd_set("rxdelay 20") == "OK"
|
|
assert cli._cmd_set("txdelay -1") == "Error, must be 0-2"
|
|
assert cli._cmd_set("txdelay 2.5") == "Error, must be 0-2"
|
|
assert cli._cmd_set("direct.txdelay -1") == "Error, must be 0-2"
|
|
assert cli._cmd_set("direct.txdelay 2.5") == "Error, must be 0-2"
|
|
|
|
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_cmd_set_txdelay_writes_delays_section():
|
|
cfg = _base_config()
|
|
stale_repeater_value = cfg["repeater"]["tx_delay_factor"]
|
|
mgr = _cfg_mgr()
|
|
cli = MeshCLI("/tmp/cfg.yaml", cfg, mgr)
|
|
|
|
assert cli._cmd_set("txdelay 2.0") == "OK"
|
|
assert cfg["delays"]["tx_delay_factor"] == 2.0
|
|
assert cfg["repeater"]["tx_delay_factor"] == stale_repeater_value
|
|
mgr.live_update_daemon.assert_called_with(["repeater", "delays"])
|
|
|
|
|
|
def test_cmd_set_direct_txdelay_writes_delays_section():
|
|
cfg = _base_config()
|
|
stale_repeater_value = cfg["repeater"]["direct_tx_delay_factor"]
|
|
mgr = _cfg_mgr()
|
|
cli = MeshCLI("/tmp/cfg.yaml", cfg, mgr)
|
|
|
|
assert cli._cmd_set("direct.txdelay 0.75") == "OK"
|
|
assert cfg["delays"]["direct_tx_delay_factor"] == 0.75
|
|
assert cfg["repeater"]["direct_tx_delay_factor"] == stale_repeater_value
|
|
mgr.live_update_daemon.assert_called_with(["repeater", "delays"])
|
|
|
|
|
|
def test_cmd_get_txdelay_and_direct_txdelay_prefer_delays_section():
|
|
cfg = _base_config()
|
|
# repeater section still carries stale values; the delays section wins.
|
|
cfg["delays"] = {"tx_delay_factor": 4.4, "direct_tx_delay_factor": 0.9}
|
|
cli = MeshCLI("/tmp/cfg.yaml", cfg, _cfg_mgr())
|
|
|
|
assert cli._cmd_get("txdelay") == "> 4.4"
|
|
assert cli._cmd_get("direct.txdelay") == "> 0.9"
|
|
|
|
|
|
def test_cmd_get_txdelay_and_direct_txdelay_default_without_delays_section():
|
|
cli = MeshCLI("/tmp/cfg.yaml", _base_config(), _cfg_mgr())
|
|
|
|
assert cli._cmd_get("txdelay") == "> 1.0"
|
|
assert cli._cmd_get("direct.txdelay") == "> 0.5"
|
|
|
|
|
|
def test_misc_commands_and_routes():
|
|
cli = MeshCLI("/tmp/cfg.yaml", _base_config(), _cfg_mgr(), enable_regions=True)
|
|
|
|
cli.storage_handler = SimpleNamespace(get_transport_keys=lambda: [])
|
|
|
|
assert cli._cmd_region("region") == "* F"
|
|
assert cli._cmd_region("region load us").startswith("Err -")
|
|
assert cli._cmd_region("region save") == "OK"
|
|
assert cli._cmd_region("region remove x") == "Err - not found"
|
|
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"
|
|
|
|
storage = SimpleNamespace(delete_neighbors_by_pubkey_prefix=MagicMock(return_value=1))
|
|
cli.storage_handler = storage
|
|
assert cli._cmd_neighbor_remove("neighbor.remove abc") == "OK"
|
|
storage.delete_neighbors_by_pubkey_prefix.assert_called_with("abc")
|
|
|
|
assert cli._cmd_neighbor_remove("neighbor.remove ") == "OK"
|
|
storage.delete_neighbors_by_pubkey_prefix.assert_called_with(None)
|
|
|
|
assert cli._cmd_neighbor_remove("neighbor.remove zz") == "ERR: bad pubkey"
|
|
|
|
no_delete = SimpleNamespace()
|
|
cli.storage_handler = no_delete
|
|
assert "not supported" in cli._cmd_neighbor_remove("neighbor.remove ab")
|
|
|
|
cli.storage_handler = None
|
|
assert cli._cmd_neighbor_remove("neighbor.remove ab") == "Error: Storage not available"
|
|
|
|
discovery = SimpleNamespace(
|
|
cleanup_sessions=MagicMock(),
|
|
create_session=MagicMock(return_value={"session_id": "s1"}),
|
|
start_session_task=MagicMock(),
|
|
)
|
|
loop = SimpleNamespace(is_running=lambda: True, call_soon_threadsafe=MagicMock())
|
|
cli._event_loop = loop
|
|
cli.config_manager.daemon = SimpleNamespace(discovery_helper=discovery)
|
|
|
|
assert cli._cmd_discover_neighbors("discover.neighbors") == "OK - Discover sent"
|
|
discovery.cleanup_sessions.assert_called_once()
|
|
assert callable(discovery.create_session.call_args.kwargs["result_enricher"])
|
|
loop.call_soon_threadsafe.assert_called_once_with(discovery.start_session_task, "s1")
|
|
|
|
assert cli._cmd_discover_neighbors("discover.neighbors foo").startswith("Err -")
|
|
|
|
cli.config_manager.daemon = SimpleNamespace(discovery_helper=None)
|
|
assert cli._cmd_discover_neighbors("discover.neighbors").startswith("Error: Discovery helper")
|
|
|
|
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"
|
|
|
|
|
|
def test_region_command_response_shapes_and_ops():
|
|
keys = [
|
|
{
|
|
"id": 1,
|
|
"name": "alpha",
|
|
"flood_policy": "allow",
|
|
"parent_id": None,
|
|
},
|
|
{
|
|
"id": 2,
|
|
"name": "#beta",
|
|
"flood_policy": "deny",
|
|
"parent_id": 1,
|
|
},
|
|
]
|
|
|
|
created = []
|
|
updated = []
|
|
deleted = []
|
|
|
|
def get_transport_keys():
|
|
return list(keys)
|
|
|
|
def update_transport_key(key_id, **kwargs):
|
|
updated.append((key_id, kwargs))
|
|
for rec in keys:
|
|
if rec["id"] == key_id:
|
|
rec.update(kwargs)
|
|
return True
|
|
return False
|
|
|
|
def create_transport_key(name, flood_policy, transport_key, parent_id, last_used):
|
|
new_id = max([r["id"] for r in keys] + [0]) + 1
|
|
keys.append(
|
|
{
|
|
"id": new_id,
|
|
"name": name,
|
|
"flood_policy": flood_policy,
|
|
"parent_id": parent_id,
|
|
}
|
|
)
|
|
created.append((name, flood_policy, parent_id))
|
|
return new_id
|
|
|
|
def delete_transport_key(key_id):
|
|
for idx, rec in enumerate(keys):
|
|
if rec["id"] == key_id:
|
|
deleted.append(key_id)
|
|
del keys[idx]
|
|
return True
|
|
return False
|
|
|
|
cfg = _base_config()
|
|
cfg["mesh"] = {"unscoped_flood_allow": True, "default_region": None}
|
|
cli = MeshCLI("/tmp/cfg.yaml", cfg, _cfg_mgr(), enable_regions=True)
|
|
cli.storage_handler = SimpleNamespace(
|
|
get_transport_keys=get_transport_keys,
|
|
update_transport_key=update_transport_key,
|
|
create_transport_key=create_transport_key,
|
|
delete_transport_key=delete_transport_key,
|
|
)
|
|
|
|
tree = cli._cmd_region("region")
|
|
assert "* F" in tree
|
|
assert " alpha F" in tree
|
|
assert " beta" in tree
|
|
|
|
assert cli._cmd_region("region get alpha") == " alpha F"
|
|
assert cli._cmd_region("region get beta") == " beta (alpha)"
|
|
assert cli._cmd_region("region get missing") == "Err - unknown region"
|
|
|
|
assert cli._cmd_region("region allowf beta") == "OK"
|
|
assert keys[1]["flood_policy"] == "allow"
|
|
assert cli._cmd_region("region denyf beta") == "OK"
|
|
assert keys[1]["flood_policy"] == "deny"
|
|
|
|
assert cli._cmd_region("region home alpha") == " home is now alpha"
|
|
assert cli._cmd_region("region home") == " home is alpha"
|
|
|
|
assert cli._cmd_region("region default") == " default scope is <null>"
|
|
assert cli._cmd_region("region default alpha") == " default scope is now alpha"
|
|
assert cfg["mesh"]["default_region"] == "alpha"
|
|
assert cli._cmd_region("region default") == " default scope is alpha"
|
|
assert cli._cmd_region("region default <null>") == " default scope is now <null>"
|
|
assert cfg["mesh"]["default_region"] is None
|
|
|
|
assert cli._cmd_region("region put gamma alpha") == "OK - (flood allowed)"
|
|
assert any(r[0] == "gamma" for r in created)
|
|
|
|
assert cli._cmd_region("region remove alpha") == "Err - not empty"
|
|
assert cli._cmd_region("region remove gamma") == "OK"
|
|
assert deleted
|
|
|
|
assert cli._cmd_region("region list denied") == "beta"
|
|
listed = cli._cmd_region("region list allowed")
|
|
assert listed.startswith("*,")
|
|
assert "alpha" in listed
|
|
assert cli._cmd_region("region list nope") == "Err - use 'allowed' or 'denied'"
|
|
|
|
|
|
def test_discovery_auto_add_skips_local_node_and_persists_remote():
|
|
identity = SimpleNamespace(get_public_key=lambda: bytes.fromhex("11" * 32))
|
|
storage = SimpleNamespace(record_advert=MagicMock())
|
|
cli = MeshCLI(
|
|
"/tmp/cfg.yaml", _base_config(), _cfg_mgr(), identity=identity, storage_handler=storage
|
|
)
|
|
|
|
self_result = cli._auto_add_discovery_result({"pub_key": "11" * 8, "node_type": 2})
|
|
assert self_result["is_self"] is True
|
|
storage.record_advert.assert_not_called()
|
|
|
|
remote_result = cli._auto_add_discovery_result(
|
|
{
|
|
"pub_key": "22" * 32,
|
|
"node_name": "Remote Repeater",
|
|
"node_type": 2,
|
|
"rssi": -70,
|
|
"response_snr": 4.25,
|
|
}
|
|
)
|
|
assert remote_result["auto_added"] is True
|
|
storage.record_advert.assert_called_once()
|
|
|
|
|
|
def test_discovery_auto_add_persists_through_the_storage_actually_wired_in():
|
|
"""MeshCLI is constructed with the SQLiteHandler, not the StorageCollector.
|
|
|
|
See repeater/main.py:516 -> TextHelper(sqlite_handler=...storage.sqlite_handler)
|
|
-> MeshCLI(storage_handler=self.sqlite_handler). SQLiteHandler exposes
|
|
store_advert but no record_advert, so a persistence path that only looked for
|
|
record_advert made `discover.neighbors` silently record nothing in production
|
|
while passing tests that injected a hand-rolled double. Spec'ing the mock off
|
|
the real class is what keeps this honest.
|
|
"""
|
|
from repeater.data_acquisition.sqlite_handler import SQLiteHandler
|
|
|
|
identity = SimpleNamespace(get_public_key=lambda: bytes.fromhex("11" * 32))
|
|
storage = MagicMock(spec=SQLiteHandler)
|
|
cli = MeshCLI(
|
|
"/tmp/cfg.yaml", _base_config(), _cfg_mgr(), identity=identity, storage_handler=storage
|
|
)
|
|
|
|
result = cli._auto_add_discovery_result(
|
|
{
|
|
"pub_key": "22" * 32,
|
|
"node_name": "Remote Repeater",
|
|
"node_type": 2,
|
|
"rssi": -70,
|
|
"response_snr": 4.25,
|
|
}
|
|
)
|
|
|
|
assert result["auto_added"] is True
|
|
storage.store_advert.assert_called_once()
|
|
record = storage.store_advert.call_args.args[0]
|
|
assert record["pubkey"] == "22" * 32
|
|
assert record["is_repeater"] is True
|
|
assert record["zero_hop"] is True
|
|
assert record["snr"] == 4.25
|
|
assert record["rssi"] == -70
|
|
|
|
|
|
def test_discovery_auto_add_prefers_record_advert_when_the_collector_is_wired_in():
|
|
"""StorageCollector.record_advert also publishes the advert; prefer it."""
|
|
from repeater.data_acquisition.storage_collector import StorageCollector
|
|
|
|
identity = SimpleNamespace(get_public_key=lambda: bytes.fromhex("11" * 32))
|
|
storage = MagicMock(spec=StorageCollector)
|
|
cli = MeshCLI(
|
|
"/tmp/cfg.yaml", _base_config(), _cfg_mgr(), identity=identity, storage_handler=storage
|
|
)
|
|
|
|
result = cli._auto_add_discovery_result({"pub_key": "22" * 32, "node_type": 2})
|
|
|
|
assert result["auto_added"] is True
|
|
storage.record_advert.assert_called_once()
|
|
|
|
|
|
def test_cmd_set_save_failure_reports_error_and_skips_live_update():
|
|
mgr = _cfg_mgr(save_ok=False)
|
|
cli = MeshCLI("/tmp/cfg.yaml", _base_config(), mgr)
|
|
|
|
for command in ("af 2.0", "name x", "freq 915", "guest.password g", "flood.max 8"):
|
|
assert cli._cmd_set(command) == "Error: Failed to save config"
|
|
mgr.live_update_daemon.assert_not_called()
|
|
|
|
|
|
def test_cmd_get_reads_repeater_security_section():
|
|
cli = MeshCLI("/tmp/cfg.yaml", _base_config(), _cfg_mgr())
|
|
|
|
# Values come from repeater.security, not the legacy top-level decoy.
|
|
assert cli._cmd_get("guest.password") == "> guest"
|
|
assert cli._cmd_get("allow.read.only") == "> on"
|
|
|
|
|
|
def test_cli_password_change_applies_to_live_repeater_acl(tmp_path):
|
|
"""A CLI password change must reach the running LoginHelper ACL, so the
|
|
next login attempt authenticates against the new password."""
|
|
from repeater.config_manager import ConfigManager
|
|
from repeater.handler_helpers.login import LoginHelper
|
|
|
|
cfg = _base_config()
|
|
login_helper = LoginHelper(identity_manager=None, config=cfg)
|
|
identity = SimpleNamespace(get_public_key=lambda: b"\x42" + b"\x00" * 31)
|
|
login_helper.register_identity(
|
|
name="repeater", identity=identity, identity_type="repeater", config=cfg
|
|
)
|
|
acl = login_helper.get_acl_for_identity(0x42)
|
|
assert acl.admin_password == "adminpw"
|
|
assert acl.guest_password == "guest"
|
|
|
|
daemon = SimpleNamespace(config=cfg, login_helper=login_helper)
|
|
manager = ConfigManager(
|
|
config_path=str(tmp_path / "config.yaml"), config=cfg, daemon_instance=daemon
|
|
)
|
|
cli = MeshCLI(str(tmp_path / "config.yaml"), cfg, manager)
|
|
|
|
assert cli._cmd_password("password fresh-secret") == "password now: fresh-secret"
|
|
assert acl.admin_password == "fresh-secret"
|
|
|
|
assert cli._cmd_set("guest.password fresh-guest") == "OK"
|
|
assert acl.guest_password == "fresh-guest"
|
|
|
|
assert cli._cmd_set("allow.read.only off") == "OK"
|
|
assert acl.allow_read_only is False
|
|
|
|
|
|
def test_cmd_set_radio_commands_stage_without_live_apply():
|
|
"""Radio changes match firmware reboot-to-apply: saved to disk, never
|
|
live-applied (a live retune would cut off the admin mid-session)."""
|
|
cfg = _base_config()
|
|
mgr = _cfg_mgr()
|
|
cli = MeshCLI("/tmp/cfg.yaml", cfg, mgr)
|
|
|
|
assert cli._cmd_set("radio 900 250 9 6") == "OK - restart repeater to apply"
|
|
assert cli._cmd_set("freq 868.5") == "OK - restart repeater to apply"
|
|
assert cli._cmd_set("tx 17") == "OK - restart repeater to apply"
|
|
|
|
assert mgr.save_to_file.call_count == 3
|
|
mgr.live_update_daemon.assert_not_called()
|
|
|
|
|
|
def test_cmd_get_flood_advert_interval_reads_engine_key():
|
|
cfg = _base_config()
|
|
cli = MeshCLI("/tmp/cfg.yaml", cfg, _cfg_mgr())
|
|
|
|
# The orphan flood_advert_interval_hours (24) must be ignored; the engine
|
|
# key is absent so the engine default (10) is reported.
|
|
assert cli._cmd_get("flood.advert.interval") == "> 10"
|
|
|
|
cfg["repeater"]["send_advert_interval_hours"] = 6
|
|
assert cli._cmd_get("flood.advert.interval") == "> 6"
|
|
|
|
|
|
def test_cli_set_commands_persist_with_real_config_manager(tmp_path):
|
|
"""Every CLI set command must work against the real ConfigManager bool save
|
|
contract: reply OK and leave the change on disk."""
|
|
import yaml
|
|
|
|
from repeater.config_manager import ConfigManager
|
|
|
|
config_path = tmp_path / "config.yaml"
|
|
cfg = _base_config()
|
|
manager = ConfigManager(config_path=str(config_path), config=cfg, daemon_instance=None)
|
|
cli = MeshCLI(str(config_path), cfg, manager)
|
|
|
|
commands = [
|
|
"af 1.5",
|
|
"name node-b",
|
|
"repeat off",
|
|
"lat 10.5",
|
|
"lon -3.25",
|
|
"radio 869.618 250 9 6",
|
|
"freq 915.125",
|
|
"tx 17",
|
|
"guest.password gpw",
|
|
"owner.info Bob|Lab",
|
|
"allow.read.only on",
|
|
"advert.interval 90",
|
|
"flood.advert.interval 12",
|
|
"flood.max 8",
|
|
"path.hash.mode 1",
|
|
"loop.detect moderate",
|
|
"rxdelay 4.5",
|
|
"txdelay 1.5",
|
|
"direct.txdelay 0.25",
|
|
"multi.acks 1",
|
|
"int.thresh -110",
|
|
"agc.reset.interval 8",
|
|
]
|
|
for command in commands:
|
|
reply = cli._cmd_set(command)
|
|
assert reply.startswith("OK"), f"set {command!r} replied {reply!r}"
|
|
|
|
assert cli._cmd_password("password newadminpw") == "password now: newadminpw"
|
|
|
|
saved = yaml.safe_load(config_path.read_text())
|
|
assert saved["repeater"]["airtime_factor"] == 1.5
|
|
assert saved["repeater"]["node_name"] == "node-b"
|
|
assert saved["repeater"]["mode"] == "monitor"
|
|
assert saved["radio"]["frequency"] == 915125000
|
|
assert saved["radio"]["bandwidth"] == 250000
|
|
assert saved["repeater"]["security"]["guest_password"] == "gpw"
|
|
assert saved["repeater"]["security"]["admin_password"] == "newadminpw"
|
|
assert saved["repeater"]["security"]["allow_read_only"] is True
|
|
assert saved["repeater"]["send_advert_interval_hours"] == 12
|
|
assert saved["delays"]["rx_delay_base"] == 4.5
|
|
assert saved["delays"]["tx_delay_factor"] == 1.5
|
|
assert saved["delays"]["direct_tx_delay_factor"] == 0.25
|