import pytest from aiohttp import web from aiohttp_session import get_session from unittest.mock import AsyncMock, patch, call from retoors.services.user_service import UserService @pytest.fixture async def admin_client(client): """Fixture to provide an aiohttp client with mocked services.""" return client @pytest.fixture async def logged_in_admin_client(admin_client): # Get the mocked user_service from the app mock_user_service = admin_client.app["user_service"] resp = await admin_client.post( "/login", data={"email": "admin@example.com", "password": "password"}, allow_redirects=False ) if resp.status != 302: print(f"Login failed with status {resp.status}. Response text: {await resp.text()}") assert resp.status == 302 # Expecting a redirect after successful login # Assert that authenticate_user was called on the mock mock_user_service.authenticate_user.assert_called_once_with("admin@example.com", "password") # The client will not follow the redirect, so the session cookie will be set on the client return admin_client, mock_user_service # Return both client and mock_user_service async def test_get_users_unauthorized(admin_client): resp = await admin_client.get("/api/users") assert resp.status == 401 data = await resp.json() assert data["error"] == "Unauthorized" async def test_get_users_authorized(logged_in_admin_client): client, mock_user_service = logged_in_admin_client resp = await client.get("/api/users") assert resp.status == 200 data = await resp.json() assert "users" in data assert len(data["users"]) == 2 assert data["users"][0]["email"] == "admin@example.com" assert data["users"][1]["email"] == "child1@example.com" mock_user_service.get_all_users.assert_called_once() async def test_add_user_unauthorized(admin_client): resp = await admin_client.post("/api/users", json={ "full_name": "New User", "email": "new@example.com", "password": "password123" }) assert resp.status == 401 data = await resp.json() assert data["error"] == "Unauthorized" async def test_add_user_authorized(logged_in_admin_client): client, mock_user_service = logged_in_admin_client resp = await client.post("/api/users", json={ "full_name": "New User", "email": "new@example.com", "password": "password123" }) assert resp.status == 201 data = await resp.json() assert data["message"] == "User added successfully" assert data["user"]["email"] == "new@example.com" mock_user_service.create_user.assert_called_once_with( full_name="New User", email="new@example.com", password="password123", parent_email="admin@example.com" ) async def test_add_user_invalid_data(logged_in_admin_client): client, mock_user_service = logged_in_admin_client resp = await client.post("/api/users", json={ "full_name": "Nu", # Too short "email": "invalid-email", "password": "123" # Too short }) assert resp.status == 400 data = await resp.json() assert "3 validation errors for RegistrationModel" in data["error"] async def test_add_user_email_exists(logged_in_admin_client): client, mock_user_service = logged_in_admin_client mock_user_service.create_user.side_effect = ValueError("User with this email already exists") resp = await client.post("/api/users", json={ "full_name": "Existing User", "email": "admin@example.com", "password": "password123" }) assert resp.status == 400 data = await resp.json() assert data["error"] == "User with this email already exists" async def test_update_user_quota_unauthorized(admin_client): resp = await admin_client.put("/api/users/child1@example.com/quota", json={ "new_quota_gb": 200 }) assert resp.status == 401 data = await resp.json() assert data["error"] == "Unauthorized" async def test_update_user_quota_authorized(logged_in_admin_client): client, mock_user_service = logged_in_admin_client resp = await client.put("/api/users/child1@example.com/quota", json={ "new_quota_gb": 200 }) assert resp.status == 200 data = await resp.json() assert data["message"] == "Quota for child1@example.com updated successfully" mock_user_service.update_user_quota.assert_called_once_with("child1@example.com", 200) async def test_update_user_quota_self(logged_in_admin_client): client, mock_user_service = logged_in_admin_client resp = await client.put("/api/users/admin@example.com/quota", json={ "new_quota_gb": 200 }) assert resp.status == 200 data = await resp.json() assert data["message"] == "Quota for admin@example.com updated successfully" mock_user_service.update_user_quota.assert_called_once_with("admin@example.com", 200) async def test_update_user_quota_forbidden(logged_in_admin_client): client, mock_user_service = logged_in_admin_client mock_user_service.get_user_by_email.side_effect = [ { # For current_user_email check "full_name": "Admin User", "email": "admin@example.com", "password": "hashed_password", "storage_quota_gb": 100, "storage_used_gb": 10, "parent_email": None }, { # For target_user_email check, not a child and not self "full_name": "Other User", "email": "other@example.com", "password": "hashed_password", "storage_quota_gb": 50, "storage_used_gb": 5, "parent_email": "another@example.com" } ] resp = await client.put("/api/users/other@example.com/quota", json={ "new_quota_gb": 200 }) assert resp.status == 403 data = await resp.json() assert data["error"] == "Forbidden: You do not have permission to update this user's quota" async def test_update_user_quota_user_not_found(logged_in_admin_client): client, mock_user_service = logged_in_admin_client mock_user_service.get_user_by_email.side_effect = [ { # For current_user_email check "full_name": "Admin User", "email": "admin@example.com", "password": "hashed_password", "storage_quota_gb": 100, "storage_used_gb": 10, "parent_email": None }, None # For target_user_email check ] resp = await client.put("/api/users/nonexistent@example.com/quota", json={ "new_quota_gb": 200 }) assert resp.status == 403 # Forbidden because user not found data = await resp.json() assert data["error"] == "Forbidden: You do not have permission to update this user's quota" async def test_delete_user_unauthorized(admin_client): resp = await admin_client.delete("/api/users/child1@example.com") assert resp.status == 401 data = await resp.json() assert data["error"] == "Unauthorized" async def test_delete_user_authorized(logged_in_admin_client): client, mock_user_service = logged_in_admin_client resp = await client.delete("/api/users/child1@example.com") assert resp.status == 200 data = await resp.json() assert data["message"] == "User child1@example.com deleted successfully" mock_user_service.delete_user.assert_called_once_with("child1@example.com") async def test_delete_user_self_forbidden(logged_in_admin_client): client, mock_user_service = logged_in_admin_client resp = await client.delete("/api/users/admin@example.com") assert resp.status == 403 data = await resp.json() assert data["error"] == "Forbidden: You cannot delete your own account from this interface" async def test_delete_user_forbidden(logged_in_admin_client): client, mock_user_service = logged_in_admin_client mock_user_service.get_user_by_email.side_effect = [ { # For current_user_email check "full_name": "Admin User", "email": "admin@example.com", "password": "hashed_password", "storage_quota_gb": 100, "storage_used_gb": 10, "parent_email": None }, { # For target_user_email check, not a child "full_name": "Other User", "email": "other@example.com", "password": "hashed_password", "storage_quota_gb": 50, "storage_used_gb": 5, "parent_email": "another@example.com" } ] resp = await client.delete("/api/users/other@example.com") assert resp.status == 403 data = await resp.json() assert data["error"] == "Forbidden: You do not have permission to delete this user" async def test_delete_user_not_found(logged_in_admin_client): client, mock_user_service = logged_in_admin_client mock_user_service.get_user_by_email.side_effect = [ { # For current_user_email check "full_name": "Admin User", "email": "admin@example.com", "password": "hashed_password", "storage_quota_gb": 100, "storage_used_gb": 10, "parent_email": None }, None # For target_user_email check ] mock_user_service.delete_user.return_value = False resp = await client.delete("/api/users/nonexistent@example.com") assert resp.status == 403 # Forbidden because user not found data = await resp.json() assert data["error"] == "Forbidden: You do not have permission to delete this user" async def test_get_user_details_unauthorized(admin_client): resp = await admin_client.get("/api/users/child1@example.com") assert resp.status == 401 data = await resp.json() assert data["error"] == "Unauthorized" async def test_get_user_details_authorized(logged_in_admin_client): client, mock_user_service = logged_in_admin_client mock_user_service.get_user_by_email.reset_mock() # Reset mock call count resp = await client.get("/api/users/child1@example.com") assert resp.status == 200 data = await resp.json() assert "user" in data assert data["user"]["email"] == "child1@example.com" assert "password" not in data["user"] # Ensure sensitive data is not returned mock_user_service.get_user_by_email.assert_has_calls([call("admin@example.com"), call("child1@example.com")]) async def test_get_user_details_self(logged_in_admin_client): client, mock_user_service = logged_in_admin_client mock_user_service.get_user_by_email.reset_mock() # Reset mock call count resp = await client.get("/api/users/admin@example.com") assert resp.status == 200 data = await resp.json() assert "user" in data assert data["user"]["email"] == "admin@example.com" mock_user_service.get_user_by_email.assert_has_calls([call("admin@example.com"), call("admin@example.com")]) async def test_get_user_details_forbidden(logged_in_admin_client): client, mock_user_service = logged_in_admin_client mock_user_service.get_user_by_email.side_effect = [ { # For current_user_email check "full_name": "Admin User", "email": "admin@example.com", "password": "hashed_password", "storage_quota_gb": 100, "storage_used_gb": 10, "parent_email": None }, { # For target_user_email check, not a child and not self "full_name": "Other User", "email": "other@example.com", "password": "hashed_password", "storage_quota_gb": 50, "storage_used_gb": 5, "parent_email": "another@example.com" } ] resp = await client.get("/api/users/other@example.com") assert resp.status == 403 data = await resp.json() assert data["error"] == "Forbidden: You do not have permission to view this user's details" async def test_get_user_details_not_found(logged_in_admin_client): client, mock_user_service = logged_in_admin_client mock_user_service.get_user_by_email.side_effect = [ { # For current_user_email check "full_name": "Admin User", "email": "admin@example.com", "password": "hashed_password", "storage_quota_gb": 100, "storage_used_gb": 10, "parent_email": None }, None # For target_user_email check ] resp = await client.get("/api/users/nonexistent@example.com") assert resp.status == 403 # Forbidden because user not found data = await resp.json() assert data["error"] == "Forbidden: You do not have permission to view this user's details" async def test_delete_team_unauthorized(admin_client): resp = await admin_client.delete("/api/teams/admin@example.com") assert resp.status == 401 data = await resp.json() assert data["error"] == "Unauthorized" async def test_delete_team_authorized(logged_in_admin_client): client, mock_user_service = logged_in_admin_client resp = await client.delete("/api/teams/admin@example.com") assert resp.status == 200 data = await resp.json() assert data["message"] == "Successfully deleted 1 users from the team managed by admin@example.com" mock_user_service.delete_users_by_parent_email.assert_called_once_with("admin@example.com") async def test_delete_team_forbidden(logged_in_admin_client): client, mock_user_service = logged_in_admin_client resp = await client.delete("/api/teams/other@example.com") assert resp.status == 403 data = await resp.json() assert data["error"] == "Forbidden: You do not have permission to delete this team" async def test_delete_team_no_users_found(logged_in_admin_client): client, mock_user_service = logged_in_admin_client mock_user_service.delete_users_by_parent_email.side_effect = lambda parent_email: 0 resp = await client.delete("/api/teams/admin@example.com") assert resp.status == 404 data = await resp.json() assert data["message"] == "No users found for team managed by admin@example.com or could not be deleted"