[test] negodata: 백엔드 테스트 스위트 구축 + 공통 픽스처(conftest) 정비

- 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>

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)
+@pytest_asyncio.fixture
-async def _dispose_app_engines():
+async def other_company_id(db_engine) -> str:
-    """테스트 세션이 끝날 때 앱 싱글톤 엔진을 정리한다.
+    """company_id 와 다른 소속사 1개(회사 스코프/IDOR 격리 테스트용)."""
-    (이벤트 루프 종료 후 커넥션이 GC 되며 나오는 'Event loop is closed' 경고 제거)
+    cid = uuid.uuid4()
-    """
+    async with db_engine.begin() as conn:
-    yield
+        await conn.execute(
-    from common.database.db_session_manager import DB_SESSION_MNG
+            text("INSERT INTO companies (company_id, name, status) VALUES (:cid, :name, :status)"),
-
+            {"cid": cid, "name": "다른회사", "status": CompanyStatus.ACTIVE.value},
-    await DB_SESSION_MNG.dispose_all()
+        )
+    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

negodata/backend/tests/test_auth.py

@@ -1,34 +1,15 @@
-"""auth 도메인 e2e 테스트 (negodata: users/companies 기반).
+"""auth 도메인 e2e — 로그인 / 내정보 / 인증거부 흐름.
 
-실행 전제: PostgreSQL(negodata_db)이 떠 있어야 한다.
+계정 생성·중복·최고관리자 스코프는 test_company_user.py. 유저 시드/로그인은 auth_headers 픽스처.
-    docker compose up -d  # 또는 로컬 postgres
-    cd negodata/backend && python -m pytest
-계정 생성은 company_id 를 요구하므로 company_id 픽스처(conftest)가 소속사를 시드한다.
 """
 
 
-async def test_create_and_login_flow(client, company_id):
+async def test_login_and_me_flow(auth_headers, client, company_id):
-    # 1) 계정 생성 (회사 하위로)
+    """검증: 시드된 유저가 로그인해 받은 토큰으로 /me 호출.
-    r = await client.post(
+    기대결과: 200, 본인 id·name·소속사(company_id)가 그대로 반환."""
-        "/v1/auth/create",
+    h = await auth_headers("user1", name="홍길동")
-        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"]
 
-    # 2) 로그인 -> 토큰 발급
+    r = await client.get("/v1/auth/me", headers=h)
-    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}"})
     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):
+async def test_login_with_wrong_password(auth_headers, client):
-    await client.post(
+    """검증: 존재하는 계정에 '틀린 비밀번호'로 로그인.
-        "/v1/auth/create",
+    기대결과: 로그인 실패 — success=False, code=1200(ACCOUNT_INVALID_INFO), 토큰 빈 문자열."""
-        json={"id": "user2", "password": "correct", "company_id": company_id, "name": "n"},
+    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)

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) 이전 라운드가 있어도, 이후 라운드의 미참여 재생성 예산을 소진하지 않는다.
+    """검증: round1=단독낙찰 + round2=전원 미참여 인 체인에서 round2 를 마감.
-    (구버전: equal_bid_yn=False 인 단독낙찰 라운드를 미참여로 세어 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()

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

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},
+        )

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):
+async def test_supplier_crud(client, auth_headers):
-    await client.post(
+    """검증: 협력사 생성 후 목록·단건 조회.
-        "/v1/auth/create",
+    기대결과: 생성 success=True, 목록 total=1, 단건 supplier_id 일치, created_at 적재."""
-        json={"id": login_id, "password": "pw1234", "company_id": company_id, "name": "n"},
+    h = await auth_headers("supuser")
-    )
-    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")
     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):
+async def test_quotation_setting_crud(client, auth_headers):
-    h = await _headers(client, company_id, "qsuser")
+    """검증: 견적 세팅 생성(마진율 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):
+async def test_quotation_create(client, auth_headers):
-    h = await _headers(client, company_id, "qtuser")
+    """검증: 견적 생성(number 는 서버 생성) 후 qt_id 로 재조회.
-    # type/status 는 int 코드(QuotationType/QuotationStatus). number 는 서버가 생성하므로 미전송.
+    기대결과: 생성 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

negodata/backend/tests/test_item.py

@@ -1,40 +1,12 @@
-"""item 도메인 e2e — CRUD + company 멀티테넌트 스코프 검증.
+"""item 도메인 e2e — 상품 CRUD. 회사 스코프(타사 격리)는 test_company_scope.py. 로그인은 auth_headers."""
-
-실행 전제: PostgreSQL(negodata_db). docker compose up -d 후 python -m pytest.
-"""
 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)
+    # soft delete → 행은 남지만 목록엔 안 잡힌다
-    r = await client.delete(f"/v1/item/delete/{item_id}", headers=h)
+    assert (await client.delete(f"/v1/item/delete/{item_id}", headers=h)).json()["result"]["success"] is True
-    assert r.json()["result"]["success"] is True
+    assert (await client.get("/v1/item/list", headers=h)).json()["total"] == 0
-
-    # 삭제 후 목록 0
-    r = await client.get("/v1/item/list", headers=h)
-    assert r.json()["total"] == 0
 
 
-async def test_item_not_found(client, company_id):
+async def test_item_not_found(client, auth_headers):
-    h = await _headers(client, company_id)
+    """검증: 존재하지 않는 상품 단건 조회.
+    기대결과: 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
+    assert body["result"]["code"] == 1300
-
-
-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

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"})},
+        )

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())

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

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.
+용어: 견적 = 한 건의 입찰 공고 / 세션 = 그 견적에 참여한 공급사별 협상 1건 / 마감 = 견적을 닫고 낙찰자를 정함.
-잡은 HTTP 엔드포인트가 없어 scheduler.jobs 함수를 직접 호출한다(앱과 같은 DB_SESSION_MNG 사용 → mock 불필요).
+
-세션 상태(DONE/REJECTED/bid_price 등)는 협상 프론트가 만드는 값이라 API 로 못 만든다 → SQL 로 직접 시드.
+마감을 자동으로 돌리는 크론 잡이 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)    # 마감시각 지남(잡① 대상)
+PAST = datetime(2020, 1, 1)    # 마감시각이 이미 지난 시점(잡①의 마감 대상)
-FUTURE = datetime(2999, 1, 1)  # 마감시각 미래(잡① 제외)
+FUTURE = datetime(2999, 1, 1)  # 마감시각이 아직 안 온 시점(잡①에서 제외)
 
 
 @pytest_asyncio.fixture
 async def clean(db_engine):
-    """conftest 의 db_engine 은 quotations 만 비우고 sessions 는 안 비운다(FK 미설정 → CASCADE 대상 아님).
+    """각 테스트 시작 전에 quotations·sessions 를 모두 비워 깨끗한 상태로 만든다.
-    잡②(close_negotiated)는 전체 견적을 스캔하므로 다른 테스트가 남긴 세션이 결과를 흔든다 → 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)