mirror of
https://github.com/pyMC-dev/pyMC_Repeater.git
synced 2026-08-07 17:33:16 +02:00
feat: enhance login handler with anonymous request support and region name formatting
- Updated the `LoginHelper` class to wrap the login handler in an `AnonRequestHandler`, allowing for proper handling of anonymous requests. - Introduced methods for formatting region names based on flood policies and added support for retrieving transport keys from SQLite storage. - Enhanced the constructor to accept additional parameters for SQLite handler and configuration, improving flexibility for owner-info and feature-flag replies. - Added tests to validate the new functionality and ensure correct behavior of region name formatting and owner/features callbacks.
This commit is contained in:
@@ -251,12 +251,16 @@ def test_login_register_identity_repeater_creates_acl_and_handler():
|
||||
identity = FakeIdentity(0x52)
|
||||
acl_obj = MagicMock()
|
||||
handler_obj = MagicMock()
|
||||
anon_obj = MagicMock()
|
||||
|
||||
with (
|
||||
patch("repeater.handler_helpers.acl.ACL", return_value=acl_obj) as acl_cls,
|
||||
patch(
|
||||
"repeater.handler_helpers.login.LoginServerHandler", return_value=handler_obj
|
||||
) as handler_cls,
|
||||
patch(
|
||||
"repeater.handler_helpers.login.AnonRequestHandler", return_value=anon_obj
|
||||
) as anon_cls,
|
||||
):
|
||||
helper.register_identity(
|
||||
name="repeater-main",
|
||||
@@ -271,11 +275,65 @@ def test_login_register_identity_repeater_creates_acl_and_handler():
|
||||
|
||||
acl_cls.assert_called_once()
|
||||
handler_cls.assert_called_once()
|
||||
handler_obj.set_send_packet_callback.assert_called_once()
|
||||
assert helper.handlers[0x52] is handler_obj
|
||||
# The login handler is wrapped in an AnonRequestHandler, and that wrapper is
|
||||
# what gets stored + wired with the send callback.
|
||||
anon_cls.assert_called_once()
|
||||
assert anon_cls.call_args.kwargs["login_handler"] is handler_obj
|
||||
anon_obj.set_send_packet_callback.assert_called_once()
|
||||
assert helper.handlers[0x52] is anon_obj
|
||||
assert helper.acls[0x52] is acl_obj
|
||||
|
||||
|
||||
class _FakeSqlite:
|
||||
def __init__(self, keys):
|
||||
self._keys = keys
|
||||
|
||||
def get_transport_keys(self):
|
||||
return self._keys
|
||||
|
||||
|
||||
def test_format_region_names_filters_and_strips():
|
||||
keys = [
|
||||
{"name": "#VHF", "flood_policy": "allow"},
|
||||
{"name": "USA", "flood_policy": "allow"},
|
||||
{"name": "secret", "flood_policy": "deny"},
|
||||
{"name": "*", "flood_policy": "allow"}, # duplicate wildcard ignored
|
||||
{"name": "", "flood_policy": "allow"},
|
||||
]
|
||||
# Default config => unscoped flood allowed => wildcard '*' present.
|
||||
helper = LoginHelper(identity_manager=MagicMock(), sqlite_handler=_FakeSqlite(keys))
|
||||
# Wildcard first (from policy), '#' stripped, deny + empty + literal '*' excluded.
|
||||
assert helper._format_region_names() == "*,VHF,USA"
|
||||
|
||||
|
||||
def test_format_region_names_wildcard_suppressed_when_unscoped_denied():
|
||||
keys = [{"name": "USA", "flood_policy": "allow"}]
|
||||
helper = LoginHelper(
|
||||
identity_manager=MagicMock(),
|
||||
sqlite_handler=_FakeSqlite(keys),
|
||||
config={"mesh": {"unscoped_flood_allow": False}},
|
||||
)
|
||||
# No wildcard when unscoped flood is denied (firmware: wildcard deny-flood).
|
||||
assert helper._format_region_names() == "USA"
|
||||
|
||||
|
||||
def test_format_region_names_without_storage_is_just_wildcard():
|
||||
# No named regions, but unscoped flood allowed by default => bare wildcard.
|
||||
helper = LoginHelper(identity_manager=MagicMock(), sqlite_handler=None)
|
||||
assert helper._format_region_names() == "*"
|
||||
|
||||
|
||||
def test_owner_and_features_callbacks_from_config():
|
||||
config = {"repeater": {"node_name": "node-x", "owner_info": "me", "mode": "monitor"}}
|
||||
helper = LoginHelper(identity_manager=MagicMock(), config=config)
|
||||
|
||||
assert helper._make_owner_info_fn("fallback", config)() == ("node-x", "me")
|
||||
# Non-forward mode sets the forwarding-disabled bit (0x80).
|
||||
assert helper._make_features_fn(config)() == 0x80
|
||||
# Forwarding mode clears it.
|
||||
assert helper._make_features_fn({"repeater": {"mode": "forward"}})() == 0x00
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_login_process_packet_routes_to_registered_handler_and_marks_no_retransmit():
|
||||
helper = LoginHelper(identity_manager=MagicMock(), packet_injector=AsyncMock())
|
||||
|
||||
Reference in New Issue
Block a user