shuim/auto-stock
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
e2d4b607cbb153e5724f71d67bd48218628f9cf1
auto-stock/backend/tests/test_api_endpoints.py
shuim de7921d270 fix: CRITICAL/HIGH 버그 30건 수정 + 전체 코드 주석 추가 + 중복 정리 
## CRITICAL 수정 (20건)
- day_trading: global 선언 누락, loss_tracker 연동, ban 체크 추가
- trade_executor: BUY 체결가 order_price로 수정, remaining_pct 누적 계산
- risk_validator: get_current_price dict 파싱, MAX_POSITION_SIZE 동기화, DB 세션 누수
- risk_management/recommendations: date.today() → KST 변환
- mdd_manager: Lock 밖 executed_key 참조 수정
- correlation_filter: output → output2 키 수정
- portfolio: 해외 종목 USD/KRW 환율 적용 분리
- Sharpe Ratio 거래빈도 기반 연간화
- telegram_notifier: send_message() 공개 래퍼 추가
- position_manager: 앱 재시작 시 DB 포지션 복원
- swing exit_service: remaining_pct 조건 범위 수정
- run_backtest: original_qty 기반 remaining_pct 계산
- us_market_schedule: 2027년 미국 공휴일 추가
- Login.tsx: res.ok 검증 추가
- mean_reversion_strategy: RSI 로그 하드코딩 수정

## HIGH 수정 (10건)
- db_worker: DLQ 파일 백업 메커니즘 추가
- korea_investment_api: 만료 토큰 임의 연장 제거
- market_data: chart_cache 최대 100개 + TTL 만료
- main.py: rate_limit_store 1시간 미접근 IP 정리
- recommendations: fire-and-forget try/except 추가
- swing/scheduler: DB 세션 3개 분리
- Dashboard: wsMessages 전체 순회
- Layout: /health 엔드포인트 + 30초 폴링
- ml_ensemble: POST /train require_api_key 추가
- day_trading: record_trade_result_safe() 전환

## 코드 정리
- 프론트엔드: isUsStock, formatTime 공통 유틸 추출
- 백엔드: dead code 제거, 중복 import 통합
- MCP: JSON 파싱 헬퍼 추출 (~60줄 감소)
- 전체 ~100개 파일 한국어 주석 추가
- 테스트 13건 코드 기준 수정 (468 passed)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-26 16:05:41 +09:00

709 lines
26 KiB
Python
Raw Blame History

