Add command-line interface, test suite, and unit tests for hardware manager and recipe models
This commit is contained in:
288
python_rewrite/tests/unit/test_hardware_manager.py
Normal file
288
python_rewrite/tests/unit/test_hardware_manager.py
Normal file
@@ -0,0 +1,288 @@
|
||||
"""
|
||||
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()
|
||||
Reference in New Issue
Block a user