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