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 # Test database @pytest.fixture(scope="session") def test_db_engine(): """Create a test database engine.""" # Create a temporary file for the test database db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False) db_file.close() # Create the database URL db_url = f"sqlite:///{db_file.name}" # Create the engine engine = create_engine(db_url) # Create tables Base.metadata.create_all(engine) yield engine # Clean up os.unlink(db_file.name) @pytest.fixture(scope="function") def db_session(test_db_engine): """Create a test database session.""" # Create a session factory TestSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=test_db_engine) # Create a session session = TestSessionLocal() try: yield session finally: session.rollback() session.close() # Test client @pytest.fixture(scope="module") def client(): """Create a test client for the FastAPI application.""" with TestClient(fastapi_app) as c: yield c # Test data @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 # Clean up 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 # Clean up db_session.delete(result) db_session.commit() # Test files @pytest.fixture(scope="session") def test_video_file(): """Create a test video file.""" # Create a temporary directory for test files test_dir = Path(tempfile.mkdtemp()) # Create a dummy video file (just a text file with .mp4 extension) 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) # Clean up os.unlink(video_file) os.rmdir(test_dir)