test_sector_tools.py•17.5 kB
"""섹터 분석 도구 테스트"""
import pytest
from datetime import datetime, timedelta
from unittest.mock import AsyncMock
from src.tools.sector_tools import SectorAnalysisTool
from src.exceptions import DataValidationError, DatabaseConnectionError
class TestSectorAnalysisTool:
"""섹터 분석 도구 테스트"""
@pytest.fixture
def mock_db_manager(self):
"""Mock 데이터베이스 매니저"""
return AsyncMock()
@pytest.fixture
def mock_cache_manager(self):
"""Mock 캐시 매니저"""
return AsyncMock()
@pytest.fixture
def sector_tool(self, mock_db_manager, mock_cache_manager):
"""섹터 분석 도구 인스턴스"""
return SectorAnalysisTool(mock_db_manager, mock_cache_manager)
@pytest.fixture
def sample_sector_data(self):
"""샘플 섹터 데이터"""
return [
{
"sector_name": "IT",
"sector_code": "IT001",
"market_cap": 450000000000000,
"total_volume": 280000000000,
"change_rate": 2.35,
"advancing_stocks": 85,
"declining_stocks": 45,
"unchanged_stocks": 12,
"total_stocks": 142,
"top_performer": "삼성전자",
"top_performer_change": 3.8,
"worst_performer": "LG유플러스",
"worst_performer_change": -2.1,
"date": datetime.now().date(),
"timestamp": datetime.now()
},
{
"sector_name": "금융",
"sector_code": "FIN001",
"market_cap": 320000000000000,
"total_volume": 180000000000,
"change_rate": -0.85,
"advancing_stocks": 35,
"declining_stocks": 65,
"unchanged_stocks": 8,
"total_stocks": 108,
"top_performer": "KB금융",
"top_performer_change": 1.2,
"worst_performer": "우리금융지주",
"worst_performer_change": -3.4,
"date": datetime.now().date(),
"timestamp": datetime.now()
},
{
"sector_name": "바이오",
"sector_code": "BIO001",
"market_cap": 180000000000000,
"total_volume": 420000000000,
"change_rate": 4.25,
"advancing_stocks": 95,
"declining_stocks": 25,
"unchanged_stocks": 5,
"total_stocks": 125,
"top_performer": "셀트리온",
"top_performer_change": 8.5,
"worst_performer": "유한양행",
"worst_performer_change": -1.8,
"date": datetime.now().date(),
"timestamp": datetime.now()
}
]
def test_tool_initialization(self, sector_tool, mock_db_manager, mock_cache_manager):
"""도구 초기화 테스트"""
assert sector_tool.name == "get_sector_analysis"
assert sector_tool.description is not None
assert "섹터" in sector_tool.description or "업종" in sector_tool.description
assert sector_tool.db_manager == mock_db_manager
assert sector_tool.cache_manager == mock_cache_manager
def test_tool_definition(self, sector_tool):
"""도구 정의 테스트"""
definition = sector_tool.get_tool_definition()
assert definition.name == "get_sector_analysis"
assert definition.description is not None
assert definition.inputSchema is not None
# 입력 스키마 검증
schema = definition.inputSchema
assert schema["type"] == "object"
assert "properties" in schema
properties = schema["properties"]
assert "market" in properties
assert "sector" in properties
assert "sort_by" in properties
assert "include_stocks" in properties
# market 파라미터 검증
market_prop = properties["market"]
assert market_prop["type"] == "string"
assert "KOSPI" in market_prop["enum"]
assert "KOSDAQ" in market_prop["enum"]
# sort_by 파라미터 검증
sort_prop = properties["sort_by"]
assert sort_prop["type"] == "string"
assert "market_cap" in sort_prop["enum"]
assert "change_rate" in sort_prop["enum"]
assert "volume" in sort_prop["enum"]
@pytest.mark.asyncio
async def test_execute_all_sectors(self, sector_tool, sample_sector_data):
"""전체 섹터 분석 조회 테스트"""
# 캐시 미스
sector_tool.cache_manager.get.return_value = None
# 데이터베이스 응답 설정
sector_tool.db_manager.fetch_all.return_value = sample_sector_data
# 실행
result = await sector_tool.execute({
"market": "ALL",
"sector": "ALL",
"sort_by": "market_cap"
})
# 결과 검증
assert len(result) == 1
content = result[0]
assert content.type == "text"
# JSON 파싱하여 내용 확인
import json
data = json.loads(content.text)
assert "timestamp" in data
assert "market" in data
assert "sector_analysis" in data
assert "market_summary" in data
# 섹터 분석 데이터 검증
sectors = data["sector_analysis"]
assert len(sectors) == 3
# 시가총액 정렬 확인 (내림차순)
it_sector = sectors[0]
assert it_sector["sector_name"] == "IT"
assert it_sector["market_cap"] == 450000000000000
assert it_sector["change_rate"] == 2.35
# 시장 요약 검증
summary = data["market_summary"]
assert "total_market_cap" in summary
assert "sector_count" in summary
assert "positive_sectors" in summary
assert "negative_sectors" in summary
@pytest.mark.asyncio
async def test_execute_specific_sector(self, sector_tool, sample_sector_data):
"""특정 섹터 분석 조회 테스트"""
# IT 섹터만 필터링
it_data = [data for data in sample_sector_data if data["sector_name"] == "IT"]
sector_tool.cache_manager.get.return_value = None
sector_tool.db_manager.fetch_all.return_value = it_data
# 실행
result = await sector_tool.execute({
"market": "KOSPI",
"sector": "IT"
})
# 결과 검증
content = result[0]
import json
data = json.loads(content.text)
sectors = data["sector_analysis"]
assert len(sectors) == 1
assert sectors[0]["sector_name"] == "IT"
# 데이터베이스 쿼리에 필터 조건 확인
call_args = sector_tool.db_manager.fetch_all.call_args[0]
query = call_args[0]
params = call_args[1:]
assert "IT" in str(params) or "IT" in query
@pytest.mark.asyncio
async def test_execute_with_stock_details(self, sector_tool, sample_sector_data):
"""주요 종목 상세 정보 포함 테스트"""
sector_tool.cache_manager.get.return_value = None
sector_tool.db_manager.fetch_all.return_value = sample_sector_data
# 주요 종목 상세 데이터 Mock
sector_tool.db_manager.fetch_many.return_value = [
[
{
"stock_code": "005930",
"stock_name": "삼성전자",
"current_price": 78500,
"change_rate": 3.8,
"volume": 15000000,
"market_cap": 480000000000000
},
{
"stock_code": "000660",
"stock_name": "SK하이닉스",
"current_price": 142000,
"change_rate": 2.1,
"volume": 8500000,
"market_cap": 103000000000000
}
]
]
# 실행
result = await sector_tool.execute({
"market": "ALL",
"sector": "ALL",
"include_stocks": True
})
# 결과 검증
content = result[0]
import json
data = json.loads(content.text)
sectors = data["sector_analysis"]
# 첫 번째 섹터에 주요 종목 정보 확인
first_sector = sectors[0]
if "major_stocks" in first_sector:
stocks = first_sector["major_stocks"]
assert len(stocks) >= 1
assert "stock_name" in stocks[0]
assert "change_rate" in stocks[0]
@pytest.mark.asyncio
async def test_sort_by_change_rate(self, sector_tool, sample_sector_data):
"""변화율 정렬 테스트"""
sector_tool.cache_manager.get.return_value = None
sector_tool.db_manager.fetch_all.return_value = sample_sector_data
# 실행
result = await sector_tool.execute({
"market": "ALL",
"sector": "ALL",
"sort_by": "change_rate"
})
# 결과 검증
content = result[0]
import json
data = json.loads(content.text)
sectors = data["sector_analysis"]
# 변화율 내림차순 정렬 확인
assert sectors[0]["sector_name"] == "바이오" # 4.25%
assert sectors[1]["sector_name"] == "IT" # 2.35%
assert sectors[2]["sector_name"] == "금융" # -0.85%
@pytest.mark.asyncio
async def test_sort_by_volume(self, sector_tool, sample_sector_data):
"""거래량 정렬 테스트"""
sector_tool.cache_manager.get.return_value = None
sector_tool.db_manager.fetch_all.return_value = sample_sector_data
# 실행
result = await sector_tool.execute({
"market": "ALL",
"sector": "ALL",
"sort_by": "volume"
})
# 결과 검증
content = result[0]
import json
data = json.loads(content.text)
sectors = data["sector_analysis"]
# 거래량 내림차순 정렬 확인
assert sectors[0]["sector_name"] == "바이오" # 420B
assert sectors[1]["sector_name"] == "IT" # 280B
assert sectors[2]["sector_name"] == "금융" # 180B
def test_sector_strength_calculation(self, sector_tool):
"""섹터 강도 계산 테스트"""
# 강한 상승 섹터
strength = sector_tool._calculate_sector_strength(85, 45, 12, 4.25)
assert strength == "매우 강함"
# 약한 상승 섹터
strength = sector_tool._calculate_sector_strength(55, 50, 15, 1.2)
assert strength == "보통"
# 하락 섹터
strength = sector_tool._calculate_sector_strength(30, 75, 10, -2.5)
assert strength == "약함"
# 강한 하락 섹터
strength = sector_tool._calculate_sector_strength(20, 90, 5, -4.8)
assert strength == "매우 약함"
def test_market_cap_formatting(self, sector_tool):
"""시가총액 포맷팅 테스트"""
# 조 단위
formatted = sector_tool._format_market_cap(450000000000000)
assert formatted == "450.0조"
# 천억 단위
formatted = sector_tool._format_market_cap(180000000000000)
assert formatted == "180.0조"
# 억 단위
formatted = sector_tool._format_market_cap(85000000000)
assert formatted == "850억"
@pytest.mark.asyncio
async def test_cache_functionality(self, sector_tool):
"""캐시 기능 테스트"""
# 캐시 히트 시나리오
cached_data = {
"timestamp": datetime.now().isoformat(),
"market": "ALL",
"sector_analysis": []
}
sector_tool.cache_manager.get.return_value = cached_data
# 실행
result = await sector_tool.execute({
"market": "ALL",
"sector": "ALL"
})
# 캐시에서 데이터 반환 확인
content = result[0]
import json
data = json.loads(content.text)
assert data == cached_data
# 데이터베이스 호출 없음 확인
sector_tool.db_manager.fetch_all.assert_not_called()
@pytest.mark.asyncio
async def test_error_handling(self, sector_tool):
"""에러 처리 테스트"""
sector_tool.cache_manager.get.return_value = None
sector_tool.db_manager.fetch_all.side_effect = DatabaseConnectionError("DB 연결 실패")
with pytest.raises(DatabaseConnectionError):
await sector_tool.execute({
"market": "ALL",
"sector": "ALL"
})
@pytest.mark.asyncio
async def test_invalid_parameters(self, sector_tool):
"""잘못된 파라미터 테스트"""
# 잘못된 시장
with pytest.raises(ValueError, match="Invalid market"):
await sector_tool.execute({
"market": "INVALID",
"sector": "ALL"
})
# 잘못된 정렬 기준
with pytest.raises(ValueError, match="Invalid sort_by"):
await sector_tool.execute({
"market": "ALL",
"sector": "ALL",
"sort_by": "invalid"
})
@pytest.mark.asyncio
async def test_empty_data_handling(self, sector_tool):
"""빈 데이터 처리 테스트"""
sector_tool.cache_manager.get.return_value = None
sector_tool.db_manager.fetch_all.return_value = []
result = await sector_tool.execute({
"market": "ALL",
"sector": "ALL"
})
content = result[0]
import json
data = json.loads(content.text)
assert data["sector_analysis"] == []
assert "message" in data
assert "데이터가 없습니다" in data["message"] or "no data" in data["message"].lower()
@pytest.mark.asyncio
async def test_market_summary_calculation(self, sector_tool, sample_sector_data):
"""시장 요약 계산 테스트"""
sector_tool.cache_manager.get.return_value = None
sector_tool.db_manager.fetch_all.return_value = sample_sector_data
result = await sector_tool.execute({
"market": "ALL",
"sector": "ALL"
})
content = result[0]
import json
data = json.loads(content.text)
summary = data["market_summary"]
assert summary["sector_count"] == 3
assert summary["positive_sectors"] == 2 # IT, 바이오
assert summary["negative_sectors"] == 1 # 금융
assert summary["total_market_cap"] == 950000000000000 # 합계
assert summary["avg_change_rate"] == 1.92 # (2.35 + (-0.85) + 4.25) / 3 = 1.92
@pytest.mark.asyncio
async def test_relative_strength_analysis(self, sector_tool, sample_sector_data):
"""섹터별 상대 강도 분석 테스트"""
sector_tool.cache_manager.get.return_value = None
sector_tool.db_manager.fetch_all.return_value = sample_sector_data
result = await sector_tool.execute({
"market": "ALL",
"sector": "ALL",
"include_relative_strength": True
})
content = result[0]
import json
data = json.loads(content.text)
# 상대 강도 분석 확인
if "relative_strength" in data:
rs_data = data["relative_strength"]
assert "strongest_sector" in rs_data
assert "weakest_sector" in rs_data
assert rs_data["strongest_sector"]["name"] == "바이오"
assert rs_data["weakest_sector"]["name"] == "금융"
@pytest.mark.asyncio
async def test_sector_rotation_analysis(self, sector_tool):
"""섹터 로테이션 분석 테스트"""
# 과거 데이터와 현재 데이터
historical_data = [
{"sector_name": "IT", "change_rate": -1.5},
{"sector_name": "금융", "change_rate": 3.2},
{"sector_name": "바이오", "change_rate": 0.8}
]
current_data = [
{"sector_name": "IT", "change_rate": 2.35},
{"sector_name": "금융", "change_rate": -0.85},
{"sector_name": "바이오", "change_rate": 4.25}
]
sector_tool.cache_manager.get.return_value = None
sector_tool.db_manager.fetch_all.side_effect = [current_data, historical_data]
result = await sector_tool.execute({
"market": "ALL",
"sector": "ALL",
"include_rotation_analysis": True
})
content = result[0]
import json
data = json.loads(content.text)
if "rotation_analysis" in data:
rotation = data["rotation_analysis"]
assert "momentum_leaders" in rotation
assert "momentum_laggards" in rotation