hbyang
3eec80d989
1) 표시가≠투찰가 (배포 서버 협상 결과 오류)
- 카운터 제시 중 멘트의 절충/중간 변수(middle_price·target_mid_price)가 vars_for 재계산으로
compute_counter 의 target 클램프·prev_customer 갱신과 어긋나, 화면엔 1,740,000 인데 실제로는
1,700,000 으로 타결되던 문제 → vars_for 에서 세 변수(counter/middle/target_mid)를 pending 으로 고정
- WC-03 최후통첩 멘트가 제시 금액을 안 보여줘 수락 시 화면에 없던 target 으로 타결되던 문제 →
멘트에 {target_price} 명시(seed init-data.sql). 운영 DB 는 별도 UPDATE 필요
2) 카드 사용 횟수 3회 초과
- quotation_settings.card_count(협상카드 사용 횟수 상한, 기본 3)가 어디서도 강제되지 않던 죽은 설정 →
에이전트가 sessions.qt_setting_id 로 card_count 를 조회해 협상카드 재생 수를 min(선택수, card_count)로 캡
(session.action_space_size 는 소진 판정 전용 — Q-table 은 카탈로그 전체로 별도 고정)
- 회귀 테스트: 표시가==타결가(test_card_tactics·test_p7_chat), card_count 로드(test_context_loader). 전체 153 pass
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
251 lines
12 KiB
Python
251 lines
12 KiB
Python
"""NegotiationContextLoader 검증 — Req_Chat 슬림화 후 세션 시작 컨텍스트 DB 조회.
|
|
|
|
backend 소유 스키마(negotiation.sessions / quotation.quotations / partner.items / partner.suppliers)에
|
|
실데이터를 넣고, agent 가 session_id 만으로 rq_type·목표가·앵커링가·품목가·매출액(total_revenue)·
|
|
유통코드·파트너 유형을 확정하는지 검증한다. 삽입 행은 테스트 종료 시 삭제.
|
|
"""
|
|
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from sqlalchemy import column, delete, insert, table
|
|
|
|
from common.database.db_session_manager import DB_SESSION_MNG
|
|
from common.enums import DBType, ErrorType
|
|
from negotiation.chat.service.chat_session_repository import ChatSessionRepository
|
|
from router.v1.chat.protocol import Req_Chat
|
|
from services.chat_service import ChatService, reset_sessions
|
|
from tenancy.config_loader import TenantConfigLoader
|
|
from tenancy.registry import TenantEngineRegistry
|
|
import os
|
|
|
|
_TENANTS_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "tenants")
|
|
|
|
# 삽입용 테이블 구성(backend 소유 스키마 — 테스트 데이터 셋업 전용).
|
|
_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_ITEMS = table(
|
|
"items",
|
|
column("item_id"), column("company_id"), column("user_id"), column("name"), column("price"),
|
|
schema="partner",
|
|
)
|
|
_T_SUPPLIERS = table(
|
|
"suppliers",
|
|
column("supplier_id"), column("company_id"), column("user_id"), column("name"), column("total_revenue"),
|
|
schema="partner",
|
|
)
|
|
_T_SUPPLIER_ITEMS = table(
|
|
"supplier_items",
|
|
column("supplier_id"), column("item_id"), column("supply_type"),
|
|
schema="partner",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_context_loaded_from_db(db_engine):
|
|
"""세션 시작 시 협상 컨텍스트가 요청이 아니라 DB 에서 확정된다."""
|
|
reset_sessions()
|
|
sid, sid2 = uuid.uuid4(), uuid.uuid4()
|
|
qid, iid = uuid.uuid4(), uuid.uuid4()
|
|
sup1, sup2 = uuid.uuid4(), uuid.uuid4()
|
|
now = datetime.now(timezone.utc)
|
|
|
|
def _ins_item(s):
|
|
return DB_SESSION_MNG.add(s, insert(_T_ITEMS).values(
|
|
item_id=iid, company_id=uuid.uuid4(), user_id=uuid.uuid4(),
|
|
name="로더 테스트 상품", price=5000,
|
|
))
|
|
|
|
def _ins_supplier(s):
|
|
return DB_SESSION_MNG.add(s, insert(_T_SUPPLIERS).values(
|
|
supplier_id=sup1, company_id=uuid.uuid4(), user_id=uuid.uuid4(),
|
|
name="로더 테스트 협력사", total_revenue=55_000_000, # 매출액(KTC total_revenue 미러)
|
|
))
|
|
|
|
def _ins_quote(s):
|
|
return DB_SESSION_MNG.add(s, insert(_T_QUOTATIONS).values(
|
|
qt_id=qid, user_id=uuid.uuid4(), qt_setting_id=uuid.uuid4(), version_id=uuid.uuid4(),
|
|
name="로더 테스트", number="QT-LOADER-TEST", type=3, status=2,
|
|
start_time=now, end_time=now + timedelta(days=1),
|
|
))
|
|
|
|
def _ins_sess(s, session_id, supplier_id):
|
|
return DB_SESSION_MNG.add(s, insert(_T_SESSIONS).values(
|
|
session_id=session_id, quotation_id=qid, item_id=iid, supplier_id=supplier_id,
|
|
qt_number="QT-LOADER-TEST", qt_round=1,
|
|
qt_type=3, # 신규협상(1:1) → 재협상 스크립트
|
|
target_price=20000, anchoring_price=19000, # 생성 시 박제된 앵커
|
|
status=2, end_time=now + timedelta(days=1),
|
|
))
|
|
|
|
def _ins_mapping(s, supplier_id, supply_type):
|
|
return DB_SESSION_MNG.add(s, insert(_T_SUPPLIER_ITEMS).values(
|
|
supplier_id=supplier_id, item_id=iid, supply_type=supply_type,
|
|
))
|
|
|
|
err = await DB_SESSION_MNG.execute_lambda_run(
|
|
[DBType.MAIN.value],
|
|
[_ins_item, _ins_supplier, _ins_quote,
|
|
lambda s: _ins_sess(s, sid, sup1),
|
|
lambda s: _ins_sess(s, sid2, sup2),
|
|
# 상품↔협력사 매핑(supplier_items): 협상 상대 sup1 은 총판(3), sup2 는 유통(1) → 취급 2곳
|
|
lambda s: _ins_mapping(s, sup1, 3),
|
|
lambda s: _ins_mapping(s, sup2, 1)],
|
|
)
|
|
assert err == ErrorType.SUCCESS
|
|
|
|
try:
|
|
reg = TenantEngineRegistry(loader=TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0))
|
|
eng = await reg.get_engine("imarketkorea")
|
|
r = await ChatService().chat(eng, Req_Chat(session_id=str(sid)))
|
|
assert r.session_id == str(sid) and r.step == "서비스안내"
|
|
|
|
saved = await ChatSessionRepository(eng.company_id).get(str(sid))
|
|
assert saved is not None
|
|
assert saved.rq_type == "재협상" # qt_type=3(신규협상 1:1)
|
|
c = saved.context
|
|
assert c["target_price"] == 20000
|
|
assert c["anchor_price"] == 19000 # sessions.anchoring_price(박제)
|
|
assert c["item_price"] == 5000
|
|
assert c["revenue_amount"] == 55_000_000.0 # 매출액 = suppliers.total_revenue
|
|
assert c["distribution_code"] == "B" # supplier_items.supply_type=3(총판) → B (매핑 우선)
|
|
assert c["partner_count"] == 2 # 매핑 기준 취급 협력사 2곳 → MULTIPLE
|
|
|
|
# 매핑 삭제 후 새 세션(sid2) → 유통코드는 ChatService 기본값, 파트너는 세션 이력 폴백
|
|
err = await DB_SESSION_MNG.execute_lambda_run(
|
|
[DBType.MAIN.value],
|
|
[lambda s: DB_SESSION_MNG.add(s, delete(_T_SUPPLIER_ITEMS).where(_T_SUPPLIER_ITEMS.c.item_id == iid))],
|
|
)
|
|
assert err == ErrorType.SUCCESS
|
|
await ChatService().chat(eng, Req_Chat(session_id=str(sid2)))
|
|
c2 = (await ChatSessionRepository(eng.company_id).get(str(sid2))).context
|
|
assert c2["distribution_code"] == "A" # 매핑 없음 → ChatService 기본값
|
|
assert c2["partner_count"] == 2 # 폴백: 세션 이력 distinct supplier 2곳
|
|
finally:
|
|
await DB_SESSION_MNG.execute_lambda_run(
|
|
[DBType.MAIN.value],
|
|
[lambda s: DB_SESSION_MNG.add(s, delete(_T_SESSIONS).where(_T_SESSIONS.c.quotation_id == qid)),
|
|
lambda s: DB_SESSION_MNG.add(s, delete(_T_QUOTATIONS).where(_T_QUOTATIONS.c.qt_id == qid)),
|
|
lambda s: DB_SESSION_MNG.add(s, delete(_T_ITEMS).where(_T_ITEMS.c.item_id == iid)),
|
|
lambda s: DB_SESSION_MNG.add(s, delete(_T_SUPPLIER_ITEMS).where(_T_SUPPLIER_ITEMS.c.item_id == iid)),
|
|
lambda s: DB_SESSION_MNG.add(s, delete(_T_SUPPLIERS).where(_T_SUPPLIERS.c.supplier_id == sup1))],
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_null_anchoring_falls_back_to_target(db_engine):
|
|
"""박제 앵커(anchoring_price)가 NULL 이면 무할인 폴백 anchor=target (앵커링 v1.2 정책 승계)."""
|
|
reset_sessions()
|
|
sid, qid, iid = uuid.uuid4(), uuid.uuid4(), uuid.uuid4()
|
|
now = datetime.now(timezone.utc)
|
|
|
|
def _ins_sess(s):
|
|
return DB_SESSION_MNG.add(s, insert(_T_SESSIONS).values(
|
|
session_id=sid, quotation_id=qid, item_id=iid, supplier_id=uuid.uuid4(),
|
|
qt_number="QT-LOADER-NULL", qt_round=1, qt_type=1,
|
|
target_price=30000, anchoring_price=None, # 박제 없음(데이터 이상 경로)
|
|
status=2, end_time=now + timedelta(days=1),
|
|
))
|
|
|
|
err = await DB_SESSION_MNG.execute_lambda_run([DBType.MAIN.value], [_ins_sess])
|
|
assert err == ErrorType.SUCCESS
|
|
try:
|
|
reg = TenantEngineRegistry(loader=TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0))
|
|
eng = await reg.get_engine("imarketkorea")
|
|
await ChatService().chat(eng, Req_Chat(session_id=str(sid)))
|
|
saved = await ChatSessionRepository(eng.company_id).get(str(sid))
|
|
assert saved is not None
|
|
assert saved.context["target_price"] == 30000
|
|
assert saved.context["anchor_price"] == 30000 # 무할인 폴백: anchor = target
|
|
finally:
|
|
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))],
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_loader_with_crud_double(db_engine):
|
|
"""INegoContextCRUD 인터페이스에 더블을 주입해 쿼리 없이 판정 로직만 검증 (backend crud 패턴)."""
|
|
from negotiation.chat.infra.repository.nego_context_crud import INegoContextCRUD
|
|
from negotiation.chat.service.negotiation_context_loader import NegotiationContextLoader
|
|
from negotiation.qtable.domain.model.snapshot import PartnerType
|
|
|
|
class _FakeCRUD(INegoContextCRUD):
|
|
async def get_session_row(self, cdb, session_id):
|
|
# (qt_type, target, anchoring_price, item_id, quotation_id, supplier_id) — 재견적(2)·앵커 미박제
|
|
return ErrorType.SUCCESS, (2, 50000, None, uuid.uuid4(), uuid.uuid4(), uuid.uuid4())
|
|
|
|
async def get_item_price(self, cdb, item_id):
|
|
return ErrorType.SUCCESS, 7000
|
|
|
|
async def get_item_lowest_price(self, cdb, item_id):
|
|
return ErrorType.SUCCESS, 6300 # 인터넷 최저가(items.internet_lowest_price)
|
|
|
|
async def get_card_count(self, cdb, session_id):
|
|
return ErrorType.SUCCESS, 3 # 협상카드 사용 횟수 상한(quotation_settings.card_count)
|
|
|
|
async def get_supplier_total_revenue(self, cdb, supplier_id):
|
|
return ErrorType.SUCCESS, 12_000_000.0
|
|
|
|
async def get_item_name(self, cdb, item_id):
|
|
return ErrorType.SUCCESS, "테스트상품"
|
|
|
|
async def get_supplier_name(self, cdb, supplier_id):
|
|
return ErrorType.SUCCESS, "테스트협력사"
|
|
|
|
async def get_supply_type(self, cdb, supplier_id, item_id):
|
|
return ErrorType.SUCCESS, 3 # sole_agency(총판) → "B"
|
|
|
|
async def count_item_suppliers(self, cdb, item_id):
|
|
return ErrorType.SUCCESS, 0 # 매핑 없음 → 세션 이력 폴백
|
|
|
|
async def count_item_session_suppliers(self, cdb, item_id):
|
|
return ErrorType.SUCCESS, 0 # 이력도 없음 → NONE
|
|
|
|
async def get_quotation_card_numbers(self, cdb, quotation_id):
|
|
return ErrorType.SUCCESS, (["NGC-003", "NGC-008"], ["WC-02"]) # 견적 선택 카드
|
|
|
|
ctx = await NegotiationContextLoader(crud=_FakeCRUD()).load(str(uuid.uuid4()))
|
|
assert ctx is not None
|
|
assert ctx.rq_type == "재견적" # qt_type=2(1:N)
|
|
assert ctx.target_price == 50000
|
|
assert ctx.anchor_price == 50000 # 미박제 → 무할인 폴백(anchor=target)
|
|
assert ctx.item_price == 7000
|
|
assert ctx.internet_lowest_price == 6300 # 인터넷 최저가 로드 확인
|
|
assert ctx.card_count == 3 # 협상카드 사용 횟수 상한 로드 확인
|
|
assert ctx.partner_name == "테스트협력사"
|
|
assert ctx.product_name == "테스트상품"
|
|
assert ctx.revenue_amount == 12_000_000.0
|
|
assert ctx.distribution_code == "B" # supply_type=3(총판) → B
|
|
assert ctx.partner_type is PartnerType.NONE
|
|
assert ctx.selected_nego_card_numbers == ["NGC-003", "NGC-008"]
|
|
assert ctx.selected_wild_card_numbers == ["WC-02"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_context_falls_back_without_db_row(db_engine):
|
|
"""DB 에 세션 행이 없으면(데모/직접 호출) 기본 컨텍스트로 폴백한다."""
|
|
reset_sessions()
|
|
reg = TenantEngineRegistry(loader=TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0))
|
|
eng = await reg.get_engine("imarketkorea")
|
|
r = await ChatService().chat(eng, Req_Chat())
|
|
saved = await ChatSessionRepository(eng.company_id).get(r.session_id)
|
|
assert saved is not None
|
|
c = saved.context
|
|
assert c["target_price"] == 10000 and c["anchor_price"] == 9900
|
|
assert c["partner_count"] == 1 and c["distribution_code"] == "A"
|