Mina Choi
82076e138e
- test DB 세션마다 자동 create/drop (팀원은 Postgres만 있으면 pytest 한 방) - auth_headers 시드 픽스처(무인증 /auth/create 제거 대응) + other_company_id - 커버: 회사 스코프(견적·상품·협력사·대시보드·세팅), 견적 마감 재견적 O/X + 알림, 견적 생성·목표가, 알림함 읽기, 회사유저 OWNER 게이팅, 기존 파일 검증/기대결과 주석 정비 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
225 lines
11 KiB
Python
225 lines
11 KiB
Python
"""견적 마감(close_and_decide) 테스트 — 마감하면 상황별로 결과가 맞게 판정되고, 그 결과가 작성자에게 알림으로 남는지 확인.
|
|
|
|
핵심은 '재견적(다음 라운드 재생성)이 나오는 경우 vs 안 나오는 경우'의 구분이다.
|
|
각 경우에 (1) 판정이 맞고 (2) 작성자 알림함에 알맞은 알림 1건이 남는지 본다:
|
|
· 단독 최저가 → 낙찰 (SUCCESS) [재견적 X]
|
|
· 협상 거부 → 결렬 (FAILURE, reason=rejected) [재견적 X]
|
|
· 동가/미참여 + 한도 남음 → 재생성 (REGENERATED) [재견적 O]
|
|
· 동가/미참여 + 한도 소진 → 결렬 (FAILURE, reason=closed) [재견적 X]
|
|
재생성 한도: 사유(동가·미참여)별로 한 체인(같은 견적번호)에서 각 1번까지만.
|
|
|
|
공급사의 협상 결과(협상완료/거부/입찰가)는 협상 화면에서만 생기는 값이라 API 로 못 만든다 → SQL 로 직접 넣는다.
|
|
마감 판정 로직 자체를 더 깊게 파는 건 test_scheduler·test_close_and_decide_fixes.
|
|
"""
|
|
import uuid
|
|
from datetime import datetime
|
|
|
|
import pytest_asyncio
|
|
from sqlalchemy import text
|
|
|
|
from common.enums import CloseOutcome, NotificationType, QuotationStatus, QuotationType, SessionStatus
|
|
from crud.quotation_crud import QuotationCRUD
|
|
from services.quotation_service import QuotationService
|
|
|
|
PAST = datetime(2020, 1, 1)
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def clean(db_engine):
|
|
"""conftest 는 notifications 를 비우지 않는다 → 알림 단언이 다른 테스트에 안 흔들리게 여기서 함께 비운다."""
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(text("TRUNCATE TABLE sessions, quotations, notifications RESTART IDENTITY CASCADE"))
|
|
return db_engine
|
|
|
|
|
|
# ----- 재견적 X (낙찰·거부) -----
|
|
async def test_award_notifies_success(clean):
|
|
"""검증: 협상완료 세션 2건(입찰 100·200) — 단독 최저가로 마감.
|
|
기대결과: 재견적 X, 판정 = 낙찰(AWARDED) + 알림 SUCCESS(winner_price=100=최저가, ref_qt_id=그 견적)."""
|
|
engine = clean
|
|
user_id = uuid.uuid4()
|
|
winner = uuid.uuid4()
|
|
qt = await _seed_quotation(engine, user_id=user_id, number="N-AWARD")
|
|
await _add_session(engine, qt, status=SessionStatus.DONE.value, bid_price=100, supplier_id=winner)
|
|
await _add_session(engine, qt, status=SessionStatus.DONE.value, bid_price=200)
|
|
|
|
outcome = await _service().close_and_decide(qt)
|
|
|
|
assert outcome == CloseOutcome.AWARDED
|
|
notis = await _notifications(engine, user_id)
|
|
assert len(notis) == 1
|
|
type_, data, ref = notis[0]
|
|
assert type_ == NotificationType.SUCCESS.value
|
|
assert data["winner_price"] == 100
|
|
assert str(ref) == str(qt)
|
|
|
|
|
|
async def test_rejected_notifies_failure(clean):
|
|
"""검증: 협상거부 세션만 있는 상태로 마감.
|
|
기대결과: 재견적 X, 판정 = 결렬(CLOSED) + 알림 FAILURE(reason=rejected)."""
|
|
engine = clean
|
|
user_id = uuid.uuid4()
|
|
qt = await _seed_quotation(engine, user_id=user_id, number="N-REJECT")
|
|
await _add_session(engine, qt, status=SessionStatus.REJECTED.value)
|
|
|
|
outcome = await _service().close_and_decide(qt)
|
|
|
|
assert outcome == CloseOutcome.CLOSED
|
|
notis = await _notifications(engine, user_id)
|
|
assert len(notis) == 1
|
|
type_, data, ref = notis[0]
|
|
assert type_ == NotificationType.FAILURE.value
|
|
assert data["reason"] == "rejected"
|
|
assert str(ref) == str(qt)
|
|
|
|
|
|
# ----- 재견적 O (동가·미참여, 한도 남음) -----
|
|
async def test_equal_bid_regenerates(clean):
|
|
"""검증: 협상완료 세션 2건이 '동가'(둘 다 100), 체인에 동가 재생성 이력 없음(한도 남음).
|
|
기대결과: 재견적 O, 판정 = 재생성(REGENERATED) + 알림 REGENERATED(reason=equal, tied_price=100, next_round=2)."""
|
|
engine = clean
|
|
user_id = uuid.uuid4()
|
|
qt = await _seed_quotation(engine, user_id=user_id, number="N-EQUAL")
|
|
await _add_session(engine, qt, status=SessionStatus.DONE.value, bid_price=100)
|
|
await _add_session(engine, qt, status=SessionStatus.DONE.value, bid_price=100)
|
|
|
|
outcome = await _service().close_and_decide(qt)
|
|
|
|
assert outcome == CloseOutcome.REGENERATED
|
|
notis = await _notifications(engine, user_id)
|
|
assert len(notis) == 1
|
|
type_, data, _ = notis[0]
|
|
assert type_ == NotificationType.REGENERATED.value
|
|
assert data["reason"] == "equal"
|
|
assert data["tied_price"] == 100
|
|
assert data["next_round"] == 2
|
|
|
|
|
|
async def test_no_show_regenerates(clean):
|
|
"""검증: 전원 미참여(미시작 세션만), 체인에 미참여 재생성 이력 없음(한도 남음).
|
|
기대결과: 재견적 O, 판정 = 재생성(REGENERATED) + 알림 REGENERATED(reason=no_show, next_round=2)."""
|
|
engine = clean
|
|
user_id = uuid.uuid4()
|
|
qt = await _seed_quotation(engine, user_id=user_id, number="N-NOSHOW")
|
|
await _add_session(engine, qt, status=SessionStatus.CREATED.value)
|
|
await _add_session(engine, qt, status=SessionStatus.CREATED.value)
|
|
|
|
outcome = await _service().close_and_decide(qt)
|
|
|
|
assert outcome == CloseOutcome.REGENERATED
|
|
notis = await _notifications(engine, user_id)
|
|
assert len(notis) == 1
|
|
type_, data, _ = notis[0]
|
|
assert type_ == NotificationType.REGENERATED.value
|
|
assert data["reason"] == "no_show"
|
|
assert data["next_round"] == 2
|
|
|
|
|
|
# ----- 재견적 X (동가·미참여지만 한도 소진 → 결렬) -----
|
|
async def test_equal_bid_limit_exhausted_fails(clean):
|
|
"""검증: 1차가 이미 '동가'로 재생성된 체인(동가 한도 1 소진)에서, 2차도 또 동가로 마감.
|
|
기대결과: 재견적 X — 판정 = 결렬(CLOSED) + 알림 FAILURE(reason=closed)."""
|
|
engine = clean
|
|
user_id = uuid.uuid4()
|
|
# 1차: 동가로 마감돼 2차를 만든 상황(equal_bid_yn=True 가 동가 재생성 표식) → 동가 한도 소진
|
|
await _seed_quotation(engine, user_id=user_id, number="N-EQUAL-LIMIT", round_=1,
|
|
status=QuotationStatus.CLOSED.value, equal_bid_yn=True)
|
|
# 2차: 또 동가
|
|
qt2 = await _seed_quotation(engine, user_id=user_id, number="N-EQUAL-LIMIT", round_=2)
|
|
await _add_session(engine, qt2, status=SessionStatus.DONE.value, bid_price=100)
|
|
await _add_session(engine, qt2, status=SessionStatus.DONE.value, bid_price=100)
|
|
|
|
outcome = await _service().close_and_decide(qt2)
|
|
|
|
assert outcome == CloseOutcome.CLOSED # 동가 한도 소진 → 재생성 없이 결렬
|
|
notis = await _notifications(engine, user_id)
|
|
assert len(notis) == 1
|
|
type_, data, ref = notis[0]
|
|
assert type_ == NotificationType.FAILURE.value
|
|
assert data["reason"] == "closed"
|
|
assert str(ref) == str(qt2)
|
|
|
|
|
|
async def test_no_show_limit_exhausted_fails(clean):
|
|
"""검증: 1차가 이미 '미참여'로 재생성된 체인(미참여 한도 1 소진)에서, 2차도 또 전원 미참여로 마감.
|
|
기대결과: 재견적 X — 판정 = 결렬(CLOSED) + 알림 FAILURE(reason=closed)."""
|
|
engine = clean
|
|
user_id = uuid.uuid4()
|
|
# 1차: 미참여로 마감돼 2차를 만든 상황(preferred_sp_yn=False·equal_bid_yn=False 가 미참여 재생성 표식) → 미참여 한도 소진
|
|
await _seed_quotation(engine, user_id=user_id, number="N-NOSHOW-LIMIT", round_=1,
|
|
status=QuotationStatus.CLOSED.value, preferred_sp_yn=False, equal_bid_yn=False)
|
|
# 2차: 또 전원 미참여
|
|
qt2 = await _seed_quotation(engine, user_id=user_id, number="N-NOSHOW-LIMIT", round_=2)
|
|
await _add_session(engine, qt2, status=SessionStatus.CREATED.value)
|
|
await _add_session(engine, qt2, status=SessionStatus.CREATED.value)
|
|
|
|
outcome = await _service().close_and_decide(qt2)
|
|
|
|
assert outcome == CloseOutcome.CLOSED # 미참여 한도 소진 → 재생성 없이 결렬
|
|
notis = await _notifications(engine, user_id)
|
|
assert len(notis) == 1
|
|
type_, data, ref = notis[0]
|
|
assert type_ == NotificationType.FAILURE.value
|
|
assert data["reason"] == "closed"
|
|
assert str(ref) == str(qt2)
|
|
|
|
|
|
# ===== 헬퍼 (위 테스트들이 쓰는 도우미. 세션 입찰값·이전 라운드 표식을 SQL 로 직접 세팅) =====
|
|
async def _seed_quotation(
|
|
engine, *, user_id, number, round_=1, status=QuotationStatus.IN_PROGRESS.value,
|
|
preferred_sp_yn=None, equal_bid_yn=None,
|
|
):
|
|
"""견적 1건 시드(작성자=user_id). preferred_sp_yn·equal_bid_yn 으로 '이전 라운드가 어떤 사유로 재생성됐는지'를 표식한다
|
|
(동가 재생성=equal_bid_yn True / 미참여 재생성=preferred_sp_yn False AND equal_bid_yn False)."""
|
|
qt_id = uuid.uuid4()
|
|
async with engine.begin() as conn:
|
|
await conn.execute(
|
|
text(
|
|
"INSERT INTO quotations "
|
|
"(qt_id, user_id, qt_setting_id, version_id, name, number, type, status, "
|
|
" round, iteration, start_time, end_time, deleted, preferred_sp_yn, equal_bid_yn) VALUES "
|
|
"(:qt_id, :user_id, :qt_setting_id, :version_id, '견적A', :number, :type, :status, "
|
|
" :round, 0, :start_time, :end_time, false, :pref, :eq)"
|
|
),
|
|
{
|
|
"qt_id": qt_id, "user_id": user_id, "qt_setting_id": uuid.uuid4(),
|
|
"version_id": uuid.uuid4(), "number": number,
|
|
"type": QuotationType.REQUOTE.value, "status": status, "round": round_,
|
|
"start_time": PAST, "end_time": PAST,
|
|
"pref": preferred_sp_yn, "eq": equal_bid_yn,
|
|
},
|
|
)
|
|
return qt_id
|
|
|
|
|
|
async def _add_session(engine, qt_id, *, status, bid_price=None, supplier_id=None):
|
|
"""세션 1건 시드(공급사 협상 1건). status/bid_price 로 협상완료·거부·입찰가를 만든다."""
|
|
async with engine.begin() as conn:
|
|
await conn.execute(
|
|
text(
|
|
"INSERT INTO sessions "
|
|
"(session_id, quotation_id, item_id, supplier_id, qt_number, qt_round, qt_type, "
|
|
" target_price, status, bid_price, end_time) VALUES "
|
|
"(:session_id, :quotation_id, :item_id, :supplier_id, 'Q', 1, :qt_type, "
|
|
" 0, :status, :bid_price, :end_time)"
|
|
),
|
|
{
|
|
"session_id": uuid.uuid4(), "quotation_id": qt_id, "item_id": uuid.uuid4(),
|
|
"supplier_id": supplier_id or uuid.uuid4(), "qt_type": QuotationType.REQUOTE.value,
|
|
"status": status, "bid_price": bid_price, "end_time": PAST,
|
|
},
|
|
)
|
|
|
|
|
|
async def _notifications(engine, user_id):
|
|
"""user_id(작성자) 인박스 알림 (type, data, ref_qt_id) — 생성순."""
|
|
async with engine.begin() as conn:
|
|
return (await conn.execute(
|
|
text("SELECT type, data, ref_qt_id FROM notifications WHERE user_id = :uid ORDER BY created_at"),
|
|
{"uid": user_id},
|
|
)).all()
|
|
|
|
|
|
def _service():
|
|
return QuotationService(QuotationCRUD())
|