Spaces:
Sleeping
Sleeping
File size: 2,337 Bytes
266d7bc |
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 73 74 75 76 77 78 79 |
import shutil
import sys
from collections.abc import Generator
from pathlib import Path
import pytest
from sqlalchemy import create_engine
from sqlalchemy.engine import Engine
from sqlalchemy.orm import Session as SQLAlchemySession
from sqlalchemy.orm import sessionmaker
# Add src to Python path for direct imports
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from config import settings
from utils.logger_util import setup_logging
logger = setup_logging()
db = settings.supabase_db
DATABASE_URL = (
f"postgresql://{db.user}:{db.password.get_secret_value()}@{db.host}:{db.port}/{db.name}"
)
@pytest.fixture(scope="session")
def db_engine() -> Generator[Engine, None, None]:
"""Create a SQLAlchemy engine for the test database session.
Disposes the engine after the test session completes.
Args:
None
Yields:
Engine: A SQLAlchemy engine connected to the test database.
"""
logger.info("Creating test database engine")
engine = create_engine(DATABASE_URL)
yield engine
logger.info("Disposing test database engine")
engine.dispose()
@pytest.fixture(scope="function")
def db_session(db_engine: Engine) -> Generator[SQLAlchemySession, None, None]:
"""Provide a SQLAlchemy session for a single test function.
Closes the session after the test finishes.
Args:
db_engine (Engine): The SQLAlchemy engine to bind the session to.
Yields:
SQLAlchemySession: A SQLAlchemy session connected to the test database.
"""
logger.info("Creating test database session")
Session = sessionmaker(bind=db_engine)
session = Session()
yield session
session.close()
logger.info("Closed test database session")
@pytest.fixture(scope="function", autouse=True)
def clear_prefect_cache() -> Generator[None, None, None]:
"""Automatically clear Prefect cache before and after each test function
to prevent interference between tests.
Args:
None
Yields:
None
"""
prefect_dir = Path(".prefect")
logger.debug("Clearing Prefect cache before test")
if prefect_dir.exists():
shutil.rmtree(prefect_dir, ignore_errors=True)
yield
if prefect_dir.exists():
shutil.rmtree(prefect_dir, ignore_errors=True)
logger.debug("Cleared Prefect cache after test")
|