166 lines
4.7 KiB
Python
166 lines
4.7 KiB
Python
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
from unittest.mock import MagicMock, AsyncMock, patch
|
|
import sys
|
|
import os
|
|
|
|
# 프로젝트 루트를 Python 경로에 추가
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
# 테스트 환경에서는 테스트 설정을 사용하도록 패치
|
|
os.environ["BE_ENV"] = "local" # 또는 테스트 설정 사용
|
|
os.environ["ENVIRONMENT"] = "local"
|
|
|
|
# .env.test 파일 로드
|
|
from dotenv import load_dotenv
|
|
test_env_path = os.path.join(os.path.dirname(__file__), '..', '.env.test')
|
|
if os.path.exists(test_env_path):
|
|
load_dotenv(test_env_path)
|
|
|
|
from app.main import app
|
|
from app.core.database import Base, get_db
|
|
from app.dependencies import get_auth_service, get_user_service, get_video_service, get_google_service
|
|
from app.core.redis.redis_manager import RedisManager
|
|
|
|
# 테스트용 데이터베이스 설정
|
|
# 환경변수에서 테스트 DB URL 가져오기 (기본값: PostgreSQL)
|
|
TEST_DATABASE_URL = os.getenv(
|
|
"TEST_DATABASE_URL",
|
|
"postgresql://postgres:postgres@localhost:5432/test_o2sound"
|
|
)
|
|
|
|
# SQLite를 사용하려면 USE_SQLITE_TEST=1 환경변수 설정
|
|
if os.getenv("USE_SQLITE_TEST") == "1":
|
|
SQLALCHEMY_DATABASE_URL = "sqlite:///:memory:"
|
|
engine = create_engine(
|
|
SQLALCHEMY_DATABASE_URL,
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
else:
|
|
# PostgreSQL 사용 (기본값)
|
|
SQLALCHEMY_DATABASE_URL = TEST_DATABASE_URL
|
|
engine = create_engine(SQLALCHEMY_DATABASE_URL, pool_pre_ping=True)
|
|
|
|
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
|
|
# 테스트용 데이터베이스 의존성 오버라이드
|
|
def override_get_db():
|
|
try:
|
|
db = TestingSessionLocal()
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
# Mock 서비스 생성
|
|
def create_mock_auth_service():
|
|
mock = MagicMock()
|
|
mock.join = MagicMock()
|
|
mock.login = MagicMock()
|
|
return mock
|
|
|
|
def create_mock_user_service():
|
|
mock = MagicMock()
|
|
mock.get_items = MagicMock()
|
|
mock.update_item = MagicMock()
|
|
mock.delete_item = MagicMock()
|
|
mock.get_user_profile = MagicMock()
|
|
mock.update_user_profile = MagicMock()
|
|
return mock
|
|
|
|
def create_mock_video_service():
|
|
mock = MagicMock()
|
|
mock.get_all_by_user_id = MagicMock()
|
|
mock.delete_by_video_id = MagicMock()
|
|
return mock
|
|
|
|
def create_mock_google_service():
|
|
mock = MagicMock()
|
|
mock.get_login_url = AsyncMock()
|
|
mock.handle_callback = AsyncMock()
|
|
mock.get_token_by_temp_id = AsyncMock()
|
|
return mock
|
|
|
|
def create_mock_redis_manager():
|
|
mock = MagicMock()
|
|
mock.get = AsyncMock(return_value=None)
|
|
mock.set = AsyncMock(return_value=True)
|
|
mock.delete = AsyncMock(return_value=True)
|
|
mock.exists = AsyncMock(return_value=False)
|
|
return mock
|
|
|
|
# 공통 Fixtures
|
|
@pytest.fixture(scope="session")
|
|
def database():
|
|
"""세션 범위의 데이터베이스 설정"""
|
|
# 테스트 시작 시 모든 테이블 생성
|
|
Base.metadata.create_all(bind=engine)
|
|
yield
|
|
# 테스트 종료 시 모든 테이블 삭제
|
|
Base.metadata.drop_all(bind=engine)
|
|
|
|
@pytest.fixture
|
|
def db_session(database):
|
|
"""각 테스트마다 새로운 트랜잭션 시작"""
|
|
connection = engine.connect()
|
|
transaction = connection.begin()
|
|
session = TestingSessionLocal(bind=connection)
|
|
|
|
yield session
|
|
|
|
session.close()
|
|
transaction.rollback()
|
|
connection.close()
|
|
|
|
@pytest.fixture
|
|
def client(database):
|
|
"""데이터베이스를 사용하는 테스트 클라이언트"""
|
|
# 의존성 오버라이드
|
|
app.dependency_overrides[get_db] = override_get_db
|
|
|
|
with TestClient(app) as test_client:
|
|
yield test_client
|
|
|
|
app.dependency_overrides.clear()
|
|
|
|
@pytest.fixture
|
|
def mock_auth_service():
|
|
return create_mock_auth_service()
|
|
|
|
@pytest.fixture
|
|
def mock_user_service():
|
|
return create_mock_user_service()
|
|
|
|
@pytest.fixture
|
|
def mock_video_service():
|
|
return create_mock_video_service()
|
|
|
|
@pytest.fixture
|
|
def mock_google_service():
|
|
return create_mock_google_service()
|
|
|
|
@pytest.fixture
|
|
def mock_redis_manager():
|
|
return create_mock_redis_manager()
|
|
|
|
@pytest.fixture
|
|
def authenticated_client(client, mock_auth_service):
|
|
"""인증된 테스트 클라이언트"""
|
|
app.dependency_overrides[get_auth_service] = lambda: mock_auth_service
|
|
|
|
# 로그인 성공 시뮬레이션
|
|
mock_auth_service.login.return_value = {
|
|
"access_token": "test_access_token",
|
|
"token_type": "bearer",
|
|
"user_id": 1
|
|
}
|
|
|
|
# 헤더에 토큰 추가
|
|
client.headers = {"Authorization": "Bearer test_access_token"}
|
|
yield client
|
|
|
|
app.dependency_overrides.clear()
|