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

"""배치 통합 테스트 (스펙 §11.5 — 실제 Postgres, Redis 없음(무Redis 폴백 경로)).

실행: cd schedules/anchoring && PYTHONPATH=src .venv/bin/python -m pytest tests/test_batch.py -q
"""
import logging
import uuid

from sqlalchemy import select, text

from anchoring import db as adb
from anchoring.batch import MarkingConflictError, _evaluate_cell, run_evaluation_batch
from anchoring.models import RateAdjustment, Session
from anchoring.reader import get_anchor_rate
from conftest import requires_db

pytestmark = requires_db


async def _run_batch(*seeders):
    """테스트 전용: 시드한 회사로 스코프 — 공유 dev DB 의 실데이터를 소비하지 않는다."""
    return await run_evaluation_batch(force=True, company_ids=[s.company_id for s in seeders])

# 시드 기본값: target 30,000 / rate 10‰ / anchor 29,700 → bracket 12 ("3만 원대" 칸)
BRACKET = 12
SUCCESS_BID = 29_000   # ≤ anchor → BID_SUCCESS
FAIL_BID = 29_999      # >  anchor → BID_FAIL


async def _adjustments(db, seeder):
    stmt = (
        select(RateAdjustment)
        .where(RateAdjustment.company_id == seeder.company_id)
        .order_by(RateAdjustment.id)
    )
    return (await db.execute(stmt)).scalars().all()


async def _marks(db, session_ids):
    stmt = select(Session.session_id, Session.anchoring_adjustment_id).where(Session.session_id.in_(session_ids))
    return dict((await db.execute(stmt)).all())


async def _seed_mixed(db, seeder, success: int, fail: int, **kw):
    ids = []
    for _ in range(success):
        ids.append(await seeder.seed_session(db, bid_price=SUCCESS_BID, **kw))
    for _ in range(fail):
        ids.append(await seeder.seed_session(db, bid_price=FAIL_BID, **kw))
    return ids


# ── §11.5: 13건 전량 평가 + 멱등 (실패 3종 혼합) + 로그 규약 ──
async def test_full_cycle_and_idempotency(seeder, caplog):
    async with adb.session_scope() as db:
        ids = await _seed_mixed(db, seeder, success=8, fail=2)   # DONE 인데 앵커 초과(와일드카드 상단 등)
        for _ in range(2):  # 가격 쓰고 결렬(REJECTED) = 실패
            ids.append(await seeder.seed_session(db, status=5, bid_price=None, last_offered_price=FAIL_BID))
        # 가격 쓰고 이탈 → 견적 마감 시 일괄 NOT_PARTICIPATED = 실패 (중간 이탈 시나리오)
        ids.append(await seeder.seed_session(db, status=4, bid_price=None, last_offered_price=FAIL_BID))
        # 합계 13건, 성공 8 → r≈0.615 → +20

    with caplog.at_level(logging.INFO, logger="anchoring"):
        result = await run_evaluation_batch(force=True, company_ids=[seeder.company_id])

    # 로그 규약: run_id 태그 + 회사별 grep 가능한 칸별 조정 라인 + 회사요약 라인
    assert result["run_id"]
    tagged = [m for m in caplog.messages if f"[batch {result['run_id']}]" in m]
    assert any(f"조정 company={seeder.company_id}" in m and "10‰→30‰" in m for m in tagged)
    assert any(f"회사요약 company={seeder.company_id}" in m and "평가=1" in m for m in tagged)

    async with adb.session_scope() as db:
        adjustments = await _adjustments(db, seeder)
        assert len(adjustments) == 1
        adj = adjustments[0]
        assert (adj.nego_count, adj.success_count) == (13, 8)
        assert (adj.anchor_rate_before, adj.anchor_rate_after) == (10, 30)
        assert sorted(adj.consumed_session_ids) == sorted(str(i) for i in ids)
        marks = await _marks(db, ids)
        assert all(v == adj.id for v in marks.values())          # 13건 모두 소비 마킹

    # 재실행 — 마킹 멱등: 우리 칸 조정은 그대로 1건
    await _run_batch(seeder)
    async with adb.session_scope() as db:
        assert len(await _adjustments(db, seeder)) == 1

        # 조회용 뷰 — rate_history(이전→새 값 리스트업) / current_rates(칸별 현재값)
        hist = (await db.execute(text(
            "SELECT anchor_rate_before, anchor_rate_after, delta_permille, success_rate "
            "FROM anchoring.rate_history WHERE company_id = :c"), {"c": seeder.company_id})).one()
        assert (hist.anchor_rate_before, hist.anchor_rate_after, hist.delta_permille) == (10, 30, 20)
        assert float(hist.success_rate) == 0.615
        cur = (await db.execute(text(
            "SELECT anchor_rate_permille FROM anchoring.current_rates "
            "WHERE company_id = :c AND supplier_type = 1 AND price_bracket_index = :b"),
            {"c": seeder.company_id, "b": BRACKET})).scalar_one()
        assert cur == 30


