""" 헤임달 (dev2-team) — 네이버 검색광고 API 클라이언트 단위 테스트 실제 API 호출 없이 unittest.mock 으로 모든 외부 의존성을 대체합니다. """ import base64 import hashlib import hmac import json import sys import unittest import urllib.error from io import BytesIO from unittest.mock import MagicMock, patch # --------------------------------------------------------------------------- # sys.path 설정 — 임포트 전에 먼저 추가해야 합니다. # --------------------------------------------------------------------------- for _p in ["/home/jay/workspace", "/home/jay/workspace/tools/naver-ads"]: if _p not in sys.path: sys.path.insert(0, _p) # --------------------------------------------------------------------------- # 외부 의존성 전체를 mock으로 등록 (실제 모듈 임포트 전) # --------------------------------------------------------------------------- # utils.env_loader — .env.keys 파일 로드 방지 _mock_load_env_keys = MagicMock() sys.modules.setdefault("utils", MagicMock()) sys.modules["utils.env_loader"] = MagicMock(load_env_keys=_mock_load_env_keys) sys.modules["utils.logger"] = MagicMock(get_logger=MagicMock(return_value=MagicMock())) # 환경변수 설정 (NaverSAClient.__init__ 에서 os.environ.get 으로 읽음) import os os.environ.setdefault("NAVER_SEARCHAD_CUSTOMER_ID", "test-customer-123") os.environ.setdefault("NAVER_SEARCHAD_API_KEY", "test-api-key-456") os.environ.setdefault("NAVER_SEARCHAD_SECRET_KEY", "test-secret-key-789") # --------------------------------------------------------------------------- # 테스트 대상 모듈 임포트 # --------------------------------------------------------------------------- from sa_api_client import NaverSAClient # type: ignore[import-not-found] # noqa: E402 from collect_sa_stats import _empty_output # type: ignore[import-not-found] # noqa: E402 # --------------------------------------------------------------------------- # 헬퍼: 테스트용 NaverSAClient 인스턴스 생성 # --------------------------------------------------------------------------- def _make_client() -> NaverSAClient: """load_env_keys를 mock한 상태로 NaverSAClient 를 생성합니다.""" with patch("sa_api_client.load_env_keys", _mock_load_env_keys): client = NaverSAClient.__new__(NaverSAClient) client._customer_id = os.environ["NAVER_SEARCHAD_CUSTOMER_ID"] client._api_key = os.environ["NAVER_SEARCHAD_API_KEY"] client._secret_key = os.environ["NAVER_SEARCHAD_SECRET_KEY"] return client # =========================================================================== # 테스트 클래스 # =========================================================================== class TestNaverSAClient(unittest.TestCase): # ----------------------------------------------------------------------- # 1. _generate_signature — HMAC-SHA256 서명 검증 # ----------------------------------------------------------------------- def test_generate_signature(self): """_generate_signature 가 올바른 HMAC-SHA256/base64 서명을 반환해야 합니다.""" client = _make_client() timestamp = "1712800000000" method = "GET" uri = "/ncc/campaigns" result = client._generate_signature(timestamp, method, uri) # 기대값을 직접 계산 message = f"{timestamp}.{method}.{uri}".encode("utf-8") expected_digest = hmac.new( client._secret_key.encode("utf-8"), message, hashlib.sha256, ).digest() expected = base64.b64encode(expected_digest).decode("utf-8") self.assertEqual(result, expected, "HMAC-SHA256 서명이 기대값과 다릅니다.") # base64 문자열인지 확인 (디코딩 성공 여부) decoded = base64.b64decode(result) self.assertEqual(len(decoded), 32, "SHA-256 다이제스트는 32바이트여야 합니다.") # ----------------------------------------------------------------------- # 2. _build_headers — 필수 헤더 존재 확인 # ----------------------------------------------------------------------- def test_build_headers(self): """_build_headers 가 필수 헤더 5개를 모두 포함해야 합니다.""" client = _make_client() headers = client._build_headers("GET", "/ncc/campaigns") required_keys = ["X-Timestamp", "X-API-KEY", "X-Customer", "X-Signature", "Content-Type"] for key in required_keys: self.assertIn(key, headers, f"필수 헤더 '{key}' 가 누락되었습니다.") self.assertEqual(headers["X-API-KEY"], client._api_key) self.assertEqual(headers["X-Customer"], client._customer_id) self.assertIn("application/json", headers["Content-Type"]) # X-Timestamp 는 숫자 문자열이어야 함 self.assertTrue(headers["X-Timestamp"].isdigit(), "X-Timestamp 가 숫자 문자열이 아닙니다.") # ----------------------------------------------------------------------- # 3. get_campaigns — _request 호출 검증 # ----------------------------------------------------------------------- def test_get_campaigns(self): """get_campaigns 가 GET /ncc/campaigns 로 _request 를 호출해야 합니다.""" client = _make_client() mock_result = [ {"nccCampaignId": "cam-001", "name": "캠페인A"}, {"nccCampaignId": "cam-002", "name": "캠페인B"}, ] with patch.object(client, "_request", return_value=mock_result) as mock_req: result = client.get_campaigns() mock_req.assert_called_once_with("GET", "/ncc/campaigns") self.assertEqual(result, mock_result, "get_campaigns 반환값이 _request 결과와 다릅니다.") # ----------------------------------------------------------------------- # 4. get_stats — 파라미터 변환 확인 (fields, time_range → JSON string) # ----------------------------------------------------------------------- def test_get_stats_params(self): """get_stats 가 fields 와 time_range 를 JSON string 으로 변환해야 합니다.""" client = _make_client() fields = ["impCnt", "clkCnt", "salesAmt"] time_range = {"since": "2026-04-01", "until": "2026-04-07"} ids = ["cam-001"] captured_params = {} def fake_request(_method, _uri, params=None): captured_params.update(params or {}) return [] with patch.object(client, "_request", side_effect=fake_request): client.get_stats(ids=ids, fields=fields, time_range=time_range) # fields 는 JSON string 이어야 함 self.assertIn("fields", captured_params) self.assertEqual( captured_params["fields"], json.dumps(fields), "fields 가 JSON string 으로 변환되지 않았습니다.", ) # time_range 는 JSON string 이어야 함 self.assertIn("timeRange", captured_params) self.assertEqual( captured_params["timeRange"], json.dumps(time_range), "time_range 가 JSON string 으로 변환되지 않았습니다.", ) # ----------------------------------------------------------------------- # 5. _request — 429 응답 시 재시도 확인 # ----------------------------------------------------------------------- def test_request_retry_on_429(self): """_request 가 429 HTTPError 를 받으면 재시도 후 성공해야 합니다.""" client = _make_client() # 두 번째 호출에서 성공 응답을 반환하는 mock 응답 객체 success_body = json.dumps({"ok": True}).encode("utf-8") mock_success_resp = MagicMock() mock_success_resp.__enter__ = MagicMock(return_value=mock_success_resp) mock_success_resp.__exit__ = MagicMock(return_value=False) mock_success_resp.read.return_value = success_body # 429 HTTPError 생성 http_429 = urllib.error.HTTPError( url="https://api.searchad.naver.com/ncc/campaigns", code=429, msg="Too Many Requests", hdrs=MagicMock(), fp=BytesIO(b"rate limited"), ) call_count = 0 def mock_urlopen(_req, timeout=30): # noqa: ARG001 nonlocal call_count call_count += 1 if call_count == 1: raise http_429 return mock_success_resp with patch("urllib.request.urlopen", side_effect=mock_urlopen), \ patch("time.sleep") as mock_sleep: result = client._request("GET", "/ncc/campaigns") self.assertEqual(call_count, 2, "urlopen 이 정확히 2번 호출되어야 합니다 (1번 실패 + 1번 성공).") mock_sleep.assert_called_once_with(1) # 2**0 = 1 self.assertEqual(result, {"ok": True}, "재시도 후 올바른 응답을 반환해야 합니다.") # ----------------------------------------------------------------------- # 6. _empty_output — 반환 구조 및 빈 값 확인 # ----------------------------------------------------------------------- def test_empty_output_graceful(self): """_empty_output 이 필수 키와 빈 summary 를 포함해야 합니다.""" since = "2026-04-01" until = "2026-04-07" output = _empty_output(since, until) # 필수 최상위 키 required_keys = ["last_collected", "summary", "keyword_groups", "daily_trend", "budget", "campaigns"] for key in required_keys: self.assertIn(key, output, f"필수 키 '{key}' 가 _empty_output 에 없습니다.") # summary.today / summary.yesterday 존재 확인 summary = output["summary"] self.assertIn("today", summary) self.assertIn("yesterday", summary) # today/yesterday 의 수치 필드가 모두 0 (빈 값) for period in ("today", "yesterday"): for metric in ("impressions", "clicks", "cost", "conversions"): self.assertEqual( summary[period][metric], 0, f"summary.{period}.{metric} 가 0이 아닙니다.", ) self.assertEqual(summary[period]["ctr"], 0.0, f"summary.{period}.ctr 가 0.0이 아닙니다.") self.assertEqual(summary[period]["cpc"], 0, f"summary.{period}.cpc 가 0이 아닙니다.") # campaigns 는 빈 리스트 self.assertEqual(output["campaigns"], [], "campaigns 가 빈 리스트여야 합니다.") # daily_trend: since~until 일수 = 7개 self.assertEqual(len(output["daily_trend"]), 7, "daily_trend 항목 수가 7이어야 합니다.") # budget 키 확인 self.assertIn("daily", output["budget"]) self.assertIn("spent_today", output["budget"]) self.assertIn("burn_rate", output["budget"]) # --------------------------------------------------------------------------- # 직접 실행 지원 # --------------------------------------------------------------------------- if __name__ == "__main__": unittest.main(verbosity=2)