cjw
79441a8a87
D4 文档/规划/历史混放收口：
- docs/TECH_DEBT.md §2 由'按批次回顾'精简为'按主题摘要'，
与§3重复内容（批次0-4详细展开）整体迁入
docs/archive/2026-09_governance_batches.md
- docs/STATUS.md 顶部 2026-09-17 之前条目迁入
docs/archive/2026-09_status_history.md，仅保留指针
- D2 历史口径补齐为 2026-09-18 批次4后续专项清偿
- D4 标已清偿
- AGENTS.md / docs/archive/README.md 同步导航
D13 锁文件流程固化（镜像引入主体已清偿，仅剩锁文件落盘）：
- 新增 deploy/generate_lockfiles.{sh,bat}：在 moldinsight conda
环境（仅项目依赖）执行 pip freeze --exclude pythonocc-core，
产出 deploy/requirements-{base,moldinsight}.lock.txt
- deploy/Dockerfile.moldinsight 注释改为指向生成脚本
- docs/OPERATIONS.md §2.1 增加完整流程说明
- tests/test_lockfile_generation.py 加锁文件存在性+体积契约；
tests/conftest.py 注册 --run-lockfile-check 选项，
默认 skip（仓库单测不阻塞），CI 镜像构建 job 显式启用 fail-fast
遗留：锁文件本身尚未落盘（本机Miniforge跨项目开发栈混装，
污染严重不能直接 pip freeze）；待 CI / 生产首次构建时按流程落锁。
测试基线：126 passed, 9 skipped（默认4原有skip + D13新增5skip；
启用 --run-lockfile-check 时严格断言2项锁文件契约）
Co-Authored-By: Claude Code <noreply@anthropic.com>
219 lines
7.3 KiB
Python
219 lines
7.3 KiB
Python
import os
|
||
import sys
|
||
import tempfile
|
||
|
||
vendor_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "_vendor"))
|
||
if os.path.isdir(vendor_dir) and vendor_dir not in sys.path:
|
||
sys.path.insert(0, vendor_dir)
|
||
|
||
src_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "src"))
|
||
if os.path.isdir(src_dir) and src_dir not in sys.path:
|
||
sys.path.insert(0, src_dir)
|
||
|
||
import pytest
|
||
from httpx import AsyncClient, ASGITransport
|
||
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
|
||
from sqlalchemy import select
|
||
|
||
from fastapi import FastAPI, APIRouter
|
||
from inventory.api import inventory_router
|
||
from shared.models.base import Base
|
||
from shared.models.identity import User
|
||
from inventory.models import Customer, Warehouse, Supplier, Product, ProductMaterial, Inventory, MaterialSupplier, SalesOrder, SalesOrderItem
|
||
from moldinsight.models import STPFile, ProcessingTask
|
||
import moldinsight.models # noqa: F401 # 全量注册：create_all 需含 moldinsight 表（stp_files 等）
|
||
from shared.database.database import get_db_session
|
||
from shared.services.auth_service import get_current_active_user, get_current_admin_user
|
||
|
||
|
||
@pytest.fixture(scope="session")
|
||
def anyio_backend():
|
||
return "asyncio"
|
||
|
||
|
||
@pytest.fixture(scope="session")
|
||
def sqlite_db_path():
|
||
fd, path = tempfile.mkstemp(prefix="gemold_test_", suffix=".db")
|
||
os.close(fd)
|
||
yield path
|
||
try:
|
||
os.remove(path)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
@pytest.fixture(scope="session")
|
||
async def async_engine(sqlite_db_path):
|
||
engine = create_async_engine(f"sqlite+aiosqlite:///{sqlite_db_path}", future=True)
|
||
async with engine.begin() as conn:
|
||
await conn.run_sync(Base.metadata.create_all)
|
||
yield engine
|
||
await engine.dispose()
|
||
|
||
|
||
@pytest.fixture(scope="function")
|
||
async def seeded_db(async_engine):
|
||
"""每个测试前清空所有表并重新播种，确保隔离。"""
|
||
session_factory = async_sessionmaker(async_engine, class_=AsyncSession, expire_on_commit=False)
|
||
async with session_factory() as session:
|
||
# 按 FK 依赖逆序清空所有表
|
||
for table in reversed(Base.metadata.sorted_tables):
|
||
await session.execute(table.delete())
|
||
await session.commit()
|
||
|
||
async with session_factory() as session:
|
||
user = User(
|
||
id=1,
|
||
username="tester",
|
||
email="tester@example.com",
|
||
hashed_password="x",
|
||
full_name="Tester",
|
||
is_active=True,
|
||
)
|
||
customer = Customer(id=1, code="C001", name="客户A", is_active=True)
|
||
supplier = Supplier(id=1, code="S001", name="供应商A", is_active=True)
|
||
warehouse = Warehouse(id=1, code="W001", name="默认仓库", is_active=True, is_default=True)
|
||
|
||
material = Product(
|
||
id=1,
|
||
sku="MAT-001",
|
||
name="钢材",
|
||
unit="kg",
|
||
item_type="material",
|
||
cost_price=10.0,
|
||
sale_price=0,
|
||
min_stock=0,
|
||
max_stock=100000,
|
||
is_active=True,
|
||
)
|
||
finished = Product(
|
||
id=2,
|
||
sku="MOLD-STD",
|
||
name="标准模具",
|
||
unit="套",
|
||
item_type="finished",
|
||
cost_price=0,
|
||
sale_price=1000.0,
|
||
min_stock=0,
|
||
max_stock=0,
|
||
is_active=True,
|
||
)
|
||
# 成品无 BOM，用于测试 BOM 缺失路径
|
||
finished_no_bom = Product(
|
||
id=3,
|
||
sku="MOLD-NB",
|
||
name="无BOM成品",
|
||
unit="套",
|
||
item_type="finished",
|
||
cost_price=0,
|
||
sale_price=500.0,
|
||
min_stock=0,
|
||
max_stock=0,
|
||
is_active=True,
|
||
)
|
||
bom = ProductMaterial(
|
||
id=1,
|
||
finished_product_id=finished.id,
|
||
material_product_id=material.id,
|
||
quantity=2.0,
|
||
loss_rate=0.05,
|
||
)
|
||
inv = Inventory(
|
||
id=1,
|
||
product_id=material.id,
|
||
warehouse_id=warehouse.id,
|
||
quantity=1000,
|
||
locked_quantity=0,
|
||
)
|
||
# 物料-供应商关联（用于采购需求推导测试）
|
||
ms = MaterialSupplier(
|
||
id=1,
|
||
product_id=material.id,
|
||
supplier_id=supplier.id,
|
||
is_primary=True,
|
||
lead_time=7,
|
||
)
|
||
|
||
stp_file = STPFile(
|
||
id=1,
|
||
original_filename="demo-mold.stp",
|
||
object_key="uploads/demo.stp",
|
||
storage_bucket="test-bucket",
|
||
file_size=123,
|
||
file_hash="hash-demo-1",
|
||
mime_type="application/step",
|
||
user_id=user.id,
|
||
volume=1000.0,
|
||
surface_area=200.0,
|
||
product_weight=50.0,
|
||
)
|
||
task = ProcessingTask(
|
||
id=1,
|
||
task_id="task-demo-1",
|
||
status="completed",
|
||
task_type="stp_parsing",
|
||
progress=100,
|
||
current_step="done",
|
||
stp_file_id=stp_file.id,
|
||
)
|
||
|
||
session.add_all([user, customer, supplier, warehouse, material, finished, finished_no_bom, bom, inv, ms, stp_file, task])
|
||
await session.commit()
|
||
|
||
async with session_factory() as session:
|
||
yield session
|
||
|
||
|
||
def pytest_addoption(parser):
|
||
"""D13 部署侧契约：仅在显式 --run-lockfile-check 时启用锁文件存在性断言。"""
|
||
parser.addoption(
|
||
"--run-lockfile-check",
|
||
action="store_true",
|
||
default=False,
|
||
help="启用 D13 锁文件部署侧契约测试（CI 镜像构建 job 使用）",
|
||
)
|
||
|
||
|
||
def pytest_collection_modifyitems(config, items):
|
||
"""默认跳过 D13 部署侧契约（仓库侧单测不应被尚未落地的锁文件阻断）。"""
|
||
if config.getoption("--run-lockfile-check", default=False):
|
||
return
|
||
skip_marker = pytest.mark.skip(
|
||
reason="D13 部署侧契约：默认 skip；CI 镜像构建 job 需传入 --run-lockfile-check 启用"
|
||
)
|
||
for item in items:
|
||
if "test_lockfile_generation" in item.nodeid:
|
||
item.add_marker(skip_marker)
|
||
|
||
|
||
@pytest.fixture(scope="function")
|
||
async def client(async_engine, seeded_db):
|
||
session_factory = async_sessionmaker(async_engine, class_=AsyncSession, expire_on_commit=False)
|
||
|
||
test_app = FastAPI()
|
||
test_app.include_router(inventory_router)
|
||
|
||
async def override_get_db_session():
|
||
async with session_factory() as session:
|
||
yield session
|
||
|
||
async def override_get_current_active_user():
|
||
async with session_factory() as session:
|
||
result = await session.execute(select(User).where(User.username == "tester"))
|
||
return result.scalar_one()
|
||
|
||
async def override_get_current_admin_user():
|
||
async with session_factory() as session:
|
||
result = await session.execute(select(User).where(User.username == "tester"))
|
||
return result.scalar_one()
|
||
|
||
test_app.dependency_overrides[get_db_session] = override_get_db_session
|
||
test_app.dependency_overrides[get_current_active_user] = override_get_current_active_user
|
||
test_app.dependency_overrides[get_current_admin_user] = override_get_current_admin_user
|
||
|
||
transport = ASGITransport(app=test_app)
|
||
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
||
yield ac
|
||
|
||
test_app.dependency_overrides.clear()
|