From 44ba74561a6b89cab166e84529f53e489e8dce96 Mon Sep 17 00:00:00 2001 From: "opencode-agent[bot]" Date: Sun, 16 Nov 2025 23:24:59 +0000 Subject: [PATCH 1/2] Guest password auth implemented with login/logoff commands and enhanced telemetry requests. All tests pass. Co-authored-by: jinglemansweep --- AGENTS.md | 53 +++- meshcore_mqtt/meshcore_worker.py | 68 ++++- tests/test_guest_password_auth.py | 452 ++++++++++++++++++++++++++++++ 3 files changed, 570 insertions(+), 3 deletions(-) create mode 100644 tests/test_guest_password_auth.py diff --git a/AGENTS.md b/AGENTS.md index 926c9c2..20eaa67 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -108,6 +108,31 @@ The bridge includes configurable message rate limiting to prevent network floodi - `--meshcore-message-initial-delay 15.0` - `--meshcore-message-send-delay 15.0` +### Guest Password Support for Telemetry Requests + +The bridge supports requesting telemetry from repeaters that require guest password authentication: + +**Two-Step Authentication Process**: +1. **Login**: Use `send_login` command with destination and password +2. **Request Telemetry**: Use `send_telemetry_req` command (with or without password) + +**Automatic Login Feature**: +- When `send_telemetry_req` includes a `password` field, the bridge automatically: + 1. Logs into the repeater using `send_login(destination, password)` + 2. Then requests telemetry using `send_telemetry_req(destination)` +- This simplifies the workflow for one-time telemetry requests + +**Manual Session Management**: +For more control over the authentication session: +1. Use `send_login` to establish the session +2. Use `send_telemetry_req` (without password) for multiple requests +3. Use `send_logoff` to terminate the session when done + +**Use Cases**: +- **Automatic**: Single telemetry request with password +- **Manual**: Multiple telemetry requests during one session +- **Session Control**: Explicit login/logout for resource management + ### MQTT Topics The bridge publishes to structured MQTT topics: @@ -146,7 +171,9 @@ The bridge supports bidirectional communication via MQTT commands. Send commands | `set_name` | Set device name | `name` | `meshcore.commands.set_name()` | | `send_advert` | Send device advertisement | None (optional: `flood`) | `meshcore.commands.send_advert()` | | `send_trace` | Send trace packet for routing diagnostics | None (optional: `auth_code`, `tag`, `flags`, `path`) | `meshcore.commands.send_trace()` | -| `send_telemetry_req` | Request telemetry data from a node | `destination` | `meshcore.commands.send_telemetry_req()` | +| `send_telemetry_req` | Request telemetry data from a node | `destination` (optional: `password`) | `meshcore.commands.send_telemetry_req()` | +| `send_login` | Login to a repeater with guest password | `destination`, `password` | `meshcore.commands.send_login()` | +| `send_logoff` | Logoff from a repeater | `destination` | `meshcore.commands.send_logoff()` | **Command Examples**: ```json @@ -176,6 +203,18 @@ The bridge supports bidirectional communication via MQTT commands. Send commands // Send trace packet with parameters {"auth_code": 12345, "tag": 67890, "flags": 1, "path": "23,5f,3a"} + +// Request telemetry data (without password) +{"destination": "node123"} + +// Request telemetry data (with guest password) +{"destination": "repeater_node", "password": "guest_password"} + +// Login to repeater with guest password +{"destination": "repeater_node", "password": "guest_password"} + +// Logoff from repeater +{"destination": "repeater_node"} ``` **Command Examples**: @@ -215,6 +254,18 @@ mosquitto_pub -h localhost -t "meshcore/command/send_trace" \ # Request telemetry data from a node mosquitto_pub -h localhost -t "meshcore/command/send_telemetry_req" \ -m '{"destination": "node123"}' + +# Request telemetry data from repeater with guest password +mosquitto_pub -h localhost -t "meshcore/command/send_telemetry_req" \ + -m '{"destination": "repeater_node", "password": "guest_password"}' + +# Login to repeater with guest password +mosquitto_pub -h localhost -t "meshcore/command/send_login" \ + -m '{"destination": "repeater_node", "password": "guest_password"}' + +# Logoff from repeater +mosquitto_pub -h localhost -t "meshcore/command/send_logoff" \ + -m '{"destination": "repeater_node"}' ``` ## Development Guidelines diff --git a/meshcore_mqtt/meshcore_worker.py b/meshcore_mqtt/meshcore_worker.py index 19dc3f0..7e237f0 100644 --- a/meshcore_mqtt/meshcore_worker.py +++ b/meshcore_mqtt/meshcore_worker.py @@ -404,6 +404,27 @@ class MeshCoreWorker: command_type, command_data ) + elif command_type == "send_login": + destination = command_data.get("destination") + password = command_data.get("password") + if not destination or not password: + self.logger.error( + "send_login requires 'destination' and 'password' fields" + ) + return + result = await self._queue_rate_limited_command( + command_type, command_data + ) + + elif command_type == "send_logoff": + destination = command_data.get("destination") + if not destination: + self.logger.error("send_logoff requires 'destination' field") + return + result = await self._queue_rate_limited_command( + command_type, command_data + ) + else: self.logger.warning(f"Unknown command type: {command_type}") return @@ -754,12 +775,55 @@ class MeshCoreWorker: ) elif command_type == "send_telemetry_req": + destination = message_data.get("destination") + password = message_data.get("password") + if not isinstance(destination, str): + raise ValueError("Invalid destination type") + + # If password is provided, login first then request telemetry + if password: + if not isinstance(password, str): + raise ValueError("Invalid password type") + # Login first + await self._rate_limited_send( + f"send_login({destination})", + self.meshcore.commands.send_login, + destination, + password, + ) + # Then request telemetry + result = await self._rate_limited_send( + f"send_telemetry_req({destination})", + self.meshcore.commands.send_telemetry_req, + destination, + ) + else: + # No password, just request telemetry directly + result = await self._rate_limited_send( + f"send_telemetry_req({destination})", + self.meshcore.commands.send_telemetry_req, + destination, + ) + + elif command_type == "send_login": + destination = message_data.get("destination") + password = message_data.get("password") + if not isinstance(destination, str) or not isinstance(password, str): + raise ValueError("Invalid destination or password type") + result = await self._rate_limited_send( + f"send_login({destination})", + self.meshcore.commands.send_login, + destination, + password, + ) + + elif command_type == "send_logoff": destination = message_data.get("destination") if not isinstance(destination, str): raise ValueError("Invalid destination type") result = await self._rate_limited_send( - f"send_telemetry_req({destination})", - self.meshcore.commands.send_telemetry_req, + f"send_logoff({destination})", + self.meshcore.commands.send_logoff, destination, ) diff --git a/tests/test_guest_password_auth.py b/tests/test_guest_password_auth.py new file mode 100644 index 0000000..9a9eae9 --- /dev/null +++ b/tests/test_guest_password_auth.py @@ -0,0 +1,452 @@ +"""Tests for guest password authentication functionality.""" + +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from meshcore_mqtt.config import Config, ConnectionType, MeshCoreConfig, MQTTConfig +from meshcore_mqtt.meshcore_worker import MeshCoreWorker +from meshcore_mqtt.message_queue import Message, MessageType + + +@pytest.fixture +def test_config() -> Config: + """Create a test configuration.""" + return Config( + mqtt=MQTTConfig(broker="localhost"), + meshcore=MeshCoreConfig( + connection_type=ConnectionType.TCP, + address="127.0.0.1", + port=5000, + ), + ) + + +@pytest.fixture +def meshcore_worker(test_config: Config) -> MeshCoreWorker: + """Create a MeshCore worker instance for testing.""" + worker = MeshCoreWorker(test_config) + # Override startup grace period to allow immediate command processing + worker._startup_grace_period = 0 + return worker + + +class TestGuestPasswordAuthentication: + """Test guest password authentication commands.""" + + async def test_send_login_command_success( + self, meshcore_worker: MeshCoreWorker + ) -> None: + """Test successful send_login command.""" + # Setup mock MeshCore instance + mock_meshcore = MagicMock() + mock_meshcore.commands = MagicMock() + mock_meshcore.commands.send_login = AsyncMock() + meshcore_worker.meshcore = mock_meshcore + + # Mock the rate limiting queue to execute immediately + mock_queue_command = AsyncMock() + meshcore_worker._queue_rate_limited_command = mock_queue_command + + # Create login command message + command_data = {"destination": "repeater_node", "password": "guest_password"} + message = Message.create( + message_type=MessageType.MQTT_COMMAND, + source="mqtt", + target="meshcore", + payload={"command_type": "send_login", **command_data}, + ) + + # Process the command + await meshcore_worker._handle_mqtt_command(message) + + # Verify the command was queued for rate-limited execution + expected_data = {"command_type": "send_login", **command_data} + mock_queue_command.assert_called_once_with( + "send_login", expected_data + ) + + async def test_send_login_command_missing_destination( + self, meshcore_worker: MeshCoreWorker + ) -> None: + """Test send_login command with missing destination.""" + # Setup mock MeshCore instance + mock_meshcore = MagicMock() + mock_meshcore.commands = MagicMock() + mock_meshcore.commands.send_login = AsyncMock() + meshcore_worker.meshcore = mock_meshcore + + # Create login command message without destination + command_data = {"password": "guest_password"} # Missing destination + message = Message.create( + message_type=MessageType.MQTT_COMMAND, + source="mqtt", + target="meshcore", + payload={"command_type": "send_login", **command_data}, + ) + + # Process the command + await meshcore_worker._handle_mqtt_command(message) + + # Verify login command was NOT called due to validation + mock_meshcore.commands.send_login.assert_not_called() + + async def test_send_login_command_missing_password( + self, meshcore_worker: MeshCoreWorker + ) -> None: + """Test send_login command with missing password.""" + # Setup mock MeshCore instance + mock_meshcore = MagicMock() + mock_meshcore.commands = MagicMock() + mock_meshcore.commands.send_login = AsyncMock() + meshcore_worker.meshcore = mock_meshcore + + # Create login command message without password + command_data = {"destination": "repeater_node"} # Missing password + message = Message.create( + message_type=MessageType.MQTT_COMMAND, + source="mqtt", + target="meshcore", + payload={"command_type": "send_login", **command_data}, + ) + + # Process the command + await meshcore_worker._handle_mqtt_command(message) + + # Verify login command was NOT called due to validation + mock_meshcore.commands.send_login.assert_not_called() + + async def test_send_logoff_command_success( + self, meshcore_worker: MeshCoreWorker + ) -> None: + """Test successful send_logoff command.""" + # Setup mock MeshCore instance + mock_meshcore = MagicMock() + mock_meshcore.commands = MagicMock() + mock_meshcore.commands.send_logoff = AsyncMock() + meshcore_worker.meshcore = mock_meshcore + + # Mock the rate limiting queue to execute immediately + mock_queue_command = AsyncMock() + meshcore_worker._queue_rate_limited_command = mock_queue_command + + # Create logoff command message + command_data = {"destination": "repeater_node"} + message = Message.create( + message_type=MessageType.MQTT_COMMAND, + source="mqtt", + target="meshcore", + payload={"command_type": "send_logoff", **command_data}, + ) + + # Process the command + await meshcore_worker._handle_mqtt_command(message) + + # Verify the command was queued for rate-limited execution + expected_data = {"command_type": "send_logoff", **command_data} + mock_queue_command.assert_called_once_with( + "send_logoff", expected_data + ) + + async def test_send_logoff_command_missing_destination( + self, meshcore_worker: MeshCoreWorker + ) -> None: + """Test send_logoff command with missing destination.""" + # Setup mock MeshCore instance + mock_meshcore = MagicMock() + mock_meshcore.commands = MagicMock() + mock_meshcore.commands.send_logoff = AsyncMock() + meshcore_worker.meshcore = mock_meshcore + + # Create logoff command message without destination + command_data = {} # Missing destination + message = Message.create( + message_type=MessageType.MQTT_COMMAND, + source="mqtt", + target="meshcore", + payload={"command_type": "send_logoff", **command_data}, + ) + + # Process the command + await meshcore_worker._handle_mqtt_command(message) + + # Verify logoff command was NOT called due to validation + mock_meshcore.commands.send_logoff.assert_not_called() + + async def test_send_telemetry_req_with_password( + self, meshcore_worker: MeshCoreWorker + ) -> None: + """Test send_telemetry_req command with password (automatic login).""" + # Setup mock MeshCore instance + mock_meshcore = MagicMock() + mock_meshcore.commands = MagicMock() + mock_meshcore.commands.send_login = AsyncMock() + mock_meshcore.commands.send_telemetry_req = AsyncMock() + meshcore_worker.meshcore = mock_meshcore + + # Mock the rate limiting methods to execute immediately + mock_queue_command = AsyncMock() + meshcore_worker._queue_rate_limited_command = mock_queue_command + + # Create telemetry command message with password + command_data = {"destination": "repeater_node", "password": "guest_password"} + message = Message.create( + message_type=MessageType.MQTT_COMMAND, + source="mqtt", + target="meshcore", + payload={"command_type": "send_telemetry_req", **command_data}, + ) + + # Process the command + await meshcore_worker._handle_mqtt_command(message) + + # Verify the command was queued for rate-limited execution + expected_data = {"command_type": "send_telemetry_req", **command_data} + mock_queue_command.assert_called_once_with( + "send_telemetry_req", expected_data + ) + + async def test_send_telemetry_req_without_password( + self, meshcore_worker: MeshCoreWorker + ) -> None: + """Test send_telemetry_req command without password (direct request).""" + # Setup mock MeshCore instance + mock_meshcore = MagicMock() + mock_meshcore.commands = MagicMock() + mock_meshcore.commands.send_telemetry_req = AsyncMock() + meshcore_worker.meshcore = mock_meshcore + + # Mock rate limiting methods to execute immediately + mock_queue_command = AsyncMock() + meshcore_worker._queue_rate_limited_command = mock_queue_command + + # Create telemetry command message without password + command_data = {"destination": "node123"} + message = Message.create( + message_type=MessageType.MQTT_COMMAND, + source="mqtt", + target="meshcore", + payload={"command_type": "send_telemetry_req", **command_data}, + ) + + # Process the command + await meshcore_worker._handle_mqtt_command(message) + + # Verify the command was queued for rate-limited execution + expected_data = {"command_type": "send_telemetry_req", **command_data} + mock_queue_command.assert_called_once_with( + "send_telemetry_req", expected_data + ) + + async def test_execute_rate_limited_login_command( + self, meshcore_worker: MeshCoreWorker + ) -> None: + """Test execution of rate-limited login command.""" + # Setup mock MeshCore instance + mock_meshcore = MagicMock() + mock_meshcore.commands = MagicMock() + mock_meshcore.commands.send_login = AsyncMock(return_value="login_success") + meshcore_worker.meshcore = mock_meshcore + + # Mock rate limiting to execute immediately + meshcore_worker._rate_limited_send = AsyncMock(return_value="login_success") + + # Create message data for rate-limited execution + message_data = { + "command_type": "send_login", + "destination": "repeater_node", + "password": "guest_password", + "future": AsyncMock(), + } + + # Execute the rate-limited command + await meshcore_worker._execute_rate_limited_message(message_data) + + # Verify the rate-limited send was called with correct parameters + meshcore_worker._rate_limited_send.assert_called_once_with( + "send_login(repeater_node)", + mock_meshcore.commands.send_login, + "repeater_node", + "guest_password", + ) + + async def test_execute_rate_limited_logoff_command( + self, meshcore_worker: MeshCoreWorker + ) -> None: + """Test execution of rate-limited logoff command.""" + # Setup mock MeshCore instance + mock_meshcore = MagicMock() + mock_meshcore.commands = MagicMock() + mock_meshcore.commands.send_logoff = AsyncMock(return_value="logoff_success") + meshcore_worker.meshcore = mock_meshcore + + # Mock rate limiting to execute immediately + meshcore_worker._rate_limited_send = AsyncMock(return_value="logoff_success") + + # Create message data for rate-limited execution + message_data = { + "command_type": "send_logoff", + "destination": "repeater_node", + "future": AsyncMock(), + } + + # Execute the rate-limited command + await meshcore_worker._execute_rate_limited_message(message_data) + + # Verify the rate-limited send was called with correct parameters + meshcore_worker._rate_limited_send.assert_called_once_with( + "send_logoff(repeater_node)", + mock_meshcore.commands.send_logoff, + "repeater_node", + ) + + async def test_execute_rate_limited_telemetry_with_password( + self, meshcore_worker: MeshCoreWorker + ) -> None: + """Test execution of rate-limited telemetry command with password.""" + # Setup mock MeshCore instance + mock_meshcore = MagicMock() + mock_meshcore.commands = MagicMock() + mock_meshcore.commands.send_login = AsyncMock(return_value="login_success") + mock_meshcore.commands.send_telemetry_req = AsyncMock(return_value="telemetry_data") + meshcore_worker.meshcore = mock_meshcore + + # Mock rate limiting to execute immediately + meshcore_worker._rate_limited_send = AsyncMock() + meshcore_worker._rate_limited_send.side_effect = [ + "login_success", # First call (login) + "telemetry_data", # Second call (telemetry) + ] + + # Create message data for rate-limited execution + message_data = { + "command_type": "send_telemetry_req", + "destination": "repeater_node", + "password": "guest_password", + "future": AsyncMock(), + } + + # Execute the rate-limited command + await meshcore_worker._execute_rate_limited_message(message_data) + + # Verify both login and telemetry were called + assert meshcore_worker._rate_limited_send.call_count == 2 + + # Check first call (login) + first_call = meshcore_worker._rate_limited_send.call_args_list[0] + assert first_call[0][0] == "send_login(repeater_node)" + assert first_call[0][1] == mock_meshcore.commands.send_login + assert first_call[0][2] == "repeater_node" + assert first_call[0][3] == "guest_password" + + # Check second call (telemetry) + second_call = meshcore_worker._rate_limited_send.call_args_list[1] + assert second_call[0][0] == "send_telemetry_req(repeater_node)" + assert second_call[0][1] == mock_meshcore.commands.send_telemetry_req + assert second_call[0][2] == "repeater_node" + + async def test_execute_rate_limited_telemetry_without_password( + self, meshcore_worker: MeshCoreWorker + ) -> None: + """Test execution of rate-limited telemetry command without password.""" + # Setup mock MeshCore instance + mock_meshcore = MagicMock() + mock_meshcore.commands = MagicMock() + mock_meshcore.commands.send_telemetry_req = AsyncMock(return_value="telemetry_data") + meshcore_worker.meshcore = mock_meshcore + + # Mock rate limiting to execute immediately + meshcore_worker._rate_limited_send = AsyncMock(return_value="telemetry_data") + + # Create message data for rate-limited execution + message_data = { + "command_type": "send_telemetry_req", + "destination": "node123", + "future": AsyncMock(), + } + + # Execute the rate-limited command + await meshcore_worker._execute_rate_limited_message(message_data) + + # Verify only telemetry was called (no login) + meshcore_worker._rate_limited_send.assert_called_once_with( + "send_telemetry_req(node123)", + mock_meshcore.commands.send_telemetry_req, + "node123", + ) + + async def test_no_meshcore_instance_handling( + self, meshcore_worker: MeshCoreWorker + ) -> None: + """Test command handling when MeshCore instance is None.""" + # Ensure meshcore is None + meshcore_worker.meshcore = None + + # Create login command message + command_data = {"destination": "repeater_node", "password": "guest_password"} + message = Message.create( + message_type=MessageType.MQTT_COMMAND, + source="mqtt", + target="meshcore", + payload={"command_type": "send_login", **command_data}, + ) + + # Process the command - should not raise exception + await meshcore_worker._handle_mqtt_command(message) + + async def test_command_error_handling( + self, meshcore_worker: MeshCoreWorker + ) -> None: + """Test error handling when authentication commands fail.""" + # Setup mock MeshCore instance that raises an exception + mock_meshcore = MagicMock() + mock_meshcore.commands = MagicMock() + mock_meshcore.commands.send_login = AsyncMock(side_effect=Exception("Login failed")) + meshcore_worker.meshcore = mock_meshcore + + # Mock the rate limiting to propagate the error + async def mock_queue_command(command_type: str, command_data: dict) -> Any: + raise Exception("Login failed") + + meshcore_worker._queue_rate_limited_command = mock_queue_command + + # Create login command message + command_data = {"destination": "repeater_node", "password": "guest_password"} + message = Message.create( + message_type=MessageType.MQTT_COMMAND, + source="mqtt", + target="meshcore", + payload={"command_type": "send_login", **command_data}, + ) + + # Process the command - should not raise exception, just log error + await meshcore_worker._handle_mqtt_command(message) + + async def test_invalid_parameter_types( + self, meshcore_worker: MeshCoreWorker + ) -> None: + """Test validation of parameter types in rate-limited execution.""" + # Setup mock MeshCore instance + mock_meshcore = MagicMock() + mock_meshcore.commands = MagicMock() + mock_meshcore.commands.send_login = AsyncMock() + meshcore_worker.meshcore = mock_meshcore + + # Create message data with invalid parameter types + message_data = { + "command_type": "send_login", + "destination": 123, # Should be string, not int + "password": "guest_password", + "future": AsyncMock(), + } + + # Execute rate-limited command - should handle error gracefully + # The error should be caught and logged, not crash the worker + await meshcore_worker._execute_rate_limited_message(message_data) + + # Verify login command was NOT called due to type validation + mock_meshcore.commands.send_login.assert_not_called() + + # The main test is that the error is handled gracefully without crashing + # The future handling may vary based on implementation details \ No newline at end of file From 2f23fec52d95a971566072528ed9622d4a0f3ad0 Mon Sep 17 00:00:00 2001 From: "opencode-agent[bot]" Date: Sun, 16 Nov 2025 23:32:03 +0000 Subject: [PATCH 2/2] Fixed pre-commit checks, all 116 tests pass. Co-authored-by: jinglemansweep --- tests/test_guest_password_auth.py | 74 ++++++++++++++++++------------- 1 file changed, 44 insertions(+), 30 deletions(-) diff --git a/tests/test_guest_password_auth.py b/tests/test_guest_password_auth.py index 9a9eae9..1771349 100644 --- a/tests/test_guest_password_auth.py +++ b/tests/test_guest_password_auth.py @@ -47,7 +47,9 @@ class TestGuestPasswordAuthentication: # Mock the rate limiting queue to execute immediately mock_queue_command = AsyncMock() - meshcore_worker._queue_rate_limited_command = mock_queue_command + meshcore_worker._queue_rate_limited_command = ( # type: ignore[method-assign] + mock_queue_command + ) # Create login command message command_data = {"destination": "repeater_node", "password": "guest_password"} @@ -63,9 +65,7 @@ class TestGuestPasswordAuthentication: # Verify the command was queued for rate-limited execution expected_data = {"command_type": "send_login", **command_data} - mock_queue_command.assert_called_once_with( - "send_login", expected_data - ) + mock_queue_command.assert_called_once_with("send_login", expected_data) async def test_send_login_command_missing_destination( self, meshcore_worker: MeshCoreWorker @@ -129,7 +129,9 @@ class TestGuestPasswordAuthentication: # Mock the rate limiting queue to execute immediately mock_queue_command = AsyncMock() - meshcore_worker._queue_rate_limited_command = mock_queue_command + meshcore_worker._queue_rate_limited_command = ( # type: ignore[method-assign] + mock_queue_command + ) # Create logoff command message command_data = {"destination": "repeater_node"} @@ -145,9 +147,7 @@ class TestGuestPasswordAuthentication: # Verify the command was queued for rate-limited execution expected_data = {"command_type": "send_logoff", **command_data} - mock_queue_command.assert_called_once_with( - "send_logoff", expected_data - ) + mock_queue_command.assert_called_once_with("send_logoff", expected_data) async def test_send_logoff_command_missing_destination( self, meshcore_worker: MeshCoreWorker @@ -160,7 +160,7 @@ class TestGuestPasswordAuthentication: meshcore_worker.meshcore = mock_meshcore # Create logoff command message without destination - command_data = {} # Missing destination + command_data: dict[str, Any] = {} # Missing destination message = Message.create( message_type=MessageType.MQTT_COMMAND, source="mqtt", @@ -187,8 +187,10 @@ class TestGuestPasswordAuthentication: # Mock the rate limiting methods to execute immediately mock_queue_command = AsyncMock() - meshcore_worker._queue_rate_limited_command = mock_queue_command - + meshcore_worker._queue_rate_limited_command = ( # type: ignore[method-assign] + mock_queue_command + ) + # Create telemetry command message with password command_data = {"destination": "repeater_node", "password": "guest_password"} message = Message.create( @@ -203,9 +205,7 @@ class TestGuestPasswordAuthentication: # Verify the command was queued for rate-limited execution expected_data = {"command_type": "send_telemetry_req", **command_data} - mock_queue_command.assert_called_once_with( - "send_telemetry_req", expected_data - ) + mock_queue_command.assert_called_once_with("send_telemetry_req", expected_data) async def test_send_telemetry_req_without_password( self, meshcore_worker: MeshCoreWorker @@ -219,8 +219,10 @@ class TestGuestPasswordAuthentication: # Mock rate limiting methods to execute immediately mock_queue_command = AsyncMock() - meshcore_worker._queue_rate_limited_command = mock_queue_command - + meshcore_worker._queue_rate_limited_command = ( # type: ignore[method-assign] + mock_queue_command + ) + # Create telemetry command message without password command_data = {"destination": "node123"} message = Message.create( @@ -235,9 +237,7 @@ class TestGuestPasswordAuthentication: # Verify the command was queued for rate-limited execution expected_data = {"command_type": "send_telemetry_req", **command_data} - mock_queue_command.assert_called_once_with( - "send_telemetry_req", expected_data - ) + mock_queue_command.assert_called_once_with("send_telemetry_req", expected_data) async def test_execute_rate_limited_login_command( self, meshcore_worker: MeshCoreWorker @@ -250,7 +250,9 @@ class TestGuestPasswordAuthentication: meshcore_worker.meshcore = mock_meshcore # Mock rate limiting to execute immediately - meshcore_worker._rate_limited_send = AsyncMock(return_value="login_success") + meshcore_worker._rate_limited_send = AsyncMock( # type: ignore[method-assign] + return_value="login_success" + ) # Create message data for rate-limited execution message_data = { @@ -282,7 +284,9 @@ class TestGuestPasswordAuthentication: meshcore_worker.meshcore = mock_meshcore # Mock rate limiting to execute immediately - meshcore_worker._rate_limited_send = AsyncMock(return_value="logoff_success") + meshcore_worker._rate_limited_send = AsyncMock( # type: ignore[method-assign] + return_value="logoff_success" + ) # Create message data for rate-limited execution message_data = { @@ -309,11 +313,13 @@ class TestGuestPasswordAuthentication: mock_meshcore = MagicMock() mock_meshcore.commands = MagicMock() mock_meshcore.commands.send_login = AsyncMock(return_value="login_success") - mock_meshcore.commands.send_telemetry_req = AsyncMock(return_value="telemetry_data") + mock_meshcore.commands.send_telemetry_req = AsyncMock( + return_value="telemetry_data" + ) meshcore_worker.meshcore = mock_meshcore # Mock rate limiting to execute immediately - meshcore_worker._rate_limited_send = AsyncMock() + meshcore_worker._rate_limited_send = AsyncMock() # type: ignore[method-assign] meshcore_worker._rate_limited_send.side_effect = [ "login_success", # First call (login) "telemetry_data", # Second call (telemetry) @@ -332,7 +338,7 @@ class TestGuestPasswordAuthentication: # Verify both login and telemetry were called assert meshcore_worker._rate_limited_send.call_count == 2 - + # Check first call (login) first_call = meshcore_worker._rate_limited_send.call_args_list[0] assert first_call[0][0] == "send_login(repeater_node)" @@ -353,11 +359,15 @@ class TestGuestPasswordAuthentication: # Setup mock MeshCore instance mock_meshcore = MagicMock() mock_meshcore.commands = MagicMock() - mock_meshcore.commands.send_telemetry_req = AsyncMock(return_value="telemetry_data") + mock_meshcore.commands.send_telemetry_req = AsyncMock( + return_value="telemetry_data" + ) meshcore_worker.meshcore = mock_meshcore # Mock rate limiting to execute immediately - meshcore_worker._rate_limited_send = AsyncMock(return_value="telemetry_data") + meshcore_worker._rate_limited_send = AsyncMock( # type: ignore[method-assign] + return_value="telemetry_data" + ) # Create message data for rate-limited execution message_data = { @@ -402,14 +412,18 @@ class TestGuestPasswordAuthentication: # Setup mock MeshCore instance that raises an exception mock_meshcore = MagicMock() mock_meshcore.commands = MagicMock() - mock_meshcore.commands.send_login = AsyncMock(side_effect=Exception("Login failed")) + mock_meshcore.commands.send_login = AsyncMock( + side_effect=Exception("Login failed") + ) meshcore_worker.meshcore = mock_meshcore # Mock the rate limiting to propagate the error async def mock_queue_command(command_type: str, command_data: dict) -> Any: raise Exception("Login failed") - meshcore_worker._queue_rate_limited_command = mock_queue_command + meshcore_worker._queue_rate_limited_command = ( # type: ignore[method-assign] + mock_queue_command + ) # Create login command message command_data = {"destination": "repeater_node", "password": "guest_password"} @@ -447,6 +461,6 @@ class TestGuestPasswordAuthentication: # Verify login command was NOT called due to type validation mock_meshcore.commands.send_login.assert_not_called() - + # The main test is that the error is handled gracefully without crashing - # The future handling may vary based on implementation details \ No newline at end of file + # The future handling may vary based on implementation details