king
8125a59366
- 옵션 1개 상품은 한 행에 썸네일·옵션값 이름·품목코드(복사)·자체코드· 추가금액·진열·판매를 두고 「저장」 한 번으로 옵션 PUT → 품목 PUT. - 옵션 없는 상품: 행 추가/삭제 후 저장 → 옵션 생성 → 카페24가 부여한 품목코드를 재시도 조회(products.wait_for_variants)로 받아 자체코드· 추가금액·썸네일까지 이어서 반영. - 순서 드래그는 품목 display_order 로 저장(옵션값 재배열 PUT 은 위치 짝맞춤 때문에 품목코드↔이름이 뒤바뀔 수 있어 사용하지 않음). - 「옵션값 불러오기」: 탭/공백 구분 텍스트(이름·판매가·추가금액·자체코드) 붙여넣기 → 행 자동 채움. 기존 옵션은 이름으로 짝지어 코드/금액만. - 옵션 2개 이상(조합) 상품은 기존 2열 화면 유지. - 모달 JS 를 app/static/cafe24-options.js 로 분리. 유닛테스트 90 통과, 통합 하네스 통과, 헤드리스 크롬 렌더 확인. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
1181 lines
49 KiB
Python
1181 lines
49 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"}}
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 디자인 보관함 FTP — 모바일 스와이프 / PC·모바일 상품상세 템플릿
|
|
# ════════════════════════════════════════════════════════════
|
|
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="")
|
|
assert cfg.configured is False
|
|
assert cfg.missing == ["CAFE24_FTP_HOST", "CAFE24_FTP_USER", "CAFE24_FTP_PASSWORD"]
|
|
full = cfgmod.Cafe24FtpConfig(host="h", port=21, user="u", password="p")
|
|
assert full.configured is True
|
|
assert full.missing == []
|
|
|
|
|
|
def test_design_file_specs_cover_all_three_files():
|
|
"""상단 탭 3개(모바일 스와이프/모바일 상품상세/PC 상품상세)가 모두 등록돼 있는가."""
|
|
assert set(cfgmod.DESIGN_FILE_SPECS) == {"swiper", "mobile_detail", "pc_detail"}
|
|
assert cfgmod.design_file_label("swiper") == "모바일 스와이프"
|
|
assert cfgmod.design_file_label("mobile_detail") == "모바일 상품상세"
|
|
assert cfgmod.design_file_label("pc_detail") == "PC 상품상세"
|
|
# 기본 경로(env 미설정 시) — 실물 확인된 값
|
|
assert cfgmod.design_file_path("mobile_detail") == "/sde_design/mobile11/product/detail.html"
|
|
assert cfgmod.design_file_path("pc_detail") == "/sde_design/skin11/product/detail.html"
|
|
|
|
|
|
def test_design_file_path_env_override():
|
|
saved = os.environ.get("CAFE24_PC_DETAIL_FTP_PATH")
|
|
os.environ["CAFE24_PC_DETAIL_FTP_PATH"] = "/sde_design/skin99/product/detail.html"
|
|
try:
|
|
assert cfgmod.design_file_path("pc_detail") == "/sde_design/skin99/product/detail.html"
|
|
finally:
|
|
if saved is None:
|
|
os.environ.pop("CAFE24_PC_DETAIL_FTP_PATH", None)
|
|
else:
|
|
os.environ["CAFE24_PC_DETAIL_FTP_PATH"] = saved
|
|
|
|
|
|
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
|
|
|
|
path = "/sde_design/mobile11/product-swiper/product-swiper.js"
|
|
_FakeFTP.files = {path: "old".encode("utf-8")}
|
|
saved = ftplib.FTP
|
|
ftplib.FTP = _FakeFTP
|
|
try:
|
|
cfg = cfgmod.Cafe24FtpConfig(host="h", port=21, user="u", password="p")
|
|
content = design_ftp.read_text_file(cfg, path)
|
|
assert content == "old"
|
|
|
|
design_ftp.write_text_file(cfg, path, "new content")
|
|
assert design_ftp.read_text_file(cfg, 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")
|
|
try:
|
|
design_ftp.read_text_file(cfg, "/missing.js")
|
|
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 save_write_snapshot(self, product_no, section, data):
|
|
self.snapshots = getattr(self, "snapshots", [])
|
|
self.snapshots.append((product_no, section, data))
|
|
|
|
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
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 카페24 읽기 지연 보정 — 마지막 쓰기가 권위
|
|
# ════════════════════════════════════════════════════════════
|
|
def test_resolve_description_no_recent_write_uses_cafe24():
|
|
html, state = store.resolve_description("cafe", last_write=None, known_digests=set(), grace_minutes=60)
|
|
assert (html, state) == ("cafe", store.SYNC_NONE)
|
|
|
|
|
|
def _write(html, minutes_ago=1):
|
|
return {"html_content": html, "created_at": now_kst() - timedelta(minutes=minutes_ago)}
|
|
|
|
|
|
def test_resolve_description_synced_when_cafe24_caught_up():
|
|
html, state = store.resolve_description(
|
|
"NEW", last_write=_write("NEW"), known_digests={store.content_digest("OLD")}, grace_minutes=60
|
|
)
|
|
assert (html, state) == ("NEW", store.SYNC_SYNCED)
|
|
|
|
|
|
def test_resolve_description_pending_when_cafe24_returns_known_old_value():
|
|
# 카페24가 아직 적용 직전 값(BACKUP 으로 남긴 값)을 돌려준다 → 마지막 쓰기를 보여준다
|
|
digests = {store.content_digest("OLD"), store.content_digest("NEW")}
|
|
html, state = store.resolve_description("OLD", last_write=_write("NEW"), known_digests=digests, grace_minutes=60)
|
|
assert (html, state) == ("NEW", store.SYNC_PENDING)
|
|
|
|
|
|
def test_resolve_description_pending_for_older_known_value_too():
|
|
# 연속 두 번 적용(A→B→C) 뒤 카페24가 A 를 돌려줘도 '아는 값'이므로 지연으로 본다
|
|
digests = {store.content_digest(v) for v in ("A", "B", "C")}
|
|
html, state = store.resolve_description("A", last_write=_write("C"), known_digests=digests, grace_minutes=60)
|
|
assert (html, state) == ("C", store.SYNC_PENDING)
|
|
|
|
|
|
def test_resolve_description_external_when_unknown_value():
|
|
# 관리자에서 직접 고친 값 — 카페24를 믿는다
|
|
digests = {store.content_digest("OLD"), store.content_digest("NEW")}
|
|
html, state = store.resolve_description("HAND", last_write=_write("NEW"), known_digests=digests, grace_minutes=60)
|
|
assert (html, state) == ("HAND", store.SYNC_EXTERNAL)
|
|
|
|
|
|
def test_resolve_description_grace_expired_trusts_cafe24():
|
|
digests = {store.content_digest("OLD"), store.content_digest("NEW")}
|
|
html, state = store.resolve_description(
|
|
"OLD", last_write=_write("NEW", minutes_ago=500), known_digests=digests, grace_minutes=60
|
|
)
|
|
assert (html, state) == ("OLD", store.SYNC_NONE)
|
|
|
|
|
|
def test_resolve_description_accepts_iso_created_at():
|
|
lw = {"html_content": "NEW", "created_at": (now_kst() - timedelta(minutes=2)).isoformat()}
|
|
html, state = store.resolve_description("OLD", last_write=lw, known_digests={store.content_digest("OLD")}, grace_minutes=60)
|
|
assert (html, state) == ("NEW", store.SYNC_PENDING)
|
|
|
|
|
|
def test_parse_grace_minutes():
|
|
assert store.parse_grace_minutes("30") == 30
|
|
assert store.parse_grace_minutes("") == store.DEFAULT_READ_LAG_GRACE_MIN
|
|
assert store.parse_grace_minutes("-5") == store.DEFAULT_READ_LAG_GRACE_MIN
|
|
assert store.parse_grace_minutes("abc") == store.DEFAULT_READ_LAG_GRACE_MIN
|
|
|
|
|
|
def test_overlay_recent_write_uses_snapshot_when_get_is_older():
|
|
fetched = {"product_name": "OLD", "price": "1000.00", "updated_date": "2026-09-18T10:00:00+09:00",
|
|
"description": "keep"}
|
|
snap = {"product_name": "NEW", "price": "2000.00", "updated_date": "2026-09-18T10:05:00+09:00"}
|
|
merged, stale = store.overlay_recent_write(fetched, snapshot=snap, written_at=now_kst(), grace_minutes=60)
|
|
assert stale is True
|
|
assert merged["product_name"] == "NEW" and merged["price"] == "2000.00"
|
|
assert merged["description"] == "keep" # 스냅샷에 없는 필드는 그대로
|
|
assert merged["updated_date"] == snap["updated_date"]
|
|
|
|
|
|
def test_overlay_recent_write_keeps_cafe24_when_caught_up():
|
|
fetched = {"product_name": "NEWER", "updated_date": "2026-09-18T10:06:00+09:00"}
|
|
snap = {"product_name": "NEW", "updated_date": "2026-09-18T10:05:00+09:00"}
|
|
merged, stale = store.overlay_recent_write(fetched, snapshot=snap, written_at=now_kst(), grace_minutes=60)
|
|
assert stale is False and merged is fetched
|
|
|
|
|
|
def test_overlay_recent_write_skips_without_dates_or_snapshot():
|
|
fetched = {"product_name": "X"}
|
|
assert store.overlay_recent_write(fetched, snapshot=None, written_at=now_kst(), grace_minutes=60) == (fetched, False)
|
|
snap = {"product_name": "NEW"}
|
|
assert store.overlay_recent_write(fetched, snapshot=snap, written_at=now_kst(), grace_minutes=60) == (fetched, False)
|
|
|
|
|
|
def test_overlay_recent_write_respects_grace():
|
|
fetched = {"product_name": "OLD", "updated_date": "2026-09-18T10:00:00+09:00"}
|
|
snap = {"product_name": "NEW", "updated_date": "2026-09-18T10:05:00+09:00"}
|
|
old = (now_kst() - timedelta(hours=10)).isoformat()
|
|
assert store.overlay_recent_write(fetched, snapshot=snap, written_at=old, grace_minutes=60) == (fetched, False)
|
|
|
|
|
|
def test_product_snapshot_keeps_only_scalar_fields():
|
|
snap = store.product_snapshot({"product_name": "A", "price": "1.00", "description": "<big>", "updated_date": None, "display": "T"})
|
|
assert snap == {"product_name": "A", "price": "1.00", "display": "T"}
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 기본 정보 · 옵션 · 품목 입력 검증
|
|
# ════════════════════════════════════════════════════════════
|
|
def test_parse_price():
|
|
assert store.parse_price("6,900") == "6900.00"
|
|
assert store.parse_price("6900.00") == "6900.00"
|
|
assert store.parse_price(6900) == "6900.00"
|
|
assert store.parse_price("") is None
|
|
assert store.parse_price(None) is None
|
|
for bad in ("abc", "-1"):
|
|
try:
|
|
store.parse_price(bad)
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
raise AssertionError(bad)
|
|
assert store.price_equal("6900.00", "6900") and not store.price_equal("6900.00", "7000")
|
|
|
|
|
|
def test_update_payload_price_and_image():
|
|
payload = products.build_update_payload(price="1000.00", supply_price="500.00", detail_image="/web/x.jpg")
|
|
assert payload["request"] == {
|
|
"price": "1000.00", "supply_price": "500.00", "detail_image": "/web/x.jpg", "image_upload_type": "A",
|
|
}
|
|
|
|
|
|
def test_parse_option_values():
|
|
assert store.parse_option_values("빨강, 파랑\n노랑,, 빨강 ") == ["빨강", "파랑", "노랑"]
|
|
|
|
|
|
def test_build_create_options_request():
|
|
body = store.build_create_options_request("색상", ["빨강", "파랑"], display_type="p")
|
|
assert body["has_option"] == "T" and body["option_type"] == "T"
|
|
assert body["options"][0]["option_display_type"] == "P"
|
|
assert [v["option_text"] for v in body["options"][0]["option_value"]] == ["빨강", "파랑"]
|
|
for name, values in (("", ["a"]), ("x", [])):
|
|
try:
|
|
store.build_create_options_request(name, values)
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
raise AssertionError((name, values))
|
|
|
|
|
|
def test_build_update_options_request_pairs_original_and_new():
|
|
original = [{"option_code": "O1", "option_name": "Color", "option_display_type": "S",
|
|
"option_value": [{"option_text": "Black", "value_no": 1}, {"option_text": "Red", "value_no": 2}]}]
|
|
edited = [{"option_name": "Colors", "option_display_type": "P",
|
|
"option_value": [{"option_text": "Jet Black", "option_image_file": "https://d/x.png"},
|
|
{"option_text": "Deep Red"}]}]
|
|
body = store.build_update_options_request(original, edited, option_list_type="S")
|
|
assert body["option_list_type"] == "S"
|
|
assert body["original_options"] == [{"option_code": "O1", "option_name": "Color",
|
|
"option_value": [{"option_text": "Black", "value_no": 1}, {"option_text": "Red", "value_no": 2}]}]
|
|
assert body["options"][0]["option_name"] == "Colors"
|
|
assert body["options"][0]["option_display_type"] == "P"
|
|
assert body["options"][0]["option_value"][0] == {"option_text": "Jet Black", "value_no": 1, "option_image_file": "https://d/x.png"}
|
|
assert body["options"][0]["option_value"][1] == {"option_text": "Deep Red", "value_no": 2}
|
|
|
|
|
|
def test_build_update_options_request_rejects_count_mismatch():
|
|
original = [{"option_name": "A", "option_value": [{"option_text": "1"}]}]
|
|
for edited in ([], [{"option_name": "A", "option_value": []}], [{"option_name": "", "option_value": [{"option_text": "1"}]}]):
|
|
try:
|
|
store.build_update_options_request(original, edited)
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
raise AssertionError(edited)
|
|
|
|
|
|
def test_build_variant_updates():
|
|
rows = [
|
|
{"variant_code": "p000000r000a", "custom_variant_code": " ABC ", "additional_amount": "1,000", "display": "on"},
|
|
{"variant_code": "P000000R000B"}, # 바뀐 것 없음 → 제외
|
|
{"variant_code": "P000000R000C", "selling": "off", "additional_amount": ""},
|
|
]
|
|
out = store.build_variant_updates(rows)
|
|
assert out == [
|
|
{"variant_code": "P000000R000A", "custom_variant_code": "ABC", "additional_amount": "1000.00", "display": "T"},
|
|
{"variant_code": "P000000R000C", "selling": "F"},
|
|
]
|
|
try:
|
|
store.build_variant_updates([{"variant_code": "bad"}])
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
raise AssertionError("bad code accepted")
|
|
# 드래그 정렬 → display_order (1~300)
|
|
assert store.build_variant_updates([{"variant_code": "P000000R000A", "display_order": "3"}]) == [
|
|
{"variant_code": "P000000R000A", "display_order": 3}
|
|
]
|
|
for bad in ("0", "301", "x"):
|
|
try:
|
|
store.build_variant_updates([{"variant_code": "P000000R000A", "display_order": bad}])
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
raise AssertionError(bad)
|
|
|
|
|
|
def test_parse_option_values_accepts_list():
|
|
assert store.parse_option_values(["빨강, 큰 것", " 파랑 ", "", "빨강, 큰 것"]) == ["빨강, 큰 것", "파랑"]
|
|
|
|
|
|
def test_wait_for_variants_retries_until_expected():
|
|
calls = {"n": 0}
|
|
|
|
class _Client:
|
|
def get(self, path, **kw):
|
|
calls["n"] += 1
|
|
return {"variants": [{"variant_code": "A"}] * (2 if calls["n"] >= 3 else 1)}
|
|
|
|
out = products.wait_for_variants(_Client(), 7, 2, attempts=5)
|
|
assert len(out) == 2 and calls["n"] == 3
|
|
|
|
|
|
class _RouteClient:
|
|
"""경로별 응답을 돌려주는 가짜 클라이언트 (get/put/post/delete)."""
|
|
|
|
def __init__(self, routes):
|
|
self.routes = routes
|
|
self.calls: list[tuple] = []
|
|
|
|
def _call(self, method, path, json=None, **_kw):
|
|
self.calls.append((method, path, json))
|
|
return self.routes.get(f"{method} {path}", {})
|
|
|
|
def get(self, path, **kw):
|
|
return self._call("GET", path, **kw)
|
|
|
|
def put(self, path, **kw):
|
|
return self._call("PUT", path, **kw)
|
|
|
|
def post(self, path, **kw):
|
|
return self._call("POST", path, **kw)
|
|
|
|
def delete(self, path, **kw):
|
|
return self._call("DELETE", path, **kw)
|
|
|
|
|
|
def test_upload_images_and_variants_wrappers():
|
|
client = _RouteClient({
|
|
"POST /admin/products/images": {"images": [{"path": "https://d/a.png"}, {"path": "https://d/b.png"}]},
|
|
"PUT /admin/products/7/variants": {"variants": [{"variant_code": "P000000R000A", "display": "F"}]},
|
|
"GET /admin/products/7/variants": {"variants": [{"variant_code": "P000000R000A"}]},
|
|
"GET /admin/products/7/options": {"option": {"has_option": "T", "options": []}},
|
|
"DELETE /admin/products/7/options": {"option": {"product_no": 7}},
|
|
})
|
|
assert products.upload_images(client, ["AAA", "BBB"]) == ["https://d/a.png", "https://d/b.png"]
|
|
assert products.upload_image_bytes(client, b"\x89PNG") == "https://d/a.png"
|
|
assert products.list_variants(client, 7) == [{"variant_code": "P000000R000A"}]
|
|
assert products.get_options(client, 7)["has_option"] == "T"
|
|
assert products.update_variants(client, 7, [{"variant_code": "P000000R000A", "display": "F"}]) == [
|
|
{"variant_code": "P000000R000A", "display": "F"}
|
|
]
|
|
sent = [c for c in client.calls if c[0] == "PUT"][-1]
|
|
assert sent[2]["requests"][0]["variant_code"] == "P000000R000A"
|
|
products.delete_options(client, 7)
|
|
assert client.calls[-1][0] == "DELETE"
|
|
|
|
|
|
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()
|