| | 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) |