o2o-negosium-original/schedules/anchoring/tests/conftest.py

"""통합 테스트 픽스처 — 실제 Postgres 필요(로컬 dev DB), 없으면 자동 스킵.

컨벤션(backend 와 동일): 전용 행을 시드하고 테스트 후 직접 정리한다.
Redis 는 초기화하지 않는다 — 클라이언트 None → get None(DB 폴백)/set no-op 로 무Redis 실행.
"""
import asyncio
import uuid
from pathlib import Path

import pytest
import pytest_asyncio
from sqlalchemy import text

from anchoring import db as adb
from anchoring.config import load_config

CFG = load_config()
_SCHEMA_SQL = Path(__file__).resolve().parents[1] / "schema.sql"


def _db_available() -> bool:
    import asyncpg

    async def _check():
        conn = await asyncpg.connect(
            host=CFG.db.host, port=CFG.db.port, user=CFG.db.user,
            password=CFG.db.password, database=CFG.db.name, timeout=2,
        )
        await conn.close()

    try:
        asyncio.run(_check())
        return True
    except Exception:
        return False


DB_OK = _db_available()
requires_db = pytest.mark.skipif(not DB_OK, reason="로컬 Postgres(negosium_db) 미가용 — 통합 테스트 스킵")


def _schema_statements() -> list[str]:
    """schema.sql 에서 psql 메타(\\connect)·주석을 제거하고 문장 단위로 분리."""
    lines = [
        line for line in _SCHEMA_SQL.read_text().splitlines()
        if not line.startswith("\\") and not line.strip().startswith("--")
    ]
    return [s.strip() for s in "\n".join(lines).split(";") if s.strip()]


@pytest_asyncio.fixture
async def db_ready():
    """테스트별 엔진(이벤트 루프 수명 일치) + 스키마 멱등 적용."""
    adb.init_engine(CFG)
    async with adb.session_scope() as s:
        for stmt in _schema_statements():
            await s.execute(text(stmt))
    yield
    await adb.dispose_engine()


class Seeder:
    """전용 시드 생성 + 정리. 한 인스턴스 = 한 회사(테넌트)."""

    def __init__(self):
        self.company_id = uuid.uuid4()
        self.user_id = uuid.uuid4()
        self.item_id = uuid.uuid4()
        self.quotation_ids: list = []
        self._item_created = False

    async def _ensure_item(self, db):
        if self._item_created:
            return
        await db.execute(text(
            "INSERT INTO partner.items (item_id, company_id, user_id, name) "
            "VALUES (:iid, :cid, :uid, 'anchoring-it-test')"
        ), {"iid": self.item_id, "cid": self.company_id, "uid": self.user_id})
        self._item_created = True

    async def seed_session(
        self, db, *,
        supplier_type=1, target_price=30_000, anchoring_price=29_700, anchoring_value=10,
        status=3, bid_price=None, last_offer_price=..., qt_type=1,
    ):
        """종료 재협상 세션 1건 시드. 반환: session_id.

        last_offer_price 기본값은 bid_price(가격 흔적 = 투찰가). None 을 명시하면 가격 흔적 없는 세션.
        """
        if last_offer_price is ...:
            last_offer_price = bid_price
        await self._ensure_item(db)
        qt_id = uuid.uuid4()
        self.quotation_ids.append(qt_id)
        await db.execute(text(
            "INSERT INTO quotation.quotations "
            "(qt_id, user_id, qt_setting_id, version_id, name, number, type, round, status, "
            " start_time, end_time, supplier_type) "
            "VALUES (:qid, :uid, :sid, :vid, 'anchoring-it-test', :num, :qtype, 1, 3, now(), now(), :stype)"
        ), {
            "qid": qt_id, "uid": self.user_id, "sid": uuid.uuid4(), "vid": uuid.uuid4(),
            "num": f"AT{uuid.uuid4().hex[:12]}", "qtype": qt_type, "stype": supplier_type,
        })
        session_id = uuid.uuid4()
        await db.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, last_offer_price, "
            " status, bid_price, end_time) "
            "VALUES (:sid, :qid, :iid, :supid, 'AT-N', 1, :qtype, :tp, :ap, :value, :lop, :status, :bid, now())"
        ), {
            "sid": session_id, "qid": qt_id, "iid": self.item_id, "supid": uuid.uuid4(),
            "qtype": qt_type, "tp": target_price, "ap": anchoring_price, "value": anchoring_value,
            "lop": last_offer_price, "status": status, "bid": bid_price,
        })
        return session_id

    async def cleanup(self, db):
        await db.execute(text(
            "DELETE FROM anchoring.adjustments WHERE company_id = :cid"
        ), {"cid": self.company_id})
        if self.quotation_ids:
            await db.execute(
                text("DELETE FROM negotiation.sessions WHERE quotation_id = ANY(:qids)"),
                {"qids": self.quotation_ids},
            )
            await db.execute(
                text("DELETE FROM quotation.quotations WHERE qt_id = ANY(:qids)"),
                {"qids": self.quotation_ids},
            )
        await db.execute(text("DELETE FROM partner.items WHERE item_id = :iid"), {"iid": self.item_id})


@pytest_asyncio.fixture
async def seeder(db_ready):
    s = Seeder()
    yield s
    async with adb.session_scope() as db:
        await s.cleanup(db)