""" 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()