88 lines
3.0 KiB
Python
88 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from src.quant_engine import kis_data_collection_v1 as kdc
|
|
|
|
|
|
class TestKisDataCollectionV1(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self._tmp_root = Path(ROOT / "Temp" / "unit_test_kis_data_collection_v1")
|
|
self._tmp_root.mkdir(parents=True, exist_ok=True)
|
|
self.seed_json = self._tmp_root / "seed.json"
|
|
self.seed_json.write_text(
|
|
json.dumps(
|
|
{"data": {"data_feed": [{"Ticker": "005930", "Name": "삼성전자", "Sector": "반도체"}]}},
|
|
ensure_ascii=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def tearDown(self) -> None:
|
|
for path in self._tmp_root.glob("*"):
|
|
try:
|
|
path.unlink()
|
|
except OSError:
|
|
pass
|
|
try:
|
|
self._tmp_root.rmdir()
|
|
except OSError:
|
|
pass
|
|
|
|
def test_build_seed_rows(self):
|
|
rows = kdc._build_seed_rows(self.seed_json)
|
|
self.assertEqual(len(rows), 1)
|
|
self.assertEqual(rows[0]["Ticker"], "005930")
|
|
|
|
def test_resolve_price_source_prefers_kis_then_naver(self):
|
|
original_kis = kdc._normalize_kis_fields
|
|
original_naver = kdc._normalize_naver_price_history
|
|
kdc._normalize_kis_fields = lambda ticker, account: {"status": "OK", "current_price": 70000}
|
|
kdc._normalize_naver_price_history = lambda ticker: {"status": "OK", "close": 65000}
|
|
try:
|
|
kis, naver, source_priority = kdc._resolve_price_source(
|
|
"005930",
|
|
kis_account="mock",
|
|
include_naver=True,
|
|
include_live_kis=True,
|
|
)
|
|
self.assertEqual(kis["status"], "OK")
|
|
self.assertEqual(naver["status"], "OK")
|
|
self.assertEqual(source_priority[0], "kis_open_api")
|
|
self.assertIn("naver_finance", source_priority)
|
|
finally:
|
|
kdc._normalize_kis_fields = original_kis
|
|
kdc._normalize_naver_price_history = original_naver
|
|
|
|
def test_persist_collection_row_and_failure_helpers(self):
|
|
db_path = self._tmp_root / "collector.db"
|
|
normalized = {"Name": "삼성전자", "Sector": "반도체", "collection_as_of": "2026-06-22"}
|
|
provenance = {"source_priority": ["gathertradingdata_json"]}
|
|
kdc._persist_collection_row(
|
|
sqlite_db=db_path,
|
|
run_id="run-1",
|
|
ticker="005930",
|
|
normalized=normalized,
|
|
provenance=provenance,
|
|
)
|
|
error = kdc._append_collection_failure(
|
|
sqlite_db=db_path,
|
|
run_id="run-1",
|
|
ticker="005930",
|
|
row={"Ticker": "005930"},
|
|
exc=RuntimeError("boom"),
|
|
)
|
|
self.assertEqual(error["ticker"], "005930")
|
|
self.assertIn("boom", error["error"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|