123 lines
4.2 KiB
Python
123 lines
4.2 KiB
Python
# 테스트 유틸리티 함수들
|
|
from typing import Dict, Any
|
|
from datetime import datetime, timedelta
|
|
from jose import jwt # python-jose 패키지
|
|
from passlib.context import CryptContext
|
|
|
|
# 비밀번호 해싱 컨텍스트
|
|
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
|
|
|
class TestUtils:
|
|
"""테스트에서 사용할 유틸리티 함수 모음"""
|
|
|
|
@staticmethod
|
|
def create_test_user(user_id: int = 1, email: str = "test@example.com") -> Dict[str, Any]:
|
|
"""테스트용 사용자 데이터 생성"""
|
|
return {
|
|
"id": user_id,
|
|
"email": email,
|
|
"name": f"Test User {user_id}",
|
|
"hashed_password": pwd_context.hash("Test1234!"),
|
|
"is_active": True,
|
|
"created_at": datetime.utcnow(),
|
|
"updated_at": datetime.utcnow()
|
|
}
|
|
|
|
@staticmethod
|
|
def create_test_token(user_id: int = 1, expires_delta: timedelta = None) -> str:
|
|
"""테스트용 JWT 토큰 생성"""
|
|
expire = datetime.utcnow() + (expires_delta or timedelta(hours=1))
|
|
payload = {
|
|
"sub": str(user_id),
|
|
"exp": expire,
|
|
"iat": datetime.utcnow()
|
|
}
|
|
return jwt.encode(payload, "test_secret_key", algorithm="HS256")
|
|
|
|
@staticmethod
|
|
def create_test_item(item_id: int = 1, user_id: int = 1) -> Dict[str, Any]:
|
|
"""테스트용 아이템 데이터 생성"""
|
|
return {
|
|
"id": item_id,
|
|
"user_id": user_id,
|
|
"name": f"Test Item {item_id}",
|
|
"address": f"Test Address {item_id}",
|
|
"url": f"https://example.com/item{item_id}",
|
|
"created_at": datetime.utcnow(),
|
|
"updated_at": datetime.utcnow()
|
|
}
|
|
|
|
@staticmethod
|
|
def create_test_video(video_id: int = 1, user_id: int = 1) -> Dict[str, Any]:
|
|
"""테스트용 비디오 데이터 생성"""
|
|
return {
|
|
"id": video_id,
|
|
"user_id": user_id,
|
|
"title": f"Test Video {video_id}",
|
|
"description": f"Description for video {video_id}",
|
|
"url": f"https://example.com/video{video_id}.mp4",
|
|
"is_uploaded": True,
|
|
"download_count": 0,
|
|
"resolution": "1920x1080",
|
|
"status": "완료됨",
|
|
"created_at": datetime.utcnow()
|
|
}
|
|
|
|
@staticmethod
|
|
def create_test_workflow_data() -> Dict[str, Any]:
|
|
"""테스트용 워크플로우 데이터 생성"""
|
|
return {
|
|
"task_id": "test_task_123",
|
|
"url": "https://example.com/test",
|
|
"status": {
|
|
"metadata": False,
|
|
"lyrics": False,
|
|
"songs": False,
|
|
"images": False,
|
|
"movies": False,
|
|
"combined": False
|
|
}
|
|
}
|
|
|
|
@staticmethod
|
|
def create_google_oauth_data() -> Dict[str, Any]:
|
|
"""테스트용 Google OAuth 데이터 생성"""
|
|
return {
|
|
"access_token": "google_access_token_test",
|
|
"refresh_token": "google_refresh_token_test",
|
|
"expires_in": 3600,
|
|
"token_type": "Bearer",
|
|
"user_info": {
|
|
"id": "google_123",
|
|
"email": "test@gmail.com",
|
|
"name": "Google Test User",
|
|
"picture": "https://example.com/picture.jpg",
|
|
"verified_email": True
|
|
}
|
|
}
|
|
|
|
@staticmethod
|
|
def assert_response_success(response, expected_data: Dict[str, Any] = None):
|
|
"""API 응답 성공 검증"""
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "success"
|
|
|
|
if expected_data:
|
|
for key, value in expected_data.items():
|
|
assert data["data"][key] == value
|
|
|
|
return data["data"]
|
|
|
|
@staticmethod
|
|
def assert_response_error(response, expected_status: int = 400, expected_message: str = None):
|
|
"""API 응답 에러 검증"""
|
|
assert response.status_code == expected_status
|
|
data = response.json()
|
|
assert data["status"] == "error"
|
|
|
|
if expected_message:
|
|
assert expected_message in data["message"]
|
|
|
|
return data
|