test(kis-token): add unit tests for SQLite-based OAuth2 token database caching and expiration

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()