"""앵커링 채팅 연동 테스트 — 박제값 소비 / 무할인 폴백 / 마지막 제시가(가격 흔적) 기록. 배치·조정 로직은 schedules/anchoring/tests 소관 — 여기는 backend 채팅 경로만 검증한다. 실제 agent 대신 결정론적 더블(_AnchorAgent)을 주입하고, dev negosium_db 에 전용 행만 시드/정리한다. (규범: schedules/anchoring/docs/개발용.md §9.2 — 표본 기준은 노출이 아니라 "가격을 써냈는가") """ import uuid import bcrypt import pytest import pytest_asyncio from sqlalchemy import text from services.agent_client import AgentTurn, IAgentClient, get_agent_client TEST_LOGIN_ID = "pytest_anchor_user" TEST_PW = "pytest1234" TEST_SUPPLIER_NAME = "파이테스트앵커공급사" MARK = "PYTESTANCHOR-" TARGET = 100_000 ANCHOR = 99_000 # negodata 가 생성 시 박제하는 값(rate 10‰) 시뮬레이션 def _parse_price(text_): digits = "".join(ch for ch in (text_ or "") if ch.isdigit()) return int(digits) if digits else None class _AnchorAgent(IAgentClient): """결정론적 더블: 서비스안내(오프닝) → 가격 입력 요청 → 합의 종료. 앵커보다 높은 가격이면 같은 step 을 반복(마지막 제시가 덮어쓰기 검증용). 매 턴 수신한 ctx.anchor_price 를 기록해 backend 의 앵커 해석을 관찰한다. (script 에 앵커가 보이는 건 테스트 관찰 편의일 뿐 — 실제 agent 는 비노출.) """ def __init__(self): self.seen_anchors: list[int] = [] async def chat(self, session_id, user_input, ctx) -> AgentTurn: self.seen_anchors.append(ctx.anchor_price) sid = session_id or "fake-session" if user_input is None: # 오프닝(턴0) 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=f"저희가 제안드리는 첫 목표 가격은 {ctx.anchor_price}원입니다. " f"제안하실 가격을 입력해 주세요.", input_mode="price") price = _parse_price(user_input) if price is not None and price <= ctx.anchor_price: return AgentTurn(session_id=sid, step="협상종료", client_step="협상종료", script=f"{price:,}원으로 합의되었습니다.", chat_end=True, outcome="success") return AgentTurn(session_id=sid, step="기존가격제시", client_step="기존가격제시", script="조금 더 조정된 가격을 부탁드립니다.", input_mode="price") @pytest.fixture(autouse=True) def _fake_agent(): from router.router import app agent = _AnchorAgent() app.dependency_overrides[get_agent_client] = lambda: agent yield agent app.dependency_overrides.pop(get_agent_client, None) @pytest_asyncio.fixture async def anchor_seed(db_engine): """공급사+유저 + 세션 2건: A(앵커 박제됨 — 정상 경로) / N(박제 NULL — 폴백 경로).""" supplier_id = uuid.uuid4() pw_hash = bcrypt.hashpw(TEST_PW.encode("utf-8"), bcrypt.gensalt()).decode("utf-8") sids = {} 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, anchor, rate in (("A", ANCHOR, 10), ("N", None, None)): item_id, qt_id, session_id = uuid.uuid4(), uuid.uuid4(), uuid.uuid4() sids[code] = session_id await conn.execute( text("INSERT INTO partner.items (item_id, company_id, user_id, name, code, price) " "VALUES (:iid, gen_random_uuid(), gen_random_uuid(), :name, :code, 100000)"), {"iid": item_id, "name": f"앵커상품 {code}", "code": f"{MARK}{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, supplier_type) " "VALUES (:qid, gen_random_uuid(), gen_random_uuid(), gen_random_uuid(), :name, :num, 1, 2, now(), now() + interval '2 hours', 1)"), {"qid": qt_id, "name": f"앵커견적 {code}", "num": f"{MARK}{code}"}, ) await conn.execute( text("INSERT INTO negotiation.sessions " "(session_id, quotation_id, item_id, supplier_id, qt_number, qt_round, qt_type, " " target_price, anchoring_price, anchoring_value, status, end_time) " "VALUES (:sesid, :qid, :iid, :sup, :qtn, 1, 1, :tp, :ap, :rate, 2, now() + interval '2 hours')"), {"sesid": session_id, "qid": qt_id, "iid": item_id, "sup": supplier_id, "qtn": f"{MARK}{code}", "tp": TARGET, "ap": anchor, "rate": rate}, ) yield {"sids": sids} 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 _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 _anchor_columns(db_engine, session_id): async with db_engine.begin() as conn: row = (await conn.execute(text( "SELECT anchoring_price, anchoring_value, last_offer_price, bid_price " "FROM negotiation.sessions WHERE session_id = :sid"), {"sid": session_id})).one() return row # ── 정상 경로: 박제값 소비 + 마지막 제시가 기록(가격 흔적) ── async def test_snapshot_consumed_and_last_offer_recorded(client, db_engine, anchor_seed, _fake_agent): sid = str(anchor_seed["sids"]["A"]) token = await _login_token(client) r = await _messages(client, token, sid) # 오프닝 seed assert r.status_code == 200 r = await _send(client, token, sid, "네, 시작할게요") # → 가격 입력 요청 (아직 가격 흔적 없음) assert r.status_code == 200 and r.json()["message"]["step"] == "기존가격제시" row = await _anchor_columns(db_engine, sid) assert row.last_offer_price is None assert _fake_agent.seen_anchors[-1] == ANCHOR # backend 가 박제값을 그대로 전달 r = await _send(client, token, sid, "99,500", "price") # 앵커 초과 → 같은 step 반복 assert r.json()["message"]["step"] == "기존가격제시" row = await _anchor_columns(db_engine, sid) assert row.last_offer_price == 99_500 # 가격 흔적 기록 # 이 시점에 이탈해 일괄마감(NOT_PARTICIPATED)돼도 last_offer_price 로 실패 표본이 된다. r = await _send(client, token, sid, "98,000", "price") # 앵커 이하 → 합의 종료 assert r.json()["session_status"] == 3 # DONE row = await _anchor_columns(db_engine, sid) assert row.last_offer_price == 98_000 # 마지막 값으로 갱신 assert row.bid_price == 98_000 assert (row.anchoring_price, row.anchoring_value) == (ANCHOR, 10) # 박제 불변 # ── 폴백 경로: 박제 NULL → 무할인(anchor=target) + 미박제 유지 ── async def test_null_snapshot_falls_back_to_target(client, db_engine, anchor_seed, _fake_agent): sid = str(anchor_seed["sids"]["N"]) token = await _login_token(client) await _messages(client, token, sid) r = await _send(client, token, sid, "네, 시작할게요") assert r.json()["message"]["step"] == "기존가격제시" assert _fake_agent.seen_anchors[-1] == TARGET # 무할인 폴백: anchor = target r = await _send(client, token, sid, "97,000", "price") # 가격 입력(폴백 앵커 이하 → 종료) assert r.json()["session_status"] == 3 row = await _anchor_columns(db_engine, sid) assert row.anchoring_price is None # backend 는 박제하지 않음(앵커 없음 → 집계 제외) assert row.anchoring_value is None assert row.last_offer_price == 97_000 # 가격 흔적 기록은 정상 동작