test_advanced.py•7.44 kB
#!/usr/bin/env python3
"""
Persona MCP Server 고급 기능 테스트
"""
import asyncio
import sys
import os
from dotenv import load_dotenv
# .env 파일 로드
load_dotenv()
# 프로젝트 루트를 경로에 추가
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from src.main import (
create_survey_impl,
CreateSurveyRequest,
next_question_impl,
NextQuestionRequest,
submit_response_impl,
SubmitResponseRequest,
predict_persona_impl,
PredictPersonaRequest,
analyze_preference_impl,
AnalyzePreferenceRequest,
get_trend_insights_impl,
GetTrendInsightsRequest,
search_responses_impl,
SearchResponsesRequest,
generate_sample_personas_impl,
GenerateSamplePersonasRequest,
)
async def test_generate_sample_personas():
"""샘플 페르소나 생성 테스트"""
print("\n" + "="*50)
print("1. 샘플 페르소나 생성 테스트")
print("="*50)
try:
req = GenerateSamplePersonasRequest(count=3, use_llm=False)
result = await generate_sample_personas_impl(req)
print(f"✅ 샘플 페르소나 생성 성공")
print(f" - 생성된 페르소나 수: {result.get('created_count', 0)}")
return True
except Exception as e:
print(f"❌ 샘플 페르소나 생성 실패: {e}")
import traceback
traceback.print_exc()
return False
async def test_survey_flow():
"""설문 플로우 테스트 (생성 -> 질문 생성 -> 응답 제출)"""
print("\n" + "="*50)
print("2. 설문 플로우 테스트")
print("="*50)
try:
# 설문 생성
print(" - 설문 생성 중...")
req = CreateSurveyRequest(category="general")
survey_result = await create_survey_impl(req)
survey_id = survey_result.get('survey_id')
if not survey_id:
print(" ❌ 설문 생성 실패")
return False
print(f" ✅ 설문 생성 성공: {survey_id}")
# 첫 번째 질문에 대한 답변
first_question = survey_result.get('first_question', {})
question_id = first_question.get('question_id', 'q1')
question_text = first_question.get('question', '')
print(f" - 첫 번째 질문: {question_text[:50]}...")
# 응답 제출 (페르소나 ID가 필요하므로 임시로 사용)
print(" - 응답 제출 중...")
submit_req = SubmitResponseRequest(
persona_id="test_persona_001",
survey_id=survey_id,
responses=[{
"question_id": question_id,
"answer": "테스트 응답입니다."
}]
)
submit_result = await submit_response_impl(submit_req)
if 'error' in submit_result:
print(f" ⚠️ 응답 제출 경고: {submit_result.get('error')}")
else:
print(f" ✅ 응답 제출 성공")
return True
except Exception as e:
print(f" ❌ 설문 플로우 실패: {e}")
import traceback
traceback.print_exc()
return False
async def test_predict_persona():
"""페르소나 예측 테스트"""
print("\n" + "="*50)
print("3. 페르소나 예측 테스트")
print("="*50)
try:
req = PredictPersonaRequest(
responses=[
{"question": "좋아하는 취미는?", "answer": "독서와 영화 감상"},
{"question": "선호하는 브랜드는?", "answer": "심플하고 미니멀한 디자인"},
]
)
result = await predict_persona_impl(req)
if 'error' in result:
print(f" ⚠️ 페르소나 예측 경고: {result.get('error')}")
return False
print(f" ✅ 페르소나 예측 성공")
persona = result.get('persona', {})
print(f" - 나이대: {persona.get('age_range', 'N/A')}")
print(f" - 성별: {persona.get('gender', 'N/A')}")
return True
except Exception as e:
print(f" ❌ 페르소나 예측 실패: {e}")
import traceback
traceback.print_exc()
return False
async def test_analyze_preference():
"""선호도 분석 테스트"""
print("\n" + "="*50)
print("4. 선호도 분석 테스트")
print("="*50)
try:
req = AnalyzePreferenceRequest(
query="스티커 디자인",
limit=5
)
result = await analyze_preference_impl(req)
if 'error' in result:
print(f" ⚠️ 선호도 분석 경고: {result.get('error')}")
return False
print(f" ✅ 선호도 분석 성공")
insights = result.get('insights', [])
print(f" - 분석된 인사이트 수: {len(insights)}")
return True
except Exception as e:
print(f" ❌ 선호도 분석 실패: {e}")
import traceback
traceback.print_exc()
return False
async def test_search_responses():
"""응답 검색 테스트"""
print("\n" + "="*50)
print("5. 응답 검색 테스트")
print("="*50)
try:
req = SearchResponsesRequest(
keyword="디자인",
limit=5
)
result = await search_responses_impl(req)
if 'error' in result:
print(f" ⚠️ 응답 검색 경고: {result.get('error')}")
return False
print(f" ✅ 응답 검색 성공")
responses = result.get('responses', [])
print(f" - 검색된 응답 수: {len(responses)}")
return True
except Exception as e:
print(f" ❌ 응답 검색 실패: {e}")
import traceback
traceback.print_exc()
return False
async def main():
"""메인 테스트 함수"""
print("\n" + "="*70)
print("Persona MCP Server 고급 기능 테스트 시작")
print("="*70)
results = []
# 테스트 실행
results.append(("샘플 페르소나 생성", await test_generate_sample_personas()))
results.append(("설문 플로우", await test_survey_flow()))
results.append(("페르소나 예측", await test_predict_persona()))
results.append(("선호도 분석", await test_analyze_preference()))
results.append(("응답 검색", await test_search_responses()))
# 결과 요약
print("\n" + "="*70)
print("테스트 결과 요약")
print("="*70)
passed = sum(1 for _, result in results if result)
total = len(results)
for name, result in results:
status = "✅ 통과" if result else "❌ 실패"
print(f" {status}: {name}")
print(f"\n총 {passed}/{total} 테스트 통과")
print("="*70)
if passed == total:
print("\n🎉 모든 테스트 통과!")
return 0
else:
print(f"\n⚠️ {total - passed}개 테스트 실패 (일부는 LLM API 키가 필요할 수 있습니다)")
return 1
if __name__ == "__main__":
try:
exit_code = asyncio.run(main())
sys.exit(exit_code)
except KeyboardInterrupt:
print("\n\n테스트가 중단되었습니다.")
sys.exit(1)
except Exception as e:
print(f"\n\n테스트 실행 중 오류 발생: {e}")
import traceback
traceback.print_exc()
sys.exit(1)