| """ |
| Extended coverage tests for admin_routes.py. |
| |
| Target: 50%+ coverage (374 statements, ~187 lines to cover) |
| Focus: Admin endpoints, user management, system operations |
| """ |
| import pytest |
| from fastapi.testclient import TestClient |
| from fastapi import FastAPI |
| from unittest.mock import Mock, patch |
|
|
| |
| from api.admin_routes import router |
|
|
|
|
| |
| |
| |
|
|
| @pytest.fixture |
| def app(): |
| """Create test FastAPI app with admin router""" |
| app = FastAPI() |
| app.include_router(router) |
| return app |
|
|
|
|
| @pytest.fixture |
| def client(app): |
| """Create test client""" |
| return TestClient(app) |
|
|
|
|
| class TestAdminEndpoints: |
| """Test admin API endpoints.""" |
|
|
| def test_list_users(self, client): |
| """Test listing all users.""" |
| response = client.get("/api/admin/users") |
|
|
| assert response.status_code in [200, 401] |
| if response.status_code == 200: |
| data = response.json() |
| assert "users" in data or isinstance(data, list) |
|
|
| def test_get_user_by_id(self, client): |
| """Test getting specific user.""" |
| response = client.get("/api/admin/users/user-123") |
|
|
| assert response.status_code in [200, 404] |
|
|
| def test_create_user(self, client): |
| """Test creating new user.""" |
| response = client.post( |
| "/api/admin/users", |
| json={ |
| "email": "test@example.com", |
| "name": "Test User", |
| "role": "user" |
| } |
| ) |
|
|
| assert response.status_code in [201, 400, 401] |
|
|
| def test_update_user(self, client): |
| """Test updating user.""" |
| response = client.put( |
| "/api/admin/users/user-123", |
| json={"name": "Updated Name"} |
| ) |
|
|
| assert response.status_code in [200, 404] |
|
|
| def test_delete_user(self, client): |
| """Test deleting user.""" |
| response = client.delete("/api/admin/users/user-123") |
|
|
| assert response.status_code in [200, 204, 404] |
|
|
| def test_list_roles(self, client): |
| """Test listing all roles.""" |
| response = client.get("/api/admin/roles") |
|
|
| assert response.status_code in [200, 401] |
|
|
| def test_create_role(self, client): |
| """Test creating new role.""" |
| response = client.post( |
| "/api/admin/roles", |
| json={ |
| "name": "custom-role", |
| "permissions": {"read": True, "write": True} |
| } |
| ) |
|
|
| assert response.status_code in [201, 400, 401] |
|
|
| def test_assign_role_to_user(self, client): |
| """Test assigning role to user.""" |
| response = client.post( |
| "/api/admin/users/user-123/roles", |
| json={"role": "admin"} |
| ) |
|
|
| assert response.status_code in [200, 404] |
|
|
|
|
| class TestSystemOperations: |
| """Test system operation endpoints.""" |
|
|
| def test_get_system_stats(self, client): |
| """Test getting system statistics.""" |
| response = client.get("/api/admin/system/stats") |
|
|
| assert response.status_code in [200, 401] |
|
|
| def test_get_system_health(self, client): |
| """Test getting system health.""" |
| response = client.get("/api/admin/system/health") |
|
|
| assert response.status_code in [200] |
|
|
| def test_trigger_backup(self, client): |
| """Test triggering system backup.""" |
| response = client.post("/api/admin/system/backup") |
|
|
| assert response.status_code in [200, 202, 401] |
|
|
| def test_get_logs(self, client): |
| """Test getting system logs.""" |
| response = client.get("/api/admin/system/logs?limit=100") |
|
|
| assert response.status_code in [200, 401] |
|
|
| def test_clear_cache(self, client): |
| """Test clearing system cache.""" |
| response = client.post("/api/admin/system/cache/clear") |
|
|
| assert response.status_code in [200, 401] |
|
|
|
|
| class TestAdminErrorHandling: |
| """Test admin endpoint error handling.""" |
|
|
| def test_handle_invalid_user_id(self, client): |
| """Test handling of invalid user ID.""" |
| response = client.get("/api/admin/users/invalid-id-!!!") |
|
|
| assert response.status_code in [400, 404] |
|
|
| def test_handle_duplicate_email(self, client): |
| """Test handling of duplicate email.""" |
| with patch('core.models.User.query') as mock_query: |
| mock_query.return_value.filter.return_value.first.return_value = Mock(email="test@example.com") |
|
|
| response = client.post( |
| "/api/admin/users", |
| json={"email": "test@example.com", "name": "Test", "role": "user"} |
| ) |
|
|
| assert response.status_code in [400, 409] |
|
|
| def test_handle_unauthorized_admin_access(self, client): |
| """Test handling of unauthorized admin access.""" |
| response = client.get( |
| "/api/admin/users", |
| headers={"Authorization": "Bearer invalid-token"} |
| ) |
|
|
| assert response.status_code in [401, 403] |
|
|