| import os |
| import pytest |
| import tempfile |
| from pathlib import Path |
| from sqlalchemy import create_engine |
| from sqlalchemy.orm import sessionmaker |
| from fastapi.testclient import TestClient |
|
|
| from app.db.base import Base |
| from app.db.models import Video, ProcessingResult |
| from app import app as fastapi_app |
|
|
| |
| @pytest.fixture(scope="session") |
| def test_db_engine(): |
| """Create a test database engine.""" |
| |
| db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False) |
| db_file.close() |
| |
| |
| db_url = f"sqlite:///{db_file.name}" |
| |
| |
| engine = create_engine(db_url) |
| |
| |
| Base.metadata.create_all(engine) |
| |
| yield engine |
| |
| |
| os.unlink(db_file.name) |
|
|
| @pytest.fixture(scope="function") |
| def db_session(test_db_engine): |
| """Create a test database session.""" |
| |
| TestSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=test_db_engine) |
| |
| |
| session = TestSessionLocal() |
| |
| try: |
| yield session |
| finally: |
| session.rollback() |
| session.close() |
|
|
| |
| @pytest.fixture(scope="module") |
| def client(): |
| """Create a test client for the FastAPI application.""" |
| with TestClient(fastapi_app) as c: |
| yield c |
|
|
| |
| @pytest.fixture(scope="function") |
| def test_video(db_session): |
| """Create a test video in the database.""" |
| video = Video( |
| id="test-video-id", |
| original_filename="test_video.mp4", |
| upload_date="2023-01-01T00:00:00", |
| file_path="test_video.mp4", |
| size=1000, |
| duration=10.0, |
| status="uploaded" |
| ) |
| db_session.add(video) |
| db_session.commit() |
| db_session.refresh(video) |
| |
| yield video |
| |
| |
| db_session.delete(video) |
| db_session.commit() |
|
|
| @pytest.fixture(scope="function") |
| def test_processing_result(db_session, test_video): |
| """Create a test processing result in the database.""" |
| result = ProcessingResult( |
| id="test-result-id", |
| video_id=test_video.id, |
| processing_date="2023-01-01T00:00:00", |
| transcript="Test transcript", |
| emotion_analysis={"test": "data"}, |
| overall_summary="Test summary", |
| transcript_analysis={"test": "data"}, |
| recommendations={"test": "data"} |
| ) |
| db_session.add(result) |
| db_session.commit() |
| db_session.refresh(result) |
| |
| yield result |
| |
| |
| db_session.delete(result) |
| db_session.commit() |
|
|
| |
| @pytest.fixture(scope="session") |
| def test_video_file(): |
| """Create a test video file.""" |
| |
| test_dir = Path(tempfile.mkdtemp()) |
| |
| |
| video_file = test_dir / "test_video.mp4" |
| with open(video_file, "w") as f: |
| f.write("This is a dummy video file for testing.") |
| |
| yield str(video_file) |
| |
| |
| os.unlink(video_file) |
| os.rmdir(test_dir) |
