125 lines
4.9 KiB
Python
125 lines
4.9 KiB
Python
"""P7 슬라이스 검증 — 대화형 /chat (스크립트 구동 + 카드선택 학습 + 와일드카드).
|
|
|
|
1. 전체 대화: 서비스안내→담당자확인→협상품목안내→가격협상→와일드카드→협상완료→협상종료(chat_end).
|
|
2. 가격협상 턴에서 카드 선택 + 학습(card_id, updated_q).
|
|
3. 와일드카드 발동(wild_card_budget) + 종료 보상(success).
|
|
4. 브랜드 치환 테넌트별(데모상사 A/B), 클린룸(스크립트에 KT 흔적 없음).
|
|
5. 경험로그 적재 + 종료 후 진행 시 에러.
|
|
6. (HTTP) 헤더로 새 세션 시작 + 한 턴 진행.
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
|
|
import pytest
|
|
|
|
from router.v1.chat.protocol import Req_Chat
|
|
from services.chat_service import ChatService, reset_sessions
|
|
from negotiation.qtable.infra.repository.learning_repository import LearningRepository
|
|
from tenancy.config_loader import TenantConfigLoader
|
|
from tenancy.registry import TenantEngineRegistry
|
|
|
|
_TENANTS_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "tenants")
|
|
_KT = re.compile(r"kt\s*commerce|nego-?wiz", re.IGNORECASE)
|
|
|
|
|
|
def _reg():
|
|
return TenantEngineRegistry(loader=TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0))
|
|
|
|
|
|
async def _run(svc, eng, turns):
|
|
"""turns: user_input 리스트(첫 None=시작). 반환: 응답 리스트."""
|
|
sid, out = None, []
|
|
for ui in turns:
|
|
r = await svc.chat(eng, Req_Chat(session_id=sid, user_input=ui, rq_type="재협상"))
|
|
sid = r.session_id
|
|
out.append(r)
|
|
if r.chat_end:
|
|
break
|
|
return out
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_full_conversation_reaches_completion(db_engine):
|
|
reset_sessions()
|
|
eng = await _reg().get_engine("ktcommerce")
|
|
svc = ChatService()
|
|
# anchor=9900, target=10000(기본). 11000(>10395)→가격협상(카드), 10200(9900~10395)→와일드카드, 9800(≤anchor)→우선협상
|
|
turns = [None, "확인", "예", "확인", "11000", "예", "10200", "예", "9800", "예",
|
|
"협상 내용을 확인했으며, 이의가 없음에 동의합니다."]
|
|
out = await _run(svc, eng, turns)
|
|
steps = [r.step for r in out]
|
|
|
|
assert steps[0] == "서비스안내"
|
|
assert "협상완료" in steps
|
|
assert out[-1].step == "협상종료" and out[-1].chat_end is True
|
|
|
|
# 가격협상 카드선택 + 학습
|
|
nego = [r for r in out if r.step == "가격협상"]
|
|
assert nego and nego[0].card_id and nego[0].card_id.startswith("NGC-A")
|
|
assert nego[0].updated_q is not None
|
|
|
|
# 와일드카드 발동
|
|
assert any(r.step == "wild_card_budget" for r in out)
|
|
|
|
# 종료 보상(success)
|
|
done = [r for r in out if r.outcome == "success"]
|
|
assert done and done[0].reward_total is not None
|
|
|
|
# 브랜드 치환 + 클린룸
|
|
assert "데모상사 A" in out[0].script
|
|
assert not _KT.search(out[0].script)
|
|
|
|
# 경험로그 적재
|
|
repo = LearningRepository("ktcommerce")
|
|
err, cnt = await repo.read(lambda s: repo.count_experience(s))
|
|
assert cnt >= 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_priority_completes_without_wildcard(db_engine):
|
|
reset_sessions()
|
|
eng = await _reg().get_engine("ktcommerce")
|
|
svc = ChatService()
|
|
# 첫 제시가가 앵커가(9900) 이하 → 우선협상 → 바로 협상완료(카드/와일드카드 없이)
|
|
out = await _run(svc, eng, [None, "확인", "예", "확인", "9800", "예",
|
|
"협상 내용을 확인했으며, 이의가 없음에 동의합니다."])
|
|
steps = [r.step for r in out]
|
|
assert "협상완료" in steps
|
|
assert "wild_card_budget" not in steps
|
|
assert "가격협상" not in steps # 우선협상이라 카드 협상 없이 타결
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tenant_brand_isolation(db_engine):
|
|
reset_sessions()
|
|
svc = ChatService()
|
|
reg = _reg()
|
|
ik = await svc.chat(await reg.get_engine("imarketkorea"), Req_Chat(session_id=None, rq_type="재협상"))
|
|
assert "데모상사 B" in ik.script
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_advance_after_end_errors(db_engine):
|
|
reset_sessions()
|
|
eng = await _reg().get_engine("ktcommerce")
|
|
svc = ChatService()
|
|
out = await _run(svc, eng, [None, "확인", "예", "확인", "1000", "예",
|
|
"협상 내용을 확인했으며, 이의가 없음에 동의합니다."])
|
|
sid = out[-1].session_id
|
|
r = await svc.chat(eng, Req_Chat(session_id=sid, user_input="확인"))
|
|
assert r.result.desc == "NEGO_INVALID_STEP"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_http_chat_start(client):
|
|
r = await client.post("/v1/chat", headers={"X-Tenant-ID": "ktcommerce"}, json={"rq_type": "재협상"})
|
|
assert r.status_code == 200
|
|
d = r.json()
|
|
assert d["step"] == "서비스안내"
|
|
assert d["input_options"] == ["확인"]
|
|
assert "데모상사 A" in d["script"]
|
|
# 헤더 없으면 400
|
|
r2 = await client.post("/v1/chat", json={"rq_type": "재협상"})
|
|
assert r2.status_code == 400
|