288 lines
12 KiB
Python
288 lines
12 KiB
Python
"""
|
|
Unit tests for hardware manager functionality.
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
from datetime import datetime
|
|
|
|
from src.tempering_machine.services.hardware.hardware_manager import (
|
|
HardwareManager, TemperatureReading, MotorStatus, MotorState, ComponentStatus
|
|
)
|
|
|
|
|
|
class TestHardwareManager:
|
|
"""Test cases for HardwareManager."""
|
|
|
|
@pytest.fixture
|
|
def hardware_manager(self):
|
|
"""Create hardware manager instance for testing."""
|
|
return HardwareManager()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initialize_success(self, hardware_manager):
|
|
"""Test successful hardware manager initialization."""
|
|
with patch('src.tempering_machine.services.hardware.modbus_client.modbus_client') as mock_modbus:
|
|
mock_modbus.connect = AsyncMock(return_value=True)
|
|
|
|
result = await hardware_manager.initialize()
|
|
|
|
assert result is True
|
|
mock_modbus.connect.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initialize_failure(self, hardware_manager):
|
|
"""Test hardware manager initialization failure."""
|
|
with patch('src.tempering_machine.services.hardware.modbus_client.modbus_client') as mock_modbus:
|
|
mock_modbus.connect = AsyncMock(return_value=False)
|
|
|
|
result = await hardware_manager.initialize()
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_temperature_valid(self, hardware_manager):
|
|
"""Test getting valid temperature reading."""
|
|
# Setup mock temperature data
|
|
hardware_manager.current_status.temperatures["tank_bottom"] = TemperatureReading(
|
|
value=25.5,
|
|
timestamp=datetime.now(),
|
|
sensor_name="tank_bottom",
|
|
is_valid=True
|
|
)
|
|
|
|
reading = await hardware_manager.get_temperature("tank_bottom")
|
|
|
|
assert reading is not None
|
|
assert reading.value == 25.5
|
|
assert reading.sensor_name == "tank_bottom"
|
|
assert reading.is_valid is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_temperature_invalid_sensor(self, hardware_manager):
|
|
"""Test getting temperature for non-existent sensor."""
|
|
reading = await hardware_manager.get_temperature("non_existent_sensor")
|
|
|
|
assert reading is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_all_temperatures(self, hardware_manager):
|
|
"""Test getting all temperature readings."""
|
|
# Setup mock temperature data
|
|
hardware_manager.current_status.temperatures = {
|
|
"tank_bottom": TemperatureReading(
|
|
value=25.5, timestamp=datetime.now(), sensor_name="tank_bottom", is_valid=True
|
|
),
|
|
"fountain": TemperatureReading(
|
|
value=30.2, timestamp=datetime.now(), sensor_name="fountain", is_valid=True
|
|
)
|
|
}
|
|
|
|
temperatures = await hardware_manager.get_all_temperatures()
|
|
|
|
assert len(temperatures) == 2
|
|
assert "tank_bottom" in temperatures
|
|
assert "fountain" in temperatures
|
|
assert temperatures["tank_bottom"].value == 25.5
|
|
assert temperatures["fountain"].value == 30.2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_average_tank_temperature(self, hardware_manager):
|
|
"""Test calculating average tank temperature."""
|
|
# Setup mock temperature data
|
|
hardware_manager.current_status.temperatures = {
|
|
"tank_bottom": TemperatureReading(
|
|
value=25.0, timestamp=datetime.now(), sensor_name="tank_bottom", is_valid=True
|
|
),
|
|
"tank_wall": TemperatureReading(
|
|
value=27.0, timestamp=datetime.now(), sensor_name="tank_wall", is_valid=True
|
|
)
|
|
}
|
|
|
|
avg_temp = await hardware_manager.get_average_tank_temperature()
|
|
|
|
assert avg_temp == 26.0 # Average of 25.0 and 27.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_average_tank_temperature_no_data(self, hardware_manager):
|
|
"""Test calculating average tank temperature with no valid data."""
|
|
hardware_manager.current_status.temperatures = {}
|
|
|
|
avg_temp = await hardware_manager.get_average_tank_temperature()
|
|
|
|
assert avg_temp is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_motor_state_success(self, hardware_manager):
|
|
"""Test successful motor state change."""
|
|
with patch('src.tempering_machine.services.hardware.modbus_client.modbus_client') as mock_modbus:
|
|
# Mock successful read and write operations
|
|
mock_modbus.read_coils = AsyncMock(return_value=MagicMock(
|
|
success=True, value=[False, False, False, False, False, False, False, False]
|
|
))
|
|
mock_modbus.write_multiple_coils = AsyncMock(return_value=MagicMock(success=True))
|
|
|
|
result = await hardware_manager.set_motor_state("mixer_motor", True)
|
|
|
|
assert result is True
|
|
mock_modbus.read_coils.assert_called_once()
|
|
mock_modbus.write_multiple_coils.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_motor_state_unknown_motor(self, hardware_manager):
|
|
"""Test setting state for unknown motor."""
|
|
result = await hardware_manager.set_motor_state("unknown_motor", True)
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_motor_state_communication_failure(self, hardware_manager):
|
|
"""Test motor state change with communication failure."""
|
|
with patch('src.tempering_machine.services.hardware.modbus_client.modbus_client') as mock_modbus:
|
|
mock_modbus.read_coils = AsyncMock(return_value=MagicMock(success=False, error="Communication error"))
|
|
|
|
result = await hardware_manager.set_motor_state("mixer_motor", True)
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enable_motor(self, hardware_manager):
|
|
"""Test enabling a motor."""
|
|
with patch.object(hardware_manager, 'set_motor_state', new=AsyncMock(return_value=True)) as mock_set:
|
|
result = await hardware_manager.enable_motor("mixer_motor")
|
|
|
|
assert result is True
|
|
mock_set.assert_called_once_with("mixer_motor", True)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disable_motor(self, hardware_manager):
|
|
"""Test disabling a motor."""
|
|
with patch.object(hardware_manager, 'set_motor_state', new=AsyncMock(return_value=True)) as mock_set:
|
|
result = await hardware_manager.disable_motor("mixer_motor")
|
|
|
|
assert result is True
|
|
mock_set.assert_called_once_with("mixer_motor", False)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disable_all_motors(self, hardware_manager):
|
|
"""Test disabling all motors."""
|
|
with patch.object(hardware_manager, 'disable_motor', new=AsyncMock(return_value=True)) as mock_disable:
|
|
result = await hardware_manager.disable_all_motors()
|
|
|
|
assert result is True
|
|
# Should be called for each motor that contains "motor" in its name
|
|
assert mock_disable.call_count > 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_emergency_stop(self, hardware_manager):
|
|
"""Test emergency stop functionality."""
|
|
with patch.object(hardware_manager, 'disable_all_motors', new=AsyncMock(return_value=True)) as mock_motors, \
|
|
patch.object(hardware_manager, 'disable_all_heaters', new=AsyncMock(return_value=True)) as mock_heaters:
|
|
|
|
result = await hardware_manager.emergency_stop()
|
|
|
|
assert result is True
|
|
mock_motors.assert_called_once()
|
|
mock_heaters.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_safe_to_operate_safe(self, hardware_manager):
|
|
"""Test safety check when system is safe."""
|
|
from src.tempering_machine.services.hardware.hardware_manager import SafetyStatus
|
|
|
|
# Setup safe conditions
|
|
hardware_manager.current_status.safety = SafetyStatus(
|
|
emergency_stop_active=False,
|
|
cover_sensor_closed=True,
|
|
temperature_alarms=[],
|
|
current_alarms=[]
|
|
)
|
|
hardware_manager.current_status.communication_health = 95.0
|
|
hardware_manager.current_status.temperatures = {
|
|
"tank_bottom": TemperatureReading(
|
|
value=25.0, timestamp=datetime.now(), sensor_name="tank_bottom", is_valid=True
|
|
)
|
|
}
|
|
|
|
is_safe, issues = await hardware_manager.is_safe_to_operate()
|
|
|
|
assert is_safe is True
|
|
assert len(issues) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_safe_to_operate_emergency_stop(self, hardware_manager):
|
|
"""Test safety check when emergency stop is active."""
|
|
from src.tempering_machine.services.hardware.hardware_manager import SafetyStatus
|
|
|
|
hardware_manager.current_status.safety = SafetyStatus(
|
|
emergency_stop_active=True,
|
|
cover_sensor_closed=True,
|
|
temperature_alarms=[],
|
|
current_alarms=[]
|
|
)
|
|
|
|
is_safe, issues = await hardware_manager.is_safe_to_operate()
|
|
|
|
assert is_safe is False
|
|
assert "Emergency stop is active" in issues
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_safe_to_operate_cover_open(self, hardware_manager):
|
|
"""Test safety check when cover is open."""
|
|
from src.tempering_machine.services.hardware.hardware_manager import SafetyStatus
|
|
|
|
hardware_manager.current_status.safety = SafetyStatus(
|
|
emergency_stop_active=False,
|
|
cover_sensor_closed=False,
|
|
temperature_alarms=[],
|
|
current_alarms=[]
|
|
)
|
|
|
|
is_safe, issues = await hardware_manager.is_safe_to_operate()
|
|
|
|
assert is_safe is False
|
|
assert "Safety cover is open" in issues
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_safe_to_operate_communication_issues(self, hardware_manager):
|
|
"""Test safety check with communication issues."""
|
|
from src.tempering_machine.services.hardware.hardware_manager import SafetyStatus
|
|
|
|
hardware_manager.current_status.safety = SafetyStatus(
|
|
emergency_stop_active=False,
|
|
cover_sensor_closed=True,
|
|
temperature_alarms=[],
|
|
current_alarms=[]
|
|
)
|
|
hardware_manager.current_status.communication_health = 30.0 # Poor communication
|
|
|
|
is_safe, issues = await hardware_manager.is_safe_to_operate()
|
|
|
|
assert is_safe is False
|
|
assert "Poor communication with hardware" in issues
|
|
|
|
def test_get_hardware_status(self, hardware_manager):
|
|
"""Test getting complete hardware status."""
|
|
status = hardware_manager.get_hardware_status()
|
|
|
|
assert status is not None
|
|
assert hasattr(status, 'temperatures')
|
|
assert hasattr(status, 'motors')
|
|
assert hasattr(status, 'safety')
|
|
assert hasattr(status, 'communication_health')
|
|
assert hasattr(status, 'system_status')
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shutdown(self, hardware_manager):
|
|
"""Test hardware manager shutdown."""
|
|
with patch.object(hardware_manager, 'stop_monitoring', new=AsyncMock()) as mock_stop, \
|
|
patch.object(hardware_manager, 'emergency_stop', new=AsyncMock(return_value=True)) as mock_emergency, \
|
|
patch('src.tempering_machine.services.hardware.modbus_client.modbus_client') as mock_modbus:
|
|
|
|
mock_modbus.disconnect = AsyncMock()
|
|
|
|
await hardware_manager.shutdown()
|
|
|
|
mock_stop.assert_called_once()
|
|
mock_emergency.assert_called_once()
|
|
mock_modbus.disconnect.assert_called_once() |