From 82076e138e7c47b1a3667ef6c12ddc6f3b0c50e4 Mon Sep 17 00:00:00 2001 From: Mina Choi Date: Wed, 1 Jul 2026 15:06:55 +0900 Subject: [PATCH] =?UTF-8?q?[test]=20negodata:=20=EB=B0=B1=EC=97=94?= =?UTF-8?q?=EB=93=9C=20=ED=85=8C=EC=8A=A4=ED=8A=B8=20=EC=8A=A4=EC=9C=84?= =?UTF-8?q?=ED=8A=B8=20=EA=B5=AC=EC=B6=95=20+=20=EA=B3=B5=ED=86=B5=20?= =?UTF-8?q?=ED=94=BD=EC=8A=A4=EC=B2=98(conftest)=20=EC=A0=95=EB=B9=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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) --- negodata/backend/conftest.py | 93 ++++++- negodata/backend/tests/test_auth.py | 69 ++--- .../tests/test_close_and_decide_fixes.py | 132 ++++----- negodata/backend/tests/test_company_scope.py | 140 ++++++++++ negodata/backend/tests/test_company_user.py | 88 ++++++ negodata/backend/tests/test_features.py | 38 ++- negodata/backend/tests/test_item.py | 77 +----- negodata/backend/tests/test_notification.py | 99 +++++++ .../tests/test_quotation_close_notify.py | 224 +++++++++++++++ .../backend/tests/test_quotation_create.py | 98 +++++++ negodata/backend/tests/test_scheduler.py | 261 ++++++++++-------- 11 files changed, 989 insertions(+), 330 deletions(-) create mode 100644 negodata/backend/tests/test_company_scope.py create mode 100644 negodata/backend/tests/test_company_user.py create mode 100644 negodata/backend/tests/test_notification.py create mode 100644 negodata/backend/tests/test_quotation_close_notify.py create mode 100644 negodata/backend/tests/test_quotation_create.py diff --git a/negodata/backend/conftest.py b/negodata/backend/conftest.py index abcfb81..655d03c 100644 --- a/negodata/backend/conftest.py +++ b/negodata/backend/conftest.py @@ -27,8 +27,47 @@ def _write_url(cfg) -> str: return f"postgresql+asyncpg://{cfg.write_id}{pw}@{cfg.write_host}:{cfg.write_port}/{cfg.name}" +def _admin_url(cfg) -> str: + """DB 생성용 관리 접속. CREATE DATABASE 는 대상 DB 안에서 못 하므로 기본 'postgres' DB 로 붙는다.""" + pw = f":{cfg.write_pw}" if cfg.write_pw else "" + return f"postgresql+asyncpg://{cfg.write_id}{pw}@{cfg.write_host}:{cfg.write_port}/postgres" + + +async def _drop_test_db(*, recreate: bool): + """test DB 를 지운다(있으면). recreate=True 면 지운 뒤 새로 만든다. + WITH (FORCE): 남아있는 커넥션을 끊고 drop (PG13+). 관리 접속은 기본 'postgres' DB.""" + engine = create_async_engine(_admin_url(main_db_config), isolation_level="AUTOCOMMIT") + try: + async with engine.connect() as conn: + await conn.execute(text(f'DROP DATABASE IF EXISTS "{main_db_config.name}" WITH (FORCE)')) + if recreate: + await conn.execute(text(f'CREATE DATABASE "{main_db_config.name}"')) + finally: + await engine.dispose() + + +@pytest_asyncio.fixture(scope="session", autouse=True) +async def _test_db_lifecycle(): + """테스트 세션 동안만 test DB 를 만들고, 끝나면 내린다. + + 매 세션 '깨끗한 새 DB'로 시작하므로 스키마 낡음(드리프트)이 원천 차단되고, 끝나면 남는 DB 도 없다. + (테이블 구조는 db_engine 의 create_all 이 현재 모델 기준으로 채운다.) + 안전가드: 이름에 'test' 있는 DB 만 만들고/지운다(dev DB 보호). + """ + assert "test" in main_db_config.name, ( + f"비-test DB('{main_db_config.name}') 는 만들거나 지우지 않는다. APP_ENV=test 로 실행하세요." + ) + await _drop_test_db(recreate=True) # 세션 시작: 깨끗한 새 DB + yield + # 세션 종료: 앱 싱글톤 커넥션부터 정리(활성 커넥션 있으면 FORCE 로 끊김) 후 DB 를 내린다. + from common.database.db_session_manager import DB_SESSION_MNG + + await DB_SESSION_MNG.dispose_all() + await _drop_test_db(recreate=False) + + @pytest_asyncio.fixture -async def db_engine(): +async def db_engine(_test_db_lifecycle): """테스트용 스키마를 보장하고, 매 테스트 시작 시 테이블을 비워 격리한다. ⚠ 이 픽스처는 TRUNCATE 한다 → dev DB(negosium_db)를 가리키면 실데이터가 날아간다. @@ -74,15 +113,16 @@ async def company_id(db_engine) -> str: return str(cid) -@pytest_asyncio.fixture(scope="session", autouse=True) -async def _dispose_app_engines(): - """테스트 세션이 끝날 때 앱 싱글톤 엔진을 정리한다. - (이벤트 루프 종료 후 커넥션이 GC 되며 나오는 'Event loop is closed' 경고 제거) - """ - yield - from common.database.db_session_manager import DB_SESSION_MNG - - await DB_SESSION_MNG.dispose_all() +@pytest_asyncio.fixture +async def other_company_id(db_engine) -> str: + """company_id 와 다른 소속사 1개(회사 스코프/IDOR 격리 테스트용).""" + cid = uuid.uuid4() + async with db_engine.begin() as conn: + await conn.execute( + text("INSERT INTO companies (company_id, name, status) VALUES (:cid, :name, :status)"), + {"cid": cid, "name": "다른회사", "status": CompanyStatus.ACTIVE.value}, + ) + return str(cid) @pytest_asyncio.fixture @@ -93,3 +133,36 @@ async def client(db_engine): transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as ac: yield ac + + +@pytest_asyncio.fixture +async def auth_headers(db_engine, client, company_id): + """테스트 유저를 시드하고 로그인 헤더(Bearer)를 돌려주는 팩토리. + + 무인증 /v1/auth/create 가 제거(최고관리자 회원관리로 일원화)돼 더는 API 로 계정을 못 만든다. + 그래서 users 행을 직접 INSERT(비번 bcrypt 해시)한 뒤 살아있는 /v1/auth/login 으로 토큰을 받는다. + company 미지정 시 기본 소속사(company_id 픽스처). role 로 OWNER 계정도 만들 수 있다. + 호출: `h = await auth_headers("user1")` / `await auth_headers("userB", other_company_id)`. + """ + from common.enums import UserRole, UserStatus + from router.v1.validator.dependencies import GetHashedPW + + async def _make(login_id, company=None, *, password="pw1234", role=UserRole.USER.value, name="n"): + cid = company or company_id + hashed = await GetHashedPW(password) + async with db_engine.begin() as conn: + # status·role 은 NOT NULL — ORM default 는 raw INSERT 에 안 먹으므로 명시(companies.status 와 동일). + await conn.execute( + text( + "INSERT INTO users (user_id, company_id, id, password, name, status, role, last_accessed_at) " + "VALUES (:uid, :cid, :id, :pw, :name, :status, :role, now())" + ), + { + "uid": uuid.uuid4(), "cid": uuid.UUID(cid), "id": login_id, "pw": hashed, + "name": name, "status": UserStatus.ACTIVE.value, "role": role, + }, + ) + r = await client.post("/v1/auth/login", json={"id": login_id, "password": password}) + return {"Authorization": f"Bearer {r.json()['access_token']}"} + + return _make diff --git a/negodata/backend/tests/test_auth.py b/negodata/backend/tests/test_auth.py index ea23cce..1c44628 100644 --- a/negodata/backend/tests/test_auth.py +++ b/negodata/backend/tests/test_auth.py @@ -1,34 +1,15 @@ -"""auth 도메인 e2e 테스트 (negodata: users/companies 기반). +"""auth 도메인 e2e — 로그인 / 내정보 / 인증거부 흐름. -실행 전제: PostgreSQL(negodata_db)이 떠 있어야 한다. - docker compose up -d # 또는 로컬 postgres - cd negodata/backend && python -m pytest -계정 생성은 company_id 를 요구하므로 company_id 픽스처(conftest)가 소속사를 시드한다. +계정 생성·중복·최고관리자 스코프는 test_company_user.py. 유저 시드/로그인은 auth_headers 픽스처. """ -async def test_create_and_login_flow(client, company_id): - # 1) 계정 생성 (회사 하위로) - r = await client.post( - "/v1/auth/create", - json={"id": "user1", "password": "pw1234", "company_id": company_id, "name": "홍길동"}, - ) - assert r.status_code == 200 - body = r.json() - assert body["result"]["success"] is True - assert body["user_id"] +async def test_login_and_me_flow(auth_headers, client, company_id): + """검증: 시드된 유저가 로그인해 받은 토큰으로 /me 호출. + 기대결과: 200, 본인 id·name·소속사(company_id)가 그대로 반환.""" + h = await auth_headers("user1", name="홍길동") - # 2) 로그인 -> 토큰 발급 - r = await client.post("/v1/auth/login", json={"id": "user1", "password": "pw1234"}) - assert r.status_code == 200 - body = r.json() - assert body["result"]["success"] is True - assert body["access_token"] - assert body["refresh_token"] - access_token = body["access_token"] - - # 3) 보호된 엔드포인트(/me) — 토큰의 유저 + 소속사 반환 - r = await client.get("/v1/auth/me", headers={"Authorization": f"Bearer {access_token}"}) + r = await client.get("/v1/auth/me", headers=h) assert r.status_code == 200 me = r.json() assert me["id"] == "user1" @@ -36,43 +17,27 @@ async def test_create_and_login_flow(client, company_id): assert me["company"]["company_id"] == company_id -async def test_login_with_wrong_password(client, company_id): - await client.post( - "/v1/auth/create", - json={"id": "user2", "password": "correct", "company_id": company_id, "name": "n"}, - ) +async def test_login_with_wrong_password(auth_headers, client): + """검증: 존재하는 계정에 '틀린 비밀번호'로 로그인. + 기대결과: 로그인 실패 — success=False, code=1200(ACCOUNT_INVALID_INFO), 토큰 빈 문자열.""" + await auth_headers("user2") # pw1234 로 시드 r = await client.post("/v1/auth/login", json={"id": "user2", "password": "wrong"}) - assert r.status_code == 200 body = r.json() assert body["result"]["success"] is False - # 자격증명 오류는 ACCOUNT_INVALID_INFO(1200) assert body["result"]["code"] == 1200 - assert body.get("access_token", "") == "" # 실패 시 토큰은 빈 문자열 + assert body.get("access_token", "") == "" async def test_login_nonexistent_account(client): + """검증: 존재하지 않는 계정으로 로그인. + 기대결과: 실패 — success=False (계정 유무를 '틀린 비번'과 구분해 흘리지 않음).""" r = await client.post("/v1/auth/login", json={"id": "ghost", "password": "whatever"}) assert r.json()["result"]["success"] is False -async def test_duplicate_account_create(client, company_id): - r1 = await client.post( - "/v1/auth/create", - json={"id": "dup", "password": "pw1234", "company_id": company_id, "name": "n"}, - ) - assert r1.json()["result"]["success"] is True - - r2 = await client.post( - "/v1/auth/create", - json={"id": "dup", "password": "pw5678", "company_id": company_id, "name": "n2"}, - ) - body = r2.json() - assert body["result"]["success"] is False - # ACCOUNT_ALREADY_EXIST(1201) - assert body["result"]["code"] == 1201 - - async def test_me_without_token_is_rejected(client): + """검증: 토큰 없이 보호 엔드포인트 /me 호출. + 기대결과: 인증 단계에서 거부 — HTTP 401 또는 403.""" r = await client.get("/v1/auth/me") - assert r.status_code in (401, 403) # HTTPBearer 가 자격증명 없음을 거부 + assert r.status_code in (401, 403) diff --git a/negodata/backend/tests/test_close_and_decide_fixes.py b/negodata/backend/tests/test_close_and_decide_fixes.py index 08b19b6..45f9b6f 100644 --- a/negodata/backend/tests/test_close_and_decide_fixes.py +++ b/negodata/backend/tests/test_close_and_decide_fixes.py @@ -6,7 +6,8 @@ - #4 재생성 사유 집계: 단독낙찰(preferred_sp_yn=True) 이전 라운드를 '미참여'로 오집계하지 않음 - #6 재생성 라운드 최소 협상기간 하한(즉시 재마감 캐스케이드 방지) -실행 전제: tests/test_scheduler.py 와 동일(PostgreSQL, APP_ENV=test). +용어: 체인 = 같은 견적번호(number)로 이어지는 라운드들 / 미참여 = 공급사가 협상에 안 들어온 채 마감됨 / + 재생성 = 결판 안 난 견적의 '다음 라운드'를 자동 생성 / 재생성 한도 = 사유(미참여·동가)별로 체인당 1번까지만. """ import asyncio import uuid @@ -29,63 +30,9 @@ async def clean(db_engine): return db_engine -async def _seed_quotation( - engine, *, number, round_, status, start_time=PAST, end_time=PAST, - preferred_sp_yn=None, equal_bid_yn=None, -): - 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, :name, :number, :type, :status, " - " :round, 0, :start_time, :end_time, false, :pref, :eq)" - ), - { - "qt_id": qt_id, "user_id": uuid.uuid4(), "qt_setting_id": uuid.uuid4(), - "version_id": uuid.uuid4(), "name": "견적", "number": number, - "type": QuotationType.REQUOTE.value, "status": status, "round": round_, - "start_time": start_time, "end_time": end_time, - "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): - 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, " - " 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_number": "Q", "qt_round": 1, - "qt_type": QuotationType.REQUOTE.value, "status": status, - "bid_price": bid_price, "end_time": PAST, - }, - ) - - -async def _rounds(engine, number): - """체인(number)의 (round, status, end_time, start_time) 목록 — round 오름차순.""" - async with engine.begin() as conn: - return (await conn.execute( - text("SELECT round, status, start_time, end_time FROM quotations " - "WHERE number = :n ORDER BY round"), - {"n": number}, - )).all() - - -# ----- #2 동시 이중 마감 가드 ----- async def test_concurrent_close_creates_only_one_next_round(clean): - """같은 견적을 5번 동시에 close_and_decide 해도 다음 라운드는 정확히 1개만 생성된다.""" + """검증: 같은 견적(전원 미참여)을 5번 동시에 close_and_decide. + 기대결과: 재생성은 1번만(REGENERATED=1), 체인은 [1,2] — 이중 재생성/충돌 없음.""" engine = clean number = "C-CONCURRENT" qt = await _seed_quotation(engine, number=number, round_=1, status=QuotationStatus.IN_PROGRESS.value) @@ -104,9 +51,9 @@ async def test_concurrent_close_creates_only_one_next_round(clean): assert round_numbers == [1, 2], f"체인은 [1,2] 여야 함(중복/충돌 없음), 실제 {round_numbers}" -# ----- #3 차수 + #6 최소 협상기간 하한 ----- async def test_next_round_numbering_and_min_duration(clean): - """다음 라운드 round = 최신+1, 협상기간이 0이어도 최소 하한(MIN_REGEN_DURATION)이 적용된다.""" + """검증: 협상기간이 0인 견적을 미참여로 재생성. + 기대결과: 체인 [1,2](round=최신+1), 새 라운드 협상기간 ≥ MIN_REGEN_DURATION(즉시 재마감 방지).""" engine = clean number = "C-DURATION" # start==end (협상기간 0) → 하한이 적용되지 않으면 새 라운드도 0 길이가 된다 @@ -129,10 +76,9 @@ async def test_next_round_numbering_and_min_duration(clean): ) -# ----- #4 재생성 사유 집계: 단독낙찰 이전 라운드를 미참여로 오집계하지 않음 ----- async def test_awarded_prior_round_not_counted_as_no_show(clean): - """체인에 '단독낙찰'(preferred_sp_yn=True) 이전 라운드가 있어도, 이후 라운드의 미참여 재생성 예산을 소진하지 않는다. - (구버전: equal_bid_yn=False 인 단독낙찰 라운드를 미참여로 세어 round2 재생성이 막혔다.)""" + """검증: round1=단독낙찰 + round2=전원 미참여 인 체인에서 round2 를 마감. + 기대결과: REGENERATED, 체인 [1,2,3] — 단독낙찰 라운드를 '미참여'로 오집계해 재생성을 막지 않는다.""" engine = clean number = "C-AWARDED-PRIOR" # round 1: 단독낙찰로 마감(preferred_sp_yn=True). 수동 재생성 등으로 체인이 이어진 상황을 가정. @@ -155,10 +101,9 @@ async def test_awarded_prior_round_not_counted_as_no_show(clean): assert round_numbers == [1, 2, 3], f"round 3 이 생성돼야 함, 실제 {round_numbers}" -# ----- #4 대비: 실제 미참여 이전 라운드는 예산을 소진(한도 1) ----- async def test_no_show_prior_round_consumes_budget(clean): - """이전 라운드가 '미참여 재생성'(preferred_sp_yn=False, equal_bid_yn=False)이면 예산(1)을 소진 → - 다음 라운드의 미참여는 재생성 없이 그냥 마감된다.""" + """검증: round1=미참여 재생성 + round2=전원 미참여 인 체인에서 round2 를 마감. + 기대결과: CLOSED, 체인 [1,2] — 미참여 재생성 한도(1) 소진돼 재생성 없이 그냥 마감(round3 없음).""" engine = clean number = "C-NOSHOW-PRIOR" # round 1: 미참여로 마감(양성 표식) → no_part 예산 1 소진 @@ -176,3 +121,60 @@ async def test_no_show_prior_round_consumes_budget(clean): rounds = await _rounds(engine, number) assert outcome == CloseOutcome.CLOSED, f"미참여 예산 소진 → 그냥 마감이어야 함, 실제 {outcome}" assert [r.round for r in rounds] == [1, 2], "재생성되면 안 됨(round 3 없음)" + + +# ===== 헬퍼 (위 테스트들이 쓰는 도우미. 세션 상태·마감 표식을 SQL 로 직접 세팅) ===== +async def _seed_quotation( + engine, *, number, round_, status, start_time=PAST, end_time=PAST, + preferred_sp_yn=None, equal_bid_yn=None, +): + """견적 1건 시드. number/round_ 로 체인을, preferred_sp_yn·equal_bid_yn 으로 '이전 라운드가 어떻게 마감됐는지'를 만든다.""" + 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, :name, :number, :type, :status, " + " :round, 0, :start_time, :end_time, false, :pref, :eq)" + ), + { + "qt_id": qt_id, "user_id": uuid.uuid4(), "qt_setting_id": uuid.uuid4(), + "version_id": uuid.uuid4(), "name": "견적", "number": number, + "type": QuotationType.REQUOTE.value, "status": status, "round": round_, + "start_time": start_time, "end_time": end_time, + "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건).""" + 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, " + " 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_number": "Q", "qt_round": 1, + "qt_type": QuotationType.REQUOTE.value, "status": status, + "bid_price": bid_price, "end_time": PAST, + }, + ) + + +async def _rounds(engine, number): + """체인(number)의 (round, status, start_time, end_time) 목록 — round 오름차순.""" + async with engine.begin() as conn: + return (await conn.execute( + text("SELECT round, status, start_time, end_time FROM quotations " + "WHERE number = :n ORDER BY round"), + {"n": number}, + )).all() diff --git a/negodata/backend/tests/test_company_scope.py b/negodata/backend/tests/test_company_scope.py new file mode 100644 index 0000000..94aa93e --- /dev/null +++ b/negodata/backend/tests/test_company_scope.py @@ -0,0 +1,140 @@ +"""회사 스코프(멀티테넌트) — 회사 소유 자원은 '내 회사 것'만 보이고, 남의 회사 것은 막힌다(보안 회귀 방지). + +회사 A 자원을 만들어 두고 회사 B 유저 토큰으로 접근하면 '없음'으로 막히는지 확인한다. +막힘 코드: 견적 1500 / 상품 1300 / 협력사 1400. 견적 하위(세션·상태·결과·카드)도 견적 통해 1500. +견적세팅만 예외 — 회사가 아니라 '유저' 스코프라, 같은 회사라도 다른 유저면 못 본다(1600). +""" +import uuid +from datetime import datetime + +from sqlalchemy import text + +from common.enums import QuotationStatus, QuotationType, SessionStatus + +PAST = datetime(2020, 1, 1) +FUTURE = datetime(2999, 1, 1) + + +# ----- 견적 ----- +async def test_quotation_hidden_across_company(client, auth_headers, other_company_id, db_engine): + """검증: 회사A 견적을 A·B 유저가 각각 단건 조회. + 기대결과: A는 success=True / B는 code=1500(없는 것처럼 막힘).""" + ha = await auth_headers("qA") + qt = await _seed_quotation(db_engine, await _user_id(db_engine, "qA")) + + assert (await client.get(f"/v1/quotation/{qt}", headers=ha)).json()["result"]["success"] is True + hb = await auth_headers("qB", other_company_id) + assert (await client.get(f"/v1/quotation/{qt}", headers=hb)).json()["result"]["code"] == 1500 + + +async def test_quotation_list_is_company_scoped(client, auth_headers, other_company_id, db_engine): + """검증: 회사A만 견적을 가진 상태에서 A·B 유저가 목록 조회. + 기대결과: A 목록 total≥1 / B 목록 total=0.""" + ha = await auth_headers("qlA") + await _seed_quotation(db_engine, await _user_id(db_engine, "qlA"), number="Q-LIST-A") + + assert (await client.get("/v1/quotation/list", headers=ha)).json()["total"] >= 1 + hb = await auth_headers("qlB", other_company_id) + assert (await client.get("/v1/quotation/list", headers=hb)).json()["total"] == 0 + + +async def test_quotation_subresources_hidden_across_company(client, auth_headers, other_company_id, db_engine): + """검증: 회사A 견적의 하위자원(세션·상태·결과·카드)을 회사B 유저가 조회. + 기대결과: 넷 다 code=1500 으로 막힘 (같은 견적을 A 는 정상 조회).""" + ha = await auth_headers("qsA") + qt = await _seed_quotation(db_engine, await _user_id(db_engine, "qsA"), number="Q-SUB") + + hb = await auth_headers("qsB", other_company_id) + for path in (f"/v1/quotation/{qt}/sessions", f"/v1/quotation/{qt}/status", + f"/v1/quotation/{qt}/result", f"/v1/quotation/{qt}/cards"): + assert (await client.get(path, headers=hb)).json()["result"]["code"] == 1500, path + assert (await client.get(f"/v1/quotation/{qt}/status", headers=ha)).json()["result"]["success"] is True + + +# ----- 상품(item) ----- +async def test_item_hidden_across_company(client, auth_headers, other_company_id): + """검증: 회사A 상품을 회사B 유저가 목록·단건 조회. + 기대결과: 목록 total=0, 단건 code=1300(ITEM_NOT_FOUND).""" + ha = await auth_headers("iA") + a_item = (await client.post("/v1/item/create", json={"name": "A상품"}, headers=ha)).json()["item"]["item_id"] + + hb = await auth_headers("iB", other_company_id) + assert (await client.get("/v1/item/list", headers=hb)).json()["total"] == 0 + assert (await client.get(f"/v1/item/{a_item}", headers=hb)).json()["result"]["code"] == 1300 + + +# ----- 협력사(supplier) ----- +async def test_supplier_hidden_across_company(client, auth_headers, other_company_id): + """검증: 회사A 협력사를 회사B 유저가 목록·단건 조회. + 기대결과: 목록 total=0, 단건 code=1400(SUPPLIER_NOT_FOUND).""" + ha = await auth_headers("sA") + a_sup = (await client.post("/v1/supplier/create", json={"name": "A협력사", "code": "SA"}, headers=ha)).json()["supplier"]["supplier_id"] + + hb = await auth_headers("sB", other_company_id) + assert (await client.get("/v1/supplier/list", headers=hb)).json()["total"] == 0 + assert (await client.get(f"/v1/supplier/{a_sup}", headers=hb)).json()["result"]["code"] == 1400 + + +# ----- 대시보드 ----- +async def test_dashboard_is_company_scoped(client, auth_headers, other_company_id, db_engine): + """검증: 회사A만 진행중 견적을 보유. A·B 유저가 각각 대시보드 요약 조회. + 기대결과: A 는 company.in_progress≥1 / B 는 0 (타사 견적이 내 회사 집계에 안 섞임).""" + ha = await auth_headers("dA") + await _seed_quotation(db_engine, await _user_id(db_engine, "dA"), number="Q-DASH") + + assert (await client.get("/v1/dashboard/summary", headers=ha)).json()["company"]["in_progress"] >= 1 + hb = await auth_headers("dB", other_company_id) + assert (await client.get("/v1/dashboard/summary", headers=hb)).json()["company"]["in_progress"] == 0 + + +# ----- 견적세팅(회사 아님 — '유저' 스코프) ----- +async def test_quotation_setting_is_user_scoped(client, auth_headers): + """검증: 유저A 견적세팅을 '같은 회사 다른 유저' B 가 목록/수정 시도. + 기대결과: B 목록엔 안 보이고(total=0), 수정은 code=1600(내 소유 아님) — 견적세팅은 유저 단위.""" + ha = await auth_headers("stA") + a_setting = (await client.post( + "/v1/quotation-setting/create", json={"target_margin_rate": 0.15}, headers=ha + )).json()["setting"]["qt_setting_id"] + + hb = await auth_headers("stB") # 같은 회사(company_id 기본), 다른 유저 + assert (await client.get("/v1/quotation-setting/list", headers=hb)).json()["total"] == 0 + r = await client.patch(f"/v1/quotation-setting/update/{a_setting}", json={"target_margin_rate": 0.2}, headers=hb) + assert r.json()["result"]["code"] == 1600 + + +# ===== 헬퍼 (위 테스트들이 쓰는 도우미) ===== +async def _user_id(engine, login_id): + """auth_headers 로 시드된 유저의 user_id.""" + async with engine.begin() as conn: + return (await conn.execute( + text("SELECT user_id FROM users WHERE id = :id"), {"id": login_id} + )).scalar_one() + + +async def _seed_quotation(engine, user_id, *, number="Q-SCOPE"): + """작성자=user_id 인 견적 1건 + 세션 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, " + " round, iteration, start_time, end_time, deleted) VALUES " + "(:qt_id, :uid, :setting, :version, '견적A', :number, :type, :status, 1, 0, :past, :future, false)" + ), + {"qt_id": qt_id, "uid": user_id, "setting": uuid.uuid4(), "version": uuid.uuid4(), + "number": number, "type": QuotationType.REQUOTE.value, + "status": QuotationStatus.IN_PROGRESS.value, "past": PAST, "future": FUTURE}, + ) + await conn.execute( + text( + "INSERT INTO sessions " + "(session_id, quotation_id, item_id, supplier_id, qt_number, qt_round, qt_type, " + " target_price, status, end_time) VALUES " + "(:sid, :qt, :item, :sup, :number, 1, :type, 0, :st, :future)" + ), + {"sid": uuid.uuid4(), "qt": qt_id, "item": uuid.uuid4(), "sup": uuid.uuid4(), + "number": number, "type": QuotationType.REQUOTE.value, + "st": SessionStatus.CREATED.value, "future": FUTURE}, + ) + return qt_id diff --git a/negodata/backend/tests/test_company_user.py b/negodata/backend/tests/test_company_user.py new file mode 100644 index 0000000..04bd3cc --- /dev/null +++ b/negodata/backend/tests/test_company_user.py @@ -0,0 +1,88 @@ +"""직원 계정 관리(/v1/company/user/*) 테스트 — '최고관리자만' 쓸 수 있고, '자기 회사'만 다뤄지는지 확인. + +- 일반 직원 계정으로는 이 기능을 못 쓴다(HTTP 403 으로 막힘). +- 최고관리자는 자기 회사 직원만 목록에 보이고, 생성도 자기 회사로 된다(남의 회사 직원은 안 보임). +- 로그인 아이디는 전체에서 유일해야 해서, 같은 아이디로 또 만들면 거부된다(코드 1201). +""" +import uuid + +from sqlalchemy import text + +from common.enums import UserRole, UserStatus + + +async def test_regular_user_forbidden_on_owner_endpoints(client, auth_headers): + """검증: 일반 USER 토큰으로 최고관리자 전용 엔드포인트(list·create) 호출. + 기대결과: 둘 다 HTTP 403(RequireOwner 차단).""" + h = await auth_headers("plainuser") # role=USER 기본 + + r = await client.get("/v1/company/user/list", headers=h) + assert r.status_code == 403 + + r = await client.post( + "/v1/company/user/create", json={"id": "x", "password": "p", "name": "n"}, headers=h + ) + assert r.status_code == 403 + + +async def test_owner_lists_only_own_company_users(client, auth_headers, company_id, other_company_id, db_engine): + """검증: 회사A OWNER + A직원 + B직원(타사)을 두고 OWNER 가 유저 목록 조회. + 기대결과: 본인·A직원은 목록에 있고 타사(B) 직원은 없음(회사 스코프).""" + owner_h = await auth_headers("ownerA", role=UserRole.OWNER.value) # 회사 A owner + await _seed_user(db_engine, company_id, "empA") # 같은 회사 직원 + await _seed_user(db_engine, other_company_id, "empB") # 다른 회사 직원 + + r = await client.get("/v1/company/user/list", headers=owner_h) + ids = {u["id"] for u in r.json()["users"]} + assert "ownerA" in ids # 본인 + assert "empA" in ids # 자기 회사 직원 + assert "empB" not in ids # 타사 직원은 안 보임 + + +async def test_owner_creates_user_in_own_company(client, auth_headers): + """검증: OWNER 가 직원 계정을 생성한 뒤 목록 조회. + 기대결과: 생성 success=True, 생성한 유저가 자기 회사 목록에 노출.""" + owner_h = await auth_headers("ownerC", role=UserRole.OWNER.value) + + r = await client.post( + "/v1/company/user/create", + json={"id": "newemp", "password": "pw1234", "name": "직원"}, + headers=owner_h, + ) + assert r.json()["result"]["success"] is True + + r = await client.get("/v1/company/user/list", headers=owner_h) + ids = {u["id"] for u in r.json()["users"]} + assert "newemp" in ids + + +async def test_duplicate_login_id_rejected(client, auth_headers): + """검증: OWNER 가 같은 로그인 ID 로 직원 계정을 2번 생성. + 기대결과: 1번째 success=True, 2번째 success=False, code=1201(ACCOUNT_ALREADY_EXIST).""" + owner_h = await auth_headers("ownerD", role=UserRole.OWNER.value) + + r1 = await client.post( + "/v1/company/user/create", json={"id": "dup", "password": "pw1234", "name": "n"}, headers=owner_h + ) + assert r1.json()["result"]["success"] is True + + r2 = await client.post( + "/v1/company/user/create", json={"id": "dup", "password": "pw5678", "name": "n2"}, headers=owner_h + ) + body = r2.json() + assert body["result"]["success"] is False + assert body["result"]["code"] == 1201 + + +# ===== 헬퍼 (위 테스트들이 쓰는 도우미) ===== +async def _seed_user(engine, company_id, login_id, *, role=UserRole.USER.value): + """로그인 안 하는 소속 직원 시드(목록 스코프 확인용). 비번은 임의값.""" + async with engine.begin() as conn: + await conn.execute( + text( + "INSERT INTO users (user_id, company_id, id, password, name, status, role, last_accessed_at) " + "VALUES (:uid, :cid, :id, 'x', 'n', :status, :role, now())" + ), + {"uid": uuid.uuid4(), "cid": uuid.UUID(company_id), "id": login_id, + "status": UserStatus.ACTIVE.value, "role": role}, + ) diff --git a/negodata/backend/tests/test_features.py b/negodata/backend/tests/test_features.py index 99e599b..f6a2e12 100644 --- a/negodata/backend/tests/test_features.py +++ b/negodata/backend/tests/test_features.py @@ -1,29 +1,22 @@ -"""supplier / quotation_setting / quotation 슬라이스 런타임 스모크. +"""협력사·견적세팅·견적을 '만들고 → 목록/단건으로 다시 조회'하는 기본 동작 확인. -create(재조회로 created_at 적재) + list + get 경로를 라이브 DB 로 확인한다. +만든 뒤 다시 읽어와, 서버가 자동으로 채우는 값(생성시각 등)이 제대로 들어갔는지까지 본다. 로그인은 auth_headers. """ import uuid from common.enums import QuotationStatus, QuotationType -async def _headers(client, company_id, login_id): - await client.post( - "/v1/auth/create", - json={"id": login_id, "password": "pw1234", "company_id": company_id, "name": "n"}, - ) - r = await client.post("/v1/auth/login", json={"id": login_id, "password": "pw1234"}) - return {"Authorization": f"Bearer {r.json()['access_token']}"} - - -async def test_supplier_crud(client, company_id): - h = await _headers(client, company_id, "supuser") +async def test_supplier_crud(client, auth_headers): + """검증: 협력사 생성 후 목록·단건 조회. + 기대결과: 생성 success=True, 목록 total=1, 단건 supplier_id 일치, created_at 적재.""" + h = await auth_headers("supuser") r = await client.post("/v1/supplier/create", json={"name": "공급사A", "code": "S1"}, headers=h) body = r.json() assert body["result"]["success"] is True sup = body["supplier"] assert sup["name"] == "공급사A" - assert sup["created_at"] # 재조회 픽스: 서버 기본값 적재 확인 + assert sup["created_at"] # 재조회로 서버 기본값 적재 확인 sid = sup["supplier_id"] r = await client.get("/v1/supplier/list", headers=h) @@ -33,8 +26,10 @@ async def test_supplier_crud(client, company_id): assert r.json()["supplier"]["supplier_id"] == sid -async def test_quotation_setting_crud(client, company_id): - h = await _headers(client, company_id, "qsuser") +async def test_quotation_setting_crud(client, auth_headers): + """검증: 견적 세팅 생성(마진율 0.15) 후 목록 조회. + 기대결과: success=True, target_margin_rate=0.15, card_count 기본 3, 목록 total≥1.""" + h = await auth_headers("qsuser") r = await client.post("/v1/quotation-setting/create", json={"target_margin_rate": 0.15}, headers=h) body = r.json() assert body["result"]["success"] is True @@ -47,9 +42,10 @@ async def test_quotation_setting_crud(client, company_id): assert r.json()["total"] >= 1 -async def test_quotation_create(client, company_id): - h = await _headers(client, company_id, "qtuser") - # type/status 는 int 코드(QuotationType/QuotationStatus). number 는 서버가 생성하므로 미전송. +async def test_quotation_create(client, auth_headers): + """검증: 견적 생성(number 는 서버 생성) 후 qt_id 로 재조회. + 기대결과: 생성 success=True, 재조회 시 name 일치·created_at 적재, 목록 total≥1.""" + h = await auth_headers("qtuser") body = { "qt_setting_id": str(uuid.uuid4()), "version_id": str(uuid.uuid4()), @@ -61,7 +57,7 @@ async def test_quotation_create(client, company_id): } r = await client.post("/v1/quotation/create", json=body, headers=h) res = r.json() - # 생성 응답은 본문(quotation)을 안 주고 qt_id/session_count 만 반환 → qt_id 로 재조회한다. + # 생성 응답엔 quotation 본문이 없고 qt_id/session_count 만 온다 → qt_id 로 재조회 assert res["result"]["success"] is True qt_id = res["qt_id"] assert qt_id @@ -69,7 +65,7 @@ async def test_quotation_create(client, company_id): r = await client.get(f"/v1/quotation/{qt_id}", headers=h) q = r.json()["quotation"] assert q["name"] == "견적A" - assert q["created_at"] # 재조회로 created_at 적재 확인 + assert q["created_at"] r = await client.get("/v1/quotation/list", headers=h) assert r.json()["total"] >= 1 diff --git a/negodata/backend/tests/test_item.py b/negodata/backend/tests/test_item.py index 619f98d..c994cab 100644 --- a/negodata/backend/tests/test_item.py +++ b/negodata/backend/tests/test_item.py @@ -1,40 +1,12 @@ -"""item 도메인 e2e — CRUD + company 멀티테넌트 스코프 검증. - -실행 전제: PostgreSQL(negodata_db). docker compose up -d 후 python -m pytest. -""" +"""item 도메인 e2e — 상품 CRUD. 회사 스코프(타사 격리)는 test_company_scope.py. 로그인은 auth_headers.""" import uuid -import pytest_asyncio -from sqlalchemy import text -from common.enums import CompanyStatus +async def test_item_crud_flow(client, auth_headers): + """검증: 상품 생성→목록→단건→부분수정→soft삭제 전체 흐름. + 기대결과: 각 단계 success, 부분수정은 준 필드만 변경(나머지 유지), soft삭제 후 목록 total=0.""" + h = await auth_headers("itemuser") - -async def _headers(client, company_id, login_id="itemuser", pw="pw1234"): - await client.post( - "/v1/auth/create", - json={"id": login_id, "password": pw, "company_id": company_id, "name": "n"}, - ) - r = await client.post("/v1/auth/login", json={"id": login_id, "password": pw}) - return {"Authorization": f"Bearer {r.json()['access_token']}"} - - -@pytest_asyncio.fixture -async def other_company_id(db_engine) -> str: - cid = uuid.uuid4() - async with db_engine.begin() as conn: - # status 는 NOT NULL(모델 default 는 ORM 전용이라 raw INSERT 엔 안 먹음) → 명시. - await conn.execute( - text("INSERT INTO companies (company_id, name, status) VALUES (:cid, :name, :status)"), - {"cid": cid, "name": "다른회사", "status": CompanyStatus.ACTIVE.value}, - ) - return str(cid) - - -async def test_item_crud_flow(client, company_id): - h = await _headers(client, company_id) - - # 등록 r = await client.post("/v1/item/create", json={"name": "상품A", "price": 1000, "code": "C1"}, headers=h) assert r.status_code == 200 body = r.json() @@ -42,47 +14,28 @@ async def test_item_crud_flow(client, company_id): item_id = body["item"]["item_id"] assert body["item"]["name"] == "상품A" - # 목록 r = await client.get("/v1/item/list", headers=h) body = r.json() assert body["total"] == 1 and len(body["items"]) == 1 - # 단건 조회 r = await client.get(f"/v1/item/{item_id}", headers=h) assert r.json()["item"]["item_id"] == item_id - # 수정 (부분) + # 부분 수정: 준 필드(price)만 바뀌고 안 준 필드(name)는 유지돼야 한다 r = await client.patch(f"/v1/item/update/{item_id}", json={"price": 2000}, headers=h) assert r.json()["item"]["price"] == 2000 - assert r.json()["item"]["name"] == "상품A" # 미지정 필드 유지 + assert r.json()["item"]["name"] == "상품A" - # 삭제 (soft) - r = await client.delete(f"/v1/item/delete/{item_id}", headers=h) - assert r.json()["result"]["success"] is True - - # 삭제 후 목록 0 - r = await client.get("/v1/item/list", headers=h) - assert r.json()["total"] == 0 + # soft delete → 행은 남지만 목록엔 안 잡힌다 + assert (await client.delete(f"/v1/item/delete/{item_id}", headers=h)).json()["result"]["success"] is True + assert (await client.get("/v1/item/list", headers=h)).json()["total"] == 0 -async def test_item_not_found(client, company_id): - h = await _headers(client, company_id) +async def test_item_not_found(client, auth_headers): + """검증: 존재하지 않는 상품 단건 조회. + 기대결과: success=False, code=1300(ITEM_NOT_FOUND).""" + h = await auth_headers("itemuser") r = await client.get(f"/v1/item/{uuid.uuid4()}", headers=h) body = r.json() assert body["result"]["success"] is False - assert body["result"]["code"] == 1300 # ITEM_NOT_FOUND - - -async def test_item_company_scope(client, company_id, other_company_id): - # 회사 A 가 상품 등록 - ha = await _headers(client, company_id, login_id="userA") - r = await client.post("/v1/item/create", json={"name": "A상품"}, headers=ha) - a_item_id = r.json()["item"]["item_id"] - - # 회사 B 유저는 A 의 상품을 목록/단건에서 볼 수 없다 - hb = await _headers(client, other_company_id, login_id="userB") - r = await client.get("/v1/item/list", headers=hb) - assert r.json()["total"] == 0 - - r = await client.get(f"/v1/item/{a_item_id}", headers=hb) - assert r.json()["result"]["code"] == 1300 # 타사 자원은 ITEM_NOT_FOUND + assert body["result"]["code"] == 1300 diff --git a/negodata/backend/tests/test_notification.py b/negodata/backend/tests/test_notification.py new file mode 100644 index 0000000..a8db585 --- /dev/null +++ b/negodata/backend/tests/test_notification.py @@ -0,0 +1,99 @@ +"""알림함 '읽는' 쪽 테스트 — 목록 조회, 안 읽은 개수, 읽음 처리(하나/전체), 그리고 남의 알림은 안 보이는지. + +'마감하면 알림이 쌓이는지'(쓰는 쪽)는 test_quotation_close_notify 가 본다. 여기선 겹치지 않게 '읽는' 동작만 본다. +알림은 원래 견적 마감 때 생기지만, 여기선 테스트를 위해 알림 행을 DB 에 직접 넣는다. +""" +import json +import uuid + +from sqlalchemy import text + +from common.enums import NotificationType + + +async def test_list_and_unread(client, auth_headers, db_engine): + """검증: 내 알림 2건을 시드하고 인박스 목록 조회. + 기대결과: total=2, unread=2, 안읽음이라 read_at 없음(None).""" + h = await auth_headers("notilist") + uid = await _user_id(db_engine, "notilist") + await _seed_notification(db_engine, uid) + await _seed_notification(db_engine, uid, ntype=NotificationType.REGENERATED.value) + + r = await client.get("/v1/notification/list", headers=h) + body = r.json() + assert body["result"]["success"] is True + assert body["total"] == 2 + assert body["unread"] == 2 + assert len(body["notifications"]) == 2 + # 안읽음은 read_at=None → RemoveNoneResponse 가 키를 제거하므로 .get() 으로 확인 + assert all(n.get("read_at") is None for n in body["notifications"]) + + +async def test_inbox_is_user_scoped(client, auth_headers, db_engine): + """검증: 내 알림 1건 + 남의 알림 1건을 시드하고 내 인박스 조회. + 기대결과: total=1, unread=1 — 내 것만 보인다(남의 알림 제외).""" + h = await auth_headers("notiscope") + me = await _user_id(db_engine, "notiscope") + await _seed_notification(db_engine, me) # 내 알림 + await _seed_notification(db_engine, uuid.uuid4()) # 남의 알림(안 보여야 함) + + r = await client.get("/v1/notification/list", headers=h) + body = r.json() + assert body["total"] == 1 and body["unread"] == 1 + + +async def test_read_all_clears_unread(client, auth_headers, db_engine): + """검증: 안읽음 2건 상태에서 read-all 호출 후 다시 목록 조회. + 기대결과: unread=0, 목록엔 그대로 남고(total=2) 모든 read_at 채워짐.""" + h = await auth_headers("notireadall") + uid = await _user_id(db_engine, "notireadall") + await _seed_notification(db_engine, uid) + await _seed_notification(db_engine, uid) + + r = await client.post("/v1/notification/read-all", headers=h) + assert r.json()["result"]["success"] is True + + r = await client.get("/v1/notification/list", headers=h) + body = r.json() + assert body["total"] == 2 and body["unread"] == 0 + assert all(n["read_at"] is not None for n in body["notifications"]) + + +async def test_read_one_decrements_unread(client, auth_headers, db_engine): + """검증: 안읽음 2건 중 1건만 읽음 처리. + 기대결과: unread 2 → 1.""" + h = await auth_headers("notireadone") + uid = await _user_id(db_engine, "notireadone") + await _seed_notification(db_engine, uid) + await _seed_notification(db_engine, uid) + + r = await client.get("/v1/notification/list", headers=h) + target_id = r.json()["notifications"][0]["notification_id"] + + r = await client.post(f"/v1/notification/{target_id}/read", headers=h) + assert r.json()["result"]["success"] is True + + r = await client.get("/v1/notification/list", headers=h) + assert r.json()["unread"] == 1 + + +# ===== 헬퍼 (위 테스트들이 쓰는 도우미) ===== +async def _user_id(engine, login_id): + """auth_headers 로 시드된 유저의 user_id(알림 시드/스코프 확인용).""" + async with engine.begin() as conn: + return (await conn.execute( + text("SELECT user_id FROM users WHERE id = :id"), {"id": login_id} + )).scalar_one() + + +async def _seed_notification(engine, user_id, *, ntype=NotificationType.SUCCESS.value, data=None): + """안읽음(read_at NULL) 알림 1건 시드.""" + async with engine.begin() as conn: + await conn.execute( + text( + "INSERT INTO notifications (notification_id, user_id, type, data, read_at) " + "VALUES (:nid, :uid, :type, CAST(:data AS JSONB), NULL)" + ), + {"nid": uuid.uuid4(), "uid": user_id, "type": ntype, + "data": json.dumps(data or {"qt_name": "견적A"})}, + ) diff --git a/negodata/backend/tests/test_quotation_close_notify.py b/negodata/backend/tests/test_quotation_close_notify.py new file mode 100644 index 0000000..95d0ce9 --- /dev/null +++ b/negodata/backend/tests/test_quotation_close_notify.py @@ -0,0 +1,224 @@ +"""견적 마감(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()) diff --git a/negodata/backend/tests/test_quotation_create.py b/negodata/backend/tests/test_quotation_create.py new file mode 100644 index 0000000..f8192ce --- /dev/null +++ b/negodata/backend/tests/test_quotation_create.py @@ -0,0 +1,98 @@ +"""견적 생성 — item×supplier 조합마다 세션이 생기고, 목표가가 산정되는지 검증. + +기존 test_features.test_quotation_create 는 item/supplier 없이 '세션 0건' 경로만 본다. +여기선 상품(인터넷최저가)을 시드해 세션 생성 + 목표가 계산(신규=인터넷최저가×(1−수수료))까지 본다. +서비스(create_quotation)를 직접 호출한다 — HTTP/auth 경로(현재 /v1/auth/create 미존재)를 안 타고 생성 로직만 격리. +""" +import uuid +from datetime import datetime + +from sqlalchemy import text + +from common.enums import QuotationType +from crud.quotation_crud import QuotationCRUD +from router.v1.quotation.protocol import Req_CreateQuotation +from services.quotation_service import QuotationService + +FUTURE = datetime(2999, 1, 1) # 마감시각 미래 — 생성 직후 크론에 안 잡히게 + + +async def test_create_builds_sessions_with_target_price(db_engine, company_id): + """검증: 신규견적을 상품2×공급사2로 생성. + 기대결과: success=True, 세션 4개, 각 목표가 = int(인터넷최저가 × (1−0.078)).""" + item1 = await _seed_item(db_engine, company_id, internet_lowest=100_000) + item2 = await _seed_item(db_engine, company_id, internet_lowest=50_000) + suppliers = [uuid.uuid4(), uuid.uuid4()] + + req = Req_CreateQuotation( + qt_setting_id=uuid.uuid4(), # FK 미설정 — 세팅 없으면 율 0(신규는 인터넷최저가만 쓰므로 무관) + name="신규견적A", + type=QuotationType.NEW_QUOTE.value, + end_time=FUTURE, + item_ids=[item1, item2], + supplier_ids=suppliers, + ) + res = await _service().create_quotation(str(uuid.uuid4()), req) + + assert res.result.success is True + assert res.session_count == 4 # 상품 2 × 공급사 2 + + fee = QuotationService.INTERNET_AVERAGE_FEE + expected = {item1: int(100_000 * (1 - fee)), item2: int(50_000 * (1 - fee))} + rows = await _session_target_prices(db_engine, res.qt_id) + assert len(rows) == 4 + for item_id, target_price in rows: + assert target_price == expected[item_id] # 상품별 목표가가 공급사 수만큼 동일 + + +async def test_create_without_price_fails(db_engine, company_id): + """검증: 가격 후보(인터넷최저가·md 등)가 전무한 상품으로 견적 생성. + 기대결과: 목표가 산정 불가로 success=False, 세션 0건(미생성).""" + item = await _seed_item(db_engine, company_id, internet_lowest=None) + + req = Req_CreateQuotation( + qt_setting_id=uuid.uuid4(), + name="가격없음", + type=QuotationType.NEW_QUOTE.value, + end_time=FUTURE, + item_ids=[item], + supplier_ids=[uuid.uuid4()], + ) + res = await _service().create_quotation(str(uuid.uuid4()), req) + + assert res.result.success is False # QUOTATION_TARGET_PRICE_UNAVAILABLE + rows = await _session_target_prices(db_engine, res.qt_id) if res.qt_id else [] + assert rows == [] + + +# ===== 헬퍼 (위 테스트들이 쓰는 도우미) ===== +def _service(): + return QuotationService(QuotationCRUD()) + + +async def _seed_item(engine, company_id, *, internet_lowest): + """상품 1건 시드(인터넷최저가만). category_type·internet_lowest_price_yn 은 NOT NULL — + ORM default 는 raw INSERT 에 안 먹으므로 명시한다(conftest companies.status 와 같은 이유).""" + item_id = uuid.uuid4() + async with engine.begin() as conn: + await conn.execute( + text( + "INSERT INTO items " + "(item_id, company_id, user_id, name, category_type, " + " internet_lowest_price_yn, internet_lowest_price) VALUES " + "(:item_id, :company_id, :user_id, '상품', 1, false, :ilp)" + ), + {"item_id": item_id, "company_id": uuid.UUID(company_id), + "user_id": uuid.uuid4(), "ilp": internet_lowest}, + ) + return item_id + + +async def _session_target_prices(engine, qt_id): + """생성된 견적의 (item_id -> target_price) 매핑.""" + async with engine.begin() as conn: + rows = (await conn.execute( + text("SELECT item_id, target_price FROM sessions WHERE quotation_id = :qt"), + {"qt": qt_id}, + )).all() + return rows diff --git a/negodata/backend/tests/test_scheduler.py b/negodata/backend/tests/test_scheduler.py index 2d26d49..0abc014 100644 --- a/negodata/backend/tests/test_scheduler.py +++ b/negodata/backend/tests/test_scheduler.py @@ -1,8 +1,15 @@ -"""scheduler 잡 e2e — '대상 선정'(어떤 견적을 고르나) + close_and_decide 위임 결과 검증. +"""scheduler(마감 크론 잡) e2e 테스트 — 어떤 견적을 고르고, 마감하면 결과가 어떻게 나오는지 확인. -실행 전제: 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 로 직접 시드. +용어: 견적 = 한 건의 입찰 공고 / 세션 = 그 견적에 참여한 공급사별 협상 1건 / 마감 = 견적을 닫고 낙찰자를 정함. + +마감을 자동으로 돌리는 크론 잡이 2개 있다(scheduler/jobs.py): + · 잡① close_expired_quotations : 마감시각(end_time)이 지났는데 아직 안 닫힌 견적을 닫는다. + · 잡② close_negotiated_quotations : 참여 세션이 전부 끝난(협상 종료) 견적을 닫는다. +두 잡 모두, 고른 견적마다 close_and_decide() 를 불러 결과(낙찰 / 다음 라운드 재생성 / 그냥 마감)를 정한다. + +이 파일은 그 두 잡이 (1) 마감할 견적을 올바로 고르는지, (2) 마감 결과가 맞는지 확인한다. +잡에는 HTTP 엔드포인트가 없어 scheduler.jobs 함수를 직접 부른다(앱과 같은 DB 연결을 써서 mock 불필요). +세션 상태(협상완료/거부/입찰가 등)는 협상 화면에서만 생기는 값이라 API 로 못 만든다 → SQL 로 직접 넣는다. """ import asyncio import uuid @@ -16,21 +23,144 @@ 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) # 마감시각 미래(잡① 제외) +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 도 비워 격리.""" + """각 테스트 시작 전에 quotations·sessions 를 모두 비워 깨끗한 상태로 만든다. + + 공용 db_engine 픽스처는 quotations 만 비운다. 그런데 잡②는 '세션이 다 끝난 견적'을 전체 견적에서 찾으므로, + 앞선 다른 테스트가 남긴 세션이 남아 있으면 엉뚱한 견적이 대상에 끼어든다 → 그래서 여기서 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 test_close_expired_picks_only_due_and_open(clean): + """검증: 잡①을 돌린다. 견적 4개를 섞어둔다 — + ① 마감시각 지난 미마감 ② 마감시각 안 지난 것 ③ 이미 마감된 것 ④ 삭제된 것. + 기대결과: ①(due) 1건만 새로 마감(CLOSED)되고, ②③④ 는 그대로 둔다.""" + 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 # 원래부터 마감 + assert (await _quotation_row(engine, deleted)).status == QuotationStatus.IN_PROGRESS.value # 삭제분은 건드리지 않음 + + +async def test_close_negotiated_picks_when_all_sessions_ended(clean): + """검증: 잡②를 돌린다. 견적 3개를 섞어둔다 — + ① 세션이 전부 끝난 것 ② 아직 진행중인 세션이 있는 것 ③ 세션이 아예 없는 것. + 기대결과: ①(세션 다 끝남)만 마감(CLOSED)되고, ②③ 은 제외.""" + 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) + # ③ 세션이 아예 없음 → 제외(끝났다고 볼 세션 자체가 없음) + 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 + + +async def test_award_single_lowest(clean): + """검증: 두 공급사가 각각 100·200 으로 협상완료(DONE)한, 마감시각 지난 견적을 잡①로 마감. + 기대결과: 마감(CLOSED)되고, 더 싼 100 공급사가 단독 낙찰(낙찰 있음 + 낙찰자=그 공급사).""" + 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): + """검증: 입찰 없이 '거부'만 있는, 마감시각 지난 견적을 잡①로 마감. + 기대결과: 마감(CLOSED)되지만 낙찰자는 없음(살 사람이 없으니 그냥 닫힘).""" + 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 # 거부뿐이라 낙찰 없이 마감 + + +async def test_scheduler_disabled_without_env(monkeypatch): + """검증: SCHEDULER_ENABLED 환경변수 없이 start_scheduler() 호출. + 기대결과: 스케줄러가 켜지지 않는다(운영에서 실수로 자동 마감이 도는 걸 막는 안전장치).""" + import scheduler + monkeypatch.delenv("SCHEDULER_ENABLED", raising=False) + scheduler._scheduler = None + scheduler.start_scheduler() + assert scheduler._scheduler is None # 환경변수가 1이 아니면 미기동 + + +async def test_scheduler_registers_both_jobs(monkeypatch): + """검증: SCHEDULER_ENABLED=1 로 start_scheduler() 호출. + 기대결과: 마감 잡 2개(close_expired·close_negotiated)가 스케줄에 등록된다.""" + 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): + """검증: 스케줄러에 잡을 걸어 실제로 발화시킨다(1초 간격으로). + 기대결과: 스케줄러가 잡을 호출해 마감시각 지난 견적이 몇 초 안에 마감(CLOSED)된다 — '스케줄러→잡→마감' 경로 확인.""" + 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): # 잡은 1초 뒤 첫 발화 → 최대 ~5초 동안 0.2초 간격으로 확인 + 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) + + +# ===== 헬퍼 (위 테스트들이 쓰는 도우미. FK 미설정이라 user/item/supplier 없이 임의 uuid 로 충분) ===== async def _add_quotation(engine, *, status=QuotationStatus.IN_PROGRESS.value, end_time=PAST, deleted=False): + """견적 1건을 DB 에 직접 넣는다(시드). status/end_time/deleted 로 '대상/제외' 상황을 만든다.""" qt_id = uuid.uuid4() async with engine.begin() as conn: await conn.execute( @@ -52,6 +182,7 @@ async def _add_quotation(engine, *, status=QuotationStatus.IN_PROGRESS.value, en async def _add_session(engine, qt_id, *, status, bid_price=None, supplier_id=None): + """세션(공급사 협상 1건)을 DB 에 직접 넣는다. status/bid_price 로 협상완료·거부·입찰가를 만든다.""" async with engine.begin() as conn: await conn.execute( text( @@ -71,119 +202,9 @@ async def _add_session(engine, qt_id, *, status, bid_price=None, supplier_id=Non async def _quotation_row(engine, qt_id): + """견적 1건을 다시 읽어온다(마감 후 status·낙찰자 확인용).""" 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)