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


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