Mina Choi
6eb4dc26f8
협상 불가 사유를 내면 500 이 나고, 거부 폼은 목록·채팅이 따로 놀았으며,
제출한 내용을 다시 볼 방법이 없었다. 결렬 건은 협력사가 낸 거부가를 계약가로
간주해 낙찰시켜 절감 통계가 음수로 뒤집힐 수 있었고, 견적 상세는 판정 가격이
세 곳에 흩어져 대화 탭에선 아예 보이지 않았다.
agent
- 결렬 종료 로깅 크래시 수정 — _log 를 action_id 기반으로 되돌리고 선택 근거
(Q·UCB·방문수)는 decision/policy 가 있을 때만 채운다. 종료 행은 카드 선택이
없고 policy.update 뒤라 값을 넣으면 학습 화면 집계가 오염된다
negosium
- 거부 폼을 목록·채팅 공용 컴포넌트 하나로 통일(사유 3종 + 공급 희망가·의견 선택)
- 거부 사유 열람 — 목록에 '거부 사유 보기'(부가정보 보기와 같은 규격), 채팅
재진입 시 대화 끝에 거부 내역 카드. 목록·채팅 init 응답에 reject_reason·reject_price 추가
- 자유 입력 거부("협상 포기합니다")가 사유 NULL 로 저장되던 문제 수정 — 폼 마커가
없으면 원문을 사유로 쓰고, 문장 속 숫자를 희망가로 오인하지 않는다
- koreanNumber 를 전역 lib 으로 이동(공용 폼이 쓴다)
negodata
- 직접 낙찰에 계약가 입력 — 결렬·미응찰 건을 오프라인으로 다시 협상한 결과를
담당자가 확정해 넣는다. 후보는 초청 협력사 전부(가격 미제출도 포함),
계약가는 sessions.custom.offline_award 에 근거·작성자·시각과 함께 남긴다
- 통계 계약가 = 담당자 확정가 우선, 없으면 투찰가. 거부가를 계약가로 치던 파생 제거.
KPI 에 오프라인 반영 건수 추가
- 견적 상세 리모델링 — 가격 레일(앵커링가/투찰현황 · 목표가 · 타결 상한가 · 결과가)을
시트에 고정해 접힘·탭 전환에도 남기고, 스펙트럼에 타결 판정선과 구간색 추가.
라벨은 폭을 실측해 두 레인으로 배치(겹침 불가). 상품·마감시각 등 전 행 동일 컬럼 제거,
협상현황에 부가정보 노출, 1:1 은 협력사·세션상태를 결과 밴드로 올림
테스트: negosium 58 · negodata 110 통과. 프론트 빌드/린트 통과.
461 lines
24 KiB
Python
461 lines
24 KiB
Python
"""채팅(chat) 도메인 e2e 테스트 — init / messages(오프닝 seed) / send(협상 진행~종료).
|
|
|
|
실제 agent(9500) 대신, 결정론적 테스트 더블(_FakeAgentClient)을 FastAPI 의존성 오버라이드로 주입한다.
|
|
(프로덕션 코드에는 mock 이 없다 — 테스트 전용 double 이다.) 더블은 backend 가 매 턴 보내는
|
|
client_step(직전 봇 step)과 user_input 으로 단계를 진행한다.
|
|
dev negosium_db 를 그대로 쓰므로 전용 테스트 행만 시드/정리한다.
|
|
"""
|
|
|
|
import uuid
|
|
|
|
import bcrypt
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import text
|
|
|
|
from services.agent_client import AgentTurn, IAgentClient, get_agent_client
|
|
from services.chat_service import ChatService
|
|
|
|
TEST_LOGIN_ID = "pytest_chat_user"
|
|
TEST_PW = "pytest1234"
|
|
TEST_SUPPLIER_NAME = "파이테스트채팅공급사"
|
|
MARK = "PYTESTCHAT-"
|
|
|
|
|
|
def _parse_price(text_):
|
|
if not text_:
|
|
return None
|
|
digits = "".join(ch for ch in text_ if ch.isdigit())
|
|
return int(digits) if digits else None
|
|
|
|
|
|
class _FakeAgentClient(IAgentClient):
|
|
"""결정론적 테스트 더블. ctx.client_step(직전 봇 step)+user_input 으로 단계를 진행한다.
|
|
|
|
플로우: (오프닝)서비스안내 → 협상품목안내 → 가격협상 → 가격제시 시 목표가 이하면 성공 종료.
|
|
'포기'/'거부' 입력은 언제든 실패 종료.
|
|
"""
|
|
|
|
async def chat(self, session_id, user_input, ctx) -> AgentTurn:
|
|
sid = session_id or "fake-session"
|
|
if user_input and ("포기" in user_input or "거부" in user_input):
|
|
return AgentTurn(
|
|
session_id=sid, step="협상종료", client_step="협상종료",
|
|
script="협상이 종료되었습니다.", chat_end=True, outcome="failure",
|
|
)
|
|
if user_input is None: # 오프닝(턴0) — 양쪽 공통
|
|
return AgentTurn(session_id=sid, step="서비스안내", client_step="서비스안내",
|
|
script="협상에 참여해 주셔서 감사합니다. 시작하시겠어요?",
|
|
input_mode="confirm", input_options=["네, 시작할게요"])
|
|
if ctx.rq_type == "재견적":
|
|
return self._requote(sid, user_input, ctx)
|
|
return self._renego(sid, user_input, ctx)
|
|
|
|
def _renego(self, sid, user_input, ctx) -> AgentTurn:
|
|
if ctx.client_step == "서비스안내":
|
|
return AgentTurn(session_id=sid, step="협상품목안내", client_step="협상품목안내",
|
|
script="협상 품목을 확인해 주세요. 가격 협상을 진행할까요?",
|
|
input_mode="confirm", input_options=["가격 협상 진행"])
|
|
if ctx.client_step == "협상품목안내":
|
|
return AgentTurn(session_id=sid, step="가격협상", client_step="가격협상",
|
|
script="희망 공급가를 입력해 주세요.", input_mode="price")
|
|
# 가격협상 단계: 목표가 이하면 합의 종료, 아니면 한 번 더 요청
|
|
price = _parse_price(user_input)
|
|
if price is not None and ctx.target_price and price <= ctx.target_price:
|
|
return AgentTurn(session_id=sid, step="협상종료", client_step="협상종료",
|
|
script=f"제안하신 {price:,}원으로 합의되었습니다. 감사합니다.",
|
|
chat_end=True, outcome="success", indicator_value=99.0)
|
|
return AgentTurn(session_id=sid, step="가격협상", client_step="가격협상",
|
|
script="조금 더 조정된 가격을 제안해 주시겠어요?", input_mode="price",
|
|
indicator_value=50.0)
|
|
|
|
def _requote(self, sid, user_input, ctx) -> AgentTurn:
|
|
# 서비스안내 → 가격제안 → 배송형태선택 → 가격협상_입력 → 결과안내(summaryCM)
|
|
if ctx.client_step == "서비스안내":
|
|
return AgentTurn(session_id=sid, step="가격제안", client_step="가격제안",
|
|
script="제시 목표가로 진행하시겠어요?", input_mode="yes_no",
|
|
input_options=["예", "아니오"])
|
|
if ctx.client_step == "가격제안":
|
|
return AgentTurn(session_id=sid, step="배송형태선택", client_step="배송형태선택",
|
|
script="배송형태를 선택해 주세요.", input_mode="delivery_type",
|
|
input_options=["협력사배송", "지정택배배송", "픽업배송"])
|
|
if ctx.client_step == "배송형태선택":
|
|
return AgentTurn(session_id=sid, step="가격협상_입력", client_step="가격협상_입력",
|
|
script="희망 공급가를 입력해 주세요.", input_mode="price")
|
|
if ctx.client_step == "가격협상_입력":
|
|
return AgentTurn(session_id=sid, step="결과안내", client_step="결과안내",
|
|
script="투찰 결과를 확인해 주세요.", input_mode="yes_no",
|
|
input_options=["투찰확정", "정보수정"])
|
|
# 결과안내 "투찰확정" → 결과제출 → 협상종료
|
|
return AgentTurn(session_id=sid, step="협상종료", client_step="협상종료",
|
|
script="투찰이 확정되었습니다. 감사합니다.",
|
|
chat_end=True, outcome="success")
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _fake_agent():
|
|
"""모든 chat 테스트에서 실제 agent 대신 결정론적 더블을 주입(의존성 오버라이드)."""
|
|
from router.router import app
|
|
|
|
app.dependency_overrides[get_agent_client] = lambda: _FakeAgentClient()
|
|
yield
|
|
app.dependency_overrides.pop(get_agent_client, None)
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def chat_seed(db_engine):
|
|
"""공급사 + 유저 + 세션 2건(본인: 협상중 P / 협상생성 C) + 1건(타 공급사 X) 시드."""
|
|
supplier_id = uuid.uuid4()
|
|
other_supplier_id = uuid.uuid4()
|
|
pw_hash = bcrypt.hashpw(TEST_PW.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
|
|
|
|
# (code, session.status, qt_type, 마감까지 h, quotation.status, 소속 공급사)
|
|
specs = [
|
|
("P", 2, 1, 2, 2, supplier_id), # 협상중 / 재협상 / +2h / 견적진행중
|
|
("C", 1, 1, 2, 1, supplier_id), # 협상생성 / 재협상 / +2h / 견적생성
|
|
("X", 2, 1, 2, 2, other_supplier_id), # 타 공급사 → 차단
|
|
("Q", 2, 2, 2, 2, supplier_id), # 협상중 / 재견적 / +2h / 견적진행중
|
|
]
|
|
sids, qids = {}, {}
|
|
|
|
async def _cleanup(conn):
|
|
await conn.execute(text(f"DELETE FROM negotiation.chats WHERE session_id IN (SELECT session_id FROM negotiation.sessions WHERE qt_number LIKE '{MARK}%')"))
|
|
await conn.execute(text(f"DELETE FROM negotiation.sessions WHERE qt_number LIKE '{MARK}%'"))
|
|
await conn.execute(text(f"DELETE FROM quotation.quotations WHERE number LIKE '{MARK}%'"))
|
|
await conn.execute(text(f"DELETE FROM partner.items WHERE code LIKE '{MARK}%'"))
|
|
await conn.execute(text("DELETE FROM supplier.supplier_users WHERE id = :id"), {"id": TEST_LOGIN_ID})
|
|
await conn.execute(text("DELETE FROM partner.suppliers WHERE name = :n"), {"n": TEST_SUPPLIER_NAME})
|
|
|
|
async with db_engine.begin() as conn:
|
|
await _cleanup(conn)
|
|
await conn.execute(
|
|
text("INSERT INTO partner.suppliers (supplier_id, company_id, user_id, name) VALUES (:sid, gen_random_uuid(), gen_random_uuid(), :name)"),
|
|
{"sid": supplier_id, "name": TEST_SUPPLIER_NAME},
|
|
)
|
|
await conn.execute(
|
|
text(
|
|
"INSERT INTO supplier.supplier_users (supplier_id, id, password, name, last_accessed_at, status, role) "
|
|
"VALUES (:sid, :id, :pw, '채팅담당자', now(), 1, 1)"
|
|
),
|
|
{"sid": supplier_id, "id": TEST_LOGIN_ID, "pw": pw_hash},
|
|
)
|
|
for code, sess_st, qt_type, hrs, quote_st, sup in specs:
|
|
item_id, qt_id, session_id = uuid.uuid4(), uuid.uuid4(), uuid.uuid4()
|
|
sids[code], qids[code] = session_id, qt_id
|
|
await conn.execute(
|
|
text(
|
|
"INSERT INTO partner.items (item_id, company_id, user_id, name, code, price, model_name, manufacturer, moq, spec) "
|
|
"VALUES (:iid, gen_random_uuid(), gen_random_uuid(), :name, :code, 100000, :model, '테스트제조사', '10', '규격A')"
|
|
),
|
|
{"iid": item_id, "name": f"상품 {code}", "code": f"{MARK}{code}", "model": f"MODEL-{code}"},
|
|
)
|
|
await conn.execute(
|
|
text(
|
|
"INSERT INTO quotation.quotations (qt_id, user_id, qt_setting_id, version_id, name, number, type, status, start_time, end_time, memo) "
|
|
"VALUES (:qid, gen_random_uuid(), gen_random_uuid(), gen_random_uuid(), :name, :num, :tp, :st, now(), now() + make_interval(hours => :hrs), '메모')"
|
|
),
|
|
{"qid": qt_id, "name": f"견적 {code}", "num": f"{MARK}{code}", "tp": qt_type, "st": quote_st, "hrs": hrs},
|
|
)
|
|
await conn.execute(
|
|
text(
|
|
"INSERT INTO negotiation.sessions "
|
|
"(session_id, quotation_id, item_id, supplier_id, qt_number, qt_round, qt_type, target_price, status, end_time) "
|
|
"VALUES (:sesid, :qid, :iid, :sup, :qtn, 1, :qtt, 100000, :st, now() + make_interval(hours => 2))"
|
|
),
|
|
{"sesid": session_id, "qid": qt_id, "iid": item_id, "sup": sup, "qtn": f"{MARK}{code}", "qtt": qt_type, "st": sess_st},
|
|
)
|
|
|
|
yield {"supplier_id": supplier_id, "sids": sids, "qids": qids}
|
|
|
|
async with db_engine.begin() as conn:
|
|
await _cleanup(conn)
|
|
|
|
|
|
async def _login_token(client):
|
|
r = await client.post("/v1/auth/login", json={"id": TEST_LOGIN_ID, "pw": TEST_PW})
|
|
return r.json()["access_token"]
|
|
|
|
|
|
def _h(token):
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
async def _init(client, token, sid):
|
|
return await client.get(f"/v1/negotiation/sessions/{sid}/chat/init", headers=_h(token))
|
|
|
|
|
|
async def _messages(client, token, sid):
|
|
return await client.get(f"/v1/negotiation/sessions/{sid}/chat/messages", headers=_h(token))
|
|
|
|
|
|
async def _send(client, token, sid, user_input, user_input_type=None):
|
|
body = {"user_input": user_input, "user_input_type": user_input_type}
|
|
return await client.post(f"/v1/negotiation/sessions/{sid}/chat/send", headers=_h(token), json=body)
|
|
|
|
|
|
async def _session_status(db_engine, session_id):
|
|
async with db_engine.begin() as conn:
|
|
return (await conn.execute(text("SELECT status FROM negotiation.sessions WHERE session_id = :sid"), {"sid": session_id})).scalar()
|
|
|
|
|
|
async def _session_bid(db_engine, session_id):
|
|
async with db_engine.begin() as conn:
|
|
return (await conn.execute(text("SELECT bid_price FROM negotiation.sessions WHERE session_id = :sid"), {"sid": session_id})).scalar()
|
|
|
|
|
|
async def _session_reject(db_engine, session_id):
|
|
async with db_engine.begin() as conn:
|
|
r = (await conn.execute(text("SELECT status, reject_reason FROM negotiation.sessions WHERE session_id = :sid"), {"sid": session_id})).first()
|
|
return r[0], r[1]
|
|
|
|
|
|
# ---- init -------------------------------------------------------------------
|
|
async def test_chat_init_returns_meta(client, chat_seed):
|
|
token = await _login_token(client)
|
|
body = (await _init(client, token, chat_seed["sids"]["P"])).json()
|
|
assert body["result"]["success"] is True
|
|
assert body["session_status"] == 2
|
|
assert body["item_name"] == "상품 P" and body["item_price"] == 100000
|
|
assert body["item_maker_name"] == "테스트제조사"
|
|
assert body["quotation_end_time"] # 타이머용 마감 시각
|
|
|
|
|
|
async def test_chat_init_returns_reject_detail(client, chat_seed, db_engine):
|
|
"""검증: 협상 거부로 끝난 세션에 재진입('결과 보기')했을 때의 init 응답.
|
|
기대결과: 대화에 남지 않는 제출 내역(reject_reason·reject_price)이 실려 열람 카드를 그릴 수 있다."""
|
|
token = await _login_token(client)
|
|
sid = chat_seed["sids"]["P"]
|
|
await client.post(
|
|
f"/v1/negotiation/sessions/{sid}/reject",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
json={"reject_reason": "단종", "reject_price": 91000, "opinion": "후속 모델로 제안 가능합니다"},
|
|
)
|
|
body = (await _init(client, token, sid)).json()
|
|
assert body["session_status"] == 5
|
|
assert body["reject_reason"] == "단종"
|
|
assert body["reject_price"] == 91000
|
|
assert body["custom"]["opinion"] == "후속 모델로 제안 가능합니다"
|
|
|
|
|
|
async def test_chat_init_forbidden_other_supplier(client, chat_seed):
|
|
token = await _login_token(client)
|
|
body = (await _init(client, token, chat_seed["sids"]["X"])).json()
|
|
assert body["result"]["code"] == 1300 # NEGO_FORBIDDEN
|
|
|
|
|
|
async def test_chat_init_vat_mode_unified_shows_excluded(client, chat_seed, db_engine):
|
|
"""검증: 부가세 전체 통일 회사(features.vat_mode=unified_excluded)의 세션 채팅 init.
|
|
기대결과: 상품에 vat_yn=true 잔존값이 있어도 item_vat_yn=False — 프론트가 'VAT별도'로 고정 표기."""
|
|
import json
|
|
|
|
company_id = uuid.uuid4()
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("INSERT INTO company.companies (company_id, name, status, settings) VALUES (:c, :n, 1, CAST(:s AS JSONB))"),
|
|
{"c": company_id, "n": f"{MARK}VAT통일사", "s": json.dumps({"features": {"vat_mode": "unified_excluded"}})},
|
|
)
|
|
await conn.execute(
|
|
text("UPDATE partner.suppliers SET company_id = :c WHERE supplier_id = :sid"),
|
|
{"c": company_id, "sid": chat_seed["supplier_id"]},
|
|
)
|
|
await conn.execute(
|
|
text("UPDATE partner.items SET vat_yn = true WHERE item_id = (SELECT item_id FROM negotiation.sessions WHERE session_id = :s)"),
|
|
{"s": chat_seed["sids"]["P"]},
|
|
)
|
|
try:
|
|
token = await _login_token(client)
|
|
body = (await _init(client, token, chat_seed["sids"]["P"])).json()
|
|
assert body["result"]["success"] is True
|
|
assert body["item_vat_yn"] is False
|
|
finally:
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(text("DELETE FROM company.companies WHERE company_id = :c"), {"c": company_id})
|
|
|
|
|
|
# ---- messages (오프닝 seed) -------------------------------------------------
|
|
async def test_messages_seeds_opening(client, chat_seed):
|
|
token = await _login_token(client)
|
|
body = (await _messages(client, token, chat_seed["sids"]["P"])).json()
|
|
assert body["result"]["success"] is True
|
|
assert len(body["items"]) == 1
|
|
msg = body["items"][0]
|
|
assert msg["sender"] == 1 # ChatSender.BOT (봇)
|
|
assert msg["next_input_mode"] == "confirm"
|
|
assert msg["script"]
|
|
|
|
|
|
# ---- send (협상 진행 → 종료) ------------------------------------------------
|
|
async def test_send_flow_to_completion(client, chat_seed, db_engine):
|
|
token = await _login_token(client)
|
|
sid = chat_seed["sids"]["P"]
|
|
|
|
await _messages(client, token, sid) # 오프닝(턴0) seed
|
|
|
|
r1 = (await _send(client, token, sid, "네, 시작할게요")).json()
|
|
assert r1["result"]["success"] is True
|
|
assert r1["message"]["next_input_mode"] == "confirm" # 품목안내
|
|
assert r1["session_status"] == 2
|
|
|
|
r2 = (await _send(client, token, sid, "가격 협상 진행")).json()
|
|
assert r2["message"]["next_input_mode"] == "price" # 가격입력 요청
|
|
|
|
r3 = (await _send(client, token, sid, "90000", user_input_type="price")).json()
|
|
assert r3["result"]["success"] is True
|
|
assert r3["message"]["chat_end"] is True
|
|
assert r3["session_status"] == 3 # 협상완료(DONE)
|
|
assert await _session_status(db_engine, sid) == 3
|
|
assert await _session_bid(db_engine, sid) == 90000 # 입찰가 확정
|
|
|
|
|
|
async def test_requote_summary_captures_delivery_type(client, chat_seed):
|
|
"""재견적: 배송형태선택에서 고른 값이 summaryCM 요약(delivery_type)에 담긴다."""
|
|
token = await _login_token(client)
|
|
sid = chat_seed["sids"]["Q"]
|
|
await _messages(client, token, sid) # 오프닝(서비스안내)
|
|
await _send(client, token, sid, "네, 시작할게요") # → 가격제안
|
|
await _send(client, token, sid, "예") # → 배송형태선택
|
|
await _send(client, token, sid, "협력사배송") # → 가격협상_입력
|
|
r = (await _send(client, token, sid, "90000", user_input_type="price")).json() # → 결과안내(summaryCM)
|
|
assert r["result"]["success"] is True
|
|
msg = r["message"]
|
|
assert msg["bot_chat_type"] == "summaryCM"
|
|
assert msg["summary"]["delivery_type"] == "협력사배송"
|
|
|
|
|
|
async def test_agent_provided_bot_chat_type_and_indicator_passthrough(client, chat_seed):
|
|
"""agent 가 bot_chat_type/indicator_value 를 직접 주면 backend 는 step 추측 없이 그대로 전달한다."""
|
|
from router.router import app
|
|
|
|
class _T(IAgentClient):
|
|
async def chat(self, session_id, user_input, ctx):
|
|
if user_input is None:
|
|
return AgentTurn(session_id=session_id, step="서비스안내", client_step="서비스안내",
|
|
script="안녕하세요", input_mode="confirm", input_options=["확인"])
|
|
return AgentTurn(session_id=session_id, step="가격협상", client_step="가격협상",
|
|
script="지표를 확인하세요", input_mode="price",
|
|
indicator_value=55.0, bot_chat_type="indicator")
|
|
|
|
app.dependency_overrides[get_agent_client] = lambda: _T()
|
|
token = await _login_token(client)
|
|
sid = chat_seed["sids"]["P"]
|
|
await _messages(client, token, sid) # 오프닝
|
|
r = (await _send(client, token, sid, "확인")).json() # → 가격협상(indicator)
|
|
assert r["result"]["success"] is True
|
|
msg = r["message"]
|
|
assert msg["bot_chat_type"] == "indicator" # agent 값 그대로
|
|
assert msg["indicator_value"] == 55.0 # 지표 전달
|
|
|
|
|
|
async def test_send_price_out_of_range(client, chat_seed):
|
|
token = await _login_token(client)
|
|
sid = chat_seed["sids"]["P"]
|
|
await _messages(client, token, sid)
|
|
# 목표가 100000 → 허용 [30000, 170000]. 10 은 하한 미만.
|
|
body = (await _send(client, token, sid, "10", user_input_type="price")).json()
|
|
assert body["result"]["code"] == 1401 # CHAT_PRICE_OUT_OF_RANGE
|
|
|
|
|
|
async def test_send_not_in_progress(client, chat_seed):
|
|
token = await _login_token(client)
|
|
sid = chat_seed["sids"]["C"] # 협상생성(미참여 전 단계)
|
|
body = (await _send(client, token, sid, "네")).json()
|
|
assert body["result"]["code"] == 1400 # CHAT_NOT_IN_PROGRESS
|
|
|
|
|
|
async def test_send_requires_auth(client, chat_seed):
|
|
sid = chat_seed["sids"]["P"]
|
|
r = await client.post(f"/v1/negotiation/sessions/{sid}/chat/send", json={"user_input": "네"})
|
|
assert r.status_code in (401, 403)
|
|
|
|
|
|
# ---- 보완: 거부 저장 / 동시전송 가드 / init 만료 정리 ------------------------
|
|
async def test_send_rejection_persists_reason(client, chat_seed, db_engine):
|
|
token = await _login_token(client)
|
|
sid = chat_seed["sids"]["P"]
|
|
await _messages(client, token, sid) # 오프닝
|
|
body = (await _send(client, token, sid, "협상 포기합니다")).json()
|
|
assert body["result"]["success"] is True
|
|
assert body["message"]["chat_end"] is True
|
|
assert body["session_status"] == 5 # 협상거부(REJECTED)
|
|
status, reason = await _session_reject(db_engine, sid)
|
|
assert status == 5 and reason == "협상 포기합니다" # 거부 사유 저장
|
|
|
|
|
|
async def test_send_blocked_when_prev_turn_pending(client, chat_seed, db_engine):
|
|
"""직전 메시지가 USER(이전 턴 처리 중)면 중복 전송을 거절한다 → CHAT_IN_PROGRESS."""
|
|
token = await _login_token(client)
|
|
sid = chat_seed["sids"]["P"]
|
|
await _messages(client, token, sid) # 오프닝(seq=1, BOT)
|
|
# 봇 응답이 아직 안 온 상태를 모사: USER 메시지를 마지막(seq=2)으로 직접 삽입
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("INSERT INTO negotiation.chats (session_id, seq, sender, target_price) VALUES (:sid, 2, 2, 0)"),
|
|
{"sid": sid},
|
|
)
|
|
body = (await _send(client, token, sid, "네")).json()
|
|
assert body["result"]["code"] == 1403 # CHAT_IN_PROGRESS
|
|
|
|
|
|
async def test_init_marks_expired_created_as_not_participated(client, chat_seed, db_engine):
|
|
"""검증: 마감시간이 지난 협상생성 세션으로 채팅 진입.
|
|
기대결과: DB 상태가 미참여(4)로 정리되고, init 자체는 열람용으로 성공한다."""
|
|
token = await _login_token(client)
|
|
sid, qid = chat_seed["sids"]["C"], chat_seed["qids"]["C"] # 협상생성(1)
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(text("UPDATE quotation.quotations SET end_time = now() - make_interval(hours => 1) WHERE qt_id = :qid"), {"qid": qid})
|
|
body = (await _init(client, token, sid)).json()
|
|
assert body["result"]["success"] is True
|
|
assert body["session_status"] == 4
|
|
assert await _session_status(db_engine, sid) == 4 # DB 도 미참여로 전이
|
|
|
|
|
|
async def test_init_allows_viewing_rejected_session(client, chat_seed, db_engine):
|
|
"""검증: 협상거부(5)로 끝난 세션에 '결과 보기'로 재진입.
|
|
기대결과: init 성공(열람 허용) — 대화 재개는 send 가 협상중만 허용해 막는다."""
|
|
token = await _login_token(client)
|
|
sid = chat_seed["sids"]["P"]
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(text("UPDATE negotiation.sessions SET status = 5 WHERE session_id = :sid"), {"sid": sid}) # 협상거부
|
|
body = (await _init(client, token, sid)).json()
|
|
assert body["result"]["success"] is True and body["session_status"] == 5
|
|
assert (await _send(client, token, sid, "네")).json()["result"]["code"] == 1400 # CHAT_NOT_IN_PROGRESS
|
|
|
|
|
|
# ---- 순수 헬퍼 단위 테스트 (DB 불필요, ChatService @staticmethod) ----------
|
|
def test_parse_price():
|
|
assert ChatService._parse_price("530,000원") == 530000 # 콤마/통화기호 제거
|
|
assert ChatService._parse_price("abc") is None
|
|
assert ChatService._parse_price("") is None
|
|
assert ChatService._parse_price(None) is None
|
|
|
|
|
|
def test_in_price_range():
|
|
assert ChatService._in_price_range(100000, 100000) is True
|
|
assert ChatService._in_price_range(29000, 100000) is False # floor(0.3) 미만
|
|
assert ChatService._in_price_range(180000, 100000) is False # ceil(1.7) 초과
|
|
assert ChatService._in_price_range(50000, None) is True # 목표가 없으면 양수면 통과
|
|
assert ChatService._in_price_range(0, None) is False
|
|
|
|
|
|
def test_input_matches_mode():
|
|
f = ChatService._input_matches_mode
|
|
assert f(None, "x", "text") is True # 직전 메타 없음 → 제약 없음
|
|
assert f({}, "x", "price") is True # input_mode 없음
|
|
assert f({"input_mode": "price"}, "100", "price") is True
|
|
assert f({"input_mode": "price"}, "예", "text") is False # price 단계에 텍스트
|
|
assert f({"input_mode": "percent"}, "5", "percent") is True
|
|
assert f({"input_mode": "yes_no"}, "예", "text") is True
|
|
assert f({"input_mode": "yes_no"}, "100", "price") is False # 버튼 단계에 가격 숫자
|
|
assert f({"input_mode": "delivery_type"}, "픽업", "text") is True
|
|
|
|
|
|
def test_resolve_bot_chat_type():
|
|
f = ChatService._resolve_bot_chat_type
|
|
assert f(1, "협상완료") == "summaryRSP" # 재협상
|
|
assert f(2, "협상완료") == "summaryCM" # 재견적
|
|
assert f(1, "협상실패") == "rejectRSP"
|
|
assert f(2, "협상실패") == "rejectCM"
|
|
assert f(2, "결과제출") == "summaryCM"
|
|
assert f(1, "가격협상") is None # 일반 step
|
|
assert f(1, None) is None
|