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