You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
236 lines
7.3 KiB
Python
236 lines
7.3 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Stock Oracle 통합 테스트 스위트
|
|
모든 테스트를 순차적으로 실행하고 결과를 요약합니다.
|
|
"""
|
|
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
# 테스트 파일들과 설명
|
|
TEST_SUITE = [
|
|
{
|
|
"file": "test_yfinance_fallback.py",
|
|
"name": "YFinance Plus 연결 테스트",
|
|
"description": "yfinance_plus 모듈이 정상적으로 작동하는지 확인"
|
|
},
|
|
{
|
|
"file": "test_simple_real_data.py",
|
|
"name": "실제 SEC 데이터 기본 테스트",
|
|
"description": "실제 SEC 재무 데이터가 정상적으로 가져와지는지 확인"
|
|
},
|
|
{
|
|
"file": "test_real_sec_vs_yfinance_plus.py",
|
|
"name": "Stock Oracle vs YFinance Plus 비교",
|
|
"description": "Stock Oracle의 실제 데이터와 yfinance_plus 직접 호출 결과 비교"
|
|
},
|
|
{
|
|
"file": "test_5year_comparison.py",
|
|
"name": "5년 분기별 정확도 테스트",
|
|
"description": "동일한 분기 데이터의 정확도를 5년간 비교 검증"
|
|
}
|
|
]
|
|
|
|
def print_header():
|
|
"""테스트 시작 헤더 출력"""
|
|
print("=" * 80)
|
|
print("🧪 Stock Oracle 통합 테스트 스위트")
|
|
print("=" * 80)
|
|
print(f"테스트 시작 시간: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
print(f"총 테스트 개수: {len(TEST_SUITE)}")
|
|
print()
|
|
|
|
def run_test(test_info: dict) -> dict:
|
|
"""단일 테스트 실행"""
|
|
test_file = test_info["file"]
|
|
test_name = test_info["name"]
|
|
test_desc = test_info["description"]
|
|
|
|
print(f"🔍 [{test_name}]")
|
|
print(f" 📄 파일: {test_file}")
|
|
print(f" 📝 설명: {test_desc}")
|
|
print(" ⏳ 실행 중...")
|
|
|
|
start_time = time.time()
|
|
|
|
try:
|
|
# 테스트 실행
|
|
result = subprocess.run(
|
|
[sys.executable, test_file],
|
|
cwd=Path(__file__).parent,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=300 # 5분 타임아웃
|
|
)
|
|
|
|
end_time = time.time()
|
|
duration = end_time - start_time
|
|
|
|
if result.returncode == 0:
|
|
status = "✅ 성공"
|
|
success = True
|
|
else:
|
|
status = "❌ 실패"
|
|
success = False
|
|
|
|
print(f" {status} (실행시간: {duration:.1f}초)")
|
|
|
|
if not success:
|
|
print(" 🚨 에러 출력:")
|
|
print(" " + "\n ".join(result.stderr.split("\n")[:10])) # 처음 10줄만
|
|
|
|
print()
|
|
|
|
return {
|
|
"name": test_name,
|
|
"file": test_file,
|
|
"success": success,
|
|
"duration": duration,
|
|
"stdout": result.stdout,
|
|
"stderr": result.stderr,
|
|
"returncode": result.returncode
|
|
}
|
|
|
|
except subprocess.TimeoutExpired:
|
|
print(" ⏰ 타임아웃 (5분 초과)")
|
|
print()
|
|
return {
|
|
"name": test_name,
|
|
"file": test_file,
|
|
"success": False,
|
|
"duration": 300,
|
|
"stdout": "",
|
|
"stderr": "Timeout after 5 minutes",
|
|
"returncode": -1
|
|
}
|
|
|
|
except Exception as e:
|
|
print(f" 💥 예외 발생: {str(e)}")
|
|
print()
|
|
return {
|
|
"name": test_name,
|
|
"file": test_file,
|
|
"success": False,
|
|
"duration": 0,
|
|
"stdout": "",
|
|
"stderr": str(e),
|
|
"returncode": -2
|
|
}
|
|
|
|
def print_summary(results: list):
|
|
"""테스트 결과 요약 출력"""
|
|
print("=" * 80)
|
|
print("📊 테스트 결과 요약")
|
|
print("=" * 80)
|
|
|
|
total_tests = len(results)
|
|
successful_tests = sum(1 for r in results if r["success"])
|
|
failed_tests = total_tests - successful_tests
|
|
total_duration = sum(r["duration"] for r in results)
|
|
|
|
print(f"총 테스트: {total_tests}")
|
|
print(f"성공: {successful_tests} ✅")
|
|
print(f"실패: {failed_tests} ❌")
|
|
print(f"성공률: {(successful_tests/total_tests)*100:.1f}%")
|
|
print(f"총 실행시간: {total_duration:.1f}초")
|
|
print()
|
|
|
|
# 개별 테스트 결과
|
|
print("📋 개별 테스트 결과:")
|
|
print("-" * 80)
|
|
for i, result in enumerate(results, 1):
|
|
status_icon = "✅" if result["success"] else "❌"
|
|
print(f"{i:2d}. {status_icon} {result['name']}")
|
|
print(f" 📄 {result['file']}")
|
|
print(f" ⏱️ {result['duration']:.1f}초")
|
|
|
|
if not result["success"]:
|
|
print(f" 🚨 실패 사유: {result['stderr'][:100]}...")
|
|
print()
|
|
|
|
# 실패한 테스트가 있다면 상세 정보 출력
|
|
if failed_tests > 0:
|
|
print("🔍 실패 테스트 상세 정보:")
|
|
print("-" * 80)
|
|
for result in results:
|
|
if not result["success"]:
|
|
print(f"❌ {result['name']} ({result['file']})")
|
|
print(" 표준 출력:")
|
|
print(" " + "\n ".join(result["stdout"].split("\n")[:5]))
|
|
print(" 표준 에러:")
|
|
print(" " + "\n ".join(result["stderr"].split("\n")[:5]))
|
|
print()
|
|
|
|
def check_prerequisites():
|
|
"""테스트 실행 전 필수 조건 확인"""
|
|
print("🔧 필수 조건 확인 중...")
|
|
|
|
# Docker 컨테이너 실행 확인
|
|
try:
|
|
result = subprocess.run(
|
|
["curl", "-s", "http://localhost:18001/docs"],
|
|
capture_output=True,
|
|
timeout=5
|
|
)
|
|
if result.returncode != 0:
|
|
print("❌ Stock Oracle API 서버가 실행되지 않고 있습니다.")
|
|
print(" docker-compose up -d 명령으로 서버를 시작해주세요.")
|
|
return False
|
|
else:
|
|
print("✅ Stock Oracle API 서버 실행 확인")
|
|
except:
|
|
print("❌ Stock Oracle API 서버 연결 실패")
|
|
return False
|
|
|
|
# 필수 Python 패키지 확인
|
|
required_packages = ["requests", "yfinance_plus", "pandas"]
|
|
for package in required_packages:
|
|
try:
|
|
__import__(package)
|
|
print(f"✅ {package} 패키지 설치 확인")
|
|
except ImportError:
|
|
print(f"❌ {package} 패키지가 설치되지 않았습니다.")
|
|
return False
|
|
|
|
print("✅ 모든 필수 조건이 만족되었습니다.")
|
|
print()
|
|
return True
|
|
|
|
def main():
|
|
"""메인 테스트 실행 함수"""
|
|
print_header()
|
|
|
|
# 필수 조건 확인
|
|
if not check_prerequisites():
|
|
print("❌ 필수 조건이 만족되지 않아 테스트를 중단합니다.")
|
|
sys.exit(1)
|
|
|
|
# 테스트 실행
|
|
results = []
|
|
|
|
for i, test_info in enumerate(TEST_SUITE, 1):
|
|
print(f"[{i}/{len(TEST_SUITE)}] ", end="")
|
|
result = run_test(test_info)
|
|
results.append(result)
|
|
|
|
# 실패한 테스트가 있으면 잠시 대기
|
|
if not result["success"]:
|
|
time.sleep(2)
|
|
|
|
# 결과 요약
|
|
print_summary(results)
|
|
|
|
# 종료 코드 설정
|
|
failed_count = sum(1 for r in results if not r["success"])
|
|
if failed_count > 0:
|
|
print(f"❌ {failed_count}개의 테스트가 실패했습니다.")
|
|
sys.exit(1)
|
|
else:
|
|
print("🎉 모든 테스트가 성공했습니다!")
|
|
sys.exit(0)
|
|
|
|
if __name__ == "__main__":
|
|
main() |