"""카드 전술 재설계 검증 — "멘트 카드 → 전술 카드" (가격 행동 실행 계층). ① 카운터 산식 결정론 + min(counter, target) 클램프 + 무의미 카운터(HOLD 강등) ② 카운터 수락 = 즉시 타결 / 거절 = 재입력 + pending 폐기 ③ 목표가 초과 타결 금지 가드(성공 스텝 진입 차단) ④ 선택형 와일드카드(WC-05 중간값 절충) 발동 — 1.02~1.05 구간 갭 해소 ⑤ E2E: 견적 선택 카드(NGC-009 조건부 가격 조정)의 카운터를 수락하면 settled=target ⑥ E2E: 협력사가 target 초과를 고수하면 종결 전술(최후통첩) 후 결렬 — 고객사 이득 가드레일 """ import os import uuid as _uuid from datetime import datetime, timedelta, timezone import pytest from negotiation.cards.domain.tactics import ( PriceAction, TacticSpec, compute_counter, tactic_available, tactic_for, ) from negotiation.chat.service.chat_engine import ChatEngine, ChatSession 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") def _engine() -> ChatEngine: cfg = TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0).load("imarketkorea") return ChatEngine(ScriptRepository(cfg, _TENANTS_DIR), rq_type="재협상") def _session(step="가격협상_확인", **ctx_over): ctx = {"input_price": 10300, "anchor_price": 10000, "target_price": 10100, "round": 1, "allow_selected_wildcards": False} ctx.update(ctx_over) return ChatSession(session_id="00000000-0000-0000-0000-00000000e001", tenant_id="imarketkorea", company_id="imarketkorea", step=step, action_space_size=0, context=ctx) # ---- ① 카운터 산식 (결정론 + 가드레일 클램프) -------------------------------- def test_counter_formulas_and_clamp(): ctx = {"input_price": 11000, "anchor_price": 9900, "target_price": 10000} assert compute_counter(tactic_for("NGC-009"), ctx) == 10000 # COUNTER_TARGET assert compute_counter(tactic_for("NGC-007"), ctx) == 9900 # COUNTER_ANCHOR assert compute_counter(tactic_for("WC-02"), ctx) == 9950 # (anchor+target)/2 # COUNTER_MID: 갑 직전 포지션 폴백 = anchor → (9900+11000)/2 = 10450 → target 클램프 assert compute_counter(tactic_for("WC-05"), ctx) == 10000 # 갑 직전 포지션이 있으면 그 기준: (9800+11000)/2 = 10400 → 역시 클램프 10000 assert compute_counter(tactic_for("WC-05"), dict(ctx, prev_customer_price=9800)) == 10000 # 클램프 미발동 구간: (9900+10050)/2 = 9975 ≤ target assert compute_counter(tactic_for("WC-05"), dict(ctx, input_price=10050)) == 9975 def test_counter_meaningless_degrades_to_hold(): """협력사 제시가가 이미 카운터 이하면 카운터가 무의미 → None(순수 설득 유지).""" ctx = {"input_price": 9950, "anchor_price": 9900, "target_price": 10000} assert compute_counter(tactic_for("NGC-009"), ctx) is None # target(10000) ≥ 제시가 assert compute_counter(TacticSpec(PriceAction.COUNTER_ANCHOR), dict(ctx, input_price=9900)) is None def test_unknown_card_falls_back_to_hold(): """미등록 카드번호(데모 NGC-B*, 커스텀 COMP-*)는 HOLD — 기존 동작 그대로.""" spec = tactic_for("NGC-B003") assert spec.price_action is PriceAction.HOLD assert compute_counter(spec, {"input_price": 11000, "target_price": 10000}) is None assert tactic_available(spec, {}) is True def test_tactic_availability_conditions(): assert tactic_available(tactic_for("WC-04"), {"round": 1}) is False # min_round=2 assert tactic_available(tactic_for("WC-04"), {"round": 2}) is True # ---- ② 카운터 수락/거절 메커니즘 (엔진) --------------------------------------- def test_accept_counter_settles_at_counter_price(): eng = _engine() s = _session(step="가격협상_카운터", pending_counter_price=10000) view = eng.advance(s, "수락") assert view.step == "협상완료" assert s.context["input_price"] == 10000 # 합의가 = 카운터가 assert "pending_counter_price" not in s.context def test_reject_counter_reenters_price_and_discards_pending(): eng = _engine() s = _session(step="가격협상_카운터", pending_counter_price=10000) view = eng.advance(s, "다른 가격 제시") assert view.step == "가격협상_재입력" # 새 가격 입력이 pending 을 폐기한다 — 이후 우선협상 타결이 옛 카운터로 오염되지 않음 view = eng.advance(s, "9900") assert "pending_counter_price" not in s.context assert s.context["input_price"] == 9900 def test_wildcard_1pct_decline_keeps_original_price(): """1% 인하 거절('아니오')도 협상완료로 가지만 합의가는 원 제시가 — pending 미적용 회귀.""" eng = _engine() s = _session(step="wild_card_1pct", input_price=10000, offer_1pct=9900, pending_counter_price=9900) view = eng.advance(s, "아니오") assert view.step == "협상완료" assert s.context["input_price"] == 10000 # 거절 → 카운터 미적용 # ---- ③ 목표가 초과 타결 금지 가드 -------------------------------------------- def test_success_step_guard_rejects_over_target(): eng = _engine() s = _session(input_price=10800, target_price=10000) view = eng.render_step(s, "협상완료") assert view.step == "협상실패" # 초과가 성공 진입 → 결렬 강제 # ---- ④ 선택형 와일드카드 발동 (1.02~1.05 구간 갭 해소) ------------------------- def test_selected_wildcard_fires_in_entry_zone(): eng = _engine() # 10300: 1pct 존(≤10200) 밖, entry 존(≤10500) 안 + WC-05 선택 s = _session(input_price=10300, allow_selected_wildcards=True, selected_wild_card_numbers=["WC-05"]) view = eng.advance(s, "예") assert view.step == "wild_card_dynamic" # 카운터 = (anchor 10000 + 10300)/2 = 10150 → target(10100) 클램프 assert s.context["pending_counter_price"] == 10100 assert s.context["active_wild_card_number"] == "WC-05" # 수락 → 그 가격으로 타결 view = eng.advance(s, "수락") assert view.step == "협상완료" and s.context["input_price"] == 10100 def test_unselected_wildcard_zone_still_falls_to_nego(): """와일드카드 미선택이면 entry 존이라도 일반 가격협상 — 기존 동작 보존.""" eng = _engine() s = _session(input_price=10300, allow_selected_wildcards=True, selected_wild_card_numbers=[]) view = eng.advance(s, "예") assert view.step == "가격협상" # ---- 인하율 표기 (회귀: 인상 제시가 "-1.3% 인하"로 표기되던 버그) ----------------- def test_discount_never_negative_and_phrase_matches_direction(): eng = _engine() # 인상 제시(기존 공급가 78000 < 제시 79000): 음수 인하율 금지 + "높은 금액" 문구 s = _session(item_price=78000, input_price=79000) v = eng.vars_for(s) assert v["discount_rate"] == "0.0" # 마이너스 인하 표기 금지 assert "높은 금액" in v["discount_phrase"] and "78000원" in v["discount_phrase"] assert "-" not in v["discount_phrase"] # 인하 제시: 상품단가(item_price) 기준 인하율 v = eng.vars_for(_session(item_price=78000, input_price=77000)) assert v["discount_rate"] == "1.3" assert "인하된 금액" in v["discount_phrase"] # 동일가 v = eng.vars_for(_session(item_price=78000, input_price=78000)) assert "동일한 수준" in v["discount_phrase"] # 기존가 미보유(신규 협상) → 문구 생략 v = eng.vars_for(_session(item_price=0, input_price=79000)) assert v["discount_phrase"] == "" and v["discount_rate"] == "0.0" def test_price_confirm_script_renders_raise_correctly(): """가격협상_확인 멘트 E2E — 인상 제시에 '인하' 표현이 나오지 않는다.""" eng = _engine() s = _session(step="기존가격제시", item_price=78000, input_price=None, round=0) s.context.pop("input_price") view = eng.advance(s, "79000") assert view.step == "가격협상_확인" assert "인하" not in view.script # 인상인데 '인하' 금지 assert "높은 금액" in view.script and "79000원" in view.script # ---- vars_for 전술 변수 치환 --------------------------------------------------- def test_vars_for_supplies_tactic_variables(): eng = _engine() # 카운터 미제시(정보성): 절충/중간 변수는 원 계산값. s0 = _session(input_price=10300, prev_customer_price=10000) # anchor=10000, target=10100 (기본) v0 = eng.vars_for(s0) assert v0["prev_partner_price"] == 10300 assert v0["prev_customer_price"] == 10000 assert v0["target_mid_price"] == 10050 # (anchor 10000 + target 10100)/2 assert v0["middle_price"] == 10150 # (prev_customer 10000 + input 10300)/2 # 카운터 제시 중: 표시 제시가(middle/target_mid/counter) == 타결가(pending) 로 고정. # 회귀(표시가≠투찰가): 예전엔 middle_price 가 재계산값 10150 을 표시하면서 10100 으로 타결됐다. s1 = _session(input_price=10300, prev_customer_price=10000, pending_counter_price=10100) v1 = eng.vars_for(s1) assert v1["counter_price"] == 10100 assert v1["middle_price"] == 10100 # 재계산 10150 이 아니라 pending assert v1["target_mid_price"] == 10100 # ---- ⑤⑥ E2E (실 DB — 견적 선택 카드 + 서비스 레이어) --------------------------- from sqlalchemy import column, delete, insert, select, table # noqa: E402 from common.database.db_session_manager import DB_SESSION_MNG # noqa: E402 from common.enums import DBType, DBWRType, ErrorType # noqa: E402 from router.v1.chat.protocol import Req_Chat # noqa: E402 from services.chat_service import ChatService, reset_sessions # noqa: E402 from tenancy.registry import TenantEngineRegistry # noqa: E402 _T_SESSIONS = table( "sessions", column("session_id"), column("quotation_id"), column("item_id"), column("supplier_id"), column("qt_number"), column("qt_round"), column("qt_type"), column("target_price"), column("anchoring_price"), column("status"), column("end_time"), schema="negotiation", ) _T_QUOTATIONS = table( "quotations", column("qt_id"), column("user_id"), column("qt_setting_id"), column("version_id"), column("name"), column("number"), column("type"), column("status"), column("start_time"), column("end_time"), schema="quotation", ) _T_VNC = table("version_nego_cards", column("vnc_id"), column("version_id"), column("nego_card_id"), schema="card") _T_NEGO = table("nego_cards", column("nego_card_id"), column("number"), column("deleted"), schema="card") async def _card_uuid(number: str): def _q(s): return DB_SESSION_MNG.execute( s, select(_T_NEGO.c.nego_card_id).where( _T_NEGO.c.number == number, _T_NEGO.c.deleted == False).limit(1)) # noqa: E712 _, rows = await DB_SESSION_MNG.execute_lambda(DBType.MAIN.value, DBWRType.DB_READ.value, _q) return rows[0] if rows else None async def _seed_quote_session(sid, selected_numbers, target=10000, anchor=9900): qid, ver_id, iid, sup = _uuid.uuid4(), _uuid.uuid4(), _uuid.uuid4(), _uuid.uuid4() now = datetime.now(timezone.utc) card_ids = {} for n in selected_numbers: card_ids[n] = await _card_uuid(n) assert card_ids[n] is not None, f"카탈로그에 {n} 없음(시드 확인)" def _seed(s_): async def run(s): e = await DB_SESSION_MNG.add(s, insert(_T_QUOTATIONS).values( qt_id=qid, user_id=_uuid.uuid4(), qt_setting_id=_uuid.uuid4(), version_id=ver_id, name="전술E2E", number=f"QT-TACTIC-{str(sid)[:8]}", type=1, status=2, start_time=now, end_time=now + timedelta(days=1))) if e != ErrorType.SUCCESS: return e for n in selected_numbers: e = await DB_SESSION_MNG.add(s, insert(_T_VNC).values( vnc_id=_uuid.uuid4(), version_id=ver_id, nego_card_id=card_ids[n])) if e != ErrorType.SUCCESS: return e return await DB_SESSION_MNG.add(s, insert(_T_SESSIONS).values( session_id=sid, quotation_id=qid, item_id=iid, supplier_id=sup, qt_number=f"QT-TACTIC-{str(sid)[:8]}", qt_round=1, qt_type=1, target_price=target, anchoring_price=anchor, status=2, end_time=now + timedelta(days=1))) return run(s_) err = await DB_SESSION_MNG.execute_lambda_run([DBType.MAIN.value], [_seed]) assert err == ErrorType.SUCCESS return qid, ver_id async def _cleanup(sid, qid, ver_id): await DB_SESSION_MNG.execute_lambda_run( [DBType.MAIN.value], [lambda s: DB_SESSION_MNG.add(s, delete(_T_SESSIONS).where(_T_SESSIONS.c.session_id == sid)), lambda s: DB_SESSION_MNG.add(s, delete(_T_VNC).where(_T_VNC.c.version_id == ver_id)), lambda s: DB_SESSION_MNG.add(s, delete(_T_QUOTATIONS).where(_T_QUOTATIONS.c.qt_id == qid))], ) @pytest.mark.asyncio async def test_e2e_counter_accept_settles_at_target(db_engine): """견적 선택 카드 NGC-009(조건부 가격 조정 → COUNTER_TARGET)의 카운터를 수락하면 합의가 = 목표가(10000) — '수락 즉시 타결' 기획 결정의 E2E 검증.""" reset_sessions() sid = _uuid.uuid4() qid, ver_id = await _seed_quote_session(sid, ["NGC-009"]) try: reg = TenantEngineRegistry(loader=TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0)) eng = await reg.get_engine(str(_uuid.uuid4())) svc = ChatService() session_id = str(sid) r = None for ui in [None, "확인", "예", "확인", "11000", "예"]: r = await svc.chat(eng, Req_Chat(session_id=session_id, user_input=ui)) # 가격협상 카드 턴 → NGC-009 카운터(target) 제시 스텝 assert r.step == "가격협상_카운터", f"카운터 스텝 기대, 실제 {r.step}" assert r.card_id == "NGC-009" assert r.input_options == ["수락", "다른 가격 제시"] r = await svc.chat(eng, Req_Chat(session_id=session_id, user_input="수락")) assert r.step == "협상완료" assert r.settled_price == 10000 # 합의가 = 목표가 (고객사 이득) finally: await _cleanup(sid, qid, ver_id) @pytest.mark.asyncio async def test_e2e_over_target_ends_in_failure_after_closing(db_engine): """협력사가 목표가 초과(11000)를 고수하면: 카드 소진 → 종결 전술(목표가 최후통첩) → 그래도 거절 → 결렬(협상실패). 목표가 초과로는 절대 타결되지 않는다.""" reset_sessions() sid = _uuid.uuid4() qid, ver_id = await _seed_quote_session(sid, ["NGC-003"]) # HOLD 카드 1장 → 빠른 소진 try: reg = TenantEngineRegistry(loader=TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0)) eng = await reg.get_engine(str(_uuid.uuid4())) svc = ChatService() session_id = str(sid) steps, r = [], None # 고수 시나리오: 가격은 항상 11000, 카운터는 전부 거절 for ui in [None, "확인", "예", "확인", "11000", "예", "11000", "예"]: r = await svc.chat(eng, Req_Chat(session_id=session_id, user_input=ui)) steps.append(r.step) # 카드(NGC-003) 소진 → 종결 국면: 목표가 최후통첩 카운터 스텝 assert r.step == "가격협상_카운터", f"종결 카운터 기대, 실제 {steps}" assert "10000" in r.script # 최후통첩 = 목표가 제시 r = await svc.chat(eng, Req_Chat(session_id=session_id, user_input="다른 가격 제시")) r = await svc.chat(eng, Req_Chat(session_id=session_id, user_input="11000")) r = await svc.chat(eng, Req_Chat(session_id=session_id, user_input="예")) assert r.step == "협상실패" # target 초과 고수 → 결렬 r = await svc.chat(eng, Req_Chat(session_id=session_id, user_input="확인")) assert r.chat_end and r.outcome == "failure" # backend REJECTED → 개찰 이관 assert r.settled_price is None # 초과가 타결 없음 finally: await _cleanup(sid, qid, ver_id)