Spaces:
Configuration error
Configuration error
| import os | |
| import sys | |
| sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) | |
| import pytest | |
| from fastapi.testclient import TestClient | |
| from sqlalchemy.ext.asyncio import AsyncSession | |
| from unittest.mock import AsyncMock, MagicMock, patch | |
| from app.main import app | |
| from app.auth.security import create_access_token | |
| from app.database.postgres import User, Email | |
| client = TestClient(app) | |
| def mock_db(): | |
| """Fixture to mock AsyncSession databases.""" | |
| db = MagicMock(spec=AsyncSession) | |
| db.execute = AsyncMock() | |
| db.commit = AsyncMock() | |
| db.refresh = AsyncMock() | |
| db.add = MagicMock() | |
| return db | |
| def test_health_check(): | |
| """Test health check / metrics scraping endpoints.""" | |
| response = client.get("/metrics") | |
| assert response.status_code == 200 | |
| assert "http_requests_total" in response.text | |
| def test_user_registration_success(mock_db): | |
| """Test registering a new user locally.""" | |
| # Mock user query to return None (user doesn't exist) | |
| mock_execute_result = MagicMock() | |
| mock_execute_result.scalars.return_value.first.return_value = None | |
| mock_db.execute.return_value = mock_execute_result | |
| with patch("app.api.auth.get_db", return_value=mock_db): | |
| payload = { | |
| "email": "test-user@example.com", | |
| "name": "Test User", | |
| "password": "testpassword123" | |
| } | |
| from app.database.postgres import get_db | |
| # Override FastAPI Dependency | |
| app.dependency_overrides[get_db] = lambda: mock_db | |
| response = client.post("/api/auth/register", json=payload) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert "access_token" in data | |
| assert data["email"] == "test-user@example.com" | |
| def test_invalid_jwt_verification(): | |
| """Test that protected endpoints reject invalid JWT tokens.""" | |
| response = client.get("/api/emails", headers={"Authorization": "Bearer invalid_jwt_token"}) | |
| assert response.status_code == 401 | |
| assert response.json()["detail"] == "Could not validate credentials" | |
| def test_google_login_redirect_url(): | |
| """Test that Google Login builds and returns redirection URLs.""" | |
| response = client.get("/api/auth/google-login") | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert "url" in data | |
| assert "callback" in data["url"] | |
| async def test_generate_mock_emails(mock_db): | |
| """Test generating mock emails for a user.""" | |
| import uuid | |
| mock_execute_result = MagicMock() | |
| mock_execute_result.scalars.return_value.first.return_value = None | |
| mock_db.execute.return_value = mock_execute_result | |
| user_id = uuid.uuid4() | |
| with patch("app.tasks.email_tasks.mongo_db") as mock_mongo, \ | |
| patch("app.tasks.email_tasks.analyze_email_task") as mock_analyze_task: | |
| mock_mongo.email_bodies.insert_one = AsyncMock() | |
| mock_analyze_task.delay = MagicMock() | |
| from app.tasks.email_tasks import generate_mock_emails | |
| count = await generate_mock_emails(mock_db, user_id) | |
| assert count == 4 | |
| assert mock_db.add.call_count == 4 | |
| assert mock_db.commit.call_count == 4 | |
| assert mock_mongo.email_bodies.insert_one.call_count == 4 | |
| assert mock_analyze_task.delay.call_count == 4 | |