"""공급사 재협상 요청/철회(IMK #15) 포털 e2e — 요청 접수 + 철회. 담당자 심사(승인/반려)는 negodata 백엔드 몫이고, 여기(포털)는 공급사가 sessions.custom.renegotiation 에 요청을 남기고(PENDING) 스스로 철회(CANCELED)하는 절반을 본다: · 개찰(OPEN_*) 마감 + 본인 마지막 라운드 세션 → 요청 기록(PENDING) + 담당자 알림 · 낙찰(AWARDED) 건 → 요청 거부 · 남의 공급사 세션 → 거부(FORBIDDEN) · 이미 대기 중인데 재요청 → 거부(중복 방지) · 대기 중 철회 → CANCELED, 이후 재요청 허용 dev negosium_db 를 그대로 쓰므로(APP_ENV=local) 전용 테스트 행만 시드하고 끝나면 지운다. """ import uuid import bcrypt import pytest_asyncio from sqlalchemy import text from common.enums import CloseReason, QuotationStatus, RenegotiationStatus, SessionStatus TEST_LOGIN_ID = "pytest_renego_user" TEST_PW = "pytest1234" TEST_SUPPLIER_NAME = "파이테스트재협상공급사" MARK = "PYTESTRENEGO-" # 시드 식별용 prefix (item code / qt number) @pytest_asyncio.fixture async def renego_seed(db_engine): """공급사 + 로그인유저 + 재협상 후보 세션들을 시드하고 (supplier_id, sids, uids) 반환. (code, quotation.status, close_reason, 소속 공급사) — 요청 자격은 견적 마감사유·소유로 갈린다. """ supplier_id = uuid.uuid4() other_supplier_id = uuid.uuid4() pw_hash = bcrypt.hashpw(TEST_PW.encode("utf-8"), bcrypt.gensalt()).decode("utf-8") specs = [ ("OPEN", QuotationStatus.CLOSED.value, CloseReason.OPEN_PRICE.value, supplier_id), # 개찰 → 요청 가능 ("AWARD", QuotationStatus.CLOSED.value, CloseReason.AWARDED.value, supplier_id), # 낙찰 → 불가 ("OTHER", QuotationStatus.CLOSED.value, CloseReason.OPEN_PRICE.value, other_supplier_id), # 남의 공급사 ] sids, uids = {}, {} async def _cleanup(conn): await conn.execute(text(f"DELETE FROM negotiation.sessions WHERE qt_number LIKE '{MARK}%'")) await conn.execute(text(f"DELETE FROM company.notifications WHERE ref_qt_id IN " f"(SELECT qt_id FROM quotation.quotations WHERE number LIKE '{MARK}%')")) await conn.execute(text(f"DELETE FROM quotation.quotations WHERE number LIKE '{MARK}%'")) await conn.execute(text(f"DELETE FROM partner.items WHERE code LIKE '{MARK}%'")) await conn.execute(text("DELETE FROM supplier.supplier_users WHERE id = :id"), {"id": TEST_LOGIN_ID}) await conn.execute(text("DELETE FROM partner.suppliers WHERE name = :n"), {"n": TEST_SUPPLIER_NAME}) async with db_engine.begin() as conn: await _cleanup(conn) await conn.execute( text("INSERT INTO partner.suppliers (supplier_id, company_id, user_id, name) " "VALUES (:sid, gen_random_uuid(), gen_random_uuid(), :name)"), {"sid": supplier_id, "name": TEST_SUPPLIER_NAME}, ) await conn.execute( text("INSERT INTO supplier.supplier_users (supplier_id, id, password, name, last_accessed_at, status, role) " "VALUES (:sid, :id, :pw, '협상담당자', now(), 1, 1)"), {"sid": supplier_id, "id": TEST_LOGIN_ID, "pw": pw_hash}, ) for code, quote_st, close_reason, sup in specs: item_id, qt_id, session_id, user_id = uuid.uuid4(), uuid.uuid4(), uuid.uuid4(), uuid.uuid4() sids[code], uids[code] = session_id, user_id await conn.execute( text("INSERT INTO partner.items (item_id, company_id, user_id, name, code, model_name, manufacturer) " "VALUES (:iid, gen_random_uuid(), gen_random_uuid(), :name, :code, :model, '테스트제조사')"), {"iid": item_id, "name": f"상품 {code}", "code": f"{MARK}{code}", "model": f"MODEL-{code}"}, ) await conn.execute( text("INSERT INTO quotation.quotations " "(qt_id, user_id, qt_setting_id, version_id, name, number, type, status, close_reason, " " round, start_time, end_time) VALUES " "(:qid, :uid, gen_random_uuid(), gen_random_uuid(), :name, :num, 2, :st, :cr, " " 1, now() - make_interval(hours => 2), now() - make_interval(hours => 1))"), {"qid": qt_id, "uid": user_id, "name": f"견적 {code}", "num": f"{MARK}{code}", "st": quote_st, "cr": close_reason}, ) await conn.execute( text("INSERT INTO negotiation.sessions " "(session_id, quotation_id, item_id, supplier_id, qt_number, qt_round, qt_type, " " target_price, status, bid_price, end_time) VALUES " "(:sesid, :qid, :iid, :sup, :qtn, 1, 2, 100000, :sst, 95000, now() - make_interval(hours => 1))"), {"sesid": session_id, "qid": qt_id, "iid": item_id, "sup": sup, "qtn": f"{MARK}{code}", "sst": SessionStatus.DONE.value}, ) yield {"supplier_id": supplier_id, "sids": sids, "uids": uids} async with db_engine.begin() as conn: await _cleanup(conn) async def _login_token(client): r = await client.post("/v1/auth/login", json={"id": TEST_LOGIN_ID, "pw": TEST_PW}) return r.json()["access_token"] async def _request(client, token, session_id, *, reason="가격 재검토", desired_price=90000): return await client.post( f"/v1/negotiation/session/{session_id}/renegotiation", headers={"Authorization": f"Bearer {token}"}, json={"reason": reason, "desired_price": desired_price}, ) async def _cancel(client, token, session_id): return await client.delete( f"/v1/negotiation/session/{session_id}/renegotiation", headers={"Authorization": f"Bearer {token}"}, ) async def _renego(db_engine, session_id): async with db_engine.begin() as conn: row = (await conn.execute( text("SELECT custom FROM negotiation.sessions WHERE session_id = :sid"), {"sid": session_id}, )).scalar() return (row or {}).get("renegotiation") or {} async def _notif_count(db_engine, qt_number): async with db_engine.begin() as conn: return (await conn.execute( text("SELECT count(*) FROM company.notifications WHERE ref_qt_id IN " "(SELECT qt_id FROM quotation.quotations WHERE number = :num)"), {"num": qt_number}, )).scalar() # ---- 요청 ------------------------------------------------------------------- async def test_request_records_pending(client, renego_seed, db_engine): """검증: 개찰(OPEN_PRICE) 마감 + 본인 마지막 라운드 세션에 재협상 요청. 기대결과: success + PENDING 기록(사유·희망가 저장) + 담당자 알림 1건.""" token = await _login_token(client) sid = renego_seed["sids"]["OPEN"] body = (await _request(client, token, sid, reason="원자재 인상 반영", desired_price=88000)).json() assert body["result"]["success"] is True assert body["status"] == RenegotiationStatus.PENDING.value saved = await _renego(db_engine, sid) assert saved["status"] == RenegotiationStatus.PENDING.value assert saved["reason"] == "원자재 인상 반영" assert saved["desired_price"] == 88000 assert await _notif_count(db_engine, f"{MARK}OPEN") == 1 async def test_request_twice_blocked(client, renego_seed, db_engine): """검증: 이미 대기(PENDING) 요청이 있는 세션에 다시 요청. 기대결과: 2번째는 거부(중복 방지) + 상태는 여전히 PENDING 1건.""" token = await _login_token(client) sid = renego_seed["sids"]["OPEN"] first = (await _request(client, token, sid)).json() second = (await _request(client, token, sid)).json() assert first["result"]["success"] is True assert second["result"]["success"] is False assert (await _renego(db_engine, sid))["status"] == RenegotiationStatus.PENDING.value async def test_request_blocked_on_awarded(client, renego_seed, db_engine): """검증: 낙찰(AWARDED)로 마감된 건에 재협상 요청. 기대결과: 거부(낙찰 건은 재협상 불가) + custom.renegotiation 미기록.""" token = await _login_token(client) sid = renego_seed["sids"]["AWARD"] body = (await _request(client, token, sid)).json() assert body["result"]["success"] is False assert await _renego(db_engine, sid) == {} async def test_request_forbidden_other_supplier(client, renego_seed, db_engine): """검증: 다른 공급사 소유 세션에 재협상 요청. 기대결과: 거부 + custom.renegotiation 미기록(소유 가드).""" token = await _login_token(client) sid = renego_seed["sids"]["OTHER"] body = (await _request(client, token, sid)).json() assert body["result"]["success"] is False assert await _renego(db_engine, sid) == {} # ---- 철회 ------------------------------------------------------------------- async def test_cancel_sets_canceled_and_allows_rerequest(client, renego_seed, db_engine): """검증: 대기 중 요청을 철회한 뒤 다시 요청. 기대결과: 철회 시 CANCELED → 재요청 시 다시 PENDING(철회 건은 재요청 허용).""" token = await _login_token(client) sid = renego_seed["sids"]["OPEN"] await _request(client, token, sid) cancelled = (await _cancel(client, token, sid)).json() assert cancelled["result"]["success"] is True assert cancelled["status"] == RenegotiationStatus.CANCELED.value assert (await _renego(db_engine, sid))["status"] == RenegotiationStatus.CANCELED.value again = (await _request(client, token, sid)).json() assert again["result"]["success"] is True assert (await _renego(db_engine, sid))["status"] == RenegotiationStatus.PENDING.value async def test_cancel_requires_pending(client, renego_seed, db_engine): """검증: 대기 요청이 없는 세션에 철회 시도. 기대결과: 거부(철회할 대기 요청 없음).""" token = await _login_token(client) sid = renego_seed["sids"]["OPEN"] body = (await _cancel(client, token, sid)).json() assert body["result"]["success"] is False