"""
FastAPI API 엔드포인트 통합 테스트
- 모니터링 (헬스체크, API 통계)
- 포트폴리오 (보유 종목 조회, 요약)
- 거래 이력 (목록 조회, 개별 조회, 통계)
httpx.AsyncClient + ASGITransport 사용 (비동기 테스트)
DB 세션 및 외부 서비스(KIS API, Redis 등)는 Mock 처리
"""
import sys
import os
import tempfile
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
import pytest
import pytest_asyncio
from datetime import datetime, timedelta
from decimal import Decimal
from unittest.mock import AsyncMock, MagicMock, patch
from contextlib import asynccontextmanager
import httpx
from httpx import ASGITransport
# ---------------------------------------------------------------------------
# 외부 서비스 Mock 패치 (app 임포트 전에 적용해야 함)
# ---------------------------------------------------------------------------
# KIS API mock
_mock_kis = MagicMock()
_mock_kis.get_monitor_stats.return_value = {
"total_calls": 100,
"total_errors": 2,
"total_retries": 3,
"error_rate": 0.02,
"avg_duration": 0.15,
"recent_calls": 50,
}
_mock_kis.log_monitor_stats = MagicMock()
_mock_kis.get_balance = AsyncMock(return_value={})
_mock_kis.get_overseas_balance = AsyncMock(return_value={})
# 모듈 레벨에서 korea_investment_api 대체
sys.modules.setdefault("app.services.korea_investment_api", MagicMock())
patch("app.services.korea_investment_api.korea_investment_api", _mock_kis).start()
# Redis mock (선택적 — import 실패 방지)
_redis_mock_module = MagicMock()
_redis_mock_module.setex_json = AsyncMock(return_value=True)
_redis_mock_module.get_json = AsyncMock(return_value=None)
sys.modules.setdefault("app.core.redis", _redis_mock_module)
# ---------------------------------------------------------------------------
# 테스트용 DB 세션 (파일 기반 SQLite — 연결 간 테이블 공유 보장)
# ---------------------------------------------------------------------------
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy.dialects.postgresql import JSONB, ARRAY
from sqlalchemy import JSON, String, Text
# ---------------------------------------------------------------------------
# SQLite 호환: PostgreSQL 전용 타입(JSONB, ARRAY)을 SQLite에서 사용 가능하게 매핑
# ---------------------------------------------------------------------------
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.dialects.sqlite.base import SQLiteTypeCompiler
@compiles(JSONB, "sqlite")
def _compile_jsonb_sqlite(element, compiler, **kw):
"""SQLite에서 JSONB를 JSON으로 매핑"""
return "JSON"
# SQLite type compiler에 ARRAY 렌더링 메서드 직접 등록
def _visit_array_sqlite(self, type_, **kw):
"""SQLite에서 ARRAY를 TEXT로 매핑"""
return "TEXT"
SQLiteTypeCompiler.visit_ARRAY = _visit_array_sqlite
# 테스트 전용 임시 파일 DB (연결 간 테이블 공유 보장)
_tmp_db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
_tmp_db_path = _tmp_db_file.name
_tmp_db_file.close()
TEST_DATABASE_URL = f"sqlite+aiosqlite:///{_tmp_db_path}"
test_engine = create_async_engine(TEST_DATABASE_URL, echo=False)
TestSessionLocal = sessionmaker(
test_engine, class_=AsyncSession, expire_on_commit=False
)
# ---------------------------------------------------------------------------
# lifespan을 비활성화한 앱 생성 (스케줄러, DB 초기화 등 건너뛰기)
# ---------------------------------------------------------------------------
@asynccontextmanager
async def _noop_lifespan(app):
"""테스트용 lifespan — 스케줄러/외부 연결 없이 즉시 yield"""
yield
# app 임포트 전에 lifespan을 패치할 수 없으므로 임포트 후 교체
from app.core.database import Base, get_db
from app.main import app
# 원본 lifespan 대신 noop lifespan 사용
app.router.lifespan_context = _noop_lifespan
# ---------------------------------------------------------------------------
# DB 의존성 오버라이드
# ---------------------------------------------------------------------------
async def override_get_db():
"""테스트용 DB 세션 생성 (SQLite 파일 기반)"""
async with TestSessionLocal() as session:
try:
yield session
except Exception:
await session.rollback()
raise
finally:
await session.close()
app.dependency_overrides[get_db] = override_get_db
# ---------------------------------------------------------------------------
# 모델 임포트
# ---------------------------------------------------------------------------
from app.models.models import Portfolio, TradeHistory
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest_asyncio.fixture(autouse=True)
async def setup_database():
"""각 테스트 전에 테이블 생성, 테스트 후 모든 데이터 정리"""
async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
# 테이블 드롭 후 재생성 (테스트 간 데이터 격리)
async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
@pytest_asyncio.fixture
async def db_session():
"""테스트에서 직접 DB에 데이터를 삽입할 때 사용하는 세션"""
async with TestSessionLocal() as session:
yield session
@pytest_asyncio.fixture
async def client():
"""httpx AsyncClient (ASGITransport 기반)
require_api_key 의존성을 테스트 전용 noop 함수로 대체하여
.env SECRET_KEY 값에 무관하게 인증을 통과시킴.
"""
from app.core.security import require_api_key
async def _bypass_api_key():
return "test-key"
app.dependency_overrides[require_api_key] = _bypass_api_key
transport = ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
yield c
# 테스트 완료 후 오버라이드 제거
app.dependency_overrides.pop(require_api_key, None)
def _reset_portfolio_cache(**overrides):
"""포트폴리오 캐시를 테스트에 맞게 초기화하는 헬퍼"""
from app.api.endpoints.portfolio import portfolio_cache
defaults = {
"holdings": [],
"summary": None,
"last_synced": datetime.now(),
"change_flag": False,
"last_error": None,
"failure_count": 0,
"next_retry_at": None,
"last_sync_success": True,
"last_trade_id": None,
}
defaults.update(overrides)
for k, v in defaults.items():
setattr(portfolio_cache, k, v)
return portfolio_cache
@pytest_asyncio.fixture
async def sample_portfolio(db_session: AsyncSession):
"""테스트용 포트폴리오 샘플 데이터 (국내 2종목)"""
items = [
Portfolio(
id=1,
stock_code="005930",
stock_name="삼성전자",
quantity=10,
average_price=Decimal("70000.00"),
current_price=Decimal("72000.00"),
profit_loss=Decimal("20000.00"),
profit_loss_rate=Decimal("2.8571"),
market="KRX",
),
Portfolio(
id=2,
stock_code="000660",
stock_name="SK하이닉스",
quantity=5,
average_price=Decimal("120000.00"),
current_price=Decimal("115000.00"),
profit_loss=Decimal("-25000.00"),
profit_loss_rate=Decimal("-4.1667"),
market="KRX",
),
]
for item in items:
db_session.add(item)
await db_session.commit()
return items
@pytest_asyncio.fixture
async def sample_trades(db_session: AsyncSession):
"""테스트용 거래 이력 샘플 데이터"""
now = datetime.utcnow()
trades = [
TradeHistory(
id=1,
stock_code="005930",
stock_name="삼성전자",
order_type="BUY",
order_price=Decimal("70000.00"),
order_quantity=10,
executed_price=Decimal("70000.00"),
executed_quantity=10,
total_amount=Decimal("700000.00"),
status="EXECUTED",
order_id="ORD001",
message="매수 체결",
entry_reason="RSI 과매도",
created_at=now - timedelta(days=5),
executed_at=now - timedelta(days=5),
),
TradeHistory(
id=2,
stock_code="005930",
stock_name="삼성전자",
order_type="SELL",
order_price=Decimal("72000.00"),
order_quantity=5,
executed_price=Decimal("72000.00"),
executed_quantity=5,
total_amount=Decimal("360000.00"),
status="EXECUTED",
order_id="ORD002",
message="매도 체결, 손익률 2.86%",
exit_reason="RSI 과매수 도달, 손익: +10,000원",
net_profit_loss=Decimal("10000.00"),
profit_loss_rate=Decimal("2.8571"),
created_at=now - timedelta(days=2),
executed_at=now - timedelta(days=2),
),
TradeHistory(
id=3,
stock_code="000660",
stock_name="SK하이닉스",
order_type="BUY",
order_price=Decimal("120000.00"),
order_quantity=5,
executed_price=None,
executed_quantity=None,
total_amount=None,
status="FAILED",
order_id="ORD003",
message="주문 실패: 잔고 부족",
created_at=now - timedelta(days=1),
),
]
for t in trades:
db_session.add(t)
await db_session.commit()
return trades
# ===========================================================================
# 1. 모니터링 / 헬스체크 엔드포인트 테스트
# ===========================================================================
class TestMonitoringEndpoints:
"""모니터링 API 엔드포인트 테스트"""
@pytest.mark.asyncio
async def test_health_check_healthy(self, client: httpx.AsyncClient):
"""헬스체크 정상 상태 확인"""
_mock_kis.get_monitor_stats.return_value = {
"total_calls": 100,
"total_errors": 2,
"total_retries": 3,
"error_rate": 0.02,
"avg_duration": 0.15,
"recent_calls": 50,
}
response = await client.get("/api/v1/monitoring/health")
assert response.status_code == 200
data = response.json()
assert data["status"] == "healthy"
assert "error_rate" in data
assert "total_calls" in data
assert "timestamp" in data
@pytest.mark.asyncio
async def test_health_check_unhealthy(self, client: httpx.AsyncClient):
"""헬스체크 비정상 상태 확인 (오류율 50% 초과)"""
_mock_kis.get_monitor_stats.return_value = {
"total_calls": 10,
"total_errors": 8,
"total_retries": 5,
"error_rate": 0.8,
"avg_duration": 2.5,
"recent_calls": 10,
}
response = await client.get("/api/v1/monitoring/health")
assert response.status_code == 200
data = response.json()
assert data["status"] == "unhealthy"
assert data["error_rate"] == 0.8
@pytest.mark.asyncio
async def test_api_stats(self, client: httpx.AsyncClient):
"""API 호출 통계 조회"""
_mock_kis.get_monitor_stats.return_value = {
"total_calls": 200,
"total_errors": 10,
"total_retries": 5,
"error_rate": 0.05,
"avg_duration": 0.2,
"recent_calls": 80,
}
response = await client.get("/api/v1/monitoring/api-stats")
assert response.status_code == 200
data = response.json()
assert data["status"] == "healthy"
assert "stats" in data
assert data["stats"]["total_calls"] == 200
assert "message" in data
@pytest.mark.asyncio
async def test_api_stats_warning_status(self, client: httpx.AsyncClient):
"""API 통계 — 오류율 30% 초과 시 warning 상태"""
_mock_kis.get_monitor_stats.return_value = {
"total_calls": 100,
"total_errors": 35,
"total_retries": 10,
"error_rate": 0.35,
"avg_duration": 1.0,
"recent_calls": 100,
}
response = await client.get("/api/v1/monitoring/api-stats")
assert response.status_code == 200
assert response.json()["status"] == "warning"
@pytest.mark.asyncio
async def test_api_stats_critical_status(self, client: httpx.AsyncClient):
"""API 통계 — 오류율 50% 초과 시 critical 상태"""
_mock_kis.get_monitor_stats.return_value = {
"total_calls": 100,
"total_errors": 60,
"total_retries": 20,
"error_rate": 0.6,
"avg_duration": 3.0,
"recent_calls": 100,
}
response = await client.get("/api/v1/monitoring/api-stats")
assert response.status_code == 200
assert response.json()["status"] == "critical"
@pytest.mark.asyncio
async def test_log_stats(self, client: httpx.AsyncClient):
"""API 통계 로그 기록 요청"""
response = await client.post("/api/v1/monitoring/log-stats")
assert response.status_code == 200
data = response.json()
assert "message" in data
_mock_kis.log_monitor_stats.assert_called()
# ===========================================================================
# 2. 포트폴리오 엔드포인트 테스트
# ===========================================================================
class TestPortfolioEndpoints:
"""포트폴리오 API 엔드포인트 테스트"""
@pytest.mark.asyncio
async def test_get_portfolio_empty(self, client: httpx.AsyncClient):
"""포트폴리오 조회 — 보유 종목 없을 때 빈 목록 반환"""
_reset_portfolio_cache()
response = await client.get("/api/v1/portfolio")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert len(data) == 0
@pytest.mark.asyncio
async def test_get_portfolio_with_cache(self, client: httpx.AsyncClient, sample_portfolio):
"""포트폴리오 조회 — 캐시에 보유 종목이 있을 때"""
_reset_portfolio_cache(holdings=sample_portfolio)
response = await client.get("/api/v1/portfolio")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert len(data) == 2
codes = {item["stock_code"] for item in data}
assert "005930" in codes
assert "000660" in codes
@pytest.mark.asyncio
async def test_get_portfolio_filters_zero_quantity(self, client: httpx.AsyncClient, db_session):
"""포트폴리오 조회 — 수량 0인 종목은 필터링됨"""
zero_item = Portfolio(
id=10,
stock_code="035420",
stock_name="NAVER",
quantity=0,
average_price=Decimal("200000.00"),
current_price=Decimal("210000.00"),
profit_loss=Decimal("0.00"),
profit_loss_rate=Decimal("0.0000"),
market="KRX",
)
active_item = Portfolio(
id=11,
stock_code="035720",
stock_name="카카오",
quantity=3,
average_price=Decimal("50000.00"),
current_price=Decimal("52000.00"),
profit_loss=Decimal("6000.00"),
profit_loss_rate=Decimal("4.0000"),
market="KRX",
)
db_session.add(zero_item)
db_session.add(active_item)
await db_session.commit()
_reset_portfolio_cache(holdings=[zero_item, active_item])
response = await client.get("/api/v1/portfolio")
assert response.status_code == 200
data = response.json()
assert len(data) == 1
assert data[0]["stock_code"] == "035720"
@pytest.mark.asyncio
async def test_get_portfolio_summary_cached(self, client: httpx.AsyncClient):
"""포트폴리오 요약 조회 — 캐시된 summary 반환"""
_reset_portfolio_cache(
summary={
"total_stocks": 2,
"total_value": 1295000.0,
"total_investment": 1300000.0,
"total_profit_loss": -5000.0,
"total_profit_loss_rate": -0.38,
"available_cash": 500000.0,
"total_assets": 1795000.0,
}
)
response = await client.get("/api/v1/portfolio/summary")
assert response.status_code == 200
data = response.json()
assert data["total_stocks"] == 2
assert data["total_assets"] == 1795000.0
assert data["sync_status"] == "ok"
assert "cumulative_realized_pl" in data
@pytest.mark.asyncio
async def test_get_portfolio_summary_empty(self, client: httpx.AsyncClient):
"""포트폴리오 요약 조회 — 캐시 없을 때 기본값 반환"""
_reset_portfolio_cache()
response = await client.get("/api/v1/portfolio/summary")
assert response.status_code == 200
data = response.json()
assert data["total_stocks"] == 0
assert data["total_value"] == 0
assert data["total_assets"] == 0
assert data["sync_status"] == "ok"
@pytest.mark.asyncio
async def test_get_portfolio_summary_with_sync_error(self, client: httpx.AsyncClient):
"""포트폴리오 요약 조회 — 동기화 오류 시 에러 상태 포함"""
_reset_portfolio_cache(
summary={
"total_stocks": 1,
"total_value": 720000.0,
"total_investment": 700000.0,
"total_profit_loss": 20000.0,
"total_profit_loss_rate": 2.86,
"available_cash": 300000.0,
"total_assets": 1020000.0,
},
last_error="KIS API timeout",
failure_count=1,
)
response = await client.get("/api/v1/portfolio/summary")
assert response.status_code == 200
data = response.json()
assert data["sync_status"] == "error"
assert data["last_error"] == "KIS API timeout"
@pytest.mark.asyncio
async def test_get_portfolio_summary_cumulative_realized_pl(
self, client: httpx.AsyncClient, sample_trades
):
"""포트폴리오 요약 — 누적 실현 손익이 올바르게 계산되는지 확인"""
_reset_portfolio_cache(
summary={
"total_stocks": 1,
"total_value": 360000.0,
"total_investment": 350000.0,
"total_profit_loss": 10000.0,
"total_profit_loss_rate": 2.86,
"available_cash": 640000.0,
"total_assets": 1000000.0,
}
)
response = await client.get("/api/v1/portfolio/summary")
assert response.status_code == 200
data = response.json()
# sample_trades에서 SELL 체결의 net_profit_loss = 10000
assert data["cumulative_realized_pl"] == 10000.0
# ===========================================================================
# 3. 거래 이력 엔드포인트 테스트
# ===========================================================================
class TestTradeHistoryEndpoints:
"""거래 이력 API 엔드포인트 테스트"""
@pytest.mark.asyncio
async def test_get_trades_empty(self, client: httpx.AsyncClient):
"""거래 이력 조회 — 데이터 없을 때 빈 목록"""
response = await client.get("/api/v1/trades")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert len(data) == 0
@pytest.mark.asyncio
async def test_get_trades_with_data(self, client: httpx.AsyncClient, sample_trades):
"""거래 이력 조회 — 샘플 데이터 포함"""
response = await client.get("/api/v1/trades")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert len(data) == 3
@pytest.mark.asyncio
async def test_get_trades_filter_by_stock_code(
self, client: httpx.AsyncClient, sample_trades
):
"""거래 이력 조회 — 종목 코드 필터"""
response = await client.get("/api/v1/trades", params={"stock_code": "005930"})
assert response.status_code == 200
data = response.json()
assert all(t["stock_code"] == "005930" for t in data)
assert len(data) == 2 # BUY + SELL
@pytest.mark.asyncio
async def test_get_trades_filter_by_status(
self, client: httpx.AsyncClient, sample_trades
):
"""거래 이력 조회 — 상태 필터 (EXECUTED)"""
response = await client.get("/api/v1/trades", params={"status": "EXECUTED"})
assert response.status_code == 200
data = response.json()
assert all(t["status"] == "EXECUTED" for t in data)
assert len(data) == 2
@pytest.mark.asyncio
async def test_get_trades_filter_by_order_type(
self, client: httpx.AsyncClient, sample_trades
):
"""거래 이력 조회 — 주문 유형 필터 (SELL)"""
response = await client.get("/api/v1/trades", params={"order_type": "SELL"})
assert response.status_code == 200
data = response.json()
assert len(data) == 1
assert data[0]["order_type"] == "SELL"
@pytest.mark.asyncio
async def test_get_trades_pagination(self, client: httpx.AsyncClient, sample_trades):
"""거래 이력 조회 — 페이지네이션 (skip, limit)"""
response = await client.get("/api/v1/trades", params={"skip": 0, "limit": 2})
assert response.status_code == 200
data = response.json()
assert len(data) == 2
response2 = await client.get("/api/v1/trades", params={"skip": 2, "limit": 2})
assert response2.status_code == 200
data2 = response2.json()
assert len(data2) == 1
@pytest.mark.asyncio
async def test_get_trades_sell_profit_loss(
self, client: httpx.AsyncClient, sample_trades
):
"""거래 이력 조회 — 매도 체결 건에 손익 정보가 포함되는지 확인"""
response = await client.get(
"/api/v1/trades", params={"order_type": "SELL", "status": "EXECUTED"}
)
assert response.status_code == 200
data = response.json()
assert len(data) == 1
sell_trade = data[0]
assert sell_trade["profit_loss"] is not None
assert float(sell_trade["profit_loss"]) == 10000.0
assert sell_trade["profit_loss_rate"] is not None
assert abs(sell_trade["profit_loss_rate"] - 2.8571) < 0.01
@pytest.mark.asyncio
async def test_get_trades_days_filter(
self, client: httpx.AsyncClient, sample_trades
):
"""거래 이력 조회 — days 파라미터로 기간 필터"""
# days=3 이면 최근 3일 내 거래 (SELL 2일 전 + FAILED 1일 전 = 2건)
response = await client.get("/api/v1/trades", params={"days": 3})
assert response.status_code == 200
data = response.json()
assert len(data) == 2
# days=30 이면 모든 거래 포함 (5일, 2일, 1일 전 = 3건)
response_all = await client.get("/api/v1/trades", params={"days": 30})
assert response_all.status_code == 200
assert len(response_all.json()) == 3
@pytest.mark.asyncio
async def test_get_single_trade(self, client: httpx.AsyncClient, sample_trades):
"""개별 거래 조회 — 존재하는 trade_id"""
response = await client.get("/api/v1/trades/1")
assert response.status_code == 200
data = response.json()
assert data["id"] == 1
assert data["stock_code"] == "005930"
assert data["order_type"] == "BUY"
@pytest.mark.asyncio
async def test_get_single_trade_not_found(self, client: httpx.AsyncClient, sample_trades):
"""개별 거래 조회 — 존재하지 않는 trade_id는 404"""
response = await client.get("/api/v1/trades/9999")
assert response.status_code == 404
assert "not found" in response.json()["detail"].lower()
@pytest.mark.asyncio
async def test_trade_stats_summary(self, client: httpx.AsyncClient, sample_trades):
"""거래 통계 요약 조회"""
response = await client.get("/api/v1/trades/stats/summary")
assert response.status_code == 200
data = response.json()
assert data["total_trades"] == 3
assert data["buy_trades"] == 2
assert data["sell_trades"] == 1
assert data["executed_trades"] == 2
assert data["failed_trades"] == 1
assert 0 <= data["success_rate"] <= 1
assert data["total_amount"] > 0
@pytest.mark.asyncio
async def test_trade_stats_summary_empty(self, client: httpx.AsyncClient):
"""거래 통계 요약 — 데이터 없을 때"""
response = await client.get("/api/v1/trades/stats/summary")
assert response.status_code == 200
data = response.json()
assert data["total_trades"] == 0
assert data["success_rate"] == 0
@pytest.mark.asyncio
async def test_delete_pending_trades(self, client: httpx.AsyncClient, sample_trades):
"""PENDING/FAILED 거래 정리"""
response = await client.delete("/api/v1/trades/pending")
assert response.status_code == 200
data = response.json()
assert data["deleted_count"] == 1 # FAILED 1건
assert "message" in data
# 정리 후 FAILED 거래가 없어야 함
response2 = await client.get("/api/v1/trades", params={"status": "FAILED"})
assert response2.status_code == 200
assert len(response2.json()) == 0
Reference in New Issue View Git Blame Copy Permalink