# ── §11.5: 이월(7건 스킵 → 누적 13건 단일 평가) ──────────
async def test_carryover(seeder):
    async with adb.session_scope() as db:
        first = await _seed_mixed(db, seeder, success=5, fail=2)  # 7건 < 10

    await _run_batch(seeder)
    async with adb.session_scope() as db:
        assert await _adjustments(db, seeder) == []
        marks = await _marks(db, first)
        assert all(v is None for v in marks.values())             # 마킹 없음 = 이월

        second = await _seed_mixed(db, seeder, success=3, fail=3)  # 누적 13건 (8S/5F)

    await _run_batch(seeder)
    async with adb.session_scope() as db:
        adjustments = await _adjustments(db, seeder)
        assert len(adjustments) == 1
        assert adjustments[0].nego_count == 13                    # 4주치 전량 1회 평가
        assert adjustments[0].anchor_rate_after == 30
        marks = await _marks(db, first + second)
        assert all(v == adjustments[0].id for v in marks.values())


# ── §11.5: 회사 격리 + 현재값 조회(무Redis DB 폴백) + δ 유형 차원 ──
async def test_company_isolation_and_reader(seeder):
    async with adb.session_scope() as db:
        await _seed_mixed(db, seeder, success=10, fail=0, supplier_type=1)  # 유통 → +20
        await _seed_mixed(db, seeder, success=10, fail=0, supplier_type=2)  # 제조 → +10 (δ 스왑 가드)

    await _run_batch(seeder)

    other_company = uuid.uuid4()
    async with adb.session_scope() as db:
        adjustments = await _adjustments(db, seeder)
        by_type = {a.supplier_type: a.anchor_rate_after for a in adjustments}
        assert by_type == {1: 30, 2: 20}
        # 조정된 칸은 새 rate, 타사 같은 (유형,구간) 칸은 정적 테이블 시작값
        assert await get_anchor_rate(db, seeder.company_id, 1, BRACKET) == 30
        assert await get_anchor_rate(db, other_company, 1, BRACKET) == 10


# ── §11.5: EXCLUDED 마킹 0 + 유효 n<10 이월 + supplier_type 미지정 ──
async def test_excluded_and_unsampleable(seeder):
    async with adb.session_scope() as db:
        # 가격 흔적 없는 종료(무가격 결렬·미참여) → EXCLUDED
        excluded = [await seeder.seed_session(db, status=5, last_offered_price=None) for _ in range(8)]
        excluded += [await seeder.seed_session(db, status=4, last_offered_price=None) for _ in range(7)]
        valid = await _seed_mixed(db, seeder, success=5, fail=0)                                # 유효 5 < 10
        untyped = [await seeder.seed_session(db, supplier_type=0, bid_price=SUCCESS_BID)]        # 칸 구성 불가
        untyped.append(await seeder.seed_session(db, supplier_type=None, bid_price=SUCCESS_BID))  # NULL 도 동일(§13-6)

    await _run_batch(seeder)

    async with adb.session_scope() as db:
        assert await _adjustments(db, seeder) == []               # 유효 5 < 10 → 평가 없음
        marks = await _marks(db, excluded + untyped)
        assert all(v == 0 for v in marks.values())                # 제외 확정 마킹(재스캔 방지)
        marks = await _marks(db, valid)
        assert all(v is None for v in marks.values())             # 유효 표본은 이월


# ── 개정 1: 마킹 rowcount ≠ n → 조정 INSERT 포함 전체 롤백 ──
async def test_marking_conflict_rolls_back(seeder):
    async with adb.session_scope() as db:
        ids = await _seed_mixed(db, seeder, success=10, fail=0)
        # 경합 시뮬레이션: 1건을 다른 실행이 먼저 소비한 상태로 만든다
        await db.execute(text(
            "UPDATE negotiation.sessions SET anchoring_adjustment_id = 999999 WHERE session_id = :sid"
        ), {"sid": ids[0]})

    samples = [(sid, 1) for sid in ids]  # 10건 전부 BID_SUCCESS 로 평가 시도
    try:
        await _evaluate_cell(seeder.company_id, 1, BRACKET, samples)
        raised = False
    except MarkingConflictError:
        raised = True
    assert raised

    async with adb.session_scope() as db:
        assert await _adjustments(db, seeder) == []               # 롤백 — 이중 조정 없음
        marks = await _marks(db, ids[1:])
        assert all(v is None for v in marks.values())             # 나머지 9건 마킹도 롤백


# ── 무증상 고장 감지: 가격 제시 흔적 0% → WARN ────────────
async def test_priced_rate_zero_warns(seeder, caplog):
    async with adb.session_scope() as db:
        for _ in range(3):  # 전부 가격 흔적 없는 종료 → priced_rate 0
            await seeder.seed_session(db, status=5, last_offered_price=None)

    with caplog.at_level(logging.WARNING, logger="anchoring"):
        await run_evaluation_batch(force=True, company_ids=[seeder.company_id])
    assert any("가격 제시 흔적 0%" in m for m in caplog.messages)


# ── dry-run: 판정·예상 조정만 로그, DB 무변경 ─────────────
async def test_dry_run_changes_nothing(seeder, caplog):
    async with adb.session_scope() as db:
        ids = await _seed_mixed(db, seeder, success=10, fail=0)
        excluded = [await seeder.seed_session(db, status=5, last_offered_price=None)]

    with caplog.at_level(logging.INFO, logger="anchoring"):
        result = await run_evaluation_batch(force=True, company_ids=[seeder.company_id], dry_run=True)

    assert result["status"] == "dry_run" and result["evaluated_cells"] == 1
    assert any("조정예정" in m and f"company={seeder.company_id}" in m for m in caplog.messages)
    async with adb.session_scope() as db:
        assert await _adjustments(db, seeder) == []               # INSERT 없음
        marks = await _marks(db, ids + excluded)
        assert all(v is None for v in marks.values())             # 마킹 없음(제외 포함)

    # 이어서 실제 실행하면 그대로 반영된다 (dry-run 이 상태를 소비하지 않았음을 증명)
    await _run_batch(seeder)
    async with adb.session_scope() as db:
        assert len(await _adjustments(db, seeder)) == 1


# ── 박제 정합 감시: 정수식과 박제 anchor 불일치 → WARN ────
async def test_snapshot_mismatch_warns(seeder, caplog):
    async with adb.session_scope() as db:
        # rate 10‰ 기준 정수식 anchor 는 29,700 — 29,000 으로 박제된 세션은 이식 오류 신호
        await seeder.seed_session(db, rate=10, anchor_price=29_000, bid_price=28_000)

    with caplog.at_level(logging.WARNING, logger="anchoring"):
        await run_evaluation_batch(force=True, company_ids=[seeder.company_id], dry_run=True)
    assert any("박제 정합 불일치 1건" in m for m in caplog.messages)