hbyang
7e0f88ca03
백엔드 close_and_decide 경로: - 동시 이중 마감 가드: 마감 판정 전 원자적 CLOSED 선점(claim)으로 두 크론 잡·수동마감 경합 직렬화 - 재생성 실패 표면화: regenerate_next_round 결과 검사 → 실패 시 REGEN_FAILED 반환(체인 끊김 은폐 방지) - 차수 충돌 방지: 다음 라운드 = 체인 최신 round+1(chain_max_round 기준) - 재생성 사유 집계 정밀화: 미참여/동가를 양성 표식으로 구분(단독낙찰·거부 오집계 제거) - 재생성 라운드 최소 협상기간 하한(즉시 재마감 캐스케이드 방지) - 잡 루프 per-item 예외 격리(한 건 실패가 배치 전체를 멈추지 않음) 프론트: - useChatController: 무권한 가드를 sessionId 별로 추적해 세션 변경 시 자연 해제 - useScrollLock: 마지막 해제를 rAF 로 지연해 재마운트 사이 일시적 잠금 해제 방지 - quotation 상세 쿼리 placeholderData 로 라운드 전환 중 시트 유지 테스트: - 신규 test_close_and_decide_fixes.py(동시성·차수·집계·기간 하한 검증) - conftest 결함 수정(존재하지 않는 tbl_account TRUNCATE 제거, companies.status 명시) - stale 테스트 갱신(test_quotation_create 를 타입드 Req/신규 응답 형식에 맞게 재작성) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
96 lines
4.1 KiB
Python
96 lines
4.1 KiB
Python
# 테스트는 APP_ENV=test 로 실행한다 (config.test.toml → negosium_test_db, dev DB 와 분리).
|
|
# 이 픽스처들은 TRUNCATE 를 하므로 dev DB(negosium_db)와 절대 공유하면 안 된다(아래 db_engine 안전가드 참고).
|
|
# config.server_configs 가 import 되는 순간 config.<APP_ENV>.toml 을 읽으므로 가장 먼저 설정.
|
|
import os
|
|
|
|
os.environ.setdefault("APP_ENV", "test")
|
|
|
|
import uuid
|
|
|
|
import pytest_asyncio
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
|
|
from common.database.model.models import MAIN_BASE
|
|
from common.enums import CompanyStatus
|
|
from config.server_configs import main_db_config
|
|
|
|
|
|
# 모델이 쓰는 스키마. test DB 는 비어 있을 수 있어 create_all 전에 직접 만든다.
|
|
# 또 TRUNCATE 가 unqualified 테이블명을 쓰므로 이 스키마들을 search_path 에 얹어 해석시킨다.
|
|
_SCHEMAS = ("company", "quotation", "card", "negotiation", "partner")
|
|
|
|
|
|
def _write_url(cfg) -> str:
|
|
pw = f":{cfg.write_pw}" if cfg.write_pw else ""
|
|
return f"postgresql+asyncpg://{cfg.write_id}{pw}@{cfg.write_host}:{cfg.write_port}/{cfg.name}"
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def db_engine():
|
|
"""테스트용 스키마를 보장하고, 매 테스트 시작 시 테이블을 비워 격리한다.
|
|
|
|
⚠ 이 픽스처는 TRUNCATE 한다 → dev DB(negosium_db)를 가리키면 실데이터가 날아간다.
|
|
그래서 test 전용 DB(이름에 'test')가 아니면 즉시 중단한다(config.test.toml / APP_ENV=test).
|
|
앱(DB_SESSION_MNG)도 APP_ENV=test 면 같은 test DB 에 접속하므로 여기서 만든 스키마를 공유한다.
|
|
"""
|
|
# 안전가드: dev DB 오염 방지. negosium_test_db 이외엔 절대 실행하지 않는다.
|
|
assert "test" in main_db_config.name, (
|
|
f"테스트가 비-test DB('{main_db_config.name}')를 가리킵니다. "
|
|
"APP_ENV=test(config.test.toml)로 실행하세요. dev DB 보호를 위해 중단합니다."
|
|
)
|
|
engine = create_async_engine(
|
|
_write_url(main_db_config),
|
|
connect_args={"server_settings": {"search_path": ",".join(_SCHEMAS) + ",public"}},
|
|
)
|
|
async with engine.begin() as conn:
|
|
for sch in _SCHEMAS:
|
|
await conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {sch}"))
|
|
await conn.run_sync(MAIN_BASE.metadata.create_all) # 이미 있으면 skip
|
|
# negodata 도메인 테이블 전부 비워 격리 (CASCADE: FK 미설정이라 안전망)
|
|
await conn.execute(
|
|
text(
|
|
"TRUNCATE TABLE users, companies, items, suppliers, "
|
|
"quotation_settings, quotations, sessions RESTART IDENTITY CASCADE"
|
|
)
|
|
)
|
|
yield engine
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def company_id(db_engine) -> str:
|
|
"""테스트용 소속사 1개를 시드하고 company_id(uuid str)를 돌려준다.
|
|
users 는 company_id 를 요구하므로 계정 생성 테스트의 선행 조건이다.
|
|
"""
|
|
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)
|
|
|
|
|
|
@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 client(db_engine):
|
|
"""앱을 실제 네트워크 없이 호출하는 httpx 클라이언트 (ASGITransport)."""
|
|
from router.router import app
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
|
yield ac
|