o2o-negosium-original/agent/tests/test_scripts_resources.py

"""대화 스크립트 리소스 검증 (Chat_server 구조 참고 적용 + KT 중립화).

1. 재협상/재견적 스크립트 step 구조 보존 (핵심 step 키 존재, next_step/모드 형태).
2. 와일드카드(wild_card_1pct/budget) 존재 + 병합.
3. 클린룸 가드: 'kt'/'commerce'/'커머스'/'Nego-Wiz' 등 특정사 표현이 남아있지 않음.
4. 브랜드 치환: {company_name}/{service_name} 가 테넌트별 값으로 치환.
5. 변수 치환: {input_price} 등 협상 변수 치환, 누락 변수는 원형 유지.
"""

import json
import os
import re

import pytest

from negotiation.chat.service.script_repository import ScriptRepository
from tenancy.config_loader import TenantConfigLoader

_TENANTS_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "tenants")
_FORBIDDEN = re.compile(r"kt\s*commerce|케이티|커머스|nego-?wiz", re.IGNORECASE)


def _repo(tenant_id="ktcommerce"):
    cfg = TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0).load(tenant_id)
    return ScriptRepository(cfg, _TENANTS_DIR)


def test_renegotiation_structure_preserved():
    s = _repo().load_scripts("재협상")
    for key in ["서비스안내", "담당자확인", "협상품목안내", "기존가격제시", "가격협상_확인", "협상완료", "협상실패", "협상종료"]:
        assert key in s, f"missing step {key}"
    # 조건 분기 보존 (가격협상_확인 예 → 조건 리스트)
    yes = s["가격협상_확인"]["next_step"]["예"]
    conds = {c["condition"] for c in yes}
    assert {"check_wildcard_entry", "check_iteration_limit", "default"} <= conds
    # 담당자확인 yes/no 분기
    assert s["담당자확인"]["next_step"] == {"예": "협상품목안내", "아니오": "담당자확인_아니오"}


def test_requote_structure_preserved():
    s = _repo().load_scripts("재견적")
    for key in ["서비스안내", "가격제안", "배송형태선택", "가격협상_확인", "결과안내", "결과제출", "협상종료"]:
        assert key in s
    assert s["배송형태선택"]["next_input_mode"] == "delivery_type"
    assert s["배송형태선택"]["input_options"] == ["협력사배송", "지정택배배송", "픽업배송"]


def test_wildcard_present_and_merged():
    repo = _repo()
    wc = repo.wildcard_scripts()
    assert "wild_card_1pct" in wc and "wild_card_budget" in wc
    # 재협상 흐름에 병합됨
    merged = repo.load_scripts("재협상")
    assert "wild_card_1pct" in merged
    assert "{offer_1pct}" in wc["wild_card_1pct"]["script"]


def test_cleanroom_no_proprietary_brand_in_any_resource():
    res_dir = os.path.join(_TENANTS_DIR, "_base", "resources")
    for fn in os.listdir(res_dir):
        if not fn.endswith(".json"):
            continue
        raw = open(os.path.join(res_dir, fn), encoding="utf-8").read()
        assert not _FORBIDDEN.search(raw), f"특정사 표현 잔존: {fn}"


def test_brand_substitution_per_tenant():
    kt = _repo("ktcommerce").get_step("서비스안내", "재협상")
    im = _repo("imarketkorea").get_step("서비스안내", "재협상")
    assert "데모상사 A" in kt["script"] and "Negosium" in kt["script"]
    assert "데모상사 B" in im["script"]
    assert "{company_name}" not in kt["script"]  # 치환 완료


def test_variable_substitution_and_missing_kept():
    repo = _repo()
    node = repo.get_step("가격협상_확인", "재협상", variables={"input_price": 950})
    assert "950" in node["script"]
    # 누락 변수는 원형 유지 (KeyError 안 남)
    budget = repo.get_step("wild_card_budget", "재협상", variables={})
    assert "{target}" in budget["script"]


def test_client_step_and_variable_mapping_load():
    repo = _repo()
    csm = repo.client_step_mapping()
    assert csm["가격협상_확인"] == "가격협상"
    vm = repo.variable_mapping()
    assert vm["인터넷 최저가"] == "internet_min_price"


@pytest.mark.asyncio
async def test_resolve_card_script_file_mode_default():
    """기본(source_type='file'): resolve_card_script 가 파일 카드 멘트를 반환(하위호환)."""
    repo = _repo()
    assert repo._config.cards.source_type == "file"
    # action 0 파일 멘트가 변수 치환되어 나온다 (DB 무접근)
    out = await repo.resolve_card_script(0, "NGC-A001", {"input_price": 9800})
    assert out and "9800" in out


class _FakeCardRepo:
    """ICardScriptRepository 더블 — DB 없이 카드코드→멘트 매핑만 흉내(세션 인자 무시)."""

    def __init__(self, by_number: dict):
        self._by = by_number

    async def get_script_by_number(self, cdb, number):
        from common.enums import ErrorType
        return ErrorType.SUCCESS, self._by.get(number)


@pytest.mark.asyncio
async def test_resolve_card_script_db_mode_prefers_db(monkeypatch):
    """source_type='backoffice_db': card.nego_cards.script(정본)를 파일보다 우선 사용 + 마커 보존."""
    cfg = TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0).load("ktcommerce")
    cfg.cards.source_type = "backoffice_db"
    fake = _FakeCardRepo({"NGC-A001": "DB 편집 멘트 **{input_price}원** 검토 중입니다."})
    repo = ScriptRepository(cfg, _TENANTS_DIR, card_repo=fake)

    # execute_lambda 를 세션 없이 콜백만 실행하도록 대체(순수 단위검증)
    async def _fake_lambda(_db, _wr, func):
        return await func(None)
    from negotiation.chat.service import script_repository as _sr
    monkeypatch.setattr(_sr.DB_SESSION_MNG, "execute_lambda", _fake_lambda)

    out = await repo.resolve_card_script(0, "NGC-A001", {"input_price": 9800})
    assert out == "DB 편집 멘트 **9800원** 검토 중입니다."  # DB 우선 + 마커(**) 불투명 보존 + 변수 치환


@pytest.mark.asyncio
async def test_resolve_card_script_db_mode_falls_back_to_file(monkeypatch):
    """DB 에 해당 카드 멘트가 없으면 파일(scripts_cards.json)로 폴백."""
    cfg = TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0).load("ktcommerce")
    cfg.cards.source_type = "backoffice_db"
    fake = _FakeCardRepo({})  # DB 미보유
    repo = ScriptRepository(cfg, _TENANTS_DIR, card_repo=fake)

    async def _fake_lambda(_db, _wr, func):
        return await func(None)
    from negotiation.chat.service import script_repository as _sr
    monkeypatch.setattr(_sr.DB_SESSION_MNG, "execute_lambda", _fake_lambda)

    out = await repo.resolve_card_script(0, "NGC-A001", {"input_price": 9800})
    assert out and "9800" in out  # 파일 폴백 멘트