259 lines
11 KiB
Python
259 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
import warnings
|
|
|
|
warnings.simplefilter("ignore", ResourceWarning)
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from src.quant_engine.kis_api_client_v1 import (
|
|
KisCredentials,
|
|
OrderEndpointBlockedError,
|
|
_assert_read_only,
|
|
)
|
|
|
|
# governance/rules/06_no_direct_api_trading.yaml — 이 테스트는 절대 약화/삭제하지 않는다.
|
|
|
|
FORBIDDEN_ORDER_PATHS = (
|
|
"/uapi/domestic-stock/v1/trading/order-cash",
|
|
"/uapi/domestic-stock/v1/trading/order-rvsecncl",
|
|
"/uapi/domestic-stock/v1/trading/order-credit",
|
|
"/uapi/domestic-stock/v1/trading/order-resv",
|
|
"/uapi/domestic-stock/v1/trading/inquire-balance", # governance/rules/07 — 계좌 보유종목 조회 금지
|
|
)
|
|
FORBIDDEN_ORDER_TR_IDS = (
|
|
"TTTC0802U", "TTTC0801U", "VTTC0802U", "VTTC0801U",
|
|
"TTTC8434R", "VTTC8434R", # governance/rules/07 — 주식잔고조회 금지
|
|
)
|
|
|
|
|
|
class TestKisApiClientV1(unittest.TestCase):
|
|
|
|
def test_order_path_is_blocked(self):
|
|
for path in FORBIDDEN_ORDER_PATHS:
|
|
with self.assertRaises(OrderEndpointBlockedError):
|
|
_assert_read_only(path, "FHKST01010100")
|
|
|
|
def test_order_tr_id_is_blocked(self):
|
|
for tr_id in FORBIDDEN_ORDER_TR_IDS:
|
|
with self.assertRaises(OrderEndpointBlockedError):
|
|
_assert_read_only("/uapi/domestic-stock/v1/quotations/inquire-price", tr_id)
|
|
|
|
def test_known_readonly_endpoints_pass(self):
|
|
_assert_read_only("/uapi/domestic-stock/v1/quotations/inquire-price", "FHKST01010100")
|
|
_assert_read_only("/uapi/domestic-stock/v1/quotations/inquire-asking-price-exp-ccn", "FHKST01010200")
|
|
_assert_read_only("/uapi/domestic-stock/v1/quotations/daily-short-sale", "FHPST04830000")
|
|
|
|
def test_no_order_endpoint_substring_anywhere_in_kis_client_source(self):
|
|
"""정적 검증 — 누군가 향후 주문 함수를 추가하더라도 경로 문자열이 소스에 남으면 즉시 탐지.
|
|
|
|
TTTC8434R/VTTC8434R(주식잔고조회)는 FORBIDDEN_TR_ID_PREFIXES 차단목록 '데이터'로
|
|
이 파일에 의도적으로 존재한다(prefix가 아닌 전체 TR_ID라 prefix-매칭으로는 막을 수
|
|
없어 명시적으로 등재) — 이 두 개는 검사에서 제외한다. 전체 코드베이스 차원의
|
|
"차단목록 외 파일에는 한 글자도 없어야 한다"는 보장은
|
|
tools/validate_no_direct_api_trading_v1.py(ALLOWLISTED_FILES 제외 전체 스캔)가 맡는다.
|
|
"""
|
|
source = (ROOT / "src" / "quant_engine" / "kis_api_client_v1.py").read_text(encoding="utf-8")
|
|
blocklist_data_exceptions = {"TTTC8434R", "VTTC8434R"}
|
|
for forbidden_path in FORBIDDEN_ORDER_PATHS:
|
|
self.assertNotIn(forbidden_path, source, f"주문 엔드포인트 경로가 소스에 존재함: {forbidden_path}")
|
|
for forbidden_tr_id in FORBIDDEN_ORDER_TR_IDS:
|
|
if forbidden_tr_id in blocklist_data_exceptions:
|
|
continue
|
|
self.assertNotIn(forbidden_tr_id, source, f"주문 TR_ID가 소스에 존재함: {forbidden_tr_id}")
|
|
|
|
def test_kis_client_module_defines_no_order_submission_function(self):
|
|
import src.quant_engine.kis_api_client_v1 as kis_module
|
|
|
|
public_names = [name for name in dir(kis_module) if not name.startswith("_")]
|
|
banned_keywords = (
|
|
"place_order", "submit_order", "cancel_order", "revise_order", "send_order",
|
|
"inquire_balance", "account_balance",
|
|
)
|
|
for name in public_names:
|
|
lowered = name.lower()
|
|
for banned in banned_keywords:
|
|
self.assertNotIn(banned, lowered, f"주문 제출/정정/취소로 의심되는 함수가 존재함: {name}")
|
|
|
|
def test_kis_credentials_load_uses_required_env_vars(self):
|
|
with patch.dict("os.environ", {
|
|
"KIS_APP_Key": "real-key",
|
|
"KIS_APP_Secret": "real-secret",
|
|
"KIS_APP_Key_TEST": "mock-key",
|
|
"KIS_APP_Secret_TEST": "mock-secret"
|
|
}):
|
|
real = KisCredentials.load("real")
|
|
mock = KisCredentials.load("mock")
|
|
|
|
self.assertEqual(real.app_key, "real-key")
|
|
self.assertEqual(real.app_secret, "real-secret")
|
|
self.assertEqual(real.account, "real")
|
|
self.assertEqual(mock.app_key, "mock-key")
|
|
self.assertEqual(mock.app_secret, "mock-secret")
|
|
self.assertEqual(mock.account, "mock")
|
|
|
|
def test_issue_or_reuse_token_with_sqlite_db_cache(self):
|
|
"""SQLite 기반 KIS 토큰 캐싱 및 만료시간 재사용 정밀 하네스 테스트."""
|
|
import tempfile
|
|
import shutil
|
|
import datetime as dt
|
|
import sqlite3
|
|
from src.quant_engine.kis_api_client_v1 import _issue_or_reuse_token, KisCredentials
|
|
|
|
tmp_dir = tempfile.mkdtemp()
|
|
db_path = Path(tmp_dir) / "kis_data_collection.db"
|
|
|
|
# 패치 대상: DB 경로와 REST API 요청
|
|
with patch("src.quant_engine.kis_api_client_v1._token_db_path", return_value=db_path), \
|
|
patch("src.quant_engine.kis_api_client_v1._requests") as mock_requests:
|
|
|
|
# API 서버 호출 시 목업 데이터 설정
|
|
creds = KisCredentials(app_key="k", app_secret="s", account="mock")
|
|
|
|
# _requests().post() 가 호출되었을 때 응답 오브젝트를 Mock화
|
|
mock_resp = mock_requests.return_value.post.return_value
|
|
mock_resp.json.return_value = {
|
|
"access_token": "new-test-token-123",
|
|
"expires_in": "3600" # 1시간 유효
|
|
}
|
|
|
|
# 1. 최초 호출 (DB가 없으므로 API 호출로 새로 발급받음)
|
|
token1 = _issue_or_reuse_token(creds)
|
|
self.assertEqual(token1, "new-test-token-123")
|
|
self.assertTrue(db_path.exists())
|
|
|
|
# DB 검사
|
|
with sqlite3.connect(db_path) as conn:
|
|
row = conn.execute("SELECT access_token, expires_at FROM kis_tokens WHERE account = 'mock'").fetchone()
|
|
self.assertIsNotNone(row)
|
|
self.assertEqual(row[0], "new-test-token-123")
|
|
|
|
# 2. 유효기간 내 두 번째 호출 (API를 호출하지 않고 DB에 저장된 토큰 재사용)
|
|
mock_requests.return_value.post.reset_mock()
|
|
token2 = _issue_or_reuse_token(creds)
|
|
self.assertEqual(token2, "new-test-token-123")
|
|
mock_requests.return_value.post.assert_not_called() # API 미호출 증빙
|
|
|
|
# 3. 인위적 만료 처리 (expires_at을 과거로 조작)
|
|
with sqlite3.connect(db_path) as conn:
|
|
past_time = dt.datetime.now(dt.timezone.utc) - dt.timedelta(hours=2)
|
|
conn.execute("UPDATE kis_tokens SET expires_at = ? WHERE account = 'mock'", (past_time.isoformat(),))
|
|
conn.commit()
|
|
|
|
# 4. 만료 후 세 번째 호출 (DB는 있으나 만료되었으므로 KIS API 다시 호출)
|
|
mock_resp.json.return_value = {
|
|
"access_token": "expired-renewed-token-456",
|
|
"expires_in": "3600"
|
|
}
|
|
token3 = _issue_or_reuse_token(creds)
|
|
self.assertEqual(token3, "expired-renewed-token-456")
|
|
mock_requests.return_value.post.assert_called_once() # API 1회 재발급 증빙
|
|
|
|
shutil.rmtree(tmp_dir, ignore_errors=True)
|
|
|
|
def test_issue_or_reuse_token_honors_token_db_override(self):
|
|
import tempfile
|
|
import shutil
|
|
import sqlite3
|
|
from src.quant_engine.kis_api_client_v1 import _issue_or_reuse_token, KisCredentials
|
|
|
|
tmp_dir = tempfile.mkdtemp()
|
|
override_db = Path(tmp_dir) / "custom_kis_tokens.db"
|
|
|
|
with patch.dict("os.environ", {"KIS_TOKEN_DB_PATH": str(override_db)}), \
|
|
patch("src.quant_engine.kis_api_client_v1._requests") as mock_requests:
|
|
creds = KisCredentials(app_key="k", app_secret="s", account="mock")
|
|
mock_resp = mock_requests.return_value.post.return_value
|
|
mock_resp.raise_for_status.return_value = None
|
|
mock_resp.json.return_value = {
|
|
"access_token": "override-token-789",
|
|
"expires_in": "3600",
|
|
}
|
|
|
|
token = _issue_or_reuse_token(creds)
|
|
self.assertEqual(token, "override-token-789")
|
|
self.assertTrue(override_db.exists())
|
|
|
|
with sqlite3.connect(override_db) as conn:
|
|
row = conn.execute(
|
|
"SELECT access_token FROM kis_tokens WHERE account = 'mock'"
|
|
).fetchone()
|
|
self.assertIsNotNone(row)
|
|
self.assertEqual(row[0], "override-token-789")
|
|
|
|
shutil.rmtree(tmp_dir, ignore_errors=True)
|
|
|
|
def test_issue_or_reuse_token_serializes_concurrent_refresh(self):
|
|
import tempfile
|
|
import shutil
|
|
import sqlite3
|
|
import threading
|
|
import time
|
|
from src.quant_engine.kis_api_client_v1 import _issue_or_reuse_token, KisCredentials
|
|
|
|
tmp_dir = tempfile.mkdtemp()
|
|
override_db = Path(tmp_dir) / "kis_tokens.db"
|
|
barrier = threading.Barrier(2)
|
|
post_calls = []
|
|
|
|
class _Response:
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def json(self):
|
|
return {
|
|
"access_token": "concurrent-token-001",
|
|
"expires_in": "3600",
|
|
}
|
|
|
|
def _post(*args, **kwargs):
|
|
post_calls.append(time.time())
|
|
time.sleep(0.2)
|
|
return _Response()
|
|
|
|
with patch.dict("os.environ", {"KIS_TOKEN_DB_PATH": str(override_db)}), \
|
|
patch("src.quant_engine.kis_api_client_v1._requests") as mock_requests:
|
|
mock_requests.return_value.post.side_effect = _post
|
|
creds = KisCredentials(app_key="k", app_secret="s", account="mock")
|
|
|
|
results: list[str] = []
|
|
errors: list[BaseException] = []
|
|
|
|
def worker() -> None:
|
|
try:
|
|
barrier.wait(timeout=5)
|
|
results.append(_issue_or_reuse_token(creds))
|
|
except BaseException as exc: # pragma: no cover - test harness diagnostic
|
|
errors.append(exc)
|
|
|
|
t1 = threading.Thread(target=worker)
|
|
t2 = threading.Thread(target=worker)
|
|
t1.start()
|
|
t2.start()
|
|
t1.join(timeout=10)
|
|
t2.join(timeout=10)
|
|
|
|
self.assertEqual(errors, [])
|
|
self.assertEqual(results, ["concurrent-token-001", "concurrent-token-001"])
|
|
self.assertEqual(len(post_calls), 1)
|
|
|
|
with sqlite3.connect(override_db) as conn:
|
|
row = conn.execute(
|
|
"SELECT access_token FROM kis_tokens WHERE account = 'mock'"
|
|
).fetchone()
|
|
self.assertIsNotNone(row)
|
|
self.assertEqual(row[0], "concurrent-token-001")
|
|
|
|
shutil.rmtree(tmp_dir, ignore_errors=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|