Spaces:
Sleeping
Sleeping
File size: 2,390 Bytes
050d8f8 1bd7131 050d8f8 1bd7131 050d8f8 1bd7131 050d8f8 1bd7131 050d8f8 1bd7131 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 |
import pytest
import os
import sys
from unittest.mock import patch, MagicMock
from fastapi.testclient import TestClient
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
# Set test environment variables BEFORE importing app
os.environ["JWT_SECRET"] = "test-secret-key-that-is-long-enough-for-security-purposes"
os.environ["GOOGLE_CLIENT_ID"] = "test-google-client-id.apps.googleusercontent.com"
os.environ["RESET_DB"] = "true" # Prevent Drive download during tests
# Add parent directory to path to allow importing app
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
# Mock the drive service before importing app
with patch("services.drive_service.DriveService") as mock_drive:
mock_instance = MagicMock()
mock_instance.download_db.return_value = False
mock_instance.upload_db.return_value = True
mock_drive.return_value = mock_instance
from app import app
from core.database import get_db, Base
# Use a file-based SQLite database for testing to ensure persistence
TEST_DATABASE_URL = "sqlite+aiosqlite:///./test_blink_data.db"
@pytest.fixture(scope="session")
def test_engine():
engine = create_async_engine(TEST_DATABASE_URL, echo=False)
yield engine
# Cleanup after session
if os.path.exists("./test_blink_data.db"):
os.remove("./test_blink_data.db")
@pytest.fixture(scope="function")
async def db_session(test_engine):
async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async_session = async_sessionmaker(
test_engine,
class_=AsyncSession,
expire_on_commit=False
)
async with async_session() as session:
yield session
@pytest.fixture(scope="function")
def client(test_engine):
async def override_get_db():
async_session = async_sessionmaker(
test_engine,
class_=AsyncSession,
expire_on_commit=False
)
async with async_session() as session:
yield session
app.dependency_overrides[get_db] = override_get_db
# Mock drive service for the test client
with patch("routers.auth.drive_service") as mock_auth_drive:
mock_auth_drive.upload_db.return_value = True
with TestClient(app) as c:
yield c
app.dependency_overrides.clear()
|