"""scheduler 잡 e2e — '대상 선정'(어떤 견적을 고르나) + close_and_decide 위임 결과 검증. 실행 전제: PostgreSQL(negodata_db). docker compose up -d 후 python -m pytest tests/test_scheduler.py. 잡은 HTTP 엔드포인트가 없어 scheduler.jobs 함수를 직접 호출한다(앱과 같은 DB_SESSION_MNG 사용 → mock 불필요). 세션 상태(DONE/REJECTED/bid_price 등)는 협상 프론트가 만드는 값이라 API 로 못 만든다 → SQL 로 직접 시드. """ import asyncio import uuid from datetime import datetime import pytest_asyncio from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.interval import IntervalTrigger from sqlalchemy import text from common.enums import QuotationStatus, QuotationType, SessionStatus from scheduler import jobs PAST = datetime(2020, 1, 1) # 마감시각 지남(잡① 대상) FUTURE = datetime(2999, 1, 1) # 마감시각 미래(잡① 제외) @pytest_asyncio.fixture async def clean(db_engine): """conftest 의 db_engine 은 quotations 만 비우고 sessions 는 안 비운다(FK 미설정 → CASCADE 대상 아님). 잡②(close_negotiated)는 전체 견적을 스캔하므로 다른 테스트가 남긴 세션이 결과를 흔든다 → sessions 도 비워 격리.""" async with db_engine.begin() as conn: await conn.execute(text("TRUNCATE TABLE sessions, quotations RESTART IDENTITY CASCADE")) return db_engine # ----- 시드 헬퍼 (FK 미설정이라 user/item/supplier 없이 임의 uuid 로 충분) ----- async def _add_quotation(engine, *, status=QuotationStatus.IN_PROGRESS.value, end_time=PAST, deleted=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) VALUES " "(:qt_id, :user_id, :qt_setting_id, :version_id, :name, :number, :type, :status, " " :round, :iteration, :start_time, :end_time, :deleted)" ), { "qt_id": qt_id, "user_id": uuid.uuid4(), "qt_setting_id": uuid.uuid4(), "version_id": uuid.uuid4(), "name": "견적", "number": f"Q-{qt_id.hex[:8]}", "type": QuotationType.REQUOTE.value, "status": status, "round": 1, "iteration": 0, "start_time": PAST, "end_time": end_time, "deleted": deleted, }, ) return qt_id async def _add_session(engine, qt_id, *, status, bid_price=None, supplier_id=None): 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, :qt_number, :qt_round, :qt_type, " " :target_price, :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_number": "Q", "qt_round": 1, "qt_type": QuotationType.REQUOTE.value, "target_price": 0, "status": status, "bid_price": bid_price, "end_time": PAST, }, ) async def _quotation_row(engine, qt_id): async with engine.begin() as conn: return (await conn.execute( text("SELECT status, preferred_sp_yn, preferred_sp_id FROM quotations WHERE qt_id = :id"), {"id": qt_id}, )).first() # ----- 잡① close_expired_quotations : 대상 선정(마감시각 지난 미마감만) ----- async def test_close_expired_picks_only_due_and_open(clean): engine = clean due = await _add_quotation(engine, status=QuotationStatus.IN_PROGRESS.value, end_time=PAST) future = await _add_quotation(engine, status=QuotationStatus.IN_PROGRESS.value, end_time=FUTURE) already = await _add_quotation(engine, status=QuotationStatus.CLOSED.value, end_time=PAST) deleted = await _add_quotation(engine, status=QuotationStatus.IN_PROGRESS.value, end_time=PAST, deleted=True) n = await jobs.close_expired_quotations() assert n == 1 # 마감 대상은 due 1건뿐 assert (await _quotation_row(engine, due)).status == QuotationStatus.CLOSED.value assert (await _quotation_row(engine, future)).status == QuotationStatus.IN_PROGRESS.value # 미래 → 안 건드림 assert (await _quotation_row(engine, already)).status == QuotationStatus.CLOSED.value # 원래부터 CLOSED assert (await _quotation_row(engine, deleted)).status == QuotationStatus.IN_PROGRESS.value # 삭제분 → 제외 # ----- 잡② close_negotiated_quotations : 대상 선정(전 세션 종결 + 세션 1개+) ----- async def test_close_negotiated_picks_when_all_sessions_ended(clean): engine = clean # 전 세션 종결(거부) → 대상 ended = await _add_quotation(engine, status=QuotationStatus.IN_PROGRESS.value, end_time=FUTURE) await _add_session(engine, ended, status=SessionStatus.REJECTED.value) # 진행중 세션 하나라도 있으면 → 제외 pending = await _add_quotation(engine, status=QuotationStatus.IN_PROGRESS.value, end_time=FUTURE) await _add_session(engine, pending, status=SessionStatus.DONE.value, bid_price=100) await _add_session(engine, pending, status=SessionStatus.IN_PROGRESS.value) # 세션 0개 → 제외 no_session = await _add_quotation(engine, status=QuotationStatus.IN_PROGRESS.value, end_time=FUTURE) await jobs.close_negotiated_quotations() assert (await _quotation_row(engine, ended)).status == QuotationStatus.CLOSED.value assert (await _quotation_row(engine, pending)).status == QuotationStatus.IN_PROGRESS.value assert (await _quotation_row(engine, no_session)).status == QuotationStatus.IN_PROGRESS.value # ----- close_and_decide 위임 결과 스모크(잡①을 통해) ----- async def test_award_single_lowest(clean): engine = clean qt = await _add_quotation(engine, end_time=PAST) winner = uuid.uuid4() 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) await jobs.close_expired_quotations() row = await _quotation_row(engine, qt) assert row.status == QuotationStatus.CLOSED.value assert row.preferred_sp_yn is True assert str(row.preferred_sp_id) == str(winner) # 최저가 단독 → 낙찰 확정 async def test_rejected_just_closes(clean): engine = clean qt = await _add_quotation(engine, end_time=PAST) await _add_session(engine, qt, status=SessionStatus.REJECTED.value) # 입찰 없는 거부만 await jobs.close_expired_quotations() row = await _quotation_row(engine, qt) assert row.status == QuotationStatus.CLOSED.value assert not row.preferred_sp_yn # 거부 → 낙찰 없이 그냥 마감 # ----- 스케줄러 와이어링(start_scheduler) : DB 불필요 ----- async def test_scheduler_disabled_without_env(monkeypatch): import scheduler monkeypatch.delenv("SCHEDULER_ENABLED", raising=False) scheduler._scheduler = None scheduler.start_scheduler() assert scheduler._scheduler is None # SCHEDULER_ENABLED != 1 → 미기동 async def test_scheduler_registers_both_jobs(monkeypatch): import scheduler monkeypatch.setenv("SCHEDULER_ENABLED", "1") scheduler._scheduler = None scheduler.start_scheduler() try: ids = {j.id for j in scheduler._scheduler.get_jobs()} assert ids == {"close_expired_quotations", "close_negotiated_quotations"} finally: scheduler.shutdown_scheduler() assert scheduler._scheduler is None # ----- 스케줄러가 실제로 잡을 호출해 마감까지 가는지(라이브) ----- async def test_scheduler_actually_runs_job_and_closes(clean): """스케줄러에 잡을 걸면 정말 호출돼 견적이 마감되는지 확인. 운영 트리거는 CronTrigger(minute='*/5')라 분 경계까지 기다려야 하므로, 여기선 1초 IntervalTrigger 로 같은 잡을 걸어 '스케줄러 → 잡 호출 → 마감' 경로만 몇 초 안에 검증한다.""" engine = clean qt = await _add_quotation(engine, end_time=PAST) await _add_session(engine, qt, status=SessionStatus.DONE.value, bid_price=100) sched = AsyncIOScheduler(timezone="Asia/Seoul") sched.add_job(jobs.close_expired_quotations, IntervalTrigger(seconds=1), max_instances=1) sched.start() try: row = None for _ in range(25): # 최대 ~5초 폴링(잡은 1초 뒤 첫 발화) await asyncio.sleep(0.2) row = await _quotation_row(engine, qt) if row.status == QuotationStatus.CLOSED.value: break assert row is not None and row.status == QuotationStatus.CLOSED.value # 크론이 잡을 호출해 마감 finally: sched.shutdown(wait=False)