125 lines
4.4 KiB
Python
125 lines
4.4 KiB
Python
"""
|
|
Tests for PLC connection functionality.
|
|
"""
|
|
|
|
import pytest
|
|
import sys
|
|
import os
|
|
from unittest.mock import Mock, patch
|
|
|
|
# Add the parent directory to the path so we can import the package
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from watermaker_plc_api.services.plc_connection import PLCConnection, get_plc_connection
|
|
from watermaker_plc_api.utils.error_handler import PLCConnectionError
|
|
|
|
|
|
class TestPLCConnection:
|
|
"""Test cases for PLCConnection class"""
|
|
|
|
def setup_method(self):
|
|
"""Reset the global connection instance before each test"""
|
|
# Clear the global singleton for clean tests
|
|
import watermaker_plc_api.services.plc_connection
|
|
watermaker_plc_api.services.plc_connection._plc_connection = None
|
|
|
|
def test_singleton_pattern(self):
|
|
"""Test that get_plc_connection returns the same instance"""
|
|
conn1 = get_plc_connection()
|
|
conn2 = get_plc_connection()
|
|
assert conn1 is conn2
|
|
|
|
@patch('watermaker_plc_api.services.plc_connection.ModbusTcpClient')
|
|
def test_successful_connection(self, mock_client_class):
|
|
"""Test successful PLC connection"""
|
|
# Setup mock
|
|
mock_client = Mock()
|
|
mock_client.connect.return_value = True
|
|
mock_client_class.return_value = mock_client
|
|
|
|
# Test connection
|
|
plc = PLCConnection()
|
|
result = plc.connect()
|
|
|
|
assert result is True
|
|
assert plc.is_connected is True
|
|
mock_client.connect.assert_called_once()
|
|
|
|
@patch('watermaker_plc_api.services.plc_connection.ModbusTcpClient')
|
|
def test_failed_connection(self, mock_client_class):
|
|
"""Test failed PLC connection"""
|
|
# Setup mock
|
|
mock_client = Mock()
|
|
mock_client.connect.return_value = False
|
|
mock_client_class.return_value = mock_client
|
|
|
|
# Test connection
|
|
plc = PLCConnection()
|
|
result = plc.connect()
|
|
|
|
assert result is False
|
|
assert plc.is_connected is False
|
|
|
|
@patch('watermaker_plc_api.services.plc_connection.ModbusTcpClient')
|
|
def test_read_input_register(self, mock_client_class):
|
|
"""Test reading input register"""
|
|
# Setup mock
|
|
mock_client = Mock()
|
|
mock_client.connect.return_value = True
|
|
mock_result = Mock()
|
|
mock_result.registers = [1234]
|
|
mock_result.isError.return_value = False
|
|
mock_client.read_input_registers.return_value = mock_result
|
|
mock_client_class.return_value = mock_client
|
|
|
|
# Test read
|
|
plc = PLCConnection()
|
|
plc.connect()
|
|
value = plc.read_input_register(1000)
|
|
|
|
assert value == 1234
|
|
mock_client.read_input_registers.assert_called_with(1000, 1, slave=1)
|
|
|
|
@patch('watermaker_plc_api.services.plc_connection.ModbusTcpClient')
|
|
def test_write_holding_register(self, mock_client_class):
|
|
"""Test writing holding register"""
|
|
# Setup mock
|
|
mock_client = Mock()
|
|
mock_client.connect.return_value = True
|
|
mock_result = Mock()
|
|
mock_result.isError.return_value = False
|
|
mock_client.write_register.return_value = mock_result
|
|
mock_client_class.return_value = mock_client
|
|
|
|
# Test write
|
|
plc = PLCConnection()
|
|
plc.connect()
|
|
success = plc.write_holding_register(1000, 5)
|
|
|
|
assert success is True
|
|
mock_client.write_register.assert_called_with(1000, 5, slave=1)
|
|
|
|
def test_write_without_connection(self):
|
|
"""Test writing register without PLC connection"""
|
|
with patch('watermaker_plc_api.services.plc_connection.ModbusTcpClient') as mock_client_class:
|
|
# Setup mock to fail connection
|
|
mock_client = Mock()
|
|
mock_client.connect.return_value = False
|
|
mock_client_class.return_value = mock_client
|
|
|
|
plc = PLCConnection()
|
|
|
|
with pytest.raises(PLCConnectionError):
|
|
plc.write_holding_register(1000, 5)
|
|
|
|
def test_get_connection_status(self):
|
|
"""Test getting connection status information"""
|
|
plc = PLCConnection()
|
|
status = plc.get_connection_status()
|
|
|
|
assert isinstance(status, dict)
|
|
assert "connected" in status
|
|
assert "ip_address" in status
|
|
assert "port" in status
|
|
assert "unit_id" in status
|
|
assert "timeout" in status |