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


class _PagingClient:
    """페이지를 넘겨가며 응답하는 가짜 클라이언트."""

    def __init__(self, count: int):
        self.count = count
        self.calls: list[dict] = []

    def get(self, path, *, params=None, json=None, product_no=None):
        self.calls.append(dict(params or {}))
        offset = int((params or {}).get("offset", 0))
        limit = int((params or {}).get("limit", 100))
        page = [{"product_no": n} for n in range(offset, min(offset + limit, self.count))]
        return {"products": page}


def test_list_all_products_walks_pages():
    client = _PagingClient(230)
    rows, truncated = products.list_all_products(client)
    assert len(rows) == 230 and truncated is False
    # 100 + 100 + 30 → 3회 호출로 끝나야 한다.
    assert len(client.calls) == 3
    assert [c["offset"] for c in client.calls] == [0, 100, 200]


def test_list_all_products_stops_at_cap():
    """상한을 넘으면 잘렸다고 알린다 — 무한 호출로 API 제한에 걸리지 않게."""
    client = _PagingClient(10_000)
    rows, truncated = products.list_all_products(client, max_items=150)
    assert len(rows) == 150 and truncated is True


def test_list_all_products_single_page():
    """현재 쇼핑몰(87개)은 1회 호출로 끝난다."""
    client = _PagingClient(87)
    rows, truncated = products.list_all_products(client)
    assert len(rows) == 87 and truncated is False
    assert len(client.calls) == 1


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"] == "락"


# ════════════════════════════════════════════════════════════
# 이미지 URL 한글 파일명 표시 ↔ 저장 (왕복 보존이 핵심)
# ════════════════════════════════════════════════════════════
_REAL_IMG = "%EC%9A%A9%EA%B8%B0%EB%83%84%EC%83%88%EC%B0%A8%EB%8B%A8(%ED%99%A9%ED%86%A0)_12.gif"
_REAL_HTML = f'<img src="/web/product/big/{_REAL_IMG}" alt="용기 냄새차단">'


def test_decode_shows_korean_filename():
    decoded = store.decode_html_urls(_REAL_HTML)
    assert "용기냄새차단(황토)_12.gif" in decoded
    assert "%EC%9A%A9" not in decoded
    # 괄호는 원본에서 인코딩돼 있지 않으므로 그대로 남아야 한다.
    assert "(황토)" in decoded


def test_url_roundtrip_is_byte_identical():
    """편집하지 않고 적용해도 카페24 저장값이 달라지면 안 된다."""
    assert store.encode_html_urls(store.decode_html_urls(_REAL_HTML)) == _REAL_HTML


def test_ascii_escapes_are_not_decoded():
    """%20·%3C 를 풀면 URL·HTML 구조가 깨진다 — 건드리지 않는다."""
    html = '<img src="/web/a%20b.png?x=1%3C2">'
    assert store.decode_html_urls(html) == html
    assert store.encode_html_urls(html) == html


def test_encode_leaves_body_text_alone():
    """본문 한글은 인코딩 대상이 아니다(URL 속성값만 바꾼다)."""
    html = '<p>여름 특가 안내</p><a href="/web/여름.html">보기</a>'
    encoded = store.encode_html_urls(html)
    assert "<p>여름 특가 안내</p>" in encoded
    assert 'href="/web/%EC%97%AC%EB%A6%84.html"' in encoded


def test_css_url_is_handled():
    html = "<style>.a{background:url(/web/upload/%ED%99%A9%ED%86%A0.png)}</style>"
    assert "황토.png" in store.decode_html_urls(html)
    assert store.encode_html_urls(store.decode_html_urls(html)) == html


def test_invalid_utf8_sequence_left_alone():
    """EUC-KR 등 UTF-8 이 아닌 이스케이프는 깨뜨리지 않고 그대로 둔다."""
    html = '<img src="/web/%C7%CF%B3%AA.gif">'
    assert store.decode_html_urls(html) == html


def test_fingerprint_detects_change():
    a = store.fingerprint("<p>A</p>")
    assert a == store.fingerprint("<p>A</p>")
    assert a != store.fingerprint("<p>B</p>")
    assert len(a) == 32


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