Spaces:
Runtime error
Runtime error
| import pytest | |
| from unittest.mock import AsyncMock, patch, MagicMock | |
| from bson import ObjectId | |
| from fastapi.testclient import TestClient | |
| # Mock data | |
| mock_user_id = str(ObjectId()) | |
| mock_org_id = str(ObjectId()) | |
| mock_db_user = { | |
| "_id": ObjectId(mock_user_id), | |
| "username": "testuser", | |
| "email": "test@test.com", | |
| "password": "hashed_password_mock", | |
| "first_name": "Test", | |
| "last_name": "User", | |
| "organization_id": mock_org_id, | |
| "role": "admin" | |
| } | |
| def mock_auth_collection(): | |
| collection_mock = AsyncMock() | |
| collection_mock.insert_one = AsyncMock() | |
| collection_mock.update_one = AsyncMock() | |
| collection_mock.delete_one = AsyncMock() | |
| collection_mock.find_one = AsyncMock() | |
| cursor_mock = AsyncMock() | |
| async def async_generator(): | |
| yield mock_db_user | |
| cursor_mock.__aiter__ = lambda self: async_generator() | |
| collection_mock.find = MagicMock(return_value=cursor_mock) | |
| with patch("app.services.auth.get_async_collection", return_value=collection_mock): | |
| yield collection_mock | |
| def test_register_user(test_client, mock_auth_collection): | |
| # Setup mocks | |
| mock_insert_result = MagicMock() | |
| mock_insert_result.inserted_id = ObjectId(mock_user_id) | |
| mock_auth_collection.insert_one.return_value = mock_insert_result | |
| # Needs to mock password generation to avoid real computing cost if we want | |
| with patch("app.api.v1.auth.generate_passwd_hash", return_value="hashed"): | |
| payload = { | |
| "username": "newuser", | |
| "email": "new@test.com", | |
| "password": "password123", | |
| "organization_id": str(ObjectId()), | |
| "department_id": str(ObjectId()), | |
| "admin_type": "org_admin" | |
| } | |
| response = test_client.post("/v1/register", json=payload) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert data["status"] is True | |
| assert "access_token" in data | |
| assert data["data"]["username"] == "newuser" | |
| mock_auth_collection.insert_one.assert_called_once() | |
| def test_login_user(test_client, mock_auth_collection): | |
| mock_auth_collection.find_one.return_value = mock_db_user | |
| with patch("app.api.v1.auth.verify_password", return_value=True): | |
| payload = { | |
| "username": "testuser", | |
| "password": "password123" | |
| } | |
| response = test_client.post("/v1/login", json=payload) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert data["status"] is True | |
| assert "access_token" in data | |
| mock_auth_collection.find_one.assert_called_once() | |
| def test_login_user_invalid(test_client, mock_auth_collection): | |
| mock_auth_collection.find_one.return_value = mock_db_user | |
| with patch("app.api.v1.auth.verify_password", return_value=False): | |
| payload = { | |
| "username": "testuser", | |
| "password": "wrongpassword" | |
| } | |
| response = test_client.post("/v1/login", json=payload) | |
| assert response.status_code == 401 | |
| assert response.json()["status"] is False | |
| def test_get_user(test_client, mock_auth_collection): | |
| mock_auth_collection.find_one.return_value = mock_db_user | |
| response = test_client.get(f"/v1/{mock_user_id}") | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert data["status"] is True | |
| assert data["data"]["username"] == "testuser" | |
| def test_update_user(test_client, mock_auth_collection): | |
| mock_update_result = MagicMock() | |
| mock_update_result.modified_count = 1 | |
| mock_auth_collection.update_one.return_value = mock_update_result | |
| payload = {"first_name": "Updated"} | |
| response = test_client.put(f"/v1/{mock_user_id}", json=payload) | |
| assert response.status_code == 200 | |
| assert response.json()["status"] is True | |
| mock_auth_collection.update_one.assert_called_once() | |
| def test_delete_user(test_client, mock_auth_collection): | |
| mock_delete_result = MagicMock() | |
| mock_delete_result.deleted_count = 1 | |
| mock_auth_collection.delete_one.return_value = mock_delete_result | |
| response = test_client.delete(f"/v1/{mock_user_id}") | |
| assert response.status_code == 200 | |
| assert response.json()["status"] is True | |
| mock_auth_collection.delete_one.assert_called_once() | |