dbx-main/app/modules/cafe24/tests/test_cafe24.py

"""카페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, products, 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} 는 거부해야 한다")


# ════════════════════════════════════════════════════════════
# 상품 엔드포인트 래퍼
# 실제 쇼핑몰 확인 결과 /admin/products/{no}/description 은 존재하지 않는다
# (`No API found.`). 상세설명은 상품 리소스의 필드다 — 경로가 되돌아가지 않게
# 여기서 고정한다.
# ════════════════════════════════════════════════════════════
class _FakeClient:
    """Cafe24Client 의 get/put 만 흉내내고 호출을 기록한다."""

    def __init__(self, payload=None):
        self.payload = payload or {}
        self.calls: list[dict] = []

    def get(self, path, *, params=None, json=None, product_no=None):
        self.calls.append({"method": "GET", "path": path, "params": params})
        return self.payload

    def put(self, path, *, params=None, json=None, product_no=None):
        self.calls.append({"method": "PUT", "path": path, "json": json})
        return self.payload


_PRODUCT = {
    "product_no": 131,
    "product_code": "P000000B",
    "product_name": "빠져락 1개(사은품)",
    "display": "T",
    "selling": "F",
    "description": "<p>PC</p>",
    "mobile_description": "<p>PC</p>",
    "separated_mobile_description": "F",
}


def test_descriptions_from_product():
    desc = products.descriptions_from_product(_PRODUCT)
    assert desc.product_no == 131
    assert desc.description == "<p>PC</p>"
    assert desc.separated_mobile is False
    assert desc.mobile_differs is False


def test_descriptions_separated_mobile_and_diff():
    desc = products.descriptions_from_product(
        {**_PRODUCT, "separated_mobile_description": "T", "mobile_description": "<p>MO</p>"}
    )
    assert desc.separated_mobile is True
    assert desc.mobile_differs is True


def test_fetch_descriptions_uses_product_resource():
    client = _FakeClient({"product": _PRODUCT})
    desc = products.fetch_descriptions(client, 131)
    assert desc.description == "<p>PC</p>"
    paths = [c["path"] for c in client.calls]
    assert paths == ["/admin/products/131"], paths
    assert not any(p.endswith("/description") for p in paths)


def test_update_descriptions_payload():
    client = _FakeClient({"product": _PRODUCT})
    products.update_descriptions(client, 131, description="<p>NEW</p>")
    call = client.calls[0]
    assert call["method"] == "PUT" and call["path"] == "/admin/products/131"
    # 준 필드만 바뀌어야 한다 — 모바일을 지정하지 않으면 보내지 않는다.
    assert call["json"] == {"request": {"description": "<p>NEW</p>"}}


def test_update_payload_optional_fields():
    both = products.build_update_payload(
        description="<p>PC</p>", mobile_description="<p>MO</p>", shop_no=1
    )
    assert both == {"shop_no": 1, "request": {"description": "<p>PC</p>", "mobile_description": "<p>MO</p>"}}
    # 빈 문자열은 "모바일을 비운다"는 뜻이므로 None 과 구분해 전달돼야 한다.
    assert products.build_update_payload(description="x", mobile_description="")["request"] == {
        "description": "x",
        "mobile_description": "",
    }


def test_normalize_product_flags():
    row = products.normalize_product(_PRODUCT)
    assert row == {
        "product_no": 131,
        "product_code": "P000000B",
        "product_name": "빠져락 1개(사은품)",
        "display": True,
        "selling": False,
    }
    # 값이 없으면 기본 True(카페24 응답에 필드가 빠진 경우 진열 중으로 본다).
    assert products.normalize_product({"product_no": "9"})["display"] is True


def test_list_products_clamps_paging():
    client = _FakeClient({"products": []})
    products.list_products(client, limit=999, offset=-5, product_name="락")
    params = client.calls[0]["params"]
    assert params["limit"] == products.PAGE_LIMIT
    assert params["offset"] == 0
    assert params["product_name"] == "락"


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