"""카페24 모듈 순수 로직 + 토큰/암호화 테스트. DB/네트워크 없이 검증한다(가짜 저장소 + refresh 함수 주입). python -m app.modules.cafe24.tests.test_cafe24 또는 pytest 로 실행 가능. """ from __future__ import annotations import os from contextlib import contextmanager from datetime import timedelta from app.integrations.cafe24 import config as cfgmod from app.integrations.cafe24 import crypto, oauth, tokens from app.integrations.cafe24.errors import Cafe24AuthError, Cafe24ConfigError from app.modules.cafe24 import store from app.timezone import now_kst SECRET = "unit-test-secret" _TEST_ENV = { "CAFE24_MALL_ID": "testmall", "CAFE24_CLIENT_ID": "cid", "CAFE24_CLIENT_SECRET": "csecret", "CAFE24_REDIRECT_URI": "http://localhost:8080/cafe24/oauth/callback", "CAFE24_TOKEN_SECRET": SECRET, } def _config(): """환경변수에 의존하지 않도록 테스트용 값을 주입해 설정을 만든다.""" saved = {k: os.environ.get(k) for k in _TEST_ENV} os.environ.update(_TEST_ENV) try: return cfgmod.load_config() finally: for key, value in saved.items(): if value is None: os.environ.pop(key, None) else: os.environ[key] = value class _FakeRepo: """Cafe24Store 의 토큰 3개 메서드만 흉내낸다.""" def __init__(self, row=None): self.row = row self.saves: list[dict] = [] def get_token_row(self, mall_id): return self.row def save_token_row(self, *, mall_id, **fields): self.saves.append(fields) if self.row is None: self.row = {"mall_id": mall_id} self.row.update(fields) @contextmanager def token_lock(self, mall_id): outer = self class Handle: row = outer.row def save(self, **fields): outer.save_token_row(mall_id=mall_id, **fields) yield Handle() def _row(**overrides): row = { "mall_id": "testmall", "access_token": crypto.encrypt("AT", secret=SECRET), "refresh_token": crypto.encrypt("RT", secret=SECRET), "access_token_expires_at": now_kst() + timedelta(hours=1), "refresh_token_expires_at": now_kst() + timedelta(days=13), "scopes": "mall.read_product,mall.write_product", "last_refreshed_at": now_kst(), "last_error": "", "connected_by": "king@dbxcorp.co.kr", } row.update(overrides) return row # ════════════════════════════════════════════════════════════ # 암호화 # ════════════════════════════════════════════════════════════ def test_crypto_roundtrip(): token = "ACCESS-TOKEN-한글-123" encrypted = crypto.encrypt(token, secret=SECRET) assert encrypted != token and token not in encrypted assert crypto.decrypt(encrypted, secret=SECRET) == token def test_crypto_empty_passthrough(): assert crypto.encrypt("", secret=SECRET) == "" assert crypto.decrypt("", secret=SECRET) == "" def test_crypto_wrong_secret_raises(): encrypted = crypto.encrypt("AT", secret=SECRET) try: crypto.decrypt(encrypted, secret="다른키") except Cafe24ConfigError: return raise AssertionError("키가 바뀌면 Cafe24ConfigError 가 나야 한다") def test_crypto_requires_secret(): try: crypto.encrypt("x", secret="") except Cafe24ConfigError: return raise AssertionError("CAFE24_TOKEN_SECRET 없으면 예외여야 한다") # ════════════════════════════════════════════════════════════ # 설정 / 인증 URL # ════════════════════════════════════════════════════════════ def test_config_basics(): config = _config() assert config.configured assert config.missing == [] assert config.api_base == "https://testmall.cafe24api.com/api/v2" assert config.scope_param == "mall.read_product,mall.write_product" def test_authorize_url_has_state_and_no_secret(): url = oauth.build_authorize_url(_config(), state="STATE123") assert url.startswith("https://testmall.cafe24api.com/api/v2/oauth/authorize?") assert "state=STATE123" in url # client_secret 은 authorize 단계에 절대 실리면 안 된다. assert "csecret" not in url # ════════════════════════════════════════════════════════════ # 토큰 상태 / 자동 갱신 # ════════════════════════════════════════════════════════════ def test_status_without_token(): status = tokens.TokenService(_FakeRepo(None), _config()).status() assert status["connected"] is False assert status["needs_reauth"] is True def test_status_never_leaks_token_values(): service = tokens.TokenService(_FakeRepo(_row()), _config()) status = service.status() assert status["connected"] is True assert "AT" not in str(status) and "RT" not in str(status) def test_valid_token_returned_without_refresh(): service = tokens.TokenService(_FakeRepo(_row()), _config()) assert service.get_access_token() == "AT" def test_expired_refresh_token_needs_reauth(): row = _row(refresh_token_expires_at=now_kst() - timedelta(days=1)) assert tokens.TokenService(_FakeRepo(row), _config()).status()["needs_reauth"] is True def test_expired_access_token_triggers_refresh(): """만료된 access token 은 refresh 후 새 값을 돌려주고, 저장은 암호문으로 한다.""" repo = _FakeRepo(_row(access_token_expires_at=now_kst() - timedelta(minutes=5))) service = tokens.TokenService(repo, _config()) seen: list[str] = [] def fake_refresh(config, *, refresh_token): seen.append(refresh_token) return oauth.TokenBundle( access_token="NEW-AT", refresh_token="NEW-RT", access_token_expires_at=now_kst() + timedelta(hours=2), refresh_token_expires_at=now_kst() + timedelta(days=14), scopes="mall.read_product,mall.write_product", ) original = tokens.refresh_tokens tokens.refresh_tokens = fake_refresh try: assert service.get_access_token() == "NEW-AT" finally: tokens.refresh_tokens = original assert seen == ["RT"] # 복호화된 refresh token 이 전달돼야 한다 saved = repo.saves[-1] assert saved["access_token"] != "NEW-AT" # 평문 저장 금지 assert crypto.decrypt(saved["access_token"], secret=SECRET) == "NEW-AT" assert saved["last_error"] == "" def test_dead_refresh_token_raises_auth_error(): row = _row( access_token_expires_at=now_kst() - timedelta(minutes=1), refresh_token_expires_at=now_kst() - timedelta(days=1), ) service = tokens.TokenService(_FakeRepo(row), _config()) try: service.get_access_token() except Cafe24AuthError as exc: assert exc.needs_reauth is True return raise AssertionError("refresh token 만료 시 Cafe24AuthError 여야 한다") # ════════════════════════════════════════════════════════════ # store.py 순수 로직 # ════════════════════════════════════════════════════════════ def test_schedule_editable_only_when_pending(): assert store.is_editable("PENDING") is True for locked in ("PROCESSING", "SUCCESS", "FAILED", "CANCELLED"): assert store.is_editable(locked) is False, locked def test_retry_budget_and_backoff(): assert all(store.can_retry(i) for i in range(store.MAX_RETRY)) assert store.can_retry(store.MAX_RETRY) is False # 무한 재시도 방지 — 간격은 증가하되 상한이 있다. assert store.retry_backoff_seconds(0) < store.retry_backoff_seconds(1) assert store.retry_backoff_seconds(99) == store.retry_backoff_seconds(2) def test_normalize_revision_type(): assert store.normalize_revision_type("backup") == store.REVISION_BACKUP assert store.normalize_revision_type("nope") == store.REVISION_DRAFT def test_parse_product_no(): assert store.parse_product_no(" 123 ") == 123 for bad in ("abc", "0", "-3", None, ""): try: store.parse_product_no(bad) except ValueError: continue raise AssertionError(f"{bad!r} 는 거부해야 한다") def _run_all(): fns = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)] for fn in fns: fn() print("PASS", fn.__name__) print(f"\n{len(fns)} tests passed.") if __name__ == "__main__": _run_all()