tests/test_suite.py

#!/usr/bin/env python3
"""
Lily LLM API 종합 테스트 스위트
모든 주요 기능을 테스트하는 통합 테스트
"""

import pytest
import requests
import time
import json
import os
import tempfile
from pathlib import Path

# 테스트 설정
BASE_URL = "http://localhost:8001"
TEST_USER_ID = "test_user_123"
TEST_SESSION_ID = "test_session_123"

class TestLilyLLMAPI:
    """Lily LLM API 종합 테스트 클래스"""
    
    def setup_method(self):
        """각 테스트 전 실행"""
        self.session = requests.Session()
        self.token = None
        
    def teardown_method(self):
        """각 테스트 후 실행"""
        self.session.close()
    
    def test_01_server_health(self):
        """서버 상태 확인"""
        response = self.session.get(f"{BASE_URL}/health")
        assert response.status_code == 200
        data = response.json()
        assert "status" in data
        assert "current_model" in data
        print("✅ 서버 상태 확인 성공")
    
    def test_02_models_endpoint(self):
        """모델 목록 조회"""
        response = self.session.get(f"{BASE_URL}/models")
        assert response.status_code == 200
        data = response.json()
        assert "available_models" in data
        assert len(data["available_models"]) > 0
        print("✅ 모델 목록 조회 성공")
    
    def test_03_text_generation(self):
        """텍스트 생성 테스트"""
        data = {
            "prompt": "안녕하세요! 간단한 인사말을 해주세요.",
            "model_id": "polyglot-ko-1.3b-chat",
            "max_length": 100,
            "temperature": 0.7,
            "top_p": 0.9,
            "do_sample": True
        }
        
        response = self.session.post(f"{BASE_URL}/generate", data=data)
        assert response.status_code == 200
        result = response.json()
        assert "generated_text" in result
        assert len(result["generated_text"]) > 0
        print("✅ 텍스트 생성 성공")
    
    def test_04_user_management(self):
        """사용자 관리 테스트"""
        # 사용자 생성
        user_data = {
            "user_id": TEST_USER_ID,
            "username": "테스트사용자",
            "email": "test@example.com"
        }
        
        response = self.session.post(f"{BASE_URL}/user/create", data=user_data)
        assert response.status_code == 200
        print("✅ 사용자 생성 성공")
        
        # 사용자 정보 조회
        response = self.session.get(f"{BASE_URL}/user/{TEST_USER_ID}")
        assert response.status_code == 200
        user_info = response.json()
        assert user_info["user_id"] == TEST_USER_ID
        print("✅ 사용자 정보 조회 성공")
    
    def test_05_authentication(self):
        """인증 시스템 테스트"""
        # 사용자 등록
        register_data = {
            "username": "testuser",
            "email": "test@example.com",
            "password": "testpassword123"
        }
        
        response = self.session.post(f"{BASE_URL}/auth/register", data=register_data)
        assert response.status_code == 200
        register_result = response.json()
        assert "access_token" in register_result
        self.token = register_result["access_token"]
        print("✅ 사용자 등록 성공")
        
        # 로그인
        login_data = {
            "username": "testuser",
            "password": "testpassword123"
        }
        
        response = self.session.post(f"{BASE_URL}/auth/login", data=login_data)
        assert response.status_code == 200
        login_result = response.json()
        assert "access_token" in login_result
        print("✅ 로그인 성공")
        
        # 현재 사용자 정보 조회
        headers = {"Authorization": f"Bearer {self.token}"}
        response = self.session.get(f"{BASE_URL}/auth/me", headers=headers)
        assert response.status_code == 200
        print("✅ 현재 사용자 정보 조회 성공")
    
    def test_06_chat_session_management(self):
        """채팅 세션 관리 테스트"""
        # 세션 생성
        session_data = {
            "user_id": TEST_USER_ID,
            "session_name": "테스트 세션"
        }
        
        response = self.session.post(f"{BASE_URL}/session/create", data=session_data)
        assert response.status_code == 200
        session_result = response.json()
        session_id = session_result["session_id"]
        print("✅ 세션 생성 성공")
        
        # 메시지 추가
        message_data = {
            "session_id": session_id,
            "user_id": TEST_USER_ID,
            "message_type": "text",
            "content": "안녕하세요! 테스트 메시지입니다."
        }
        
        response = self.session.post(f"{BASE_URL}/chat/message", data=message_data)
        assert response.status_code == 200
        print("✅ 메시지 추가 성공")
        
        # 채팅 기록 조회
        response = self.session.get(f"{BASE_URL}/chat/history/{session_id}")
        assert response.status_code == 200
        history = response.json()
        assert len(history) > 0
        print("✅ 채팅 기록 조회 성공")
    
    def test_07_document_upload(self):
        """문서 업로드 테스트"""
        # 임시 PDF 파일 생성
        with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp_file:
            tmp_file.write(b"%PDF-1.4\n1 0 obj\n<<\n/Type /Catalog\n/Pages 2 0 R\n>>\nendobj\n2 0 obj\n<<\n/Type /Pages\n/Kids [3 0 R]\n/Count 1\n>>\nendobj\n3 0 obj\n<<\n/Type /Page\n/Parent 2 0 R\n/MediaBox [0 0 612 792]\n/Contents 4 0 R\n>>\nendobj\n4 0 obj\n<<\n/Length 44\n>>\nstream\nBT\n/F1 12 Tf\n72 720 Td\n(Test PDF) Tj\nET\nendstream\nendobj\nxref\n0 5\n0000000000 65535 f \n0000000009 00000 n \n0000000058 00000 n \n0000000115 00000 n \n0000000204 00000 n \ntrailer\n<<\n/Size 5\n/Root 1 0 R\n>>\nstartxref\n297\n%%EOF\n")
            tmp_file_path = tmp_file.name
        
        try:
            with open(tmp_file_path, 'rb') as f:
                files = {'file': f}
                data = {'user_id': TEST_USER_ID}
                response = self.session.post(f"{BASE_URL}/document/upload", files=files, data=data)
            
            assert response.status_code == 200
            upload_result = response.json()
            assert "document_id" in upload_result
            print("✅ 문서 업로드 성공")
            
            # 사용자 문서 목록 조회
            response = self.session.get(f"{BASE_URL}/documents/db/{TEST_USER_ID}")
            assert response.status_code == 200
            documents = response.json()
            assert len(documents) > 0
            print("✅ 문서 목록 조회 성공")
            
        finally:
            os.unlink(tmp_file_path)
    
    def test_08_rag_functionality(self):
        """RAG 기능 테스트"""
        # RAG 쿼리
        rag_data = {
            "query": "테스트 쿼리입니다.",
            "user_id": TEST_USER_ID,
            "max_length": 200,
            "temperature": 0.7
        }
        
        response = self.session.post(f"{BASE_URL}/rag/generate", data=rag_data)
        assert response.status_code == 200
        rag_result = response.json()
        assert "response" in rag_result
        print("✅ RAG 쿼리 성공")
    
    def test_09_background_tasks(self):
        """백그라운드 작업 테스트"""
        # 문서 처리 작업 시작
        task_data = {
            "file_path": "/tmp/test_document.pdf",
            "user_id": TEST_USER_ID
        }
        
        response = self.session.post(f"{BASE_URL}/tasks/document/process", data=task_data)
        assert response.status_code == 200
        task_result = response.json()
        assert "task_id" in task_result
        task_id = task_result["task_id"]
        print("✅ 백그라운드 작업 시작 성공")
        
        # 작업 상태 확인
        response = self.session.get(f"{BASE_URL}/tasks/{task_id}")
        assert response.status_code == 200
        status_result = response.json()
        assert "status" in status_result
        print("✅ 작업 상태 확인 성공")
    
    def test_10_monitoring(self):
        """모니터링 기능 테스트"""
        # 성능 모니터링 시작
        response = self.session.post(f"{BASE_URL}/monitoring/start")
        assert response.status_code == 200
        print("✅ 성능 모니터링 시작 성공")
        
        # 모니터링 상태 조회
        response = self.session.get(f"{BASE_URL}/monitoring/status")
        assert response.status_code == 200
        status_data = response.json()
        assert "current_metrics" in status_data
        print("✅ 모니터링 상태 조회 성공")
        
        # 시스템 건강 상태 조회
        response = self.session.get(f"{BASE_URL}/monitoring/health")
        assert response.status_code == 200
        health_data = response.json()
        assert "status" in health_data
        print("✅ 시스템 건강 상태 조회 성공")
        
        # 성능 모니터링 중지
        response = self.session.post(f"{BASE_URL}/monitoring/stop")
        assert response.status_code == 200
        print("✅ 성능 모니터링 중지 성공")
    
    def test_11_websocket_connection(self):
        """WebSocket 연결 테스트"""
        try:
            import websockets
            import asyncio
            
            async def test_websocket():
                uri = f"ws://localhost:8001/ws/{TEST_USER_ID}"
                async with websockets.connect(uri) as websocket:
                    # 연결 테스트
                    await websocket.send(json.dumps({
                        "type": "ping",
                        "message": "테스트 메시지"
                    }))
                    
                    # 응답 대기
                    response = await asyncio.wait_for(websocket.recv(), timeout=5.0)
                    data = json.loads(response)
                    assert "type" in data
                    print("✅ WebSocket 연결 성공")
            
            asyncio.run(test_websocket())
            
        except ImportError:
            print("⚠️ websockets 라이브러리가 설치되지 않아 WebSocket 테스트를 건너뜁니다.")
        except Exception as e:
            print(f"⚠️ WebSocket 테스트 실패: {e}")
    
    def test_12_error_handling(self):
        """오류 처리 테스트"""
        # 잘못된 모델 ID로 요청
        data = {
            "prompt": "테스트",
            "model_id": "invalid-model-id",
            "max_length": 100
        }
        
        response = self.session.post(f"{BASE_URL}/generate", data=data)
        assert response.status_code in [400, 404, 422]
        print("✅ 잘못된 모델 ID 오류 처리 성공")
        
        # 잘못된 사용자 ID로 요청
        response = self.session.get(f"{BASE_URL}/user/invalid-user-id")
        assert response.status_code in [404, 422]
        print("✅ 잘못된 사용자 ID 오류 처리 성공")
    
    def test_13_performance_metrics(self):
        """성능 메트릭 테스트"""
        # 여러 요청을 보내서 성능 측정
        start_time = time.time()
        
        for i in range(5):
            data = {
                "prompt": f"테스트 요청 {i+1}",
                "model_id": "polyglot-ko-1.3b-chat",
                "max_length": 50,
                "temperature": 0.7
            }
            
            response = self.session.post(f"{BASE_URL}/generate", data=data)
            assert response.status_code == 200
        
        end_time = time.time()
        total_time = end_time - start_time
        
        print(f"✅ 성능 테스트 완료: 5개 요청을 {total_time:.2f}초에 처리")
        assert total_time < 60  # 60초 이내에 완료되어야 함
    
    def test_14_data_persistence(self):
        """데이터 지속성 테스트"""
        # 사용자 생성
        user_data = {
            "user_id": f"persist_test_user_{int(time.time())}",
            "username": "지속성테스트사용자",
            "email": "persist@test.com"
        }
        
        response = self.session.post(f"{BASE_URL}/user/create", data=user_data)
        assert response.status_code == 200
        user_id = user_data["user_id"]
        
        # 서버 재시작 시뮬레이션 (실제로는 재시작하지 않음)
        # 대신 동일한 사용자 정보를 다시 조회
        response = self.session.get(f"{BASE_URL}/user/{user_id}")
        assert response.status_code == 200
        user_info = response.json()
        assert user_info["user_id"] == user_id
        print("✅ 데이터 지속성 확인 성공")

def run_comprehensive_test():
    """종합 테스트 실행"""
    print("🚀 Lily LLM API 종합 테스트 시작")
    print("=" * 60)
    
    test_instance = TestLilyLLMAPI()
    
    # 테스트 메서드 목록
    test_methods = [
        test_instance.test_01_server_health,
        test_instance.test_02_models_endpoint,
        test_instance.test_03_text_generation,
        test_instance.test_04_user_management,
        test_instance.test_05_authentication,
        test_instance.test_06_chat_session_management,
        test_instance.test_07_document_upload,
        test_instance.test_08_rag_functionality,
        test_instance.test_09_background_tasks,
        test_instance.test_10_monitoring,
        test_instance.test_11_websocket_connection,
        test_instance.test_12_error_handling,
        test_instance.test_13_performance_metrics,
        test_instance.test_14_data_persistence
    ]
    
    passed_tests = 0
    failed_tests = 0
    
    for i, test_method in enumerate(test_methods, 1):
        try:
            print(f"\n📋 테스트 {i}/{len(test_methods)}: {test_method.__name__}")
            print("-" * 50)
            
            test_method()
            passed_tests += 1
            
        except Exception as e:
            print(f"❌ 테스트 실패: {e}")
            failed_tests += 1
    
    # 결과 요약
    print("\n" + "=" * 60)
    print("📊 테스트 결과 요약")
    print("=" * 60)
    print(f"✅ 통과: {passed_tests}")
    print(f"❌ 실패: {failed_tests}")
    print(f"📈 성공률: {passed_tests/(passed_tests+failed_tests)*100:.1f}%")
    
    if failed_tests == 0:
        print("\n🎉 모든 테스트가 성공했습니다!")
        return True
    else:
        print(f"\n⚠️ {failed_tests}개의 테스트가 실패했습니다.")
        return False

if __name__ == "__main__":
    success = run_comprehensive_test()
    exit(0 if success else 1)