"""개찰 견적 직접 낙찰(award_quotation) 테스트 — 담당자가 개찰(낙찰자 미정 마감) 견적의 낙찰자를 직접 지정. 직접 낙찰은 자동 낙찰(close_and_decide)과 결과 컬럼은 같되(close_reason=AWARDED, preferred_sp_*), 알림에 manual 플래그로 '직접' 낙찰임을 남긴다. 다음을 본다: · 개찰 + 투찰(DONE) 협력사 지정 → 낙찰 확정 + 알림 SUCCESS(manual=True) · 개찰 아님(이미 낙찰) → 거부(INVALID_REQUEST_DATA), 알림 없음 · 후보 아닌 협력사 지정 → 거부, close_reason 유지 · 낙찰 후 재지정(재클릭) → 거부(동시성 가드), 알림 1건 유지 세션의 협상 결과(협상완료/입찰가)와 개찰 상태(close_reason)는 협상/마감에서만 생기는 값이라 SQL 로 직접 넣는다. """ import uuid from datetime import datetime import pytest_asyncio from sqlalchemy import text from common.enums import AwardType, CloseReason, ErrorType, QuotationStatus, QuotationType, SessionStatus, UserRole from crud.quotation_crud import QuotationCRUD from services.quotation 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 async def test_award_opened_sets_winner_and_notifies(clean): """검증: 개찰(동가) 견적 + 투찰 협력사 2건(A=100, B=120) 중 A 를 직접 낙찰. 기대결과: close_reason→낙찰, preferred_sp=A, 동가플래그 해제 + 알림 SUCCESS(manual=True, winner_price=100).""" engine = clean user_id, supplier_a = uuid.uuid4(), uuid.uuid4() qt = await _seed_opened(engine, user_id=user_id, number="A-WIN", close_reason=CloseReason.OPEN_EQUAL.value) await _add_session(engine, qt, status=SessionStatus.DONE.value, bid_price=100, supplier_id=supplier_a) await _add_session(engine, qt, status=SessionStatus.DONE.value, bid_price=120) res = await _service().award_quotation(str(qt), None, user_id, UserRole.USER.value, supplier_a, 100) assert res.result.success is True row = await _quotation(engine, qt) assert row.close_reason == CloseReason.AWARDED.value assert row.award_type == AwardType.MANUAL.value # 직접 낙찰 — 통계에서 자동낙찰과 구분 assert row.preferred_sp_yn is True assert str(row.preferred_sp_id) == str(supplier_a) assert row.equal_bid_yn is False notis = await _notifications(engine, user_id) assert len(notis) == 1 type_, data, ref = notis[0] assert type_ == 1 # NotificationType.SUCCESS assert data["manual"] is True assert data["winner_price"] == 100 assert str(ref) == str(qt) async def test_award_rejects_when_not_opened(clean): """검증: 이미 낙찰된 견적(close_reason=AWARDED)에 직접 낙찰을 다시 시도. 기대결과: 거부(INVALID_REQUEST_DATA) + 새 알림 없음.""" engine = clean user_id, supplier_a = uuid.uuid4(), uuid.uuid4() qt = await _seed_opened(engine, user_id=user_id, number="A-DONE", close_reason=CloseReason.AWARDED.value) await _add_session(engine, qt, status=SessionStatus.DONE.value, bid_price=100, supplier_id=supplier_a) res = await _service().award_quotation(str(qt), None, user_id, UserRole.USER.value, supplier_a, 100) assert res.result.success is False assert res.result.code == ErrorType.INVALID_REQUEST_DATA.value assert len(await _notifications(engine, user_id)) == 0 async def test_award_rejects_unknown_supplier(clean): """검증: 개찰 견적에, 이 견적에 초청되지 않은(세션이 없는) 협력사 id 를 지정. 기대결과: 거부 + close_reason 은 개찰(OPEN_PRICE) 그대로 유지, 알림 없음.""" engine = clean user_id, bidder, stranger = uuid.uuid4(), uuid.uuid4(), uuid.uuid4() qt = await _seed_opened(engine, user_id=user_id, number="A-STRANGER", close_reason=CloseReason.OPEN_PRICE.value) await _add_session(engine, qt, status=SessionStatus.DONE.value, bid_price=100, supplier_id=bidder) res = await _service().award_quotation(str(qt), None, user_id, UserRole.USER.value, stranger, 100) assert res.result.success is False row = await _quotation(engine, qt) assert row.close_reason == CloseReason.OPEN_PRICE.value assert row.preferred_sp_id is None assert len(await _notifications(engine, user_id)) == 0 async def test_award_is_idempotent(clean): """검증: 직접 낙찰 성공 후 같은 견적에 재지정(재클릭/경합). 기대결과: 2번째는 거부(이미 낙찰) + 알림은 1건만 유지(동시성 가드가 한 번만 통과).""" engine = clean user_id, supplier_a = uuid.uuid4(), uuid.uuid4() qt = await _seed_opened(engine, user_id=user_id, number="A-IDEMP", close_reason=CloseReason.OPEN_REJECT.value) await _add_session(engine, qt, status=SessionStatus.DONE.value, bid_price=100, supplier_id=supplier_a) first = await _service().award_quotation(str(qt), None, user_id, UserRole.USER.value, supplier_a, 100) second = await _service().award_quotation(str(qt), None, user_id, UserRole.USER.value, supplier_a, 100) assert first.result.success is True assert second.result.success is False assert len(await _notifications(engine, user_id)) == 1 async def test_award_rejects_non_owner(clean): """검증: 남의 개찰 견적을 일반 유저(비소유·USER)가 직접 낙찰 시도. 기대결과: 거부(ACCOUNT_FORBIDDEN) + close_reason 개찰 유지 + 알림 없음.""" engine = clean owner, other, supplier_a = uuid.uuid4(), uuid.uuid4(), uuid.uuid4() qt = await _seed_opened(engine, user_id=owner, number="A-NONOWNER", close_reason=CloseReason.OPEN_PRICE.value) await _add_session(engine, qt, status=SessionStatus.DONE.value, bid_price=100, supplier_id=supplier_a) res = await _service().award_quotation(str(qt), None, other, UserRole.USER.value, supplier_a, 100) assert res.result.success is False assert res.result.code == ErrorType.ACCOUNT_FORBIDDEN.value row = await _quotation(engine, qt) assert row.close_reason == CloseReason.OPEN_PRICE.value assert row.preferred_sp_id is None assert len(await _notifications(engine, owner)) == 0 async def test_award_allows_owner_role(clean): """검증: 남의 개찰 견적을 최고관리자(OWNER)가 직접 낙찰. 기대결과: 낙찰 성공 + 알림은 견적 작성자(owner) 인박스에 남는다(호출자가 아니라).""" engine = clean creator, admin, supplier_a = uuid.uuid4(), uuid.uuid4(), uuid.uuid4() qt = await _seed_opened(engine, user_id=creator, number="A-OWNER", close_reason=CloseReason.OPEN_EQUAL.value) await _add_session(engine, qt, status=SessionStatus.DONE.value, bid_price=100, supplier_id=supplier_a) res = await _service().award_quotation(str(qt), None, admin, UserRole.OWNER.value, supplier_a, 100) assert res.result.success is True row = await _quotation(engine, qt) assert row.close_reason == CloseReason.AWARDED.value assert str(row.preferred_sp_id) == str(supplier_a) assert len(await _notifications(engine, creator)) == 1 assert len(await _notifications(engine, admin)) == 0 async def test_award_records_contract_price_and_reason(clean): """검증: 협력사 제출가(100)와 다른 계약가(88)로 직접 낙찰 — 오프라인 재협상 결과 반영. 기대결과: 계약가는 낙찰 세션 contract_price 컬럼에, 사유·처리자는 견적 custom.award 에 남고, 알림 winner_price 도 계약가.""" engine = clean user_id, supplier_a = uuid.uuid4(), uuid.uuid4() qt = await _seed_opened(engine, user_id=user_id, number="A-OFFLINE", close_reason=CloseReason.OPEN_PRICE.value) await _add_session(engine, qt, status=SessionStatus.REJECTED.value, bid_price=None, supplier_id=supplier_a) res = await _service().award_quotation( str(qt), None, user_id, UserRole.USER.value, supplier_a, 88, "오프라인 협상, 8/12 통화 합의", ) assert res.result.success is True async with engine.begin() as conn: # 계약가 = 세션 컬럼(협력사 가격 — bid_price/reject_price 와 같은 축) sess_price = (await conn.execute( text("SELECT contract_price FROM sessions WHERE quotation_id = :qt AND supplier_id = :sp"), {"qt": qt, "sp": supplier_a}, )).scalar() # 사유·처리자 = 견적 custom.award(견적 단위 결정) qt_custom = (await conn.execute( text("SELECT custom FROM quotations WHERE qt_id = :qt"), {"qt": qt}, )).scalar() assert sess_price == 88 award = qt_custom["award"] assert award["reason"] == "오프라인 협상, 8/12 통화 합의" assert award["by"] == str(user_id) notis = await _notifications(engine, user_id) assert notis[0][1]["winner_price"] == 88 async def test_award_allows_supplier_without_price(clean): """검증: 시스템에 가격을 한 번도 안 낸(미참여) 협력사를 계약가와 함께 직접 낙찰. 기대결과: 낙찰 성공 — 전원 미응찰 견적도 오프라인 협상 결과를 반영할 수 있다.""" engine = clean user_id, supplier_a = uuid.uuid4(), uuid.uuid4() qt = await _seed_opened(engine, user_id=user_id, number="A-NOSHOW", close_reason=CloseReason.OPEN_NOSHOW.value) await _add_session(engine, qt, status=SessionStatus.NOT_PARTICIPATED.value, bid_price=None, supplier_id=supplier_a) res = await _service().award_quotation(str(qt), None, user_id, UserRole.USER.value, supplier_a, 77_000) assert res.result.success is True row = await _quotation(engine, qt) assert row.close_reason == CloseReason.AWARDED.value assert str(row.preferred_sp_id) == str(supplier_a) async def test_award_rejects_without_contract_price(clean): """검증: 계약가 없이(0) 직접 낙찰 시도. 기대결과: 거부(INVALID_REQUEST_DATA) — 계약가는 절감 통계의 기준이라 필수다.""" engine = clean user_id, supplier_a = uuid.uuid4(), uuid.uuid4() qt = await _seed_opened(engine, user_id=user_id, number="A-NOPRICE", close_reason=CloseReason.OPEN_PRICE.value) await _add_session(engine, qt, status=SessionStatus.DONE.value, bid_price=100, supplier_id=supplier_a) res = await _service().award_quotation(str(qt), None, user_id, UserRole.USER.value, supplier_a, 0) assert res.result.success is False assert res.result.code == ErrorType.INVALID_REQUEST_DATA.value row = await _quotation(engine, qt) assert row.close_reason == CloseReason.OPEN_PRICE.value # ===== 헬퍼 ===== async def _seed_opened(engine, *, user_id, number, close_reason, round_=1): """개찰/낙찰 상태(status=CLOSED + close_reason)로 견적 1건 시드. 낙찰자 컬럼은 비운 채 시작.""" 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, close_reason, " " round, iteration, start_time, end_time, deleted) VALUES " "(:qt_id, :user_id, :qt_setting_id, :version_id, '견적A', :number, :type, :status, :close_reason, " " :round, 0, :start_time, :end_time, false)" ), { "qt_id": qt_id, "user_id": user_id, "qt_setting_id": uuid.uuid4(), "version_id": uuid.uuid4(), "number": number, "type": QuotationType.REQUOTE.value, "status": QuotationStatus.CLOSED.value, "close_reason": close_reason, "round": round_, "start_time": PAST, "end_time": PAST, }, ) 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 _quotation(engine, qt_id): """견적 1행(마감 결과 컬럼 확인용).""" async with engine.begin() as conn: return (await conn.execute( text("SELECT close_reason, award_type, preferred_sp_yn, preferred_sp_id, equal_bid_yn " "FROM quotations WHERE qt_id = :qt_id"), {"qt_id": qt_id}, )).one() 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())