50 lines
2.2 KiB
Python
50 lines
2.2 KiB
Python
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
from src.core.connection_manager import ConnectionManager
|
|
|
|
@pytest.fixture
|
|
def connection_manager():
|
|
return ConnectionManager()
|
|
|
|
def test_connect_calls_client_connect(connection_manager):
|
|
with patch("src.core.connection_manager.mqtt.Client") as mock_client_cls:
|
|
mock_client_instance = mock_client_cls.return_value
|
|
|
|
connection_manager.connect("127.0.0.1", "test_client")
|
|
|
|
mock_client_cls.assert_called_with(client_id="test_client", protocol=4) # protocol=4 is MQTTv311 usually, check paho constants if strict
|
|
mock_client_instance.connect.assert_called_with("127.0.0.1", 1883, keepalive=60)
|
|
mock_client_instance.loop_start.assert_called_once()
|
|
|
|
def test_disconnect_calls_client_disconnect(connection_manager):
|
|
with patch("src.core.connection_manager.mqtt.Client") as mock_client_cls:
|
|
mock_client_instance = mock_client_cls.return_value
|
|
connection_manager.connect("127.0.0.1", "test_client")
|
|
|
|
connection_manager.disconnect()
|
|
|
|
mock_client_instance.loop_stop.assert_called_once()
|
|
mock_client_instance.disconnect.assert_called_once()
|
|
assert connection_manager.client is None
|
|
assert connection_manager.connected is False
|
|
|
|
def test_on_connect_callback(connection_manager):
|
|
with patch("src.core.connection_manager.mqtt.Client") as mock_client_cls:
|
|
# We don't need to mock the instance methods deeply, just the flow.
|
|
connection_manager.connect("127.0.0.1", "test_client")
|
|
|
|
# Simulate successful connection callback
|
|
# on_connect signature: client, userdata, flags, rc
|
|
connection_manager._on_connect(None, None, None, 0)
|
|
|
|
assert connection_manager.connected is True
|
|
|
|
def test_on_connect_callback_failure(connection_manager):
|
|
with patch("src.core.connection_manager.mqtt.Client") as mock_client_cls:
|
|
connection_manager.connect("127.0.0.1", "test_client")
|
|
|
|
# Simulate failed connection callback (rc != 0)
|
|
connection_manager._on_connect(None, None, None, 5)
|
|
|
|
assert connection_manager.connected is False
|