king
7a2e933c16
1) 한글 파일명이 %EC%9A%A9… 으로 보이던 문제
카페24는 상세페이지 안 이미지 경로를 퍼센트 인코딩해서 저장한다. 화면에서는
읽을 수 없으므로 store.decode_html_urls 로 풀어 보여주고, 저장할 때
encode_html_urls 로 되돌린다. 두 함수는 서로의 역이며 왕복이 보존된다 —
편집하지 않고 적용해도 카페24 저장값이 한 바이트도 달라지지 않는다.
깨뜨리지 않기 위한 두 가지 제약을 뒀다. 디코딩은 non-ASCII(%80~%FF)만 한다.
%20·%3C 를 풀면 URL·HTML 구조가 깨진다. 인코딩은 src/href/poster/data-src 와
CSS url() 안의 값만 한다. 본문 한글 텍스트를 인코딩하면 페이지가 망가진다.
UTF-8 로 해석되지 않는 이스케이프(EUC-KR 등)는 건드리지 않고 그대로 둔다.
2) 편집 후 적용
POST /cafe24/products/{no}/apply 는 이 순서를 지킨다.
카페24 현재값 재조회 → BACKUP revision → 지문 대조 → PUT → MANUAL revision
현재값을 다시 읽는 것은 로컬 DB 의 마지막 버전이 지금 카페24에 올라간 값이라고
믿을 수 없기 때문이다(관리자 페이지에서 직접 고쳤을 수 있다). 지문(sha256 앞
32자)은 편집 중 남이 바꾼 내용을 조용히 덮어쓰는 것을 막는 낙관적 잠금이다.
미분리 상품(separated_mobile_description='F')은 모바일 필드도 같은 HTML 로
함께 쓴다. PC 만 바꾸면 모바일 상세가 어긋난다. 분리 상품은 모바일을 건드리지
않고 화면에 별도 반영 안내를 띄운다.
빈 내용은 거부한다(상세페이지를 통째로 날리는 실수 방지). 변경이 없으면 API 를
호출하지 않는다. 실패 시에도 BACKUP 은 남아 있으므로 오류 메시지에 버전 번호를
알려준다. 편집 중 페이지 이탈 경고도 넣었다.
버전 이력 표를 상세 화면에 붙였다(목록 조회는 html_content 를 제외하고 길이만
계산한다 — 수 MB 가 될 수 있다). 버전 선택 복원은 Phase 6.
검증: 유닛테스트 30개 통과(신규 7개 — 실제 파일명으로 왕복 동일성, ASCII
이스케이프 미변환, 본문 한글 보존, CSS url(), 잘못된 UTF-8 무시, 지문).
실제 쓰기(PUT)는 서버 배포 후 테스트 상품 1건으로 확인 필요.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
409 lines
16 KiB
Python
409 lines
16 KiB
Python
"""카페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"] == "락"
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 이미지 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()
|