"""Unit tests for SystemService. Tests the SystemService business logic layer, ensuring: - System information queries - Shutdown/restart operations - Error handling and error codes - Command execution and output parsing """ import pytest from unittest.mock import AsyncMock, Mock, patch from app.backend.services.system_service import SystemService class TestSystemServiceInfo: """Test system information queries.""" @pytest.mark.asyncio async def test_get_info_success(self): """Test successful system info retrieval.""" # Arrange service = SystemService() # Act with patch('psutil.cpu_percent', return_value=25.5), \ patch('psutil.virtual_memory') as mock_memory, \ patch('psutil.disk_usage') as mock_disk, \ patch('psutil.cpu_count', return_value=4), \ patch('psutil.boot_time', return_value=1000000000.0), \ patch('psutil.time.time', return_value=1000086400.0), \ patch('platform.node', return_value='dangerous-pi'), \ patch('platform.system', return_value='Linux'), \ patch('platform.release', return_value='6.1.21'): # Mock memory stats mock_memory.return_value = Mock( total=4294967296, # 4GB used=2147483648, # 2GB percent=50.0 ) # Mock disk stats mock_disk.return_value = Mock( total=32212254720, # 30GB used=10737418240, # 10GB percent=33.3 ) result = await service.get_info() # Assert assert result.success is True assert result.data["hostname"] == "dangerous-pi" assert result.data["platform"] == "Linux" assert result.data["cpu"]["count"] == 4 assert result.data["cpu"]["percent"] == 25.5 assert result.data["memory"]["total"] == 4294967296 assert result.data["memory"]["used"] == 2147483648 assert result.data["memory"]["percent"] == 50.0 assert result.data["disk"]["total"] == 32212254720 assert result.data["uptime"] == 86400.0 # 1 day @pytest.mark.asyncio async def test_get_info_with_temperature(self): """Test system info with CPU temperature.""" # Arrange service = SystemService() # Act with patch('psutil.cpu_percent', return_value=50.0), \ patch('psutil.virtual_memory'), \ patch('psutil.disk_usage'), \ patch('psutil.cpu_count', return_value=4), \ patch('psutil.boot_time', return_value=1000000000.0), \ patch('psutil.time.time', return_value=1000000100.0), \ patch('platform.node', return_value='test'), \ patch('platform.system', return_value='Linux'), \ patch('platform.release', return_value='6.1'), \ patch('pathlib.Path.exists', return_value=True), \ patch('pathlib.Path.read_text', return_value='45678'): # Mock memory and disk patch('psutil.virtual_memory', return_value=Mock(total=1, used=1, percent=1)) patch('psutil.disk_usage', return_value=Mock(total=1, used=1, percent=1)) result = await service.get_info() # Assert assert result.success is True assert result.data["cpu"]["temperature"] == 45.678 @pytest.mark.asyncio async def test_get_info_exception_handling(self): """Test system info retrieval with exception.""" # Arrange service = SystemService() # Act with patch('psutil.cpu_percent', side_effect=Exception("psutil error")): result = await service.get_info() # Assert assert result.success is False assert result.error.code == "system_info_error" assert "psutil error" in result.error.details class TestSystemServiceShutdown: """Test shutdown operations.""" @pytest.mark.asyncio async def test_shutdown_immediate(self): """Test immediate shutdown.""" # Arrange service = SystemService() # Act with patch.object(service, '_run_command', new_callable=AsyncMock) as mock_cmd: mock_cmd.return_value = "" result = await service.shutdown(delay=0) # Assert assert result.success is True assert "initiated" in result.data["message"].lower() mock_cmd.assert_called_once_with("sudo shutdown -h now") @pytest.mark.asyncio async def test_shutdown_with_delay(self): """Test scheduled shutdown.""" # Arrange service = SystemService() # Act with patch.object(service, '_run_command', new_callable=AsyncMock) as mock_cmd: mock_cmd.return_value = "" result = await service.shutdown(delay=300) # 5 minutes # Assert assert result.success is True assert "scheduled" in result.data["message"].lower() assert result.data["delay"] == 300 # 300 seconds = 5 minutes mock_cmd.assert_called_once_with("sudo shutdown -h +5") @pytest.mark.asyncio async def test_shutdown_failure(self): """Test shutdown command failure.""" # Arrange service = SystemService() # Act with patch.object(service, '_run_command', new_callable=AsyncMock) as mock_cmd: mock_cmd.side_effect = RuntimeError("Permission denied") result = await service.shutdown(delay=0) # Assert assert result.success is False assert result.error.code == "shutdown_error" assert "Permission denied" in result.error.details class TestSystemServiceRestart: """Test restart operations.""" @pytest.mark.asyncio async def test_restart_immediate(self): """Test immediate restart.""" # Arrange service = SystemService() # Act with patch.object(service, '_run_command', new_callable=AsyncMock) as mock_cmd: mock_cmd.return_value = "" result = await service.restart(delay=0) # Assert assert result.success is True assert "initiated" in result.data["message"].lower() mock_cmd.assert_called_once_with("sudo shutdown -r now") @pytest.mark.asyncio async def test_restart_with_delay(self): """Test scheduled restart.""" # Arrange service = SystemService() # Act with patch.object(service, '_run_command', new_callable=AsyncMock) as mock_cmd: mock_cmd.return_value = "" result = await service.restart(delay=600) # 10 minutes # Assert assert result.success is True assert "scheduled" in result.data["message"].lower() assert result.data["delay"] == 600 mock_cmd.assert_called_once_with("sudo shutdown -r +10") class TestSystemServiceCancelShutdown: """Test shutdown cancellation.""" @pytest.mark.asyncio async def test_cancel_shutdown_success(self): """Test successful shutdown cancellation.""" # Arrange service = SystemService() # Act with patch.object(service, '_run_command', new_callable=AsyncMock) as mock_cmd: mock_cmd.return_value = "" result = await service.cancel_shutdown() # Assert assert result.success is True assert "cancelled" in result.data["message"].lower() mock_cmd.assert_called_once_with("sudo shutdown -c") class TestSystemServiceLogs: """Test log retrieval.""" @pytest.mark.asyncio async def test_get_logs_success(self): """Test successful log retrieval.""" # Arrange service = SystemService() log_content = "Jan 01 12:00:00 test systemd[1]: Started dangerous-pi" # Act with patch.object(service, '_run_command', new_callable=AsyncMock) as mock_cmd: mock_cmd.return_value = log_content result = await service.get_logs(service="dangerous-pi", lines=100) # Assert assert result.success is True assert result.data["service"] == "dangerous-pi" assert result.data["lines"] == 100 assert result.data["logs"] == log_content mock_cmd.assert_called_once_with( "sudo journalctl -u dangerous-pi -n 100 --no-pager" ) @pytest.mark.asyncio async def test_get_logs_failure(self): """Test log retrieval failure.""" # Arrange service = SystemService() # Act with patch.object(service, '_run_command', new_callable=AsyncMock) as mock_cmd: mock_cmd.side_effect = RuntimeError("Service not found") result = await service.get_logs(service="nonexistent") # Assert assert result.success is False assert result.error.code == "logs_error" class TestSystemServiceStatus: """Test service status queries.""" @pytest.mark.asyncio async def test_get_service_status_active(self): """Test getting status of active service.""" # Arrange service = SystemService() status_output = "● dangerous-pi.service - Dangerous Pi Backend\n Loaded: loaded\n Active: active (running)" # Act with patch.object(service, '_run_command', new_callable=AsyncMock) as mock_cmd, \ patch.object(service, '_is_service_enabled', new_callable=AsyncMock) as mock_enabled: mock_cmd.return_value = status_output mock_enabled.return_value = True result = await service.get_service_status("dangerous-pi") # Assert assert result.success is True assert result.data["service"] == "dangerous-pi" assert result.data["active"] is True assert result.data["enabled"] is True assert "active (running)" in result.data["status"] @pytest.mark.asyncio async def test_get_service_status_inactive(self): """Test getting status of inactive service.""" # Arrange service = SystemService() status_output = "● test.service\n Loaded: loaded\n Active: inactive (dead)" # Act with patch.object(service, '_run_command', new_callable=AsyncMock) as mock_cmd, \ patch.object(service, '_is_service_enabled', new_callable=AsyncMock) as mock_enabled: mock_cmd.return_value = status_output mock_enabled.return_value = False result = await service.get_service_status("test") # Assert assert result.success is True assert result.data["active"] is False assert result.data["enabled"] is False