o2o-negosium-original/agent/tests/test_card_selection_e2e.py

"""협상카드 선택 E2E — 실 DB 왕복으로 "견적에서 고른 카드만 뽑히는지" 검증.

시나리오: 견적 생성 시 협상카드 2장(NGC-003, NGC-007)만 선택 →
version_nego_cards 로 연결 → 협상 세션 시작 → 가격협상 턴 2회 진행.
검증: ① 뽑힌 카드가 선택 2장 안에서만 나옴(선택 마스크) ② 세션 내 중복 없음(사용 마스크)
③ 카탈로그(DB, NGC-001~011) 기준 action space ④ 선택 없으면 전체 카탈로그 허용(폴백).
"""

import os
import uuid as _uuid
from datetime import datetime, timedelta, timezone

import pytest
from sqlalchemy import column, delete, insert, select, table

from common.database.db_session_manager import DB_SESSION_MNG
from common.enums import DBType, DBWRType, ErrorType
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

_TENANTS_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "tenants")

_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


@pytest.mark.asyncio
async def test_selected_cards_only_are_played(db_engine):
    reset_sessions()
    sid, qid, ver_id = _uuid.uuid4(), _uuid.uuid4(), _uuid.uuid4()
    iid, sup = _uuid.uuid4(), _uuid.uuid4()
    now = datetime.now(timezone.utc)
    selected = ["NGC-003", "NGC-007"]
    card_ids = {}
    for n in selected:
        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="QT-CARDSEL-E2E", type=1, status=2,
                start_time=now, end_time=now + timedelta(days=1)))
            if e != ErrorType.SUCCESS:
                return e
            for n in selected:  # 견적 생성 시 선택한 카드 2장
                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="QT-CARDSEL-E2E", qt_round=1, qt_type=1,
                target_price=10000, anchoring_price=9900, 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

    try:
        reg = TenantEngineRegistry(loader=TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0))
        eng = await reg.get_engine(str(_uuid.uuid4()))   # 자동 온보딩(_base type:db → 실 DB 카탈로그)
        assert eng.action_space_size == 11               # 카탈로그 11장(NGC-001~011)

        svc = ChatService()
        played = []
        session_id = str(sid)
        # 적응형 진행: 카드 전술 재설계 후 카운터 제시 카드(NGC-007 등)는 수락/거절 스텝
        # (가격협상_카운터)으로 전환된다 — 거절하고 새 가격을 제시하며 카드 2턴을 유도한다.
        prices = iter(["11000", "10600", "10400"])
        r = await svc.chat(eng, Req_Chat(session_id=session_id))
        for _ in range(14):
            if r.step in ("가격협상", "가격협상_카운터") and r.card_id:
                played.append(r.card_id)
                if len(played) == 2:
                    break
            if r.chat_end:
                break
            if r.input_mode == "price":
                ui = next(prices)
            elif r.step == "가격협상_카운터":
                ui = "다른 가격 제시"
            elif r.input_options:
                ui = "예" if "예" in r.input_options else r.input_options[0]
            else:
                ui = "확인"
            r = await svc.chat(eng, Req_Chat(session_id=session_id, user_input=ui))

        assert len(played) == 2, f"가격협상 카드 턴 2회 기대, 실제 {played}"
        # ① 선택한 카드 안에서만 뽑힘 ② 세션 내 중복 없음
        assert set(played) <= set(selected), f"선택 밖 카드 발동: {played}"
        assert len(set(played)) == 2, f"카드 중복 사용: {played}"
    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)),
             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_no_selection_allows_full_catalog(db_engine):
    """선택 카드가 없으면(직접호출/데모) 전체 카탈로그가 허용된다 — 카드가 정상적으로 뽑히는지 기본 검증."""
    reset_sessions()
    reg = TenantEngineRegistry(loader=TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0))
    eng = await reg.get_engine(str(_uuid.uuid4()))   # _base type:db → DB 카탈로그
    catalog = {eng.mapper.get_card_id(a) for a in range(eng.action_space_size)}

    svc = ChatService()
    played = []
    sid = None
    for ui in [None, "확인", "예", "확인", "11000", "예", "10600", "예", "10600", "예"]:
        r = await svc.chat(eng, Req_Chat(session_id=sid, user_input=ui))
        sid = r.session_id
        if r.step == "가격협상" and r.card_id:
            played.append(r.card_id)
        if r.chat_end:
            break

    assert played, "가격협상 카드 턴이 발생해야 함"
    assert set(played) <= catalog                     # 카탈로그(NGC-001~011) 내에서만
    assert len(played) == len(set(played))            # 세션 내 중복 없음