diff --git a/tests/unit/test_kis_api_client_v1.py b/tests/unit/test_kis_api_client_v1.py index 8d90cce..57fdc6d 100644 --- a/tests/unit/test_kis_api_client_v1.py +++ b/tests/unit/test_kis_api_client_v1.py @@ -95,6 +95,65 @@ class TestKisApiClientV1(unittest.TestCase): 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) + if __name__ == "__main__": unittest.main()