|
| 1 | +from unittest.mock import AsyncMock |
| 2 | + |
| 3 | +from httpx import AsyncClient |
| 4 | +import pytest |
| 5 | +import pytest_asyncio |
| 6 | + |
| 7 | +from api.dependencies import create_user_use_case_factory |
| 8 | +from api.domain.organization.errors import OrganizationNotFoundError |
| 9 | +from api.domain.role.errors import RoleNotFoundError |
| 10 | +from api.domain.user.errors import UserAlreadyExistsError, UserIsNotAdminError |
| 11 | +from api.tests.helpers import create_key |
| 12 | +from api.tests.integration.factories.sql import RoleSQLFactory, UserSQLFactory |
| 13 | +from api.utils.variables import EndpointRoute |
| 14 | + |
| 15 | +URL = f"/v1{EndpointRoute.ADMIN_USERS}" |
| 16 | + |
| 17 | + |
| 18 | +def _valid_body(role_id: int, **overrides) -> dict: |
| 19 | + body = { |
| 20 | + "email": "newuser@test.com", |
| 21 | + "password": "s3cr3t", |
| 22 | + "role": role_id, |
| 23 | + } |
| 24 | + body.update(overrides) |
| 25 | + return body |
| 26 | + |
| 27 | + |
| 28 | +@pytest.mark.asyncio(loop_scope="session") |
| 29 | +class TestCreateUser: |
| 30 | + @pytest_asyncio.fixture(autouse=True) |
| 31 | + async def setup(self, db_session): |
| 32 | + self.admin_user = UserSQLFactory(admin_user=True) |
| 33 | + self.token = await create_key(db_session, name="admin_token", user=self.admin_user) |
| 34 | + |
| 35 | + async def test_happy_path(self, client: AsyncClient, db_session): |
| 36 | + role = RoleSQLFactory() |
| 37 | + await db_session.flush() |
| 38 | + |
| 39 | + response = await client.post( |
| 40 | + url=URL, |
| 41 | + headers={"Authorization": f"Bearer {self.token.token}"}, |
| 42 | + json=_valid_body(role_id=role.id), |
| 43 | + ) |
| 44 | + |
| 45 | + assert response.status_code == 201, response.text |
| 46 | + data = response.json() |
| 47 | + assert data["email"] == "newuser@test.com" |
| 48 | + assert isinstance(data["id"], int) |
| 49 | + assert data["role"] == role.id |
| 50 | + |
| 51 | + @pytest.mark.parametrize( |
| 52 | + "use_case_result,expected_status,expected_detail", |
| 53 | + [ |
| 54 | + ( |
| 55 | + UserAlreadyExistsError(email="existing@test.com"), |
| 56 | + 409, |
| 57 | + "User existing@test.com already exists.", |
| 58 | + ), |
| 59 | + ( |
| 60 | + RoleNotFoundError(id=99), |
| 61 | + 404, |
| 62 | + "Role 99 not found.", |
| 63 | + ), |
| 64 | + ( |
| 65 | + OrganizationNotFoundError(id=99), |
| 66 | + 404, |
| 67 | + "Organization 99 not found.", |
| 68 | + ), |
| 69 | + ( |
| 70 | + UserIsNotAdminError(), |
| 71 | + 403, |
| 72 | + "User has no admin rights.", |
| 73 | + ), |
| 74 | + ], |
| 75 | + ) |
| 76 | + async def test_error_maps_to_correct_http_status(self, client: AsyncClient, app, use_case_result, expected_status, expected_detail): |
| 77 | + mock_use_case = AsyncMock() |
| 78 | + mock_use_case.execute.return_value = use_case_result |
| 79 | + app.dependency_overrides[create_user_use_case_factory] = lambda: mock_use_case |
| 80 | + |
| 81 | + response = await client.post( |
| 82 | + url=URL, |
| 83 | + headers={"Authorization": f"Bearer {self.token.token}"}, |
| 84 | + json=_valid_body(role_id=1), |
| 85 | + ) |
| 86 | + |
| 87 | + assert response.status_code == expected_status |
| 88 | + assert response.json().get("detail") == expected_detail |
| 89 | + |
| 90 | + @pytest.mark.parametrize( |
| 91 | + "headers,expected_status,expected_detail", |
| 92 | + [ |
| 93 | + ({}, 401, "Not authenticated"), |
| 94 | + ({"Authorization": "Bearer invalid-token"}, 401, "Invalid API key."), |
| 95 | + ], |
| 96 | + ) |
| 97 | + async def test_auth(self, client: AsyncClient, headers, expected_status, expected_detail): |
| 98 | + response = await client.post(url=URL, headers=headers, json=_valid_body(role_id=1)) |
| 99 | + |
| 100 | + assert response.status_code == expected_status |
| 101 | + assert response.json().get("detail") == expected_detail |
0 commit comments