diff --git a/dashboard/backend/tests/integration/test_system.py b/dashboard/backend/tests/integration/test_system.py new file mode 100644 index 0000000..2a515b4 --- /dev/null +++ b/dashboard/backend/tests/integration/test_system.py @@ -0,0 +1,196 @@ +""" +Integration tests for system operations. +""" +import pytest +import os +from unittest.mock import patch, Mock + + +class TestSystemStats: + """Test system statistics endpoint.""" + + def test_get_system_stats(self, test_app, admin_headers, temp_volume_root): + """Test getting system statistics.""" + # Mock disk_usage to avoid /volume1 dependency + mock_disk = Mock() + mock_disk.total = 1000000000000 + mock_disk.used = 500000000000 + mock_disk.free = 500000000000 + + with patch('shutil.disk_usage', return_value=mock_disk): + response = test_app.get("/api/system/stats", headers=admin_headers) + + assert response.status_code == 200 + data = response.json() + + # Verify expected fields are present + assert "cpu_percent" in data + assert "cpu_count" in data + assert "load_avg" in data + assert "memory" in data + assert "disk" in data + assert "uptime" in data + assert "platform" in data + + # Verify data types + assert isinstance(data["cpu_count"], int) + assert isinstance(data["load_avg"], list) + assert len(data["load_avg"]) == 3 + assert isinstance(data["memory"], dict) + assert "total" in data["memory"] + assert "used" in data["memory"] + assert "percent" in data["memory"] + + def test_get_system_stats_without_auth(self, test_app): + """Test getting stats without authentication.""" + response = test_app.get("/api/system/stats") + + assert response.status_code == 401 + + +class TestAuditLog: + """Test audit log endpoint.""" + + def test_get_audit_log_empty(self, test_app, admin_headers, temp_volume_root): + """Test getting audit log when file doesn't exist.""" + response = test_app.get("/api/system/audit-log", headers=admin_headers) + + assert response.status_code == 200 + data = response.json() + assert "entries" in data + assert isinstance(data["entries"], list) + + def test_get_audit_log_with_entries(self, test_app, admin_headers, temp_volume_root): + """Test getting audit log with sample entries.""" + # Create audit log file + log_dir = os.path.join(temp_volume_root, "docker/nas-dashboard") + os.makedirs(log_dir, exist_ok=True) + log_file = os.path.join(log_dir, "audit.log") + + # Write sample log entries + with open(log_file, "w") as f: + f.write("2024-01-01T10:00:00 192.168.1.100 admin GET /api/docker/containers 200 50ms\n") + f.write("2024-01-01T10:01:00 192.168.1.101 user POST /api/auth/login 401 100ms\n") + + response = test_app.get("/api/system/audit-log", headers=admin_headers) + + assert response.status_code == 200 + data = response.json() + assert "entries" in data + assert len(data["entries"]) >= 1 + + def test_get_audit_log_with_limit(self, test_app, admin_headers, temp_volume_root): + """Test getting audit log with line limit.""" + response = test_app.get( + "/api/system/audit-log", + headers=admin_headers, + params={"lines": 50} + ) + + assert response.status_code == 200 + data = response.json() + assert "entries" in data + + def test_get_audit_log_without_auth(self, test_app): + """Test getting audit log without authentication.""" + response = test_app.get("/api/system/audit-log") + + assert response.status_code == 401 + + +class TestNotification: + """Test notification endpoint.""" + + @pytest.mark.skip(reason="Requires Telegram bot token and makes external API call") + def test_send_notification_success(self, test_app, admin_headers): + """Test sending a notification.""" + response = test_app.post( + "/api/system/notify", + headers=admin_headers, + json={"message": "Test notification"} + ) + + # This will fail without proper Telegram config + assert response.status_code in [200, 400] + + def test_send_notification_without_message(self, test_app, admin_headers): + """Test sending notification without message.""" + response = test_app.post( + "/api/system/notify", + headers=admin_headers, + json={} + ) + + assert response.status_code == 200 + data = response.json() + assert data["ok"] is False + + def test_send_notification_without_auth(self, test_app): + """Test sending notification without authentication.""" + response = test_app.post( + "/api/system/notify", + json={"message": "Test"} + ) + + assert response.status_code == 401 + + +class TestOpenClawToken: + """Test OpenClaw token endpoint.""" + + def test_get_openclaw_token_as_admin(self, test_app, admin_headers, mock_config, monkeypatch): + """Test getting OpenClaw token as admin from LAN.""" + # Mock the IP check to return a LAN IP + def mock_get_client_ip(request): + return "192.168.1.100" + + def mock_is_lan_ip(ip): + return True + + import config + monkeypatch.setattr(config, "get_client_ip", mock_get_client_ip) + monkeypatch.setattr(config, "is_lan_ip", mock_is_lan_ip) + + response = test_app.get("/api/system/openclaw-token", headers=admin_headers) + + assert response.status_code == 200 + data = response.json() + assert "token" in data + + def test_get_openclaw_token_non_lan(self, test_app, admin_headers, monkeypatch): + """Test getting OpenClaw token from non-LAN IP.""" + # Mock the IP check to return a non-LAN IP + def mock_get_client_ip(request): + return "1.2.3.4" + + def mock_is_lan_ip(ip): + return False + + import config + monkeypatch.setattr(config, "get_client_ip", mock_get_client_ip) + monkeypatch.setattr(config, "is_lan_ip", mock_is_lan_ip) + + response = test_app.get("/api/system/openclaw-token", headers=admin_headers) + + assert response.status_code == 403 + + def test_get_openclaw_token_as_member(self, test_app, mock_config, temp_auth_file, temp_rbac_file): + """Test that non-admin cannot get OpenClaw token.""" + from auth_service import create_access_token + from datetime import timedelta + + member_token = create_access_token( + data={"sub": "memberuser", "role": "member"}, + expires_delta=timedelta(minutes=30) + ) + headers = {"Authorization": f"Bearer {member_token}"} + + response = test_app.get("/api/system/openclaw-token", headers=headers) + + assert response.status_code == 403 + + def test_get_openclaw_token_without_auth(self, test_app): + """Test getting OpenClaw token without authentication.""" + response = test_app.get("/api/system/openclaw-token") + + assert response.status_code == 401