"""测试配置与 Fixtures。 提供测试数据库、认证头、数据工厂等公共 fixtures。 使用 SQLite 内存数据库进行隔离测试，每个测试用例结束后自动回滚。 """ from __future__ import annotations import os import sys import asyncio import pytest from datetime import datetime, timedelta, timezone from typing import Generator # 添加项目根目录到 Python 路径 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) # 设置测试环境变量，必须在导入 app 之前 os.environ["APP_ENV"] = "test" os.environ["MYSQL_DATABASE"] = ":memory:" os.environ["SECRET_KEY"] = "test-secret-key-for-testing-only" os.environ["JWT_EXPIRE_MINUTES"] = "60" os.environ["AI_PROVIDER"] = "mock" os.environ["AUTH_TOKEN_STORE_PATH"] = "D:/tmp/order-flow-test/revoked_tokens.json" import uuid from sqlalchemy import create_engine, event from sqlalchemy.orm import Session, sessionmaker from backend.app.core.security import create_access_token, hash_password from backend.app.db import Base from backend.app.main import app # ============================================================ # 兼容 httpx 0.28+ 的同步测试客户端 # ============================================================ class SyncTestClient: """同步测试客户端，封装 httpx.AsyncClient + ASGITransport。 兼容 httpx 0.28+ (移除了 Client 的 app 参数)。 提供与 starlette.testclient.TestClient 相同的同步接口。 """ def __init__(self, app_obj, raise_server_exceptions=True): import httpx self._app = app_obj self._raise = raise_server_exceptions self._transport = httpx.ASGITransport(app=app_obj, raise_app_exceptions=raise_server_exceptions) self._client = httpx.AsyncClient( transport=self._transport, base_url="http://testserver", follow_redirects=True, ) self._loop = asyncio.new_event_loop() def _run(self, coro): return self._loop.run_until_complete(coro) def request(self, method, url, **kwargs): return self._run(self._client.request(method, url, **kwargs)) def get(self, url, **kwargs): return self.request("GET", url, **kwargs) def post(self, url, **kwargs): return self.request("POST", url, **kwargs) def put(self, url, **kwargs): return self.request("PUT", url, **kwargs) def delete(self, url, **kwargs): return self.request("DELETE", url, **kwargs) def patch(self, url, **kwargs): return self.request("PATCH", url, **kwargs) def options(self, url, **kwargs): return self.request("OPTIONS", url, **kwargs) def head(self, url, **kwargs): return self.request("HEAD", url, **kwargs) def close(self): self._run(self._client.aclose()) self._loop.close() def __enter__(self): return self def __exit__(self, *args): self.close() # ============================================================ # 数据库 Fixtures # ============================================================ @pytest.fixture(scope="session") def engine(): """创建 SQLite 内存引擎，整个测试会话共享。 使用 check_same_thread=False 允许多线程访问。 """ eng = create_engine( "sqlite:///:memory:", echo=False, connect_args={"check_same_thread": False}, ) Base.metadata.create_all(eng) yield eng Base.metadata.drop_all(eng) eng.dispose() @pytest.fixture(scope="function") def db_session(engine) -> Generator[Session, None, None]: """每个测试函数独立的数据库会话，测试结束后自动回滚。""" connection = engine.connect() transaction = connection.begin() session = Session(bind=connection) yield session session.close() transaction.rollback() connection.close() @pytest.fixture(scope="function") def client(db_session, seed_data): """FastAPI 测试客户端，使用测试数据库会话。 使用 SyncTestClient (封装 httpx.AsyncClient + ASGITransport)， 兼容 httpx 0.28+ 版本。 """ from backend.app.api.deps import get_db_session def override_get_db(): try: yield db_session finally: pass app.dependency_overrides[get_db_session] = override_get_db with SyncTestClient(app, raise_server_exceptions=False) as c: yield c app.dependency_overrides.clear() # ============================================================ # 基础数据 Fixtures # ============================================================ @pytest.fixture def seed_roles(db_session): """预置系统角色。""" from backend.app.models.system import Role roles = [ Role(id=1, role_name="管理员", role_code="admin", status=1), Role(id=2, role_name="业务员", role_code="salesman", status=1), Role(id=3, role_name="管理层", role_code="manager", status=1), Role(id=4, role_name="司机", role_code="driver", status=1), Role(id=5, role_name="秘书", role_code="secretary", status=1), ] db_session.add_all(roles) db_session.flush() return roles @pytest.fixture def seed_users(db_session, seed_roles): """预置测试用户。""" from backend.app.models.system import User users = [ User(id=1, username="admin01", password_hash=hash_password("admin123"), real_name="管理员", mobile="13800000001", role_id=1, status=1), User(id=2, username="sales01", password_hash=hash_password("sales123"), real_name="业务员A", mobile="13800000002", role_id=2, status=1), User(id=3, username="sales02", password_hash=hash_password("sales123"), real_name="业务员B", mobile="13800000003", role_id=2, status=1), User(id=4, username="manager01", password_hash=hash_password("manager123"), real_name="管理层", mobile="13800000004", role_id=3, status=1), User(id=5, username="driver01", password_hash=hash_password("driver123"), real_name="司机A", mobile="13800000005", role_id=4, status=1), User(id=6, username="driver02", password_hash=hash_password("driver123"), real_name="司机B", mobile="13800000006", role_id=4, status=1), User(id=7, username="secretary01", password_hash=hash_password("secretary123"), real_name="秘书", mobile="13800000007", role_id=5, status=1), ] db_session.add_all(users) db_session.flush() return users @pytest.fixture def seed_menus(db_session): """预置菜单权限。""" from backend.app.models.system import Menu, RoleMenu menus = [ # === 订单相关 === Menu(id=1, menu_name="订单管理", menu_path="/orders", menu_type="page", permission_code="order:list", sort_no=1, status=1), Menu(id=2, menu_name="创建订单", menu_path="/orders/create", menu_type="button", permission_code="order:create", parent_id=1, sort_no=1, status=1), Menu(id=3, menu_name="审批订单", menu_path="/orders/approve", menu_type="button", permission_code="order:approve", parent_id=1, sort_no=2, status=1), Menu(id=12, menu_name="提交订单", menu_path="", menu_type="button", permission_code="order:submit", parent_id=1, sort_no=3, status=1), Menu(id=13, menu_name="取消订单", menu_path="", menu_type="button", permission_code="order:cancel", parent_id=1, sort_no=4, status=1), Menu(id=14, menu_name="取消审批", menu_path="", menu_type="button", permission_code="order:cancel-approve", parent_id=1, sort_no=5, status=1), Menu(id=15, menu_name="供应商文本", menu_path="", menu_type="button", permission_code="order:supplier-text", parent_id=1, sort_no=6, status=1), Menu(id=16, menu_name="更新订单", menu_path="", menu_type="button", permission_code="order:update", parent_id=1, sort_no=7, status=1), # === 客户相关 === Menu(id=4, menu_name="客户管理", menu_path="/customers", menu_type="page", permission_code="customer:list", sort_no=2, status=1), Menu(id=22, menu_name="创建客户", menu_path="", menu_type="button", permission_code="customer:create", parent_id=4, sort_no=1, status=1), Menu(id=23, menu_name="更新客户", menu_path="", menu_type="button", permission_code="customer:update", parent_id=4, sort_no=2, status=1), # === 产品相关 === Menu(id=5, menu_name="产品管理", menu_path="/products", menu_type="page", permission_code="product:list", sort_no=3, status=1), # === 报表相关 === Menu(id=7, menu_name="报表统计", menu_path="/reports", menu_type="page", permission_code="report:performance:view", sort_no=5, status=1), Menu(id=24, menu_name="导出报表", menu_path="", menu_type="button", permission_code="report:performance:export", parent_id=7, sort_no=1, status=1), # === 司机任务相关 === Menu(id=8, menu_name="司机任务", menu_path="/driver/tasks", menu_type="page", permission_code="driver:task:list", sort_no=6, status=1), Menu(id=9, menu_name="接单", menu_path="", menu_type="button", permission_code="driver:task:accept", parent_id=8, sort_no=1, status=1), Menu(id=10, menu_name="揽货", menu_path="", menu_type="button", permission_code="driver:task:pickup", parent_id=8, sort_no=2, status=1), Menu(id=11, menu_name="送达", menu_path="", menu_type="button", permission_code="driver:task:deliver", parent_id=8, sort_no=3, status=1), # === 物流管理相关 === Menu(id=25, menu_name="物流任务查看", menu_path="", menu_type="button", permission_code="logistics:task:list", parent_id=8, sort_no=4, status=1), Menu(id=26, menu_name="创建物流任务", menu_path="", menu_type="button", permission_code="logistics:task:create", parent_id=8, sort_no=5, status=1), Menu(id=27, menu_name="取消物流任务", menu_path="", menu_type="button", permission_code="logistics:task:cancel", parent_id=8, sort_no=6, status=1), Menu(id=28, menu_name="物流轨迹查看", menu_path="", menu_type="button", permission_code="logistics:trace:list", parent_id=8, sort_no=7, status=1), Menu(id=29, menu_name="新增物流轨迹", menu_path="", menu_type="button", permission_code="logistics:trace:create", parent_id=8, sort_no=8, status=1), # === 系统管理相关 === Menu(id=6, menu_name="系统管理", menu_path="/system", menu_type="page", permission_code="system:view", sort_no=10, status=1), Menu(id=30, menu_name="用户列表", menu_path="", menu_type="button", permission_code="system:user:list", parent_id=6, sort_no=1, status=1), Menu(id=31, menu_name="创建用户", menu_path="", menu_type="button", permission_code="system:user:create", parent_id=6, sort_no=2, status=1), Menu(id=32, menu_name="更新用户", menu_path="", menu_type="button", permission_code="system:user:update", parent_id=6, sort_no=3, status=1), Menu(id=33, menu_name="重置密码", menu_path="", menu_type="button", permission_code="system:user:reset-password", parent_id=6, sort_no=4, status=1), Menu(id=34, menu_name="角色列表", menu_path="", menu_type="button", permission_code="system:role:list", parent_id=6, sort_no=5, status=1), Menu(id=35, menu_name="创建角色", menu_path="", menu_type="button", permission_code="system:role:create", parent_id=6, sort_no=6, status=1), Menu(id=36, menu_name="更新角色", menu_path="", menu_type="button", permission_code="system:role:update", parent_id=6, sort_no=7, status=1), Menu(id=37, menu_name="分配菜单", menu_path="", menu_type="button", permission_code="system:role:assign-menus", parent_id=6, sort_no=8, status=1), Menu(id=38, menu_name="菜单列表", menu_path="", menu_type="button", permission_code="system:menu:list", parent_id=6, sort_no=9, status=1), Menu(id=39, menu_name="创建菜单", menu_path="", menu_type="button", permission_code="system:menu:create", parent_id=6, sort_no=10, status=1), Menu(id=40, menu_name="更新菜单", menu_path="", menu_type="button", permission_code="system:menu:update", parent_id=6, sort_no=11, status=1), # === 主数据相关 === Menu(id=41, menu_name="主数据列表", menu_path="", menu_type="button", permission_code="master-data:list", parent_id=5, sort_no=1, status=1), Menu(id=42, menu_name="主数据更新", menu_path="", menu_type="button", permission_code="master-data:update", parent_id=5, sort_no=2, status=1), # === AI 相关 === Menu(id=43, menu_name="AI识别", menu_path="", menu_type="button", permission_code="ai:recognize", sort_no=8, status=1), Menu(id=44, menu_name="AI修正", menu_path="", menu_type="button", permission_code="ai:correct", parent_id=43, sort_no=1, status=1), Menu(id=45, menu_name="AI解析订单", menu_path="", menu_type="button", permission_code="ai:parse-order", parent_id=43, sort_no=2, status=1), ] db_session.add_all(menus) db_session.flush() # 角色-菜单关联 role_menus = [ # 管理员：全部权限 *[RoleMenu(role_id=1, menu_id=m.id) for m in menus], # 业务员：订单 CRUD、客户、AI RoleMenu(role_id=2, menu_id=1), # order:list RoleMenu(role_id=2, menu_id=2), # order:create RoleMenu(role_id=2, menu_id=12), # order:submit RoleMenu(role_id=2, menu_id=13), # order:cancel RoleMenu(role_id=2, menu_id=16), # order:update RoleMenu(role_id=2, menu_id=4), # customer:list RoleMenu(role_id=2, menu_id=22), # customer:create RoleMenu(role_id=2, menu_id=23), # customer:update RoleMenu(role_id=2, menu_id=45), # ai:parse-order RoleMenu(role_id=2, menu_id=28), # logistics:trace:list # 管理层：订单、审批、物流、报表、主数据、AI RoleMenu(role_id=3, menu_id=1), # order:list RoleMenu(role_id=3, menu_id=3), # order:approve RoleMenu(role_id=3, menu_id=14), # order:cancel-approve RoleMenu(role_id=3, menu_id=15), # order:supplier-text RoleMenu(role_id=3, menu_id=7), # report:performance:view RoleMenu(role_id=3, menu_id=24), # report:performance:export RoleMenu(role_id=3, menu_id=25), # logistics:task:list RoleMenu(role_id=3, menu_id=26), # logistics:task:create RoleMenu(role_id=3, menu_id=27), # logistics:task:cancel RoleMenu(role_id=3, menu_id=28), # logistics:trace:list RoleMenu(role_id=3, menu_id=29), # logistics:trace:create RoleMenu(role_id=3, menu_id=41), # master-data:list RoleMenu(role_id=3, menu_id=42), # master-data:update RoleMenu(role_id=3, menu_id=43), # ai:recognize RoleMenu(role_id=3, menu_id=44), # ai:correct # 司机：司机任务、物流轨迹 RoleMenu(role_id=4, menu_id=8), # driver:task:list RoleMenu(role_id=4, menu_id=9), # driver:task:accept RoleMenu(role_id=4, menu_id=10), # driver:task:pickup RoleMenu(role_id=4, menu_id=11), # driver:task:deliver RoleMenu(role_id=4, menu_id=28), # logistics:trace:list RoleMenu(role_id=4, menu_id=29), # logistics:trace:create # 秘书：AI 相关 RoleMenu(role_id=5, menu_id=43), # ai:recognize RoleMenu(role_id=5, menu_id=44), # ai:correct RoleMenu(role_id=5, menu_id=45), # ai:parse-order ] db_session.add_all(role_menus) db_session.flush() return menus @pytest.fixture def seed_data(db_session, seed_users, seed_menus): """预置完整基础数据，返回数据库会话。""" from backend.app.models.system import SystemConfig # 预置系统配置 configs = [ SystemConfig(config_key="logistics_timeout_days", config_value="2", config_name="物流超时天数", status=1), SystemConfig(config_key="inactive_customer_days", config_value="30", config_name="沉默客户天数", status=1), SystemConfig(config_key="arrears_generate_mode", config_value="delivered", config_name="欠款生成模式", status=1), SystemConfig(config_key="arrears_reminder_start_type", config_value="order_date", config_name="欠款起算口径", status=1), SystemConfig(config_key="inactive_order_amount_threshold", config_value="1000", config_name="沉默客户金额阈值", status=1), ] db_session.add_all(configs) db_session.flush() return db_session # ============================================================ # 认证头 Fixtures # ============================================================ @pytest.fixture(autouse=True) def _reset_test_state(): """每个测试前重置全局状态，防止跨测试污染。""" from backend.app.services.auth_service import auth_service auth_service.revoked_tokens.clear() # 清除 Redis 缓存，防止缓存导致的测试间数据泄漏 try: from backend.app.core.cache import get_redis r = get_redis() if r is not None: r.flushdb() except Exception: pass yield auth_service.revoked_tokens.clear() @pytest.fixture def admin_headers(seed_data) -> dict: """管理员认证头（每次生成唯一 token）。""" token = create_access_token({"user_id": 1, "role_code": "admin", "username": "admin01", "jti": uuid.uuid4().hex}) return {"Authorization": f"Bearer {token}"} @pytest.fixture def salesman_headers(seed_data) -> dict: """业务员认证头（每次生成唯一 token）。""" token = create_access_token({"user_id": 2, "role_code": "salesman", "username": "sales01", "jti": uuid.uuid4().hex}) return {"Authorization": f"Bearer {token}"} @pytest.fixture def salesman2_headers(seed_data) -> dict: """业务员B认证头（每次生成唯一 token）。""" token = create_access_token({"user_id": 3, "role_code": "salesman", "username": "sales02", "jti": uuid.uuid4().hex}) return {"Authorization": f"Bearer {token}"} @pytest.fixture def manager_headers(seed_data) -> dict: """管理层认证头（每次生成唯一 token）。""" token = create_access_token({"user_id": 4, "role_code": "manager", "username": "manager01", "jti": uuid.uuid4().hex}) return {"Authorization": f"Bearer {token}"} @pytest.fixture def driver_headers(seed_data) -> dict: """司机认证头（每次生成唯一 token）。""" token = create_access_token({"user_id": 5, "role_code": "driver", "username": "driver01", "jti": uuid.uuid4().hex}) return {"Authorization": f"Bearer {token}"} @pytest.fixture def driver2_headers(seed_data) -> dict: """司机B认证头（每次生成唯一 token）。""" token = create_access_token({"user_id": 6, "role_code": "driver", "username": "driver02", "jti": uuid.uuid4().hex}) return {"Authorization": f"Bearer {token}"} @pytest.fixture def secretary_headers(seed_data) -> dict: """秘书认证头（每次生成唯一 token）。""" token = create_access_token({"user_id": 7, "role_code": "secretary", "username": "secretary01", "jti": uuid.uuid4().hex}) return {"Authorization": f"Bearer {token}"} # ============================================================ # 数据工厂 Fixtures # ============================================================ @pytest.fixture def make_customer(db_session, seed_data): """客户工厂函数。""" from backend.app.models.business import Customer _counter = [0] def _make(**kwargs): _counter[0] += 1 defaults = { "customer_name": f"测试客户{_counter[0]}", "mobile": f"1380000{_counter[0]:04d}", "settlement_type": "immediate", "settlement_days": 0, "salesman_id": 2, "credit_limit": 0, "deleted": 0, } defaults.update(kwargs) customer = Customer(**defaults) db_session.add(customer) db_session.flush() return customer return _make @pytest.fixture def make_product(db_session, seed_data): """产品工厂函数。""" from backend.app.models.business import Product _counter = [0] def _make(**kwargs): _counter[0] += 1 defaults = { "product_name": f"测试产品{_counter[0]}", "specification": f"{100 + _counter[0]}x{100 + _counter[0]}cm", "unit": "张", "category": "工业品", "cost_price": 60.0, "sale_price": 100.0, "status": 1, } defaults.update(kwargs) product = Product(**defaults) db_session.add(product) db_session.flush() return product return _make @pytest.fixture def make_supplier(db_session, seed_data): """供应商工厂函数。""" from backend.app.models.business import Supplier _counter = [0] def _make(**kwargs): _counter[0] += 1 defaults = { "supplier_name": f"测试工厂{_counter[0]}", "supplier_type": "factory", "contact_name": f"联系人{_counter[0]}", "contact_mobile": f"1390000{_counter[0]:04d}", "status": 1, } defaults.update(kwargs) supplier = Supplier(**defaults) db_session.add(supplier) db_session.flush() return supplier return _make @pytest.fixture def make_order(db_session, seed_data, make_customer, make_product, make_supplier): """订单工厂函数，支持指定订单状态。""" from backend.app.models.business import SalesOrder, SalesOrderItem _counter = [0] def _make(order_status="draft", salesman_id=2, **kwargs): _counter[0] += 1 customer = make_customer() product = make_product() supplier = make_supplier() order = SalesOrder( order_no=f"TEST-ORD-{_counter[0]:06d}", customer_id=customer.id, customer_name=customer.customer_name, customer_mobile=customer.mobile, salesman_id=salesman_id, order_status=order_status, factory_id=supplier.id, sale_price_total=1000.0, cost_price_total=600.0, profit_total=400.0, profit_rate=40.0, deleted=0, ) for k, v in kwargs.items(): if hasattr(order, k): setattr(order, k, v) db_session.add(order) db_session.flush() item = SalesOrderItem( order_id=order.id, product_id=product.id, product_name=product.product_name, specification=product.specification, unit=product.unit, quantity=10, sale_price=100.0, cost_price=60.0, deleted=0, ) db_session.add(item) db_session.flush() # 附加关联对象到 order 上，方便测试使用 order._test_customer = customer order._test_product = product order._test_supplier = supplier return order return _make @pytest.fixture def make_task(db_session, seed_data, make_order): """司机任务工厂函数。""" from backend.app.models.business import LogisticsTask _counter = [0] def _make(status="pending", driver_id=5, **kwargs): _counter[0] += 1 order = make_order(order_status="approved") task = LogisticsTask( task_no=f"TASK-{_counter[0]:06d}", order_id=order.id, driver_id=driver_id, factory_id=order.factory_id, pickup_address="泰兴工厂", delivery_address="上海仓库", pickup_content="测试货物", quantity=10, status=status, created_by=1, ) for k, v in kwargs.items(): if hasattr(task, k): setattr(task, k, v) db_session.add(task) db_session.flush() task._test_order = order return task return _make