king
93037411fb
- product-swiper.js는 상품이 아니라 카페24 "디자인 보관함" 스킨 파일이라
Admin API(OAuth)로는 접근 불가 — 실물 확인(스크린샷) 결과 디자인 보관함
FTP 계정(호스트/포트/ID/PW, OAuth와 별개)으로만 읽기/쓰기 가능.
app/integrations/cafe24/design_ftp.py 를 표준 ftplib 로 새로 추가.
- 상단 탭에 "모바일 스와이프" 버튼 추가. 편집 화면(swiper.html)은 상세페이지
편집기(_editor.html)와 완전히 같은 문법강조·색상·단축키 JS를 그대로 옮겨
씀(요청사항) — 상품 전용 UI(목록·진열/판매·예약)는 제외.
- 적용 순서도 상세페이지와 동일한 원칙: FTP에서 현재값 재조회 → BACKUP →
지문 대조(충돌 거부) → FTP 쓰기 → MANUAL 버전 + 감사로그(apply_swiper).
textarea의 CRLF는 적용 전 LF로 정규화(안 하면 매번 "변경됨"으로 오판).
- 버전 이력은 cafe24_product_revisions(product_no NOT NULL)에 넣을 수 없어
새 테이블 cafe24_swiper_revisions 추가(scripts/sql/cafe24_db_003_*.sql,
서버에서 별도 실행 필요).
- 신규 env: CAFE24_FTP_HOST(미설정 시 {mall_id}.ftp.cafe24.com)/PORT/USER/
PASSWORD, CAFE24_SWIPER_FTP_PATH. .env.example·문서 갱신.
- 유닛테스트 4건 추가(FTP config 기본값, 가짜 FTP로 읽기/쓰기 왕복·오류 처리).
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
898 lines
36 KiB
Python
898 lines
36 KiB
Python
"""카페24 모듈 순수 로직 + 토큰/암호화 테스트.
|
|
|
|
DB/네트워크 없이 검증한다(가짜 저장소 + refresh 함수 주입).
|
|
python -m app.modules.cafe24.tests.test_cafe24
|
|
또는 pytest 로 실행 가능.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ftplib
|
|
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 Cafe24ApiError, Cafe24AuthError, Cafe24ConfigError
|
|
from app.modules.cafe24 import store, worker
|
|
from app.timezone import now_kst
|
|
|
|
SECRET = "unit-test-secret"
|
|
|
|
# wait_for_description 은 실물 카페24 API 의 쓰기 직후 읽기 지연을 흡수하려고
|
|
# 실제로 sleep 한다. 가짜 클라이언트는 그 지연을 재현하지 않으므로(항상 같은
|
|
# 값을 돌려줌) 재시도 예산을 다 채우게 되는데, 테스트에서까지 그 시간을 그대로
|
|
# 기다릴 필요는 없다.
|
|
products.time.sleep = lambda *_a, **_k: None
|
|
|
|
_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"
|
|
# 요청 scope 는 DEFAULT_SCOPES 한 곳에서 나온다(authorize 와 클라이언트가
|
|
# 어긋나지 않게). 늘어날 때 이 테스트도 함께 갱신할 것.
|
|
assert config.scope_param == ",".join(cfgmod.DEFAULT_SCOPES)
|
|
assert "mall.read_product" in config.scope_param
|
|
assert "mall.write_product" in config.scope_param
|
|
|
|
|
|
def test_product_url_uses_shop_url_env():
|
|
"""다이렉트 주소는 CAFE24_SHOP_URL 기준. 스킴·끝 슬래시가 없어도 맞춘다."""
|
|
saved = os.environ.get("CAFE24_SHOP_URL")
|
|
try:
|
|
for value in ("https://miras.co.kr", "miras.co.kr", "https://miras.co.kr/"):
|
|
os.environ["CAFE24_SHOP_URL"] = value
|
|
config = _config()
|
|
assert (
|
|
config.product_url(119)
|
|
== "https://miras.co.kr/product/detail.html?product_no=119"
|
|
), value
|
|
finally:
|
|
if saved is None:
|
|
os.environ.pop("CAFE24_SHOP_URL", None)
|
|
else:
|
|
os.environ["CAFE24_SHOP_URL"] = saved
|
|
|
|
|
|
def test_product_url_falls_back_to_cafe24_domain():
|
|
"""CAFE24_SHOP_URL 이 없어도 항상 유효한 주소가 나와야 한다."""
|
|
saved = os.environ.pop("CAFE24_SHOP_URL", None)
|
|
try:
|
|
assert _config().product_url(119) == (
|
|
"https://testmall.cafe24.com/product/detail.html?product_no=119"
|
|
)
|
|
finally:
|
|
if saved is not None:
|
|
os.environ["CAFE24_SHOP_URL"] = saved
|
|
|
|
|
|
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"
|
|
# mobile_description 은 절대 보내지 않는다 — 보내면 카페24가 모바일 상세설명
|
|
# 설정을 "직접 등록"으로 바꿔버린다(실물 확인). separated_mobile_description="F"
|
|
# 만 지정해 "PC 상세설명과 동일"을 강제한다.
|
|
assert call["json"] == {"request": {"description": "<p>NEW</p>", "separated_mobile_description": "F"}}
|
|
|
|
|
|
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
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 소스 정리(포맷) — 렌더링을 바꾸지 않는 것이 최우선
|
|
# ════════════════════════════════════════════════════════════
|
|
_MESSY = (
|
|
'<style>\n\t/* 주석 */\n\t.v{max-width:100%}\n</style>'
|
|
'<div class="wrap"><p>안녕<span>하세요</span> 여름 특가</p>'
|
|
'<img src="/web/a.gif"><img src="/web/b.gif">'
|
|
"<table><tr><td>1</td><td>2</td></tr></table></div>"
|
|
)
|
|
|
|
|
|
def test_format_breaks_block_tags():
|
|
out = store.format_html(_MESSY)
|
|
lines = out.split("\n")
|
|
assert '<div class="wrap">' in lines
|
|
assert "</div>" in lines
|
|
# 블록 안쪽은 들여쓴다.
|
|
assert any(line.startswith(" <table>") for line in lines)
|
|
assert any(line.startswith(" <td>") for line in lines)
|
|
|
|
|
|
def test_format_keeps_inline_elements_together():
|
|
"""이미지 사이에 줄바꿈이 들어가면 화면에 공백이 생긴다 — 붙여둬야 한다."""
|
|
out = store.format_html(_MESSY)
|
|
assert '<img src="/web/a.gif"><img src="/web/b.gif">' in out
|
|
assert "<p>안녕<span>하세요</span> 여름 특가</p>" in out
|
|
|
|
|
|
def test_format_preserves_style_content_verbatim():
|
|
out = store.format_html(_MESSY)
|
|
assert "\t/* 주석 */" in out
|
|
assert "\t.v{max-width:100%}" in out
|
|
|
|
|
|
def test_format_is_idempotent():
|
|
"""편집하지 않고 다시 적용해도 저장값이 계속 바뀌면 안 된다."""
|
|
once = store.format_html(_MESSY)
|
|
assert store.format_html(once) == once
|
|
assert store.format_html(store.format_html(once)) == once
|
|
|
|
|
|
_REAL_DETAIL = """<div style="width: 1000px; margin: 0 auto;">
|
|
|
|
<img src="../img/promo/dadamam_detail1.jpg">
|
|
<img src="../img/promo/dadamam_detail2.jpg">
|
|
|
|
<!-- 대파_타임랩스----------------><img contenteditable="false" src="../img/gif/NEW_1.gif">
|
|
<img contenteditable="false" src="../img/2+1/2+1_02.jpg">
|
|
</div>"""
|
|
|
|
|
|
def test_format_indents_every_line_of_a_run():
|
|
"""원문 줄바꿈을 살리고 **모든 줄**을 들여쓴다.
|
|
|
|
예전에는 첫 줄만 들여쓰고 나머지가 1열에 붙어 나왔다.
|
|
"""
|
|
lines = store.format_html(_REAL_DETAIL).split("\n")
|
|
img_lines = [line for line in lines if "<img" in line]
|
|
assert len(img_lines) == 4, img_lines
|
|
assert all(line.startswith(" <") for line in img_lines), img_lines
|
|
|
|
|
|
def test_format_keeps_comment_with_its_element():
|
|
"""`<!-- 라벨 --><img>` 는 붙여둔다 — 나누면 라벨과 대상이 떨어진다."""
|
|
out = store.format_html(_REAL_DETAIL)
|
|
assert "<!-- 대파_타임랩스----------------><img contenteditable=" in out
|
|
|
|
|
|
def test_format_keeps_single_blank_line():
|
|
"""구획용 빈 줄은 한 줄까지 유지한다(여러 줄은 하나로)."""
|
|
out = store.format_html("<div>\n\n\n<img src=\"a.gif\">\n\n\n<img src=\"b.gif\">\n</div>")
|
|
assert "\n\n" in out
|
|
assert "\n\n\n" not in out
|
|
|
|
|
|
def test_format_real_detail_is_idempotent():
|
|
once = store.format_html(_REAL_DETAIL)
|
|
assert store.format_html(once) == once
|
|
|
|
|
|
def test_format_collapses_short_blocks():
|
|
assert store.format_html("<td>1</td>") == "<td>1</td>"
|
|
# 길면 나눈다.
|
|
long_text = "가" * 200
|
|
assert "\n" in store.format_html("<td>%s</td>" % long_text)
|
|
|
|
|
|
def test_format_survives_broken_html():
|
|
"""닫는 태그 누락·꺾쇠 조각이 있어도 예외 없이 뭔가를 돌려준다."""
|
|
for bad in ("<div><p>열고 안 닫음", "a < b 그리고 c > d", "<<>>", "<div", ""):
|
|
assert isinstance(store.format_html(bad), str)
|
|
|
|
|
|
def test_format_does_not_touch_urls():
|
|
"""포맷은 속성값을 건드리지 않는다(인코딩과 서로 간섭하지 않게)."""
|
|
html = '<div><img src="/web/%EC%9A%A9%EA%B8%B0(a)_1.gif"></div>'
|
|
assert "/web/%EC%9A%A9%EA%B8%B0(a)_1.gif" in store.format_html(html)
|
|
|
|
|
|
def test_format_then_encode_roundtrip():
|
|
"""화면 표시(디코딩+정리) → 저장(인코딩+정리) 순서에서 URL 이 원형을 지킨다."""
|
|
raw = store.format_html(_REAL_HTML)
|
|
shown = store.format_html(store.decode_html_urls(raw))
|
|
saved = store.format_html(store.encode_html_urls(shown))
|
|
assert saved == raw
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 예약 — 입력 검증
|
|
# ════════════════════════════════════════════════════════════
|
|
def test_parse_tristate():
|
|
assert store.parse_tristate("on") is True
|
|
assert store.parse_tristate("off") is False
|
|
for keep in ("", None, "keep", "이상한값"):
|
|
assert store.parse_tristate(keep) is None, keep
|
|
|
|
|
|
def test_parse_schedule_at_accepts_future_kst():
|
|
base = now_kst()
|
|
target = (base + timedelta(hours=3)).replace(second=0, microsecond=0)
|
|
parsed = store.parse_schedule_at(target.strftime("%Y-%m-%dT%H:%M"), now=base)
|
|
assert parsed.utcoffset() == timedelta(hours=9) # 타임존 표기 없는 입력을 KST 로 해석
|
|
assert parsed.hour == target.hour and parsed.minute == target.minute
|
|
|
|
|
|
def test_parse_schedule_at_rejects_past_and_garbage():
|
|
base = now_kst()
|
|
past = (base - timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M")
|
|
for bad in (past, "", "어제", "2026-13-45T99:99"):
|
|
try:
|
|
store.parse_schedule_at(bad, now=base)
|
|
except ValueError:
|
|
continue
|
|
raise AssertionError(f"{bad!r} 는 거부해야 한다")
|
|
|
|
|
|
def test_parse_schedule_at_allows_one_minute_grace():
|
|
"""`datetime-local` 은 초를 버린다 — 지금 이 분(分)을 고른 것을 거부하면 안 된다.
|
|
|
|
now 의 초를 고정해 실행 시각에 따라 결과가 달라지지 않게 한다(예전에 이 테스트가
|
|
초에 따라 실패했다).
|
|
"""
|
|
base = now_kst().replace(second=40, microsecond=0)
|
|
this_minute = base.strftime("%Y-%m-%dT%H:%M") # 초가 잘려 base 보다 40초 과거
|
|
assert store.parse_schedule_at(this_minute, now=base) is not None
|
|
|
|
|
|
def test_describe_schedule_action():
|
|
assert store.describe_schedule_action(
|
|
has_html=True, set_display=True, set_selling=False
|
|
) == "상세페이지 · 진열 · 판매중지"
|
|
assert store.describe_schedule_action(
|
|
has_html=False, set_display=None, set_selling=True
|
|
) == "판매"
|
|
assert store.describe_schedule_action(
|
|
has_html=False, set_display=None, set_selling=None
|
|
) == "없음"
|
|
|
|
|
|
def test_normalize_schedule_status():
|
|
assert store.normalize_schedule_status("success") == store.STATUS_SUCCESS
|
|
assert store.normalize_schedule_status("없는상태") == store.STATUS_PENDING
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 예약 — 상품 수정 payload (진열/판매 포함)
|
|
# ════════════════════════════════════════════════════════════
|
|
def test_update_payload_flags():
|
|
payload = products.build_update_payload(display=True, selling=False)
|
|
assert payload == {"request": {"display": "T", "selling": "F"}}
|
|
|
|
|
|
def test_update_payload_skips_none():
|
|
"""None 인 항목은 아예 보내지 않는다 = 그 필드를 건드리지 않는다."""
|
|
payload = products.build_update_payload(description="<p>x</p>")
|
|
assert payload["request"] == {"description": "<p>x</p>"}
|
|
|
|
|
|
def test_update_product_skips_empty_request():
|
|
"""바꿀 것이 없으면 API 를 호출하지 않는다."""
|
|
client = _FakeClient({"product": _PRODUCT})
|
|
assert products.update_product(client, 131) == {}
|
|
assert client.calls == []
|
|
|
|
|
|
def test_update_product_sends_flags_and_html():
|
|
client = _FakeClient({"product": _PRODUCT})
|
|
products.update_product(client, 131, description="<p>새</p>", display=False)
|
|
call = client.calls[0]
|
|
assert call["method"] == "PUT" and call["path"] == "/admin/products/131"
|
|
assert call["json"] == {"request": {"description": "<p>새</p>", "display": "F"}}
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 모바일 스와이프(product-swiper.js) — 디자인 보관함 FTP
|
|
# ════════════════════════════════════════════════════════════
|
|
def test_ftp_config_host_falls_back_to_mall_id():
|
|
"""CAFE24_FTP_HOST 미설정 시 카페24 규칙({mall_id}.ftp.cafe24.com)을 쓴다."""
|
|
saved = os.environ.pop("CAFE24_FTP_HOST", None)
|
|
try:
|
|
cfg = cfgmod.load_ftp_config(mall_id="testmall")
|
|
assert cfg.host == "testmall.ftp.cafe24.com"
|
|
assert cfg.port == cfgmod.DEFAULT_FTP_PORT
|
|
finally:
|
|
if saved is not None:
|
|
os.environ["CAFE24_FTP_HOST"] = saved
|
|
|
|
|
|
def test_ftp_config_missing_lists_absent_vars():
|
|
cfg = cfgmod.Cafe24FtpConfig(host="", port=21, user="", password="", swiper_path="")
|
|
assert cfg.configured is False
|
|
assert cfg.missing == [
|
|
"CAFE24_FTP_HOST", "CAFE24_FTP_USER", "CAFE24_FTP_PASSWORD", "CAFE24_SWIPER_FTP_PATH",
|
|
]
|
|
full = cfgmod.Cafe24FtpConfig(host="h", port=21, user="u", password="p", swiper_path="/x.js")
|
|
assert full.configured is True
|
|
assert full.missing == []
|
|
|
|
|
|
class _FakeFTP:
|
|
"""ftplib.FTP 대역 — 실제 소켓 없이 RETR/STOR 명령만 흉내낸다."""
|
|
|
|
files: dict[str, bytes] = {}
|
|
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def connect(self, host, port, timeout=None):
|
|
self.calls.append(("connect", host, port))
|
|
|
|
def login(self, user, password):
|
|
self.calls.append(("login", user, password))
|
|
|
|
def set_pasv(self, value):
|
|
pass
|
|
|
|
def retrbinary(self, cmd, callback):
|
|
path = cmd.split(" ", 1)[1]
|
|
if path not in self.files:
|
|
raise ftplib.error_perm("550 No such file")
|
|
callback(self.files[path])
|
|
|
|
def storbinary(self, cmd, fp):
|
|
path = cmd.split(" ", 1)[1]
|
|
self.files[path] = fp.read()
|
|
|
|
def quit(self):
|
|
self.calls.append(("quit",))
|
|
|
|
|
|
def test_design_ftp_read_and_write_roundtrip():
|
|
from app.integrations.cafe24 import design_ftp
|
|
|
|
_FakeFTP.files = {"/sde_design/mobile11/product-swiper/product-swiper.js": "old".encode("utf-8")}
|
|
saved = ftplib.FTP
|
|
ftplib.FTP = _FakeFTP
|
|
try:
|
|
cfg = cfgmod.Cafe24FtpConfig(
|
|
host="h", port=21, user="u", password="p",
|
|
swiper_path="/sde_design/mobile11/product-swiper/product-swiper.js",
|
|
)
|
|
content = design_ftp.read_text_file(cfg, cfg.swiper_path)
|
|
assert content == "old"
|
|
|
|
design_ftp.write_text_file(cfg, cfg.swiper_path, "new content")
|
|
assert design_ftp.read_text_file(cfg, cfg.swiper_path) == "new content"
|
|
finally:
|
|
ftplib.FTP = saved
|
|
|
|
|
|
def test_design_ftp_missing_file_raises_cafe24_error():
|
|
from app.integrations.cafe24 import design_ftp
|
|
from app.integrations.cafe24.errors import Cafe24FtpError
|
|
|
|
_FakeFTP.files = {}
|
|
saved = ftplib.FTP
|
|
ftplib.FTP = _FakeFTP
|
|
try:
|
|
cfg = cfgmod.Cafe24FtpConfig(host="h", port=21, user="u", password="p", swiper_path="/missing.js")
|
|
try:
|
|
design_ftp.read_text_file(cfg, cfg.swiper_path)
|
|
raise AssertionError("Cafe24FtpError 가 났어야 한다")
|
|
except Cafe24FtpError:
|
|
pass
|
|
finally:
|
|
ftplib.FTP = saved
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 예약 worker — 성공/재시도/최종실패
|
|
# ════════════════════════════════════════════════════════════
|
|
class _FakeStore:
|
|
"""worker 가 쓰는 저장소 메서드만 흉내낸다."""
|
|
|
|
def __init__(self, rows):
|
|
self.rows = list(rows)
|
|
self.revisions = {}
|
|
self.added = []
|
|
self.finished = []
|
|
self.audits = []
|
|
|
|
@contextmanager
|
|
def claim_due_schedule(self, *, now):
|
|
yield self.rows.pop(0) if self.rows else None
|
|
|
|
def get_revision(self, revision_id):
|
|
return self.revisions.get(int(revision_id), {})
|
|
|
|
def add_revision(self, **fields):
|
|
self.added.append(fields)
|
|
return 900 + len(self.added)
|
|
|
|
def finish_schedule(self, schedule_id, *, status, error="", next_retry_at=None, retry_count=None):
|
|
self.finished.append(
|
|
{"id": schedule_id, "status": status, "error": error,
|
|
"next_retry_at": next_retry_at, "retry_count": retry_count}
|
|
)
|
|
|
|
def log_audit(self, **fields):
|
|
self.audits.append(fields)
|
|
|
|
|
|
class _FakeApi:
|
|
def __init__(self, client):
|
|
self.client = client
|
|
|
|
|
|
class _WorkerClient(_FakeClient):
|
|
"""PUT 을 실패시킬 수 있는 클라이언트."""
|
|
|
|
def __init__(self, payload=None, fail_put=None):
|
|
super().__init__(payload)
|
|
self.fail_put = fail_put
|
|
|
|
def put(self, path, *, params=None, json=None, product_no=None):
|
|
if self.fail_put:
|
|
raise self.fail_put
|
|
return super().put(path, params=params, json=json, product_no=product_no)
|
|
|
|
|
|
def _schedule_row(**overrides):
|
|
row = {
|
|
"id": 7, "product_no": 131, "revision_id": 55,
|
|
"set_display": True, "set_selling": None, "retry_count": 0,
|
|
}
|
|
row.update(overrides)
|
|
return row
|
|
|
|
|
|
def test_worker_applies_html_and_flags():
|
|
st = _FakeStore([_schedule_row()])
|
|
st.revisions[55] = {"html_content": "<p>예약 내용</p>"}
|
|
client = _WorkerClient({"product": _PRODUCT})
|
|
assert worker.process_once(st, _FakeApi(client)) == 1
|
|
|
|
# 쓰기 직전 현재값을 읽어 BACKUP 을 남겼는가
|
|
assert any(r["revision_type"] == store.REVISION_BACKUP for r in st.added)
|
|
# HTML 과 진열 상태를 한 번의 PUT 으로 보냈는가
|
|
put = [c for c in client.calls if c["method"] == "PUT"][0]
|
|
assert put["json"]["request"]["description"] == "<p>예약 내용</p>"
|
|
# mobile_description 은 보내지 않는다 — separated_mobile_description="F" 로
|
|
# "PC 상세설명과 동일"을 강제한다(직접 보내면 "직접 등록"으로 바뀌어버린다).
|
|
assert "mobile_description" not in put["json"]["request"]
|
|
assert put["json"]["request"]["separated_mobile_description"] == "F"
|
|
assert put["json"]["request"]["display"] == "T"
|
|
assert "selling" not in put["json"]["request"] # 변경 없음이면 보내지 않는다
|
|
assert st.finished == [
|
|
{"id": 7, "status": store.STATUS_SUCCESS, "error": "",
|
|
"next_retry_at": None, "retry_count": None}
|
|
]
|
|
|
|
|
|
def test_worker_flags_only_skips_backup():
|
|
"""HTML 없이 진열/판매만 바꾸는 예약은 상세설명을 읽거나 백업하지 않는다."""
|
|
st = _FakeStore([_schedule_row(revision_id=None, set_selling=False)])
|
|
client = _WorkerClient({"product": _PRODUCT})
|
|
assert worker.process_once(st, _FakeApi(client)) == 1
|
|
assert st.added == [] # 백업 없음
|
|
put = [c for c in client.calls if c["method"] == "PUT"][0]
|
|
assert "description" not in put["json"]["request"]
|
|
assert put["json"]["request"] == {"display": "T", "selling": "F"}
|
|
|
|
|
|
def test_worker_retries_then_fails():
|
|
"""실패는 재시도 예산 안에서 다시 시도하고, 소진되면 FAILED 로 확정한다."""
|
|
boom = Cafe24ApiError("서버 오류", status=500)
|
|
|
|
st = _FakeStore([_schedule_row(retry_count=0)])
|
|
st.revisions[55] = {"html_content": "<p>x</p>"}
|
|
worker.process_once(st, _FakeApi(_WorkerClient({"product": _PRODUCT}, fail_put=boom)))
|
|
first = st.finished[0]
|
|
assert first["status"] == store.STATUS_PENDING # 다시 대기로
|
|
assert first["retry_count"] == 1
|
|
assert first["next_retry_at"] is not None
|
|
|
|
st2 = _FakeStore([_schedule_row(retry_count=store.MAX_RETRY)])
|
|
st2.revisions[55] = {"html_content": "<p>x</p>"}
|
|
worker.process_once(st2, _FakeApi(_WorkerClient({"product": _PRODUCT}, fail_put=boom)))
|
|
assert st2.finished[0]["status"] == store.STATUS_FAILED
|
|
assert any(a["result"] == "FAIL" for a in st2.audits)
|
|
|
|
|
|
def test_worker_missing_revision_is_failure_not_crash():
|
|
st = _FakeStore([_schedule_row(revision_id=999, retry_count=store.MAX_RETRY)])
|
|
worker.process_once(st, _FakeApi(_WorkerClient({"product": _PRODUCT})))
|
|
assert st.finished[0]["status"] == store.STATUS_FAILED
|
|
assert "999" in st.finished[0]["error"]
|
|
|
|
|
|
def test_worker_stops_when_nothing_due():
|
|
st = _FakeStore([])
|
|
assert worker.process_once(st, _FakeApi(_WorkerClient())) == 0
|
|
|
|
|
|
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()
|