an0n.eu/meshcore-mqtt
1
0
Fork 0
forked from iarv/meshcore-mqtt
Code Pull Requests Activity

Fixed pre-commit checks, all 116 tests pass.

Browse Source 
Co-authored-by: jinglemansweep <jinglemansweep@users.noreply.github.com>
This commit is contained in:
opencode-agent[bot]
2025-11-16 23:32:03 +00:00
parent 44ba74561a
commit 2f23fec52d
1 changed files with 44 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
tests/test_guest_password_auth.py
View File

@@ -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,7 +187,9 @@ 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"}
@@ -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,7 +219,9 @@ 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"}
@@ -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)
@@ -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"}