"""카드 전술 검증 — "스크립트에 꽂힌 변수가 곧 전술" (파싱 + 변수별 유효조건 + tactic JSONB). ① 제안가 파싱(마지막 제안가 변수) + 변수별 계산식 결정론 ② 변수 공통 유효조건 — 목표가 초과·제시가 이상이면 미발동(클램프 아님 — IMK 8AB0 회귀) ③ 카운터 수락 = 즉시 타결 / 거절 = 재입력 + pending 폐기 ④ 목표가 초과 타결 금지 가드(성공 스텝 진입 차단) ⑤ 와일드 진입 — 종결 전용 카드 예약(중반 미발동) + 카드 이력 공유(중복 발동 차단, IMK BB9A 회귀) ⑥ E2E: 견적 선택 카드(NGC-010 목표가 제안)의 카운터를 수락하면 settled=target ⑦ E2E: 협력사가 target 초과를 고수하면 종결 전술(최후통첩) 후 결렬 — 고객사 이득 가드레일 """ import os import uuid as _uuid from datetime import datetime, timedelta, timezone import pytest from negotiation.cards.domain.tactics import ( CardSpec, HOLD, available, build_card_spec, compute_offer, is_played, mark_played, parse_offer_variable, playable, spec_from_context, ) 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") # 엔진 단위 테스트용 카드 스펙(로더가 DB 스크립트 파싱으로 만드는 것과 같은 형태). _SPECS = { "WC-02": {"offer_variable": "target_mid_price", "min_round": 1, "closing": False}, "WC-05": {"offer_variable": "middle_price", "min_round": 1, "closing": True}, } 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, "card_specs": dict(_SPECS)} 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_parse_offer_variable_last_offer_wins(): """제안가 변수가 여럿이면 마지막 것 — 카드 문장은 배경을 먼저, 제안을 마지막에 한다(WC-04).""" assert parse_offer_variable("적정가는 {anchoring_price}원이었으나 {target_price}원으로 제안") == "target_price" assert parse_offer_variable("{target_price}원을 제안 드립니다") == "target_price" # 읽어주기 변수만 있으면 설득 카드 — 제안가 없음 assert parse_offer_variable("시장가 {internet_lowest_price}원 안팎, 제시가 {prev_partner_price}원") is None assert parse_offer_variable("가격 변수 없는 설득 멘트") is None assert parse_offer_variable(None) is None # WC-05 정본: 읽어주기(직전 제안·제시가) 뒤 절충가 제안 assert parse_offer_variable("당사 제안 {prev_customer_price}원과 귀사 제안 {prev_partner_price}원을 절반씩, {middle_price}원으로") == "middle_price" # negodata 에디터 칩 표기(anchor_price)도 앵커가 제안으로 인식 — DB 시드 표기(anchoring_price)의 별칭 assert parse_offer_variable("예산 한도는 {anchor_price}원입니다") == "anchor_price" def test_build_card_spec_merges_script_and_tactic(): spec = build_card_spec("{target_price}원으로 제안", {"min_round": 2, "closing": True}) assert spec == CardSpec(offer_variable="target_price", min_round=2, closing=True) # tactic 없음 → 기본값. offer_variable override 는 파싱보다 우선. assert build_card_spec("설득 멘트", None) == HOLD assert build_card_spec("멘트", {"offer_variable": "anchoring_price"}).offer_variable == "anchoring_price" def test_offer_formulas(): ctx = {"input_price": 11000, "anchor_price": 9900, "target_price": 10000} offer = lambda var, c=None: compute_offer(CardSpec(offer_variable=var), c or ctx) # noqa: E731 assert offer("target_price") == 10000 assert offer("anchoring_price") == 9900 assert offer("target_mid_price") == 9950 # (anchor+target)/2 # 절충가: 갑 직전 포지션 폴백 = anchor → (8900+9500)/2 = 9200 assert offer("middle_price", dict(ctx, input_price=9500, anchor_price=8900)) == 9200 # 갑 직전 포지션이 있으면 그 기준: (9000+9500)/2 = 9250 assert offer("middle_price", dict(ctx, input_price=9500, prev_customer_price=9000)) == 9250 # ---- ② 변수 공통 유효조건 — 미발동(클램프 아님) -------------------------------- def test_offer_over_target_does_not_fire_imk_8ab0(): """IMK 8AB0 회귀: 목표가 9,000 / 제시가 9,500 → 절충가 (8,910+9,500)/2 = 9,205 > 목표가. 구현이 목표가로 깎아 부르면 '중간에서 만나자며 목표가를 부르는' 모순 — 클램프가 아니라 미발동이 정답.""" ctx = {"input_price": 9500, "anchor_price": 8910, "target_price": 9000} assert compute_offer(CardSpec(offer_variable="middle_price"), ctx) is None def test_offer_at_or_above_input_price_does_not_fire(): """협력사 제시가가 이미 제안가 이하면 부를 이유가 없다 → 미발동.""" ctx = {"input_price": 9950, "anchor_price": 9900, "target_price": 10000} assert compute_offer(CardSpec(offer_variable="target_price"), ctx) is None # target ≥ 제시가 assert compute_offer(CardSpec(offer_variable="anchoring_price"), dict(ctx, input_price=9900)) is None def test_offer_without_materials_does_not_fire(): """재료 결측(목표가·제시가·앵커) — 어떤 변수도 미발동.""" assert compute_offer(CardSpec(offer_variable="target_price"), {"input_price": 11000}) is None # 목표가 없음 assert compute_offer(CardSpec(offer_variable="anchoring_price"), {"input_price": 11000, "target_price": 10000}) is None # 앵커 없음 assert compute_offer(HOLD, {"input_price": 11000, "target_price": 10000}) is None # 설득 카드 assert compute_offer(CardSpec(offer_variable="없는변수"), {"input_price": 11000, "target_price": 10000}) is None def test_available_min_round_and_closing_phase(): spec2 = CardSpec(offer_variable="target_price", min_round=2) assert available(spec2, {"round": 1}) is False # min_round 미만 assert available(spec2, {"round": 2}) is True closing = CardSpec(offer_variable="middle_price", closing=True) assert available(closing, {"round": 1}) is False # 종결 전용 — 중반 미발동(예약) assert available(closing, {"round": 1}, closing_phase=True) is True assert available(spec2, {"round": 3}, closing_phase=True) is False # 종결 국면엔 종결 카드만 def test_tactic_offer_variable_overrides_parse(): """검증: tactic.offer_variable 명시 지정(negodata 셀렉트) — 파싱(마지막 변수) 대신 지정 변수 사용. 기대결과: 멘트 마지막이 target_price 여도 지정한 anchoring_price 가 제안가 변수가 된다.""" script = "적정가는 {anchoring_price}원이었으나 {target_price}원으로 제안 드립니다." assert build_card_spec(script).offer_variable == "target_price" # 자동: 마지막 변수 spec = build_card_spec(script, {"offer_variable": "anchoring_price"}) assert spec.offer_variable == "anchoring_price" # 명시 지정이 우선 def test_available_requires_context_value(): """검증: 시장가 인용 카드(NGC-008류)의 requires 게이트 — build_card_spec 이 스크립트에서 잡아내고, 기대결과: 컨텍스트에 인터넷 최저가가 없으면(0/결측) 미발동, 있으면 발동(퍼즈 #3·13·23·40 회귀).""" spec = build_card_spec("유사 거래는 {internet_lowest_price}원 안팎에서 합의되고 있습니다.") assert spec.requires == ("internet_lowest_price",) assert available(spec, {"round": 1}) is False # 결측 assert available(spec, {"round": 1, "internet_lowest_price": 0}) is False # 미수집(0) assert available(spec, {"round": 1, "internet_lowest_price": 6300}) is True # 일반 카드는 requires 없음 — 기존 동작 그대로. assert build_card_spec("귀사와의 협력을 소중히 생각합니다.").requires == () def test_offer_monotonic_no_regression(): """검증: 역행 금지(IMK 논의 — 절충 16,980 후 예산 상한 16,810 제시) 재현. 기대결과: 직전 당사 제안보다 낮은 제안가 카드는 미발동(설득 폴백으로도 안 나감). 직전 제안이 없으면 앵커 제시 허용, 같은 금액 재제시 허용, 더 높은 제안은 정상.""" anchor_card = CardSpec(offer_variable="anchoring_price") ctx = {"round": 2, "target_price": 17_300, "anchor_price": 16_810, "input_price": 17_500} assert compute_offer(anchor_card, ctx) == 16_810 # 첫 카운터 전(포지션=앵커): 같은 금액 → 허용 ctx["prev_customer_price"] = 16_980 # 절충 카드가 이미 16,980 을 부른 상태 assert compute_offer(anchor_card, ctx) is None # 앵커 16,810 은 역행 → 미발동 assert playable(anchor_card, ctx) is False # 멘트에 금액이 박히므로 설득 폴백도 금지 assert compute_offer(CardSpec(offer_variable="target_price"), ctx) == 17_300 # 상향 제안은 정상 def test_played_history_is_shared_by_number(): ctx = {} assert is_played(ctx, "WC-05") is False mark_played(ctx, "WC-05") assert is_played(ctx, "WC-05") is True mark_played(ctx, "WC-05") # 재기록해도 1건 유지 assert ctx["played_card_numbers"] == ["WC-05"] mark_played(ctx, None) # no-op(폴백 최후통첩) assert ctx["played_card_numbers"] == ["WC-05"] def test_spec_from_context_reads_snapshot_and_falls_back_to_hold(): ctx = {"card_specs": dict(_SPECS)} assert spec_from_context(ctx, "WC-05") == CardSpec(offer_variable="middle_price", min_round=1, closing=True) assert spec_from_context(ctx, "NGC-B003") == HOLD # 미등록 카드(데모) 폴백 assert spec_from_context({}, "WC-05") == HOLD # 스펙 미적재(구세션·데모) 폴백 # ---- ③ 카운터 수락/거절 메커니즘 (엔진) --------------------------------------- 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 == "협상실패" # 초과가 성공 진입 → 결렬 강제 # ---- ⑤ 와일드 진입 — 종결 예약 + 중복 차단 (IMK BB9A 회귀) --------------------- def test_selected_wildcard_fires_in_entry_zone_and_records_position(): eng = _engine() # 10300: 1pct 존(≤10200) 밖, entry 존(≤10500) 안 + 비종결 WC-02 선택 s = _session(input_price=10300, allow_selected_wildcards=True, selected_wild_card_numbers=["WC-02"]) view = eng.advance(s, "예") assert view.step == "wild_card_dynamic" # 제안가 = (anchor 10000 + target 10100)/2 = 10050 ≤ target — 그대로 제시(클램프 없음) assert s.context["pending_counter_price"] == 10050 assert s.context["prev_customer_price"] == 10050 # 갑 포지션 기록 — "당사 제안" 멘트 정합(BB9A ③) assert s.context["active_wild_card_number"] == "WC-02" assert is_played(s.context, "WC-02") # 카드 이력 기록 # 수락 → 그 가격으로 타결 view = eng.advance(s, "수락") assert view.step == "협상완료" and s.context["input_price"] == 10050 def test_closing_card_is_reserved_never_fires_mid_negotiation(): """종결 전용 카드(WC-05)는 entry 존이라도 중반에 안 나간다 — 종결 국면의 마지막 한 방으로 예약. (BB9A 중복의 절반: 중반에 당겨 쓴 카드를 종결에서 또 쓰던 경로 차단.)""" eng = _engine() s = _session(input_price=10300, allow_selected_wildcards=True, selected_wild_card_numbers=["WC-05"]) view = eng.advance(s, "예") assert view.step == "가격협상" # 종결 카드뿐 → 일반 카드 플레이로 assert "active_wild_card_number" not in s.context assert not is_played(s.context, "WC-05") # 안 나갔으니 이력도 없음 # 종결 국면에선 발동 가능 + 이력 없음 — 서비스 종결 루프가 이 카드를 쓴다 spec = spec_from_context(s.context, "WC-05") assert available(spec, s.context, closing_phase=True) is True def test_played_wildcard_is_skipped_on_reentry(): """이미 쓴 카드는 같은 협상에서 다시 안 나간다 — 다음 후보로 넘어간다.""" eng = _engine() s = _session(input_price=10300, allow_selected_wildcards=True, wildcard_used=False, selected_wild_card_numbers=["WC-02"], played_card_numbers=["WC-02"]) view = eng.advance(s, "예") assert view.step == "가격협상" # 유일 후보가 사용됨 → 발동 없음 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") _T_VWC = table("version_wild_cards", column("vwc_id"), column("version_id"), column("wild_card_id"), schema="card") _T_WILD = table("wild_cards", column("wild_card_id"), column("number"), column("deleted"), schema="card") async def _card_uuid(number: str, *, wild=False): tbl, pk = (_T_WILD, _T_WILD.c.wild_card_id) if wild else (_T_NEGO, _T_NEGO.c.nego_card_id) def _q(s): return DB_SESSION_MNG.execute( s, select(pk).where(tbl.c.number == number, tbl.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, wild_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, wild_ids = {}, {} for n in selected_numbers: card_ids[n] = await _card_uuid(n) assert card_ids[n] is not None, f"카탈로그에 {n} 없음(시드 확인)" for n in wild_numbers: wild_ids[n] = await _card_uuid(n, wild=True) assert wild_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 for n in wild_numbers: e = await DB_SESSION_MNG.add(s, insert(_T_VWC).values( vwc_id=_uuid.uuid4(), version_id=ver_id, wild_card_id=wild_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_VWC).where(_T_VWC.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-010(향후 거래 연계 — 스크립트 {target_price} 파싱 → 목표가 제안)의 카운터를 수락하면 합의가 = 목표가(10000) — '수락 즉시 타결' 기획 결정의 E2E 검증.""" reset_sessions() sid = _uuid.uuid4() qid, ver_id = await _seed_quote_session(sid, ["NGC-010"]) 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-010 카운터(target) 제시 스텝 assert r.step == "가격협상_카운터", f"카운터 스텝 기대, 실제 {r.step}" assert r.card_id == "NGC-010" 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"]) # 설득 카드 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) @pytest.mark.asyncio async def test_e2e_bb9a_no_duplicate_wildcard_and_real_middle(db_engine): """IMK BB9A 재현 E2E — 와일드카드 2장(WC-02·WC-05) + 설득 카드 1장. 기대 흐름(수정 후): · 중반 와일드 진입 = 비종결 WC-02 (종결 전용 WC-05 는 예약 — 구현 전엔 WC-05 가 먼저 나갔다) · 종결 국면 = WC-05, 절충가 = (당사 직전 제안 + 협력사 제시가)/2 실계산 (구현 전엔 목표가로 클램프) · 같은 카드 2회 발동 없음 + 종결 발동도 card_id 기록 """ reset_sessions() sid = _uuid.uuid4() qid, ver_id = await _seed_quote_session(sid, ["NGC-003"], wild_numbers=["WC-02", "WC-05"]) 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 # 10300: 1pct 존(≤ 9900×1.02=10098) 밖, entry 존(≤ 10395) 안 → 선택형 와일드 발동 구간 for ui in [None, "확인", "예", "확인", "10300", "예"]: r = await svc.chat(eng, Req_Chat(session_id=session_id, user_input=ui)) assert r.step == "wild_card_dynamic" assert r.card_id == "WC-02" # 종결 전용 WC-05 가 아니라 비종결 카드 # WC-02 제안가 = (anchor 9900 + target 10000)/2 = 9950 r = await svc.chat(eng, Req_Chat(session_id=session_id, user_input="다른 가격 제시")) # 10010 재제시 → 설득 카드(NGC-003) 1장 소진 r = await svc.chat(eng, Req_Chat(session_id=session_id, user_input="10010")) r = await svc.chat(eng, Req_Chat(session_id=session_id, user_input="예")) assert r.step == "가격협상" and r.card_id == "NGC-003" # 10005 재제시 → 카드 소진 → 종결 국면: WC-05 절충가 = (9950 + 10005)/2 = 9980 (≤ target) r = await svc.chat(eng, Req_Chat(session_id=session_id, user_input="10005")) r = await svc.chat(eng, Req_Chat(session_id=session_id, user_input="예")) assert r.step == "가격협상_카운터" assert r.card_id == "WC-05" # 종결 발동도 카드 기록(구현 전 null) assert "9980" in r.script # 실제 절충가 — 목표가(10000) 클램프 아님 r = await svc.chat(eng, Req_Chat(session_id=session_id, user_input="수락")) assert r.step == "협상완료" assert r.settled_price == 9980 # 표시가 = 타결가 finally: await _cleanup(sid, qid, ver_id)