Files

T

2026-09-22 13:39:18 +08:00

105 KiB

Raw Blame History

测试数据批次隔离与级联清理实施计划

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 为六类基础资料、五类托盘业务、六类报表和首页增加正式/测试批次隔离，并允许超级管理员在完整预览后原子纳入或整批清理测试数据。

Architecture: 以可空 test_batch_id 作为 16 张业务表的统一租界键，请求通过 X-Test-Batch-Id 解析成受信任的 TestBatchScope，控制器显式传入服务，DAO 对所有读路径追加同一作用域条件。批次管理、旧数据关联闭包和清理事务集中在独立 controller/service/DAO 中；预览令牌的哈希、闭包摘要和失效时间持久化到批次行，执行时锁行、重算并一次提交。

Tech Stack: FastAPI、SQLAlchemy AsyncSession、Pydantic v2、MySQL、PostgreSQL、Python unittest、Vue 3、Pinia、Element Plus、uni-app、Node.js test runner

Spec: docs/superpowers/specs/2026-09-22-test-data-batch-cleanup-design.md

Global Constraints

test_batch_id IS NULL 永远表示正式数据；正式数据不允许通过本功能或旧 DELETE API 物理删除。
只有 current_user.user.admin is True 才能管理批次、创建/编辑测试主数据、纳入旧数据和清理；菜单权限不能替代后端硬校验。
已授权普通用户只能进入 ACTIVE 批次执行入库、装填、发货、使用和退回，现有仓库/现场数据权限继续叠加生效。
请求体不接收可信的 test_batch_id；业务写入只能从服务端解析出的 TestBatchScope 继承。
正式、测试及不同测试批次之间禁止关联；通用编辑接口不得修改既有批次标记。
旧数据纳入和批次清理均使用一次性预览令牌，令牌原文只返回一次，数据库只保存 SHA-256 摘要，默认有效期 600 秒。
清理必须在一个数据库事务内完成，不配置 ON DELETE CASCADE，任何错误都回滚并保留 CLOSED 状态。
TEST_BATCH_ONLINE_PURGE_MAX_ROWS 默认 0，表示在线清理关闭；目标环境完成备份恢复演练与压测并配置正整数后才允许清理非空批次。
同步维护 MySQL/PostgreSQL 全量 SQL 与各自增量补丁；本计划不执行任何数据库迁移、生产写入或真实清理。
保留现有全局编码、单号和二维码唯一约束；测试数据仍使用全局唯一值。
测试与格式检查统一带 PYTHONDONTWRITEBYTECODE=1，避免继续产生或改动仓库内缓存文件。
当前工作区已有用户改动；每次提交只使用计划中列出的 git commit --only 文件清单，不得清理、取消暂存或覆盖其他文件。

Task 1: 数据模型、VO 与双数据库迁移契约

Files:

Create: shierfeng-fastapi-backend/module_business/entity/do/test_batch_do.py
Create: shierfeng-fastapi-backend/module_business/entity/vo/test_batch_vo.py
Create: shierfeng-fastapi-backend/sql/patches/20260922_test_data_batch_mysql.sql
Create: shierfeng-fastapi-backend/sql/patches/20260922_test_data_batch_pg.sql
Modify: shierfeng-fastapi-backend/module_business/entity/do/master_do.py
Modify: shierfeng-fastapi-backend/module_business/entity/do/operation_do.py
Modify: shierfeng-fastapi-backend/module_business/domain/enums.py
Modify: shierfeng-fastapi-backend/sql/shierfeng-fastapi.sql
Modify: shierfeng-fastapi-backend/sql/shierfeng-fastapi-pg.sql
Modify: shierfeng-fastapi-backend/tests/module_business/test_model_contract.py
Modify: shierfeng-fastapi-backend/tests/module_business/test_vo_contract.py
Modify: shierfeng-fastapi-backend/tests/module_business/test_sql_contract.py

Interfaces:

Consumes: 现有 Base、DataBaseConfig.db_type、16 张业务表及 sys_user.user_id。
Produces: TestBatchStatus、TestBatchPreviewAction、BizTestBatch、BizTestBatchUser，以及所有批次业务模型上的 test_batch_id: int | None。
Produces: TestBatchCreateModel、TestBatchEditModel、TestBatchUserAssignModel、TestBatchModel、TestBatchOptionModel、TestBatchDetailModel、BatchImpactModel、TestBatchPreviewModel、PurgePreviewModel、AdoptionRootModel、AdoptionPreviewRequestModel、AdoptionExecuteRequestModel、PurgeExecuteRequestModel。
Step 1: 先写模型、VO 和 SQL 失败契约

在三个现有契约测试中加入以下断言，完整列出 18 张业务表和 16 个 test_batch_id 字段，不用模糊包含判断：

EXPECTED_TEST_BATCH_TABLES = {'biz_test_batch', 'biz_test_batch_user'}
TEST_BATCH_OWNED_TABLES = {
    'biz_strain', 'biz_customer', 'biz_customer_site', 'biz_production_line',
    'biz_warehouse', 'biz_pallet', 'biz_inbound_order', 'biz_inbound_item',
    'biz_filling_record', 'biz_shipment_order', 'biz_shipment_item',
    'biz_usage_record', 'biz_return_order', 'biz_return_item',
    'biz_pallet_inventory', 'biz_pallet_event',
}

def test_all_owned_models_have_nullable_indexed_batch_fk(self) -> None:
    for table_name in TEST_BATCH_OWNED_TABLES:
        table = Base.metadata.tables[table_name]
        column = table.c.test_batch_id
        self.assertTrue(column.nullable, table_name)
        self.assertTrue(column.index, table_name)
        fk = next(iter(column.foreign_keys))
        self.assertEqual('biz_test_batch.batch_id', fk.target_fullname)
        self.assertIsNone(fk.ondelete)

def test_test_batch_status_rejects_unknown_value(self) -> None:
    with self.assertRaises(ValidationError) as context:
        TestBatchModel(
            batchId=1,
            batchCode='TB-1',
            batchName='验收批次',
            status='UNKNOWN',
            createBy='admin',
            createTime=datetime(2026, 9, 22),
            updateBy='admin',
            updateTime=datetime(2026, 9, 22),
        )
    self.assertIn(('status',), {tuple(error['loc']) for error in context.exception.errors()})

test_sql_contract.py 还要逐个验证两个全量脚本和两个补丁均创建批次表、为 16 张表新增字段/索引/外键、未出现 ON DELETE CASCADE，且补丁没有把既有行更新为非空批次。

Step 2: 运行契约并确认因模型/DDL 缺失而失败

Run:

cd shierfeng-fastapi-backend
PYTHONDONTWRITEBYTECODE=1 conda run -n ShiErFeng python -m unittest \
  tests.module_business.test_model_contract \
  tests.module_business.test_vo_contract \
  tests.module_business.test_sql_contract -v

Expected: FAIL，明确报告 biz_test_batch、TestBatchModel 或 test_batch_id 不存在。

Step 3: 增加枚举、模型和请求/响应 VO

domain/enums.py 增加：

class TestBatchStatus(str, Enum):
    ACTIVE = 'ACTIVE'
    CLOSED = 'CLOSED'
    PURGED = 'PURGED'


class TestBatchPreviewAction(str, Enum):
    ADOPT = 'ADOPT'
    PURGE = 'PURGE'

test_batch_do.py 使用以下字段和约束；purge_summary 存规范 JSON 文本以保持 MySQL/PostgreSQL 一致：

class BizTestBatch(Base):
    __tablename__ = 'biz_test_batch'
    __table_args__ = (
        CheckConstraint("status IN ('ACTIVE', 'CLOSED', 'PURGED')", name='ck_biz_test_batch_status'),
        CheckConstraint("preview_action IS NULL OR preview_action IN ('ADOPT', 'PURGE')", name='ck_biz_test_batch_preview_action'),
        {'comment': '测试数据批次'},
    )

    batch_id = Column(BigInteger, primary_key=True, autoincrement=True)
    batch_code = Column(String(64), nullable=False, unique=True)
    batch_name = Column(String(100), nullable=False)
    status = Column(String(16), nullable=False, server_default='ACTIVE')
    create_by = Column(String(64), nullable=False, server_default='')
    create_time = Column(DateTime, nullable=False, default=datetime.now)
    update_by = Column(String(64), nullable=False, server_default='')
    update_time = Column(DateTime, nullable=False, default=datetime.now)
    closed_by = Column(BigInteger, ForeignKey('sys_user.user_id'), nullable=True)
    closed_at = Column(DateTime, nullable=True)
    purged_by = Column(BigInteger, ForeignKey('sys_user.user_id'), nullable=True)
    purged_at = Column(DateTime, nullable=True)
    purge_summary = Column(Text, nullable=True)
    preview_action = Column(String(16), nullable=True)
    preview_token_hash = Column(CHAR(64), nullable=True)
    preview_scope_hash = Column(CHAR(64), nullable=True)
    preview_expires_at = Column(DateTime, nullable=True)
    remark = Column(String(500), nullable=True)


class BizTestBatchUser(Base):
    __tablename__ = 'biz_test_batch_user'
    batch_id = Column(BigInteger, ForeignKey('biz_test_batch.batch_id'), primary_key=True)
    user_id = Column(BigInteger, ForeignKey('sys_user.user_id'), primary_key=True)
    create_by = Column(String(64), nullable=False, server_default='')
    create_time = Column(DateTime, nullable=False, default=datetime.now)

16 张业务表统一增加：

test_batch_id = Column(
    BigInteger,
    ForeignKey('biz_test_batch.batch_id'),
    nullable=True,
    index=True,
    comment='测试数据批次ID；NULL表示正式数据',
)

VO 统一继承一个严格基类，同时接受 Python snake_case 和 API camelCase，并固定以下关键输入：

PositiveId = Annotated[int, Field(gt=0)]


class TestBatchApiModel(BaseModel):
    model_config = ConfigDict(
        alias_generator=to_camel,
        from_attributes=True,
        populate_by_name=True,
        extra='forbid',
    )


class TestBatchCreateModel(TestBatchApiModel):
    batch_name: str = Field(min_length=1, max_length=100)
    remark: str | None = Field(default=None, max_length=500)

class TestBatchEditModel(TestBatchApiModel):
    remark: str | None = Field(default=None, max_length=500)

class TestBatchUserAssignModel(TestBatchApiModel):
    user_ids: list[PositiveId]

class TestBatchModel(TestBatchApiModel):
    batch_id: PositiveId
    batch_code: str
    batch_name: str
    status: TestBatchStatus
    remark: str | None = None
    create_by: str
    create_time: datetime
    update_by: str
    update_time: datetime
    closed_by: int | None = None
    closed_at: datetime | None = None
    purged_by: int | None = None
    purged_at: datetime | None = None
    purge_summary: dict[str, int] | None = None

class TestBatchOptionModel(TestBatchApiModel):
    batch_id: PositiveId
    batch_code: str
    batch_name: str
    status: TestBatchStatus

class BatchImpactModel(TestBatchApiModel):
    table_name: str
    label: str
    count: int = Field(ge=0)
    keys: list[str]

class TestBatchDetailModel(TestBatchModel):
    user_ids: list[int]
    data_summary: list[BatchImpactModel]

class TestBatchPreviewModel(TestBatchApiModel):
    batch_id: PositiveId
    batch_name: str
    preview_token: str
    expires_at: datetime
    total_count: int = Field(ge=0)
    impacts: list[BatchImpactModel]

class PurgePreviewModel(TestBatchPreviewModel):
    online_purge_allowed: bool
    online_purge_limit: int = Field(ge=0)
    blocking_reason: str | None = None

AdoptionRootType = Literal[
    'strain', 'customer', 'site', 'line', 'warehouse', 'pallet',
    'inbound', 'filling', 'shipment', 'usage', 'return',
]

class AdoptionRootModel(TestBatchApiModel):
    resource_type: AdoptionRootType
    resource_ids: list[PositiveId] = Field(min_length=1)

class AdoptionPreviewRequestModel(TestBatchApiModel):
    roots: list[AdoptionRootModel] = Field(min_length=1)

class AdoptionExecuteRequestModel(AdoptionPreviewRequestModel):
    batch_name: str = Field(min_length=1, max_length=100)
    preview_token: str = Field(min_length=32, max_length=128)

class PurgeExecuteRequestModel(TestBatchApiModel):
    batch_name: str = Field(min_length=1, max_length=100)
    preview_token: str = Field(min_length=32, max_length=128)

TestBatchEditModel 故意不包含 batch_name 和 status：批次名称创建后不可修改，状态只能经关闭/重新开启/清理专用命令变化。

Step 4: 更新 MySQL/PostgreSQL 全量 SQL 与增量补丁

四份 SQL 使用同一结构语义：先建 biz_test_batch，再建 biz_test_batch_user，最后给 16 张表增加可空字段、普通索引和无级联动作的外键。MySQL 补丁的字段形态固定为：

ALTER TABLE biz_strain
  ADD COLUMN test_batch_id BIGINT NULL COMMENT '测试数据批次ID；NULL表示正式数据',
  ADD INDEX idx_biz_strain_test_batch_id (test_batch_id),
  ADD CONSTRAINT fk_biz_strain_test_batch_id
    FOREIGN KEY (test_batch_id) REFERENCES biz_test_batch (batch_id);

PostgreSQL 补丁的字段形态固定为：

ALTER TABLE biz_strain ADD COLUMN test_batch_id BIGINT NULL;
COMMENT ON COLUMN biz_strain.test_batch_id IS '测试数据批次ID；NULL表示正式数据';
CREATE INDEX idx_biz_strain_test_batch_id ON biz_strain (test_batch_id);
ALTER TABLE biz_strain ADD CONSTRAINT fk_biz_strain_test_batch_id
  FOREIGN KEY (test_batch_id) REFERENCES biz_test_batch (batch_id);

对其余 15 张表逐表展开相同 DDL，不使用动态 SQL，不加入任何修改既有行 test_batch_id 的 UPDATE 语句。

Step 5: 运行模型、VO 和 SQL 契约

Run: Step 2 的命令。

Expected: PASS；两个方言均能编译 18 张业务表，所有批次外键的 ondelete 为 None。

Step 6: 提交本任务

git add shierfeng-fastapi-backend/module_business/entity/do/test_batch_do.py \
  shierfeng-fastapi-backend/module_business/entity/vo/test_batch_vo.py \
  shierfeng-fastapi-backend/module_business/entity/do/master_do.py \
  shierfeng-fastapi-backend/module_business/entity/do/operation_do.py \
  shierfeng-fastapi-backend/module_business/domain/enums.py \
  shierfeng-fastapi-backend/sql/shierfeng-fastapi.sql \
  shierfeng-fastapi-backend/sql/shierfeng-fastapi-pg.sql \
  shierfeng-fastapi-backend/sql/patches/20260922_test_data_batch_mysql.sql \
  shierfeng-fastapi-backend/sql/patches/20260922_test_data_batch_pg.sql \
  shierfeng-fastapi-backend/tests/module_business/test_model_contract.py \
  shierfeng-fastapi-backend/tests/module_business/test_vo_contract.py \
  shierfeng-fastapi-backend/tests/module_business/test_sql_contract.py
git commit --only shierfeng-fastapi-backend/module_business/entity/do/test_batch_do.py \
  shierfeng-fastapi-backend/module_business/entity/vo/test_batch_vo.py \
  shierfeng-fastapi-backend/module_business/entity/do/master_do.py \
  shierfeng-fastapi-backend/module_business/entity/do/operation_do.py \
  shierfeng-fastapi-backend/module_business/domain/enums.py \
  shierfeng-fastapi-backend/sql/shierfeng-fastapi.sql \
  shierfeng-fastapi-backend/sql/shierfeng-fastapi-pg.sql \
  shierfeng-fastapi-backend/sql/patches/20260922_test_data_batch_mysql.sql \
  shierfeng-fastapi-backend/sql/patches/20260922_test_data_batch_pg.sql \
  shierfeng-fastapi-backend/tests/module_business/test_model_contract.py \
  shierfeng-fastapi-backend/tests/module_business/test_vo_contract.py \
  shierfeng-fastapi-backend/tests/module_business/test_sql_contract.py \
  -m "数据：增加测试批次模型与迁移契约"

Task 2: 受信任批次上下文与超级管理员硬门禁

Files:

Create: shierfeng-fastapi-backend/module_business/domain/test_batch_scope.py
Create: shierfeng-fastapi-backend/module_business/dependency/__init__.py
Create: shierfeng-fastapi-backend/module_business/dependency/test_batch_dependency.py
Create: shierfeng-fastapi-backend/module_business/dao/test_batch_dao.py
Create: shierfeng-fastapi-backend/module_business/service/test_batch_access_service.py
Create: shierfeng-fastapi-backend/tests/module_business/test_batch_fixtures.py
Create: shierfeng-fastapi-backend/tests/module_business/test_test_batch_access.py
Modify: shierfeng-fastapi-backend/common/context.py
Modify: shierfeng-fastapi-backend/middlewares/context_middleware.py

Interfaces:

Consumes: BizTestBatch、BizTestBatchUser、CurrentUserModel、请求头 X-Test-Batch-Id。
Produces: TestBatchScope(batch_id, mode, status, writable)、TestBatchScopeDependency(write: bool = False)、TestBatchAccessService.assert_super_admin(current_user: CurrentUserModel) -> None、lock_active_batch(db: AsyncSession, scope: TestBatchScope) -> BizTestBatch | None。
Error contract: 不合法头返回 ServiceWarning('测试批次标识必须为正整数')；未授权或批次已清理返回 PermissionException(data={'reason': 'TEST_BATCH_UNAVAILABLE'}, message='当前测试批次不可用')；写请求命中已关闭批次时返回同一 reason 与明确文案 当前测试批次已关闭。
Test fixtures: make_current_user(user_id=7, permissions=None)、make_admin()、make_test_scope(batch_id, status='ACTIVE', writable=True) 供后续测试统一使用。
Step 1: 写上下文解析与越权失败测试

class TestBatchAccessTest(unittest.IsolatedAsyncioTestCase):
    async def test_permission_character_cannot_replace_super_admin_flag(self) -> None:
        current_user = make_current_user(user_id=2, permissions=['system:testbatch:purge'])
        with self.assertRaisesRegex(PermissionException, '仅超级管理员'):
            TestBatchAccessService.assert_super_admin(current_user)

    async def test_authorized_active_batch_resolves_writable_scope(self) -> None:
        TestBatchDao.get_batch = AsyncMock(return_value=SimpleNamespace(batch_id=7, status='ACTIVE'))
        TestBatchDao.user_is_authorized = AsyncMock(return_value=True)
        scope = await TestBatchAccessService.resolve_scope(self.db, make_current_user(8), 7, write=True)
        self.assertEqual(TestBatchScope(batch_id=7, mode='TEST', status=TestBatchStatus.ACTIVE, writable=True), scope)

    async def test_closed_batch_rejects_business_write(self) -> None:
        TestBatchDao.get_batch = AsyncMock(return_value=SimpleNamespace(batch_id=7, status='CLOSED'))
        with self.assertRaisesRegex(PermissionException, '已关闭'):
            await TestBatchAccessService.resolve_scope(self.db, make_admin(), 7, write=True)

另写中间件异常路径测试：call_next 抛异常后，RequestContext.get_current_test_batch_scope() 必须回到 None。

test_batch_fixtures.py 提供统一构造器：

def make_current_user(
    user_id: int = 7,
    permissions: list[str] | None = None,
    *,
    admin: bool = False,
) -> CurrentUserModel:
    return CurrentUserModel(
        permissions=permissions or [],
        roles=[],
        user=UserInfoModel(
            userId=user_id,
            userName='admin' if user_id == 1 else f'user-{user_id}',
            admin=admin,
        ),
    )

def make_admin() -> CurrentUserModel:
    return make_current_user(1, ['*:*:*'], admin=True)

def make_test_scope(batch_id: int, status: str = 'ACTIVE', writable: bool = True) -> TestBatchScope:
    return TestBatchScope(batch_id=batch_id, mode='TEST', status=TestBatchStatus(status), writable=writable)

Step 2: 运行新测试并确认失败

Run:

cd shierfeng-fastapi-backend
PYTHONDONTWRITEBYTECODE=1 conda run -n ShiErFeng python -m unittest \
  tests.module_business.test_test_batch_access -v

Expected: FAIL，报告新模块或 TestBatchScope 不存在。

Step 3: 实现作用域值对象、DAO 和访问服务

@dataclass(frozen=True, slots=True)
class TestBatchScope:
    batch_id: int | None
    mode: Literal['FORMAL', 'TEST']
    status: TestBatchStatus | None
    writable: bool

    @classmethod
    def formal(cls) -> 'TestBatchScope':
        return cls(batch_id=None, mode='FORMAL', status=None, writable=True)


def batch_scope_condition(column: InstrumentedAttribute, scope: TestBatchScope) -> ColumnElement[bool]:
    return column.is_(None) if scope.batch_id is None else column == scope.batch_id


def assert_record_in_scope(record: Any, scope: TestBatchScope, label: str) -> None:
    if record is None or record.test_batch_id != scope.batch_id:
        raise ServiceException(message=f'{label}不属于当前数据模式或测试批次')

TestBatchDao 实现并保持无 commit()：

async def get_batch(db: AsyncSession, batch_id: int) -> BizTestBatch | None:
    result = await db.execute(select(BizTestBatch).where(BizTestBatch.batch_id == batch_id))
    return result.scalar_one_or_none()

async def get_batch_for_update(db: AsyncSession, batch_id: int) -> BizTestBatch | None:
    result = await db.execute(
        select(BizTestBatch)
        .where(BizTestBatch.batch_id == batch_id)
        .with_for_update()
        .execution_options(populate_existing=True)
    )
    return result.scalar_one_or_none()

async def user_is_authorized(db: AsyncSession, batch_id: int, user_id: int) -> bool:
    result = await db.execute(
        select(BizTestBatchUser.batch_id).where(
            BizTestBatchUser.batch_id == batch_id,
            BizTestBatchUser.user_id == user_id,
        )
    )
    return result.scalar_one_or_none() is not None

async def list_available_batches(db: AsyncSession, user_id: int, is_admin: bool) -> list[BizTestBatch]:
    statement = select(BizTestBatch).where(BizTestBatch.status == TestBatchStatus.ACTIVE.value)
    if not is_admin:
        statement = statement.join(BizTestBatchUser).where(BizTestBatchUser.user_id == user_id)
    result = await db.execute(statement.order_by(BizTestBatch.batch_id.desc()))
    return list(result.scalars().all())

resolve_scope 的正式模式直接返回 TestBatchScope.formal()；测试模式每次查询批次、校验授权和状态。只有管理页专用服务可以读取 CLOSED 汇总，普通业务依赖永远只接受 ACTIVE。

Step 4: 实现 FastAPI 依赖和请求级清理

class TestBatchScopeResolver:
    def __init__(self, *, write: bool = False) -> None:
        self.write = write

    async def __call__(
        self,
        request: Request,
        db: AsyncSession = Depends(get_db),
        current_user: CurrentUserModel = CurrentUserDependency(),
    ) -> TestBatchScope:
        raw = request.headers.get('X-Test-Batch-Id')
        batch_id = parse_optional_positive_batch_id(raw)
        scope = await TestBatchAccessService.resolve_scope(db, current_user, batch_id, write=self.write)
        RequestContext.set_current_test_batch_scope(scope)
        return scope


def TestBatchScopeDependency(*, write: bool = False) -> params.Depends:  # noqa: N802
    return Depends(TestBatchScopeResolver(write=write))

将 ContextCleanupMiddleware.dispatch 改成 try/finally，确保正常响应和异常都调用 RequestContext.clear_all()；clear_all() 同时清空测试批次 ContextVar。

Step 5: 运行访问测试及现有认证测试

Run:

cd shierfeng-fastapi-backend
PYTHONDONTWRITEBYTECODE=1 conda run -n ShiErFeng python -m unittest \
  tests.module_business.test_test_batch_access -v

Expected: PASS；普通用户即使持有 system:testbatch:purge 也不能通过硬门禁。

Step 6: 提交本任务

git add shierfeng-fastapi-backend/module_business/domain/test_batch_scope.py \
  shierfeng-fastapi-backend/module_business/dependency/__init__.py \
  shierfeng-fastapi-backend/module_business/dependency/test_batch_dependency.py \
  shierfeng-fastapi-backend/module_business/dao/test_batch_dao.py \
  shierfeng-fastapi-backend/module_business/service/test_batch_access_service.py \
  shierfeng-fastapi-backend/common/context.py \
  shierfeng-fastapi-backend/middlewares/context_middleware.py \
  shierfeng-fastapi-backend/tests/module_business/test_batch_fixtures.py \
  shierfeng-fastapi-backend/tests/module_business/test_test_batch_access.py
git commit --only shierfeng-fastapi-backend/module_business/domain/test_batch_scope.py \
  shierfeng-fastapi-backend/module_business/dependency/__init__.py \
  shierfeng-fastapi-backend/module_business/dependency/test_batch_dependency.py \
  shierfeng-fastapi-backend/module_business/dao/test_batch_dao.py \
  shierfeng-fastapi-backend/module_business/service/test_batch_access_service.py \
  shierfeng-fastapi-backend/common/context.py \
  shierfeng-fastapi-backend/middlewares/context_middleware.py \
  shierfeng-fastapi-backend/tests/module_business/test_batch_fixtures.py \
  shierfeng-fastapi-backend/tests/module_business/test_test_batch_access.py \
  -m "安全：增加测试批次上下文硬校验"

Task 3: 批次生命周期与用户授权服务

Files:

Create: shierfeng-fastapi-backend/module_business/service/test_batch_service.py
Create: shierfeng-fastapi-backend/tests/module_business/test_test_batch_service.py
Modify: shierfeng-fastapi-backend/module_business/dao/test_batch_dao.py
Modify: shierfeng-fastapi-backend/module_business/entity/vo/test_batch_vo.py

Interfaces:

Consumes: Task 1 的批次模型/VO、Task 2 的管理员硬门禁和批次 DAO。
Produces: TestBatchService.list_available_batches/list_batches/get_batch_detail/create_batch/edit_batch/replace_users/close_batch/reopen_batch。
Produces: 供 Task 11 映射为 /system/test-batch/* 路由的稳定 service/VO 契约；本任务不提前注册 controller，避免破坏当前精确路由契约。
Step 1: 写生命周期与 service 签名失败测试

async def test_only_active_batch_can_close(self) -> None:
    batch = SimpleNamespace(status='ACTIVE', closed_by=None, closed_at=None)
    TestBatchDao.get_batch_for_update = AsyncMock(return_value=batch)
    result = await TestBatchService.close_batch(self.db, 3, make_admin())
    self.assertEqual('CLOSED', batch.status)
    self.db.commit.assert_awaited_once()

async def test_purged_batch_cannot_reopen(self) -> None:
    TestBatchDao.get_batch_for_update = AsyncMock(return_value=SimpleNamespace(status='PURGED'))
    with self.assertRaisesRegex(ServiceWarning, '已清理'):
        await TestBatchService.reopen_batch(self.db, 3, make_admin())
    self.db.rollback.assert_awaited_once()

async def test_regular_user_options_only_include_own_active_batches(self) -> None:
    await TestBatchService.list_available_batches(self.db, make_current_user(8))
    TestBatchDao.list_available_batches.assert_awaited_once_with(self.db, 8, False)

Step 2: 运行生命周期测试并确认失败

Run:

cd shierfeng-fastapi-backend
PYTHONDONTWRITEBYTECODE=1 conda run -n ShiErFeng python -m unittest \
  tests.module_business.test_test_batch_service -v

Expected: FAIL，报告 TestBatchService 不存在。

Step 3: 实现事务边界和合法状态转换

创建批次编码使用时间戳和 secrets.token_hex(4)，数据库唯一约束冲突时回滚并返回明确错误：

@classmethod
async def create_batch(cls, db, command, current_user) -> CrudResponseModel:
    cls._assert_admin(current_user)
    batch_name = command.batch_name.strip()
    if not batch_name:
        raise ServiceWarning(message='测试批次名称不能为空')
    now = datetime.now()
    batch = BizTestBatch(
        batch_code=f'TB-{now:%Y%m%d%H%M%S}-{secrets.token_hex(4).upper()}',
        batch_name=batch_name,
        status=TestBatchStatus.ACTIVE.value,
        create_by=current_user.user.user_name,
        create_time=now,
        update_by=current_user.user.user_name,
        update_time=now,
        remark=command.remark,
    )
    try:
        await TestBatchDao.add_batch(db, batch)
        await db.commit()
        return CrudResponseModel(is_success=True, message='测试批次创建成功')
    except IntegrityError:
        await db.rollback()
        raise ServiceException(message='测试批次创建失败，批次编码冲突') from None
    except Exception:
        await db.rollback()
        raise

close_batch、reopen_batch 和 replace_users 都先 get_batch_for_update()；replace_users 只接受存在且未停用的 sys_user，去重后整表替换，并且只允许 ACTIVE 批次。关闭时写 closed_by/closed_at 并清空旧纳入预览，重新开启时清空关闭信息和所有预览字段。get_batch_detail 同时调用 TestBatchDao.count_owned_rows(db, batch_id)，返回 16 张表的分类数量，供 ACTIVE/CLOSED/PURGED 管理页只读查看；数据库中的 purge_summary 文本由 service 使用 json.loads 转换为 dict[str, int] 后再构造 VO。

Step 4: 固定供 Task 11 注册的路由契约并验证 service 签名

GET  /system/test-batch/options                    登录用户可用批次，无菜单权限
GET  /system/test-batch/list                       system:testbatch:list
GET  /system/test-batch/{batch_id}                 system:testbatch:query
POST /system/test-batch                            system:testbatch:add
PUT  /system/test-batch/{batch_id}                 system:testbatch:edit
PUT  /system/test-batch/{batch_id}/users           system:testbatch:assign
PUT  /system/test-batch/{batch_id}/close           system:testbatch:close
PUT  /system/test-batch/{batch_id}/reopen          system:testbatch:close

静态路径 options 和 list 必须在动态 /{batch_id} 之前注册。所有写路由使用 @Log，日志仅记录批次 ID、状态和响应摘要，不记录预览令牌或大批业务编号；分类数量保存在批次元数据中。

本任务在 test_test_batch_service.py 通过 inspect.signature 固定上述八个公开 service 方法及输入输出，不创建 controller；Task 11 一次注册完整 12 条路由并同步四层权限契约。

Step 5: 运行生命周期与 service 契约测试

Run:

cd shierfeng-fastapi-backend
PYTHONDONTWRITEBYTECODE=1 conda run -n ShiErFeng python -m unittest \
  tests.module_business.test_test_batch_service -v

Expected: PASS；现有全局精确路由契约仍保持绿色。

Step 6: 提交本任务

git add shierfeng-fastapi-backend/module_business/service/test_batch_service.py \
  shierfeng-fastapi-backend/module_business/dao/test_batch_dao.py \
  shierfeng-fastapi-backend/module_business/entity/vo/test_batch_vo.py \
  shierfeng-fastapi-backend/tests/module_business/test_test_batch_service.py
git commit --only shierfeng-fastapi-backend/module_business/service/test_batch_service.py \
  shierfeng-fastapi-backend/module_business/dao/test_batch_dao.py \
  shierfeng-fastapi-backend/module_business/entity/vo/test_batch_vo.py \
  shierfeng-fastapi-backend/tests/module_business/test_test_batch_service.py \
  -m "功能：增加测试批次生命周期管理"

Task 4: 六类主数据的隔离、继承与旧删除封闭

Files:

Modify: shierfeng-fastapi-backend/module_business/controller/strain_controller.py
Modify: shierfeng-fastapi-backend/module_business/controller/customer_controller.py
Modify: shierfeng-fastapi-backend/module_business/controller/site_controller.py
Modify: shierfeng-fastapi-backend/module_business/controller/line_controller.py
Modify: shierfeng-fastapi-backend/module_business/controller/warehouse_controller.py
Modify: shierfeng-fastapi-backend/module_business/controller/pallet_controller.py
Modify: shierfeng-fastapi-backend/module_business/dao/master_dao.py
Modify: shierfeng-fastapi-backend/module_business/entity/vo/master_vo.py
Modify: shierfeng-fastapi-backend/module_business/service/master_service.py
Modify: shierfeng-fastapi-backend/module_business/service/pallet_service.py
Modify: shierfeng-fastapi-backend/tests/module_business/test_master_service.py
Modify: shierfeng-fastapi-backend/tests/module_business/test_pallet_service.py
Modify: shierfeng-fastapi-backend/tests/module_business/test_vo_contract.py

Interfaces:

Consumes: TestBatchScopeDependency、batch_scope_condition、assert_record_in_scope、lock_active_batch。
Produces: 六类主数据的列表/详情/二维码/用户分配均只访问当前 scope；测试主数据写入仅允许超级管理员。
Compatibility: 六类 DELETE 路由路径保留，但 service 固定抛出 ServiceWarning('不支持单条物理删除，请通过测试数据批次清理')。
Step 1: 写正式/测试隔离和删除拒绝测试

async def test_list_passes_formal_scope_to_dao(self) -> None:
    scope = TestBatchScope.formal()
    await MasterService.list_strain(self.db, StrainPageQueryModel(), scope, is_page=True)
    MasterDao.list_strain.assert_awaited_once_with(self.db, ANY, scope, True)

async def test_test_master_create_inherits_scope_and_requires_admin(self) -> None:
    scope = make_test_scope(9)
    await MasterService.add_strain(self.db, StrainModel(strainCode='T-1', strainName='测试'), make_admin(), scope)
    self.assertEqual(9, MasterDao.add_strain.await_args.args[2])
    self.assertNotIn('test_batch_id', StrainModel.model_fields)

async def test_regular_user_cannot_create_test_master(self) -> None:
    with self.assertRaisesRegex(PermissionException, '仅超级管理员'):
        await MasterService.add_strain(self.db, make_strain(), make_current_user(8), make_test_scope(9))

async def test_all_legacy_master_deletes_are_rejected_before_dao(self) -> None:
    for resource, value in deletion_cases():
        with self.subTest(resource=resource), self.assertRaisesRegex(ServiceWarning, '测试数据批次清理'):
            await getattr(MasterService, f'remove_{resource}')(self.db, value)
    MasterDao._delete.assert_not_awaited()

补充父子跨批次测试：测试现场只能引用同批次客户，测试产线只能引用同批次现场；正式记录引用测试父项和反向混用均失败。

Step 2: 运行主数据服务测试并确认失败

Run:

cd shierfeng-fastapi-backend
PYTHONDONTWRITEBYTECODE=1 conda run -n ShiErFeng python -m unittest \
  tests.module_business.test_master_service \
  tests.module_business.test_pallet_service -v

Expected: FAIL，显示 service/DAO 尚未接收 TestBatchScope。

Step 3: 在 DAO 的所有读路径增加显式作用域参数

统一签名并在根模型上加条件：

async def list_strain(db, query_object, scope: TestBatchScope, is_page=True):
    conditions = [batch_scope_condition(BizStrain.test_batch_id, scope)]
    if query_object.strain_code:
        conditions.append(BizStrain.strain_code.like(f'%{query_object.strain_code}%'))
    if query_object.strain_name:
        conditions.append(BizStrain.strain_name.like(f'%{query_object.strain_name}%'))
    if query_object.status:
        conditions.append(BizStrain.status == query_object.status)
    statement = select(BizStrain).where(*conditions).order_by(BizStrain.strain_id.desc())
    return await MasterDao._paginate(db, statement, query_object, is_page)

async def get_strain(db, strain_id: int, scope: TestBatchScope, *, for_update=False):
    return await MasterDao._get(
        db, BizStrain, BizStrain.strain_id, strain_id,
        batch_scope_condition(BizStrain.test_batch_id, scope),
        for_update=for_update,
    )

async def _add(db, model, id_field, page_object, test_batch_id: int | None):
    payload = page_object.model_dump(exclude={id_field, 'can_remove', 'test_batch_id'})
    payload['test_batch_id'] = test_batch_id
    record = model(**payload)
    db.add(record)
    await db.flush()
    return record

对 strain/customer/site/line/warehouse/pallet 的 list/get、托盘二维码、仓库/现场用户分配入口逐个应用。find_*_by_code 保持全局不带 scope，以继续执行全局唯一性检查。

Step 4: 在服务和控制器中继承并重新校验 scope

所有 GET 注入 Annotated[TestBatchScope, TestBatchScopeDependency()]，所有 POST/PUT 注入 TestBatchScopeDependency(write=True)。测试 scope 的主数据写入先执行：

TestBatchAccessService.assert_super_admin(current_user)
await TestBatchAccessService.lock_active_batch(db, scope)
safe_object = page_object.model_copy(update={config['id_field']: None})
await getattr(MasterDao, f'add_{resource}')(db, safe_object, scope.batch_id)

编辑时从 scope 内查询当前记录，更新 payload 永远排除 test_batch_id。VO 不声明可写的 test_batch_id，DAO 只接受 service 单独传入的受信任参数。正式模式保持现有角色权限行为；测试模式的新增、编辑、仓库/现场用户分配均要求超级管理员。

Step 5: 将六类旧删除服务改为固定拒绝

@staticmethod
async def reject_physical_delete() -> NoReturn:
    raise ServiceWarning(message='不支持单条物理删除，请通过测试数据批次清理')

六类 controller 保留 URL 和 HTTP 方法，调用该拒绝逻辑，不再计算 can_remove。从主数据 VO、DAO 和页面响应中移除 can_remove 及 get_master_removability，避免继续暗示正式数据可删除。

Step 6: 运行主数据测试和静态检查

Run:

cd shierfeng-fastapi-backend
PYTHONDONTWRITEBYTECODE=1 conda run -n ShiErFeng python -m unittest \
  tests.module_business.test_master_service \
  tests.module_business.test_pallet_service \
  tests.module_business.test_vo_contract -v
PYTHONDONTWRITEBYTECODE=1 conda run -n ShiErFeng python -m ruff check \
  module_business/controller module_business/dao/master_dao.py \
  module_business/service/master_service.py module_business/service/pallet_service.py \
  tests/module_business/test_master_service.py tests/module_business/test_pallet_service.py

Expected: PASS；断言没有主数据 DELETE 路径能触发 DAO 删除。

Step 7: 提交本任务

git add shierfeng-fastapi-backend/module_business/controller/{strain,customer,site,line,warehouse,pallet}_controller.py \
  shierfeng-fastapi-backend/module_business/dao/master_dao.py \
  shierfeng-fastapi-backend/module_business/entity/vo/master_vo.py \
  shierfeng-fastapi-backend/module_business/service/master_service.py \
  shierfeng-fastapi-backend/module_business/service/pallet_service.py \
  shierfeng-fastapi-backend/tests/module_business/test_master_service.py \
  shierfeng-fastapi-backend/tests/module_business/test_pallet_service.py \
  shierfeng-fastapi-backend/tests/module_business/test_vo_contract.py
git commit --only shierfeng-fastapi-backend/module_business/controller/{strain,customer,site,line,warehouse,pallet}_controller.py \
  shierfeng-fastapi-backend/module_business/dao/master_dao.py \
  shierfeng-fastapi-backend/module_business/entity/vo/master_vo.py \
  shierfeng-fastapi-backend/module_business/service/master_service.py \
  shierfeng-fastapi-backend/module_business/service/pallet_service.py \
  shierfeng-fastapi-backend/tests/module_business/test_master_service.py \
  shierfeng-fastapi-backend/tests/module_business/test_pallet_service.py \
  shierfeng-fastapi-backend/tests/module_business/test_vo_contract.py \
  -m "安全：隔离主数据并封闭单条删除"

Task 5: 五类业务流程的批次一致性与自动继承

Files:

Modify: shierfeng-fastapi-backend/module_business/controller/inbound_controller.py
Modify: shierfeng-fastapi-backend/module_business/controller/filling_controller.py
Modify: shierfeng-fastapi-backend/module_business/controller/shipment_controller.py
Modify: shierfeng-fastapi-backend/module_business/controller/usage_controller.py
Modify: shierfeng-fastapi-backend/module_business/controller/return_controller.py
Modify: shierfeng-fastapi-backend/module_business/dao/operation_dao.py
Modify: shierfeng-fastapi-backend/module_business/dao/inventory_dao.py
Modify: shierfeng-fastapi-backend/module_business/service/inbound_service.py
Modify: shierfeng-fastapi-backend/module_business/service/filling_service.py
Modify: shierfeng-fastapi-backend/module_business/service/shipment_service.py
Modify: shierfeng-fastapi-backend/module_business/service/usage_service.py
Modify: shierfeng-fastapi-backend/module_business/service/return_service.py
Modify: shierfeng-fastapi-backend/tests/module_business/test_inbound_service.py
Modify: shierfeng-fastapi-backend/tests/module_business/test_filling_service.py
Modify: shierfeng-fastapi-backend/tests/module_business/test_shipment_service.py
Modify: shierfeng-fastapi-backend/tests/module_business/test_usage_service.py
Modify: shierfeng-fastapi-backend/tests/module_business/test_return_service.py
Modify: shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py

Interfaces:

Consumes: Task 2 的 TestBatchScope 和 lock_active_batch，现有仓库/现场授权与托盘状态机。
Produces: 五类列表、详情、选项、扫码及写入的 scope 隔离；订单、明细、装填、使用、库存和事件自动继承同一 test_batch_id。
Ordering contract: 测试业务写入先锁 biz_test_batch 并确认 ACTIVE，再按现有顺序锁仓库/现场、库存、托盘及单据。
Compatibility: 入库和发货 DELETE 路由保留，但统一返回批次清理提示，不再删除草稿。
Step 1: 为五条链路增加失败测试矩阵

每个业务测试文件至少覆盖正式模式、同批次成功、正式/测试混用、跨批次拒绝、关闭后拒绝、继承字段和回滚。共用断言形态如下：

async def test_test_scope_write_locks_batch_before_business_rows(self) -> None:
    calls = []
    TestBatchAccessService.lock_active_batch = AsyncMock(side_effect=lambda *_: calls.append('batch'))
    InventoryDao.get_inventories_for_update = AsyncMock(side_effect=lambda *_: calls.append('inventory') or inventories)
    await InboundService.add_inbound_services(self.db, command, current_user, make_test_scope(4))
    self.assertLess(calls.index('batch'), calls.index('inventory'))

async def test_created_records_inherit_batch_id(self) -> None:
    await FillingService.fill_pallet_services(self.db, command, current_user, make_test_scope(4))
    filling = OperationDao.add_filling_record.await_args.args[1]
    event = InventoryDao.add_event.await_args.args[2]
    self.assertEqual(4, filling.test_batch_id)
    self.assertEqual(4, event.test_batch_id)
    self.assertEqual(4, inventory.test_batch_id)

async def test_cross_batch_resource_is_rejected_and_rolled_back(self) -> None:
    pallet.test_batch_id = 5
    with self.assertRaisesRegex(ServiceException, '不属于当前数据模式或测试批次'):
        await ShipmentService.add_shipment_services(self.db, command, current_user, make_test_scope(4))
    self.db.rollback.assert_awaited_once()
    self.db.commit.assert_not_awaited()

为入库/发货旧删除增加测试，断言 OperationDao.delete_inbound、delete_shipment 从未执行。

Step 2: 运行五类业务测试并确认失败

Run:

cd shierfeng-fastapi-backend
PYTHONDONTWRITEBYTECODE=1 conda run -n ShiErFeng python -m unittest \
  tests.module_business.test_inbound_service \
  tests.module_business.test_filling_service \
  tests.module_business.test_shipment_service \
  tests.module_business.test_usage_service \
  tests.module_business.test_return_service \
  tests.module_business.test_inventory_dao -v

Expected: FAIL，至少显示 service 方法缺少 scope 参数或新记录未写入 test_batch_id。

Step 3: 给 OperationDao 和 InventoryDao 的所有读取增加 scope

所有 _get_by_id、_get_for_update、列表、详情、幂等查询、扫码、库存和事件查询均显式接收 scope，查询根表时追加：

batch_scope_condition(model.test_batch_id, scope)

关键签名固定为：

OperationDao.get_inbound_list(db, query, warehouse_ids, scope, is_page=True) -> PageModel
OperationDao.get_inbound_for_update(db, inbound_id, scope) -> BizInboundOrder | None
OperationDao.get_active_filling_for_update(db, pallet_id, scope) -> BizFillingRecord | None
InventoryDao.get_inventory_for_update(db, pallet_id, scope) -> BizPalletInventory | None
InventoryDao.get_inventories_for_update(db, pallet_ids, scope) -> list[BizPalletInventory]
InventoryDao.get_pallet_by_qr_token(db, qr_token, scope) -> BizPallet | None
InventoryDao.get_event_by_idempotency_key(db, pallet_id, idempotency_key, scope) -> BizPalletEvent | None

被当前 scope 隐藏的 ID 不得回退到无条件查询；需要区分错误时由服务用专用 get_*_unscoped_for_validation 查询并只返回归属字段，不返回业务内容。

Step 4: 给五类控制器和服务传入读写 scope

列表、详情、选项和扫码使用：

test_scope: Annotated[TestBatchScope, TestBatchScopeDependency()]

新增、编辑、确认、使用和退回使用：

test_scope: Annotated[TestBatchScope, TestBatchScopeDependency(write=True)]

所有服务公开方法把 test_scope 放在 current_user 之后；测试写操作进入 try 后第一条数据库语句必须是：

await TestBatchAccessService.lock_active_batch(query_db, test_scope)

随后对仓库、现场、产线、菌种、托盘、活动装填、库存和已有单据逐一执行 assert_record_in_scope，再执行现有数据权限与状态机校验。

Step 5: 让所有业务派生记录继承 scope

每个构造器都显式赋值，不能依赖数据库默认值：

inbound = BizInboundOrder(
    inbound_no=inbound_no,
    warehouse_id=page_object.warehouse_id,
    status=DocumentStatus.DRAFT.value,
    create_by=user_name,
    create_time=current_time,
    update_by=user_name,
    update_time=current_time,
    test_batch_id=test_scope.batch_id,
)
items = [
    BizInboundItem(
        inbound_id=inbound.inbound_id,
        pallet_id=pallet_id,
        create_by=user_name,
        create_time=current_time,
        test_batch_id=test_scope.batch_id,
    )
    for pallet_id in pallet_ids
]
inventory.test_batch_id = test_scope.batch_id
event.test_batch_id = test_scope.batch_id

同样覆盖 BizFillingRecord、BizShipmentOrder/Item、BizUsageRecord、BizReturnOrder/Item。已有库存只能保持原 scope，不允许一次业务操作把 test_batch_id 从 NULL 改为值或从一个批次改到另一个批次。

Step 6: 封闭入库和发货旧删除逻辑

delete_inbound_services 与 delete_shipment_services 在访问 DAO 前固定抛出：

raise ServiceWarning(message='不支持单条物理删除，请通过测试数据批次清理')

DAO 私有删除方法暂时保留给 Task 8 的集中清理 DAO 使用，但不得从旧 controller/service 调用。

Step 7: 运行五类业务回归和格式检查

Run: Step 2 命令，然后运行：

cd shierfeng-fastapi-backend
PYTHONDONTWRITEBYTECODE=1 conda run -n ShiErFeng python -m ruff format --check \
  module_business/controller module_business/dao module_business/service tests/module_business
PYTHONDONTWRITEBYTECODE=1 conda run -n ShiErFeng python -m ruff check \
  module_business/controller module_business/dao module_business/service tests/module_business

Expected: 全部 PASS；测试证明批次锁早于业务锁，任一跨 scope 输入都会回滚。

Step 8: 提交本任务

git add shierfeng-fastapi-backend/module_business/controller/{inbound,filling,shipment,usage,return}_controller.py \
  shierfeng-fastapi-backend/module_business/dao/operation_dao.py \
  shierfeng-fastapi-backend/module_business/dao/inventory_dao.py \
  shierfeng-fastapi-backend/module_business/service/{inbound,filling,shipment,usage,return}_service.py \
  shierfeng-fastapi-backend/tests/module_business/test_{inbound,filling,shipment,usage,return}_service.py \
  shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py
git commit --only shierfeng-fastapi-backend/module_business/controller/{inbound,filling,shipment,usage,return}_controller.py \
  shierfeng-fastapi-backend/module_business/dao/operation_dao.py \
  shierfeng-fastapi-backend/module_business/dao/inventory_dao.py \
  shierfeng-fastapi-backend/module_business/service/{inbound,filling,shipment,usage,return}_service.py \
  shierfeng-fastapi-backend/tests/module_business/test_{inbound,filling,shipment,usage,return}_service.py \
  shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py \
  -m "功能：隔离五类测试业务链路"

Task 6: 报表、选项、扫码与首页数据隔离

Files:

Modify: shierfeng-fastapi-backend/module_business/controller/report_controller.py
Modify: shierfeng-fastapi-backend/module_business/dao/access_dao.py
Modify: shierfeng-fastapi-backend/module_business/dao/report_dao.py
Modify: shierfeng-fastapi-backend/module_business/service/access_service.py
Modify: shierfeng-fastapi-backend/module_business/service/report_service.py
Modify: shierfeng-fastapi-backend/tests/module_business/test_access_service.py
Modify: shierfeng-fastapi-backend/tests/module_business/test_report_service.py
Create: shierfeng-fastapi-backend/tests/module_business/test_test_batch_query_scope_contract.py
Modify: shierfeng-fastapi-frontend/scripts/verify-business-contract.mjs

Interfaces:

Consumes: Tasks 4-5 的 scoped 主数据、业务、库存查询。
Produces: 六类报表全部显式接收 TestBatchScope；用户仓库/现场授权只在当前 scope 的目标资源上生效。
Dashboard contract: src/views/dashboard/index.vue 继续调用 listInbound/listShipment/listReturn/listPalletInventory/listShipmentReport，不新增聚合 API；统一请求头自然隔离其指标、趋势和状态分布。
Step 1: 写 DAO 查询盘点和报表隔离失败测试

test_test_batch_query_scope_contract.py 用 AST 固定所有公开读方法都包含 scope: TestBatchScope，并对 SQLAlchemy statement 做条件检查：

SCOPED_DAO_METHODS = {
    'master_dao.py': {'list_strain', 'get_strain', 'list_customer', 'get_customer', 'list_site', 'get_site',
                      'list_line', 'get_line', 'list_warehouse', 'get_warehouse', 'list_pallet', 'get_pallet'},
    'operation_dao.py': {'get_inbound_list', 'get_inbound_by_id', 'get_filling_list', 'get_filling_by_id',
                         'get_shipment_list', 'get_shipment_by_id', 'get_usage_list', 'get_usage_by_id',
                         'get_return_list', 'get_return_by_id'},
    'report_dao.py': {'get_pallet_inventory_list', 'get_filling_report_list', 'get_shipment_report_list',
                      'get_consumption_report_list', 'get_return_report_list', 'get_pallet_trace_list'},
}

async def test_formal_report_does_not_return_test_rows(self) -> None:
    await ReportService.get_pallet_inventory_list_services(self.db, query, admin, TestBatchScope.formal(), True)
    self.assertEqual(TestBatchScope.formal(), ReportDao.get_pallet_inventory_list.await_args.args[-2])

补充普通用户测试：批次 scope 与 list_user_warehouse_ids/list_user_site_ids 的交集为空时返回空页，不放宽原权限。

Step 2: 运行报表与查询盘点测试并确认失败

Run:

cd shierfeng-fastapi-backend
PYTHONDONTWRITEBYTECODE=1 conda run -n ShiErFeng python -m unittest \
  tests.module_business.test_access_service \
  tests.module_business.test_report_service \
  tests.module_business.test_test_batch_query_scope_contract -v

Expected: FAIL，显示报表方法和授权 DAO 尚未接收 scope。

Step 3: 将批次条件加到六类报表的根表和关键 join

ReportDao 每个入口增加 scope，除根表外，对能独立产生行的 join 表也增加相同归属条件。例如托盘库存：

conditions = [batch_scope_condition(BizPalletInventory.test_batch_id, scope)]
statement = statement.join(BizPallet, BizPallet.pallet_id == BizPalletInventory.pallet_id).where(
    *conditions,
    batch_scope_condition(BizPallet.test_batch_id, scope),
)

装填、发货、消耗、退回、追溯分别校验 record/order/item/event/pallet 的批次字段。任何不一致行不应被静默拼入报表；测试数据构造器另加一条不一致记录并断言查询排除。

Step 4: 让访问范围、业务选项和扫码同时收窄

AccessDao.list_user_warehouse_ids/list_user_site_ids 增加 scope 并 join 目标主表过滤 test_batch_id。所有仓库、现场、菌种、产线、托盘选项沿用 scope；扫码按二维码全局定位后必须比较归属，跨 scope 返回“不属于当前数据模式或测试批次”。

Step 5: 固化首页只依赖 scoped API

在 verify-business-contract.mjs 增加：

test('首页指标全部来自受批次作用域保护的业务与报表 API', async () => {
  const source = await readFile(path.join(frontendRoot, 'src/views/dashboard/index.vue'), 'utf8')
  for (const api of ['listInbound', 'listReturn', 'listShipment', 'listPalletInventory', 'listShipmentReport']) {
    assert.match(source, new RegExp(`\\b${api}\\b`))
  }
  assert.doesNotMatch(source, /axios\.|fetch\(/)
})

Step 6: 运行后端与 Web 契约测试

Run:

cd shierfeng-fastapi-backend
PYTHONDONTWRITEBYTECODE=1 conda run -n ShiErFeng python -m unittest \
  tests.module_business.test_access_service \
  tests.module_business.test_report_service \
  tests.module_business.test_test_batch_query_scope_contract -v
cd ../shierfeng-fastapi-frontend
npm run test:business

Expected: PASS；查询盘点覆盖六类主数据、五类业务、六类报表和首页所依赖的全部 API。

Step 7: 提交本任务

git add shierfeng-fastapi-backend/module_business/controller/report_controller.py \
  shierfeng-fastapi-backend/module_business/dao/{access,report}_dao.py \
  shierfeng-fastapi-backend/module_business/service/{access,report}_service.py \
  shierfeng-fastapi-backend/tests/module_business/test_access_service.py \
  shierfeng-fastapi-backend/tests/module_business/test_report_service.py \
  shierfeng-fastapi-backend/tests/module_business/test_test_batch_query_scope_contract.py \
  shierfeng-fastapi-frontend/scripts/verify-business-contract.mjs
git commit --only shierfeng-fastapi-backend/module_business/controller/report_controller.py \
  shierfeng-fastapi-backend/module_business/dao/{access,report}_dao.py \
  shierfeng-fastapi-backend/module_business/service/{access,report}_service.py \
  shierfeng-fastapi-backend/tests/module_business/test_access_service.py \
  shierfeng-fastapi-backend/tests/module_business/test_report_service.py \
  shierfeng-fastapi-backend/tests/module_business/test_test_batch_query_scope_contract.py \
  shierfeng-fastapi-frontend/scripts/verify-business-contract.mjs \
  -m "安全：隔离报表选项扫码与首页数据"

Task 7: 旧数据双向关联闭包与纳入事务

Files:

Create: shierfeng-fastapi-backend/module_business/domain/test_batch_graph.py
Create: shierfeng-fastapi-backend/module_business/dao/test_batch_adoption_dao.py
Create: shierfeng-fastapi-backend/module_business/service/test_batch_adoption_service.py
Create: shierfeng-fastapi-backend/tests/module_business/test_test_batch_adoption.py
Modify: shierfeng-fastapi-backend/module_business/dao/test_batch_dao.py
Modify: shierfeng-fastapi-backend/module_business/entity/vo/test_batch_vo.py

Interfaces:

Consumes: 11 类 AdoptionRootModel、Task 1 的 16 个批次模型、Task 2 的 admin 门禁。
Produces: BatchSelection = dict[str, frozenset[int]]、TestBatchGraph.compute_closure(db: AsyncSession, roots: list[AdoptionRootModel]) -> BatchSelection、TestBatchAdoptionService.preview(db, batch_id, command, current_user)、execute(db, batch_id, command, current_user)。
Route contract for Task 11: POST /system/test-batch/{batch_id}/adoption/preview 与 /adoption/execute，权限均为 system:testbatch:adopt 且 service 硬校验 admin。
Step 1: 写图闭包、预览过期和事务回滚失败测试

async def test_closure_walks_foreign_keys_in_both_directions_until_stable(self) -> None:
    roots = [AdoptionRootModel(resourceType='pallet', resourceIds=[10])]
    closure = await TestBatchGraph.compute_closure(self.db, roots)
    self.assertEqual({10}, closure['biz_pallet'])
    self.assertEqual({21}, closure['biz_filling_record'])
    self.assertEqual({31}, closure['biz_shipment_item'])
    self.assertEqual({30}, closure['biz_shipment_order'])
    self.assertEqual({41}, closure['biz_pallet_event'])

async def test_closure_rejects_any_already_tagged_record(self) -> None:
    self.rows['biz_usage_record'][50].test_batch_id = 8
    with self.assertRaisesRegex(ServiceWarning, '已属于其他测试批次'):
        await TestBatchAdoptionService.preview(self.db, 7, command, make_admin())

async def test_execute_recomputes_and_rejects_changed_scope(self) -> None:
    preview = await service.preview(self.db, 7, command, make_admin())
    add_new_connected_event()
    with self.assertRaisesRegex(ServiceWarning, '预览范围已变化'):
        await service.execute(self.db, 7, execute_command(preview), make_admin())
    self.db.rollback.assert_awaited_once()

另覆盖循环去重、事件 source_type/source_id 的多态反向边、批次名称不匹配、令牌过期、令牌重复、普通用户越权和任一 UPDATE 失败后全部回滚。

Step 2: 运行纳入测试并确认失败

Run:

cd shierfeng-fastapi-backend
PYTHONDONTWRITEBYTECODE=1 conda run -n ShiErFeng python -m unittest \
  tests.module_business.test_test_batch_adoption -v

Expected: FAIL，报告图和纳入服务不存在。

Step 3: 定义完整图目录并从 SQLAlchemy 外键生成双向边

test_batch_graph.py 固定模型、主键、展示编号和根类型映射：

BATCH_MODELS = {
    'biz_strain': (BizStrain, BizStrain.strain_id, BizStrain.strain_code),
    'biz_customer': (BizCustomer, BizCustomer.customer_id, BizCustomer.customer_code),
    'biz_customer_site': (BizCustomerSite, BizCustomerSite.site_id, BizCustomerSite.site_code),
    'biz_production_line': (BizProductionLine, BizProductionLine.line_id, BizProductionLine.line_code),
    'biz_warehouse': (BizWarehouse, BizWarehouse.warehouse_id, BizWarehouse.warehouse_code),
    'biz_pallet': (BizPallet, BizPallet.pallet_id, BizPallet.pallet_code),
    'biz_inbound_order': (BizInboundOrder, BizInboundOrder.inbound_id, BizInboundOrder.inbound_no),
    'biz_inbound_item': (BizInboundItem, BizInboundItem.inbound_item_id, BizInboundItem.inbound_item_id),
    'biz_filling_record': (BizFillingRecord, BizFillingRecord.filling_id, BizFillingRecord.filling_no),
    'biz_shipment_order': (BizShipmentOrder, BizShipmentOrder.shipment_id, BizShipmentOrder.shipment_no),
    'biz_shipment_item': (BizShipmentItem, BizShipmentItem.shipment_item_id, BizShipmentItem.shipment_item_id),
    'biz_usage_record': (BizUsageRecord, BizUsageRecord.usage_id, BizUsageRecord.usage_no),
    'biz_return_order': (BizReturnOrder, BizReturnOrder.return_id, BizReturnOrder.return_no),
    'biz_return_item': (BizReturnItem, BizReturnItem.return_item_id, BizReturnItem.return_item_id),
    'biz_pallet_inventory': (BizPalletInventory, BizPalletInventory.pallet_id, BizPalletInventory.pallet_id),
    'biz_pallet_event': (BizPalletEvent, BizPalletEvent.event_id, BizPalletEvent.event_id),
}

ROOT_TABLES = {
    'strain': 'biz_strain', 'customer': 'biz_customer', 'site': 'biz_customer_site',
    'line': 'biz_production_line', 'warehouse': 'biz_warehouse', 'pallet': 'biz_pallet',
    'inbound': 'biz_inbound_order', 'filling': 'biz_filling_record',
    'shipment': 'biz_shipment_order', 'usage': 'biz_usage_record', 'return': 'biz_return_order',
}

遍历 Base.metadata 中仅连接 BATCH_MODELS 的外键，排除 test_batch_id，生成正反两个查询方向。额外定义 BizPalletEvent.source_type/source_id 到 INBOUND/FILLING/SHIPMENT/USAGE/RETURN 根表的多态边。BFS 每轮按表名和主键排序，直到没有新增 ID。

Step 4: 实现规范摘要和一次性预览令牌

def canonical_selection(selection: BatchSelection) -> bytes:
    payload = {table: sorted(ids) for table, ids in sorted(selection.items()) if ids}
    return json.dumps(payload, ensure_ascii=True, separators=(',', ':'), sort_keys=True).encode()

def selection_hash(selection: BatchSelection) -> str:
    return hashlib.sha256(canonical_selection(selection)).hexdigest()

def issue_preview_token() -> tuple[str, str]:
    raw = secrets.token_urlsafe(32)
    return raw, hashlib.sha256(raw.encode()).hexdigest()

预览将 preview_action='ADOPT'、令牌哈希、scope 哈希和 now + 600s 写入已锁定批次行并提交；响应按 16 张表返回 table_name/label/count/keys，keys 全量排序返回，便于管理员识别被闭包带入的正式编号。

Step 5: 实现锁定、重算和原子纳入

执行顺序固定：硬校验 admin -> 锁 ACTIVE 批次 -> 比较批次名称 -> 常量时间比较令牌摘要 -> 校验 action/过期时间 -> 按表名和主键锁根记录 -> 计算闭包 -> 按稳定顺序锁闭包全部记录 -> 再算一次闭包 -> 比较 scope 哈希 -> 确认所有记录 test_batch_id is None -> 16 张表批量更新 -> 清空预览字段 -> 单次 commit。

for table_name in sorted(selection):
    await TestBatchAdoptionDao.set_test_batch_id(
        db,
        table_name,
        sorted(selection[table_name]),
        batch_id,
    )
await TestBatchDao.clear_preview(db, batch)
await db.commit()

任何异常均 rollback()；不允许跳过某张表、不允许前端提交最终 ID 列表、不允许部分纳入。

Step 6: 固定两个纳入 service 签名并运行测试

在测试中用 inspect.signature 固定 preview(db, batch_id, command, current_user) 和 execute(db, batch_id, command, current_user)；路由统一留到 Task 11 注册，确保精确 API 权限契约不会在中间任务失配。

Run:

cd shierfeng-fastapi-backend
PYTHONDONTWRITEBYTECODE=1 conda run -n ShiErFeng python -m unittest \
  tests.module_business.test_test_batch_adoption \
  tests.module_business.test_test_batch_service -v

Expected: PASS；确认 scope 改变、令牌重复和批次关闭都不会更新任何业务行。

Step 7: 提交本任务

git add shierfeng-fastapi-backend/module_business/domain/test_batch_graph.py \
  shierfeng-fastapi-backend/module_business/dao/test_batch_adoption_dao.py \
  shierfeng-fastapi-backend/module_business/service/test_batch_adoption_service.py \
  shierfeng-fastapi-backend/module_business/dao/test_batch_dao.py \
  shierfeng-fastapi-backend/module_business/entity/vo/test_batch_vo.py \
  shierfeng-fastapi-backend/tests/module_business/test_test_batch_adoption.py
git commit --only shierfeng-fastapi-backend/module_business/domain/test_batch_graph.py \
  shierfeng-fastapi-backend/module_business/dao/test_batch_adoption_dao.py \
  shierfeng-fastapi-backend/module_business/service/test_batch_adoption_service.py \
  shierfeng-fastapi-backend/module_business/dao/test_batch_dao.py \
  shierfeng-fastapi-backend/module_business/entity/vo/test_batch_vo.py \
  shierfeng-fastapi-backend/tests/module_business/test_test_batch_adoption.py \
  -m "功能：增加旧测试数据闭包纳入"

Task 8: 清理预览、外键顺序、原子回滚与幂等

Files:

Create: shierfeng-fastapi-backend/module_business/dao/test_batch_purge_dao.py
Create: shierfeng-fastapi-backend/module_business/service/test_batch_purge_service.py
Create: shierfeng-fastapi-backend/tests/module_business/test_test_batch_purge.py
Modify: shierfeng-fastapi-backend/config/env.py
Modify: shierfeng-fastapi-backend/module_business/dao/test_batch_dao.py
Modify: shierfeng-fastapi-backend/module_business/domain/test_batch_graph.py
Modify: shierfeng-fastapi-backend/module_business/entity/vo/test_batch_vo.py

Interfaces:

Consumes: CLOSED 批次、Task 7 的规范 scope 摘要/令牌机制、SQLAlchemy 外键元数据。
Produces: TestBatchPurgeService.preview(db, batch_id, user)、purge(db, batch_id, command, user)。
Route contract for Task 11: POST /system/test-batch/{batch_id}/purge/preview 与 POST /system/test-batch/{batch_id}/purge，权限 system:testbatch:purge 且 service 硬校验 admin。
Configuration: TestBatchConfig.test_batch_online_purge_max_rows、TestBatchConfig.test_batch_preview_ttl_seconds=600。
Step 1: 写删除拓扑、回滚、并发与幂等失败测试

EXPECTED_DELETE_ORDER = [
    'biz_pallet_inventory', 'biz_usage_record', 'biz_return_item',
    'biz_shipment_item', 'biz_inbound_item', 'biz_return_order',
    'biz_shipment_order', 'biz_inbound_order', 'biz_pallet_event',
    'biz_filling_record', 'biz_production_line', 'biz_customer_site',
    'biz_customer', 'biz_warehouse', 'biz_strain', 'biz_pallet',
]

def test_delete_order_is_derived_from_real_foreign_keys(self) -> None:
    order = derive_batch_delete_order(Base.metadata)
    self.assertEqual(EXPECTED_DELETE_ORDER, order)
    assert_child_tables_precede_parents(self, order, Base.metadata)

async def test_active_batch_cannot_be_purged(self) -> None:
    with self.assertRaisesRegex(ServiceWarning, '必须先关闭'):
        await service.preview(self.db, 7, make_admin())

async def test_delete_failure_rolls_back_and_keeps_closed(self) -> None:
    TestBatchPurgeDao.delete_table_rows = AsyncMock(side_effect=[3, RuntimeError('fk failure')])
    with self.assertRaisesRegex(RuntimeError, 'fk failure'):
        await service.purge(self.db, 7, command, make_admin())
    self.db.rollback.assert_awaited_once()
    self.assertEqual('CLOSED', batch.status)

async def test_repeated_purge_of_purged_batch_is_stable(self) -> None:
    batch.status = 'PURGED'
    result = await service.purge(self.db, 7, command, make_admin())
    self.assertEqual('该测试批次已清理', result.message)
    TestBatchPurgeDao.delete_table_rows.assert_not_awaited()

另覆盖预览变化、名称错误、令牌过期/重复、在线上限未配置、超过上限、两次并发请求只执行一次、用户仓库/现场关联先删；日志装饰器契约在 Task 11 注册 controller 时验证。

Step 2: 运行清理测试并确认失败

Run:

cd shierfeng-fastapi-backend
PYTHONDONTWRITEBYTECODE=1 conda run -n ShiErFeng python -m unittest \
  tests.module_business.test_test_batch_purge -v

Expected: FAIL，报告清理服务、配置或删除顺序函数不存在。

Step 3: 增加安全配置并从实际外键图生成删除顺序

class TestBatchSettings(BaseSettings):
    test_batch_preview_ttl_seconds: int = 600
    test_batch_online_purge_max_rows: int = 0

# GetConfig 增加 get_test_batch_config()，模块底部实例化：
TestBatchConfig = get_config.get_test_batch_config()

derive_batch_delete_order() 读取 16 张表间的真实 ForeignKey，执行确定性的子表优先拓扑排序；若检测到环或输出与安全表集合不一致则 fail-fast。biz_user_warehouse 和 biz_user_site 不进入带批次字段的拓扑，执行时在删除仓库/现场之前按目标 ID 显式删除。

Step 4: 实现预览和在线阈值保护

预览锁定批次并要求 CLOSED，按 16 张表统计数量和关键编号，生成 PURGE 令牌及 scope 摘要。总行数计算为 16 张业务表数量之和，不包含授权关联和批次元数据。即使上限为 0 或总数超限，预览仍返回 onlinePurgeAllowed=false 和明确 blockingReason，以便管理员检查范围；只有执行接口在锁行并重算后应用以下阻断：

if TestBatchConfig.test_batch_online_purge_max_rows <= 0:
    raise ServiceWarning(message='尚未配置在线测试批次清理上限')
if total_count > TestBatchConfig.test_batch_online_purge_max_rows:
    raise ServiceWarning(message='批次数据量超过在线清理上限，请按运维流程处理')

Step 5: 实现单事务清理和审计汇总

执行时锁批次行，PURGED 直接返回稳定结果；其他状态必须为 CLOSED。校验名称、令牌、过期时间和重算摘要后：

await TestBatchPurgeDao.delete_user_warehouse_links(db, selection['biz_warehouse'])
await TestBatchPurgeDao.delete_user_site_links(db, selection['biz_customer_site'])
for table_name in derive_batch_delete_order(Base.metadata):
    deleted = await TestBatchPurgeDao.delete_table_rows(db, table_name, batch_id)
    if deleted != len(selection.get(table_name, ())):
        raise ServiceException(message=f'{table_name} 清理数量发生变化')
await TestBatchDao.delete_batch_users(db, batch_id)
batch.status = TestBatchStatus.PURGED.value
batch.purged_by = current_user.user.user_id
batch.purged_at = datetime.now()
batch.purge_summary = json.dumps(counts, ensure_ascii=False, sort_keys=True)
await TestBatchDao.clear_preview(db, batch)
await db.commit()

不在 DAO 内 commit；任何异常统一由 service rollback。分类计数保存在 purge_summary；操作日志仅记录批次 ID、结果状态和响应摘要。

Step 6: 固定清理 service 签名并运行清理/SQL 契约

在测试中固定 preview(db, batch_id, current_user) 与 purge(db, batch_id, command, current_user)；Task 11 再将其映射到完整路由。

Run:

cd shierfeng-fastapi-backend
PYTHONDONTWRITEBYTECODE=1 TEST_BATCH_ONLINE_PURGE_MAX_ROWS=10000 \
  conda run -n ShiErFeng python -m unittest \
  tests.module_business.test_test_batch_purge \
  tests.module_business.test_sql_contract -v

Expected: PASS；失败回滚保持 CLOSED，并发重复请求不执行第二次删除，所有外键均无 cascade。

Step 7: 提交本任务

git add shierfeng-fastapi-backend/config/env.py \
  shierfeng-fastapi-backend/module_business/dao/test_batch_dao.py \
  shierfeng-fastapi-backend/module_business/dao/test_batch_purge_dao.py \
  shierfeng-fastapi-backend/module_business/domain/test_batch_graph.py \
  shierfeng-fastapi-backend/module_business/entity/vo/test_batch_vo.py \
  shierfeng-fastapi-backend/module_business/service/test_batch_purge_service.py \
  shierfeng-fastapi-backend/tests/module_business/test_test_batch_purge.py
git commit --only shierfeng-fastapi-backend/config/env.py \
  shierfeng-fastapi-backend/module_business/dao/test_batch_dao.py \
  shierfeng-fastapi-backend/module_business/dao/test_batch_purge_dao.py \
  shierfeng-fastapi-backend/module_business/domain/test_batch_graph.py \
  shierfeng-fastapi-backend/module_business/entity/vo/test_batch_vo.py \
  shierfeng-fastapi-backend/module_business/service/test_batch_purge_service.py \
  shierfeng-fastapi-backend/tests/module_business/test_test_batch_purge.py \
  -m "安全：增加测试批次原子清理"

Task 9: Web 批次 Store、请求头和模式切换

Files:

Create: shierfeng-fastapi-frontend/src/api/system/testBatch.js
Create: shierfeng-fastapi-frontend/src/store/modules/testBatch.js
Create: shierfeng-fastapi-frontend/src/utils/testBatchContext.js
Create: shierfeng-fastapi-frontend/src/components/TestBatchModeSwitch/index.vue
Modify: shierfeng-fastapi-frontend/src/store/modules/user.js
Modify: shierfeng-fastapi-frontend/src/utils/request.js
Modify: shierfeng-fastapi-frontend/src/layout/components/Navbar.vue
Modify: shierfeng-fastapi-frontend/src/layout/components/AppMain.vue
Modify: shierfeng-fastapi-frontend/scripts/verify-business-contract.mjs

Interfaces:

Consumes: GET /system/test-batch/options 与后端错误数据 data.reason === 'TEST_BATCH_UNAVAILABLE'。
Produces: useTestBatchStore、currentBatchId/currentBatchName/isTestMode/scopeVersion、请求头 X-Test-Batch-Id。
UI contract: 顶栏始终显示“正式数据”或醒目的“测试数据 · 批次名”；切换后 scopeVersion 递增并重建当前路由组件，清空未提交表单和缓存列表。
Step 1: 写 Web 模式契约失败测试

在 verify-business-contract.mjs 增加静态契约：

const source = relativePath => readFile(path.join(frontendRoot, relativePath), 'utf8')

test('Web 测试模式统一附加请求头并通过路由重建清空页面状态', async () => {
  const requestSource = await source('src/utils/request.js')
  const appMainSource = await source('src/layout/components/AppMain.vue')
  const navbarSource = await source('src/layout/components/Navbar.vue')
  assert.match(requestSource, /X-Test-Batch-Id/)
  assert.match(requestSource, /TEST_BATCH_UNAVAILABLE/)
  assert.match(appMainSource, /scopeVersion/)
  assert.match(navbarSource, /TestBatchModeSwitch/)
  assert.match(navbarSource, /测试数据/)
})

另验证 store 只持久化批次 ID，不缓存授权结果；退出登录调用 clearStoredTestBatchId()。

Step 2: 运行 Web 契约并确认失败

Run:

cd shierfeng-fastapi-frontend
npm run test:business

Expected: FAIL，报告批次 store、组件或请求头不存在。

Step 3: 实现批次 API 和 Pinia Store

API：

export function listAvailableTestBatches() {
  return request({ url: '/system/test-batch/options', method: 'get' })
}

Store 固定行为：

const useTestBatchStore = defineStore('testBatch', {
  state: () => ({
    options: [],
    currentBatchId: getStoredTestBatchId(),
    currentBatchName: '',
    scopeVersion: 0
  }),
  getters: {
    isTestMode: state => Boolean(state.currentBatchId)
  },
  actions: {
    async refreshOptions() {
      const response = await listAvailableTestBatches()
      this.options = Array.isArray(response.data) ? response.data : []
      const current = this.options.find(item => String(item.batchId) === String(this.currentBatchId))
      if (this.currentBatchId && (!current || current.status !== 'ACTIVE')) this.leaveTestMode()
      else this.currentBatchName = current?.batchName || ''
    },
    selectBatch(batch) {
      const allowed = this.options.some(item => item.batchId === batch.batchId && item.status === 'ACTIVE')
      if (!allowed) throw new Error('所选测试批次当前不可用')
      this.currentBatchId = String(batch.batchId)
      this.currentBatchName = batch.batchName
      setStoredTestBatchId(this.currentBatchId)
      this.scopeVersion += 1
    },
    leaveTestMode() {
      this.currentBatchId = ''
      this.currentBatchName = ''
      clearStoredTestBatchId()
      this.scopeVersion += 1
    },
    reset() {
      this.options = []
      this.leaveTestMode()
    }
  }
})

testBatchContext.js 只封装 sessionStorage 的同步读写；store 引用该工具，request.js 不静态导入 store，从而避免 store -> api -> request -> store 循环依赖。

user.js 保存 user.admin === true 为 admin state，登出时调用 clearStoredTestBatchId()；页面随后按现有流程重新加载，不静态导入 test batch store。

Step 4: 在统一请求层注入并处理失效上下文

在传输加密前读取 store：

const testBatchId = getStoredTestBatchId()
if (getToken() && testBatchId) {
  config.headers['X-Test-Batch-Id'] = String(testBatchId)
}

成功响应和异常响应都识别 TEST_BATCH_UNAVAILABLE，先清理 storage，再通过 await import('@/store/modules/testBatch') 延迟取得 store 并退回正式模式，随后展示服务端原因并拒绝原请求；不得把失败响应转成成功，不得在页面内乐观删除记录。

Step 5: 实现顶部模式控件与页面重建

TestBatchModeSwitch 使用正式/测试分段选择和批次下拉；测试模式使用 el-tag type="danger" effect="dark" 持续显示 测试数据 · {batchName}。AppMain.vue 的页面 key 改为：

<component
  v-if="!route.meta.link"
  :is="Component"
  :key="`${route.path}:${testBatchStore.scopeVersion}`"
/>

每次切换前统一弹出“切换数据模式将清空当前页面未提交内容”的确认框；确认后变更 store，组件重建会清空扫描结果、草稿表单、列表和首页图表状态。

Step 6: 运行契约和生产构建

Run:

cd shierfeng-fastapi-frontend
npm run test:business
npm run build:prod

Expected: PASS；构建无循环依赖，正式模式不发送 X-Test-Batch-Id。

Step 7: 提交本任务

git add shierfeng-fastapi-frontend/src/api/system/testBatch.js \
  shierfeng-fastapi-frontend/src/store/modules/testBatch.js \
  shierfeng-fastapi-frontend/src/utils/testBatchContext.js \
  shierfeng-fastapi-frontend/src/store/modules/user.js \
  shierfeng-fastapi-frontend/src/utils/request.js \
  shierfeng-fastapi-frontend/src/components/TestBatchModeSwitch/index.vue \
  shierfeng-fastapi-frontend/src/layout/components/Navbar.vue \
  shierfeng-fastapi-frontend/src/layout/components/AppMain.vue \
  shierfeng-fastapi-frontend/scripts/verify-business-contract.mjs
git commit --only shierfeng-fastapi-frontend/src/api/system/testBatch.js \
  shierfeng-fastapi-frontend/src/store/modules/testBatch.js \
  shierfeng-fastapi-frontend/src/utils/testBatchContext.js \
  shierfeng-fastapi-frontend/src/store/modules/user.js \
  shierfeng-fastapi-frontend/src/utils/request.js \
  shierfeng-fastapi-frontend/src/components/TestBatchModeSwitch/index.vue \
  shierfeng-fastapi-frontend/src/layout/components/Navbar.vue \
  shierfeng-fastapi-frontend/src/layout/components/AppMain.vue \
  shierfeng-fastapi-frontend/scripts/verify-business-contract.mjs \
  -m "前端：增加测试数据模式切换"

Task 10: Web 批次管理页与“纳入测试批次”操作

Files:

Create: shierfeng-fastapi-frontend/src/views/system/testBatch/index.vue
Create: shierfeng-fastapi-frontend/src/components/TestBatchAdoptionDialog/index.vue
Modify: shierfeng-fastapi-frontend/src/api/system/testBatch.js
Modify: shierfeng-fastapi-frontend/src/views/master/strain/index.vue
Modify: shierfeng-fastapi-frontend/src/views/master/customer/index.vue
Modify: shierfeng-fastapi-frontend/src/views/master/site/index.vue
Modify: shierfeng-fastapi-frontend/src/views/master/line/index.vue
Modify: shierfeng-fastapi-frontend/src/views/master/warehouse/index.vue
Modify: shierfeng-fastapi-frontend/src/views/master/pallet/index.vue
Modify: shierfeng-fastapi-frontend/src/views/business/inbound/index.vue
Modify: shierfeng-fastapi-frontend/src/views/business/filling/index.vue
Modify: shierfeng-fastapi-frontend/src/views/business/shipment/index.vue
Modify: shierfeng-fastapi-frontend/src/views/business/usage/index.vue
Modify: shierfeng-fastapi-frontend/src/views/business/return/index.vue
Modify: shierfeng-fastapi-frontend/scripts/verify-business-contract.mjs

Interfaces:

Consumes: Tasks 3、7、8 的 12 个批次 API；userStore.admin 与 testBatchStore.isTestMode。
Produces: 仅超级管理员可见的管理页和正式模式下 11 类来源数据的“纳入测试批次”入口。
Dialog contract: 先预览完整分类/编号，再要求输入批次全名，执行只提交 roots、批次名和 previewToken。
Step 1: 写管理 API 和界面可见性失败契约

const testBatchRequests = [
  ['listTestBatches', "'/system/test-batch/list'", 'get'],
  ['getTestBatch', "'/system/test-batch/' + batchId", 'get'],
  ['addTestBatch', "'/system/test-batch'", 'post'],
  ['updateTestBatch', "'/system/test-batch/' + batchId", 'put'],
  ['assignTestBatchUsers', "'/system/test-batch/' + batchId + '/users'", 'put'],
  ['closeTestBatch', "'/system/test-batch/' + batchId + '/close'", 'put'],
  ['reopenTestBatch', "'/system/test-batch/' + batchId + '/reopen'", 'put'],
  ['previewTestBatchAdoption', "'/system/test-batch/' + batchId + '/adoption/preview'", 'post'],
  ['executeTestBatchAdoption', "'/system/test-batch/' + batchId + '/adoption/execute'", 'post'],
  ['previewTestBatchPurge', "'/system/test-batch/' + batchId + '/purge/preview'", 'post'],
  ['purgeTestBatch', "'/system/test-batch/' + batchId + '/purge'", 'post']
]

契约还要扫描 11 个来源页面：存在 TestBatchAdoptionDialog、对应 root-type，按钮同时受 userStore.admin 和正式模式限制；六个主数据页的测试模式写按钮受 userStore.admin 限制；六个报表页不得出现该组件。system:testbatch:adopt 指令在 Task 11 与完整权限契约同时接入。

Step 2: 运行 Web 契约并确认失败

Run:

cd shierfeng-fastapi-frontend
npm run test:business

Expected: FAIL，报告管理页面、API 或纳入对话框缺失。

Step 3: 实现批次管理页全部状态和操作

页面提供批次编码、名称、状态、参与用户、创建/关闭/清理时间和清理汇总；按钮矩阵固定为：

const actionsFor = row => ({
  edit: row.status !== 'PURGED',
  assign: row.status === 'ACTIVE',
  close: row.status === 'ACTIVE',
  reopen: row.status === 'CLOSED',
  purgePreview: row.status === 'CLOSED',
  audit: row.status === 'PURGED'
})

创建、备注编辑、用户授权、关闭和重新开启均等待后端成功后重新拉取列表。清理对话框先展示所有 BatchImpactModel，要求输入完整批次名；同时展示 onlinePurgeLimit，onlinePurgeAllowed=false 时显示 blockingReason 并禁用最终清理提交。提交成功后退出当前测试模式并刷新批次列表。预览过期时清空令牌，要求重新预览。

Step 4: 实现可复用的纳入对话框

<TestBatchAdoptionDialog
  ref="adoptionDialogRef"
  :root-type="'strain'"
  :resource-ids="selectedIds"
  @success="getList"
/>

组件只列出 ACTIVE 批次；打开时调用 preview，按表显示数量与全部关键编号；确认输入必须严格等于响应中的 batchName。若预览范围变化、目标批次关闭或令牌失效，保留选择但清空预览和确认文本，不执行第二次请求。

Step 5: 接入 11 个来源页面

根类型映射固定为：

const adoptionRootTypes = {
  strain: 'strain', customer: 'customer', site: 'site', line: 'line',
  warehouse: 'warehouse', pallet: 'pallet', inbound: 'inbound',
  filling: 'filling', shipment: 'shipment', usage: 'usage', return: 'return'
}

只有 userStore.admin === true && !testBatchStore.isTestMode 时展示纳入操作。没有表格多选的业务页增加 selection 列；点击后必须至少一个正整数 ID。六个主数据页在测试模式还要用 userStore.admin 限制新增、编辑和仓库/现场用户分配，授权普通测试用户只能读主数据并执行五类业务。纳入成功后重新查询当前正式列表，因为整条闭包会从正式模式消失。

Step 6: 运行契约与生产构建

Run:

cd shierfeng-fastapi-frontend
npm run test:business
npm run build:prod

Expected: PASS；报表没有纳入入口，普通用户和测试模式均看不到按钮。

Step 7: 提交本任务

git add shierfeng-fastapi-frontend/src/api/system/testBatch.js \
  shierfeng-fastapi-frontend/src/views/system/testBatch/index.vue \
  shierfeng-fastapi-frontend/src/components/TestBatchAdoptionDialog/index.vue \
  shierfeng-fastapi-frontend/src/views/master/{strain,customer,site,line,warehouse,pallet}/index.vue \
  shierfeng-fastapi-frontend/src/views/business/{inbound,filling,shipment,usage,return}/index.vue \
  shierfeng-fastapi-frontend/scripts/verify-business-contract.mjs
git commit --only shierfeng-fastapi-frontend/src/api/system/testBatch.js \
  shierfeng-fastapi-frontend/src/views/system/testBatch/index.vue \
  shierfeng-fastapi-frontend/src/components/TestBatchAdoptionDialog/index.vue \
  shierfeng-fastapi-frontend/src/views/master/{strain,customer,site,line,warehouse,pallet}/index.vue \
  shierfeng-fastapi-frontend/src/views/business/{inbound,filling,shipment,usage,return}/index.vue \
  shierfeng-fastapi-frontend/scripts/verify-business-contract.mjs \
  -m "前端：增加批次管理与旧数据纳入"

Task 11: 移除物理删除入口并收紧权限契约

Files:

Create: shierfeng-fastapi-backend/module_business/controller/test_batch_controller.py
Modify: shierfeng-fastapi-backend/module_business/controller/strain_controller.py
Modify: shierfeng-fastapi-backend/module_business/controller/customer_controller.py
Modify: shierfeng-fastapi-backend/module_business/controller/site_controller.py
Modify: shierfeng-fastapi-backend/module_business/controller/line_controller.py
Modify: shierfeng-fastapi-backend/module_business/controller/warehouse_controller.py
Modify: shierfeng-fastapi-backend/module_business/controller/pallet_controller.py
Modify: shierfeng-fastapi-backend/module_business/controller/inbound_controller.py
Modify: shierfeng-fastapi-backend/module_business/controller/shipment_controller.py
Modify: shierfeng-fastapi-backend/sql/shierfeng-fastapi.sql
Modify: shierfeng-fastapi-backend/sql/shierfeng-fastapi-pg.sql
Modify: shierfeng-fastapi-backend/sql/patches/20260922_test_data_batch_mysql.sql
Modify: shierfeng-fastapi-backend/sql/patches/20260922_test_data_batch_pg.sql
Modify: shierfeng-fastapi-backend/tests/module_business/test_api_permission_contract.py
Modify: shierfeng-fastapi-backend/tests/module_business/test_sql_contract.py
Modify: shierfeng-fastapi-frontend/src/api/master/strain.js
Modify: shierfeng-fastapi-frontend/src/api/master/customer.js
Modify: shierfeng-fastapi-frontend/src/api/master/site.js
Modify: shierfeng-fastapi-frontend/src/api/master/line.js
Modify: shierfeng-fastapi-frontend/src/api/master/warehouse.js
Modify: shierfeng-fastapi-frontend/src/api/master/pallet.js
Modify: shierfeng-fastapi-frontend/src/api/business/inbound.js
Modify: shierfeng-fastapi-frontend/src/api/business/shipment.js
Modify: shierfeng-fastapi-frontend/src/views/master/strain/index.vue
Modify: shierfeng-fastapi-frontend/src/views/master/customer/index.vue
Modify: shierfeng-fastapi-frontend/src/views/master/site/index.vue
Modify: shierfeng-fastapi-frontend/src/views/master/line/index.vue
Modify: shierfeng-fastapi-frontend/src/views/master/warehouse/index.vue
Modify: shierfeng-fastapi-frontend/src/views/master/pallet/index.vue
Modify: shierfeng-fastapi-frontend/src/views/business/inbound/index.vue
Modify: shierfeng-fastapi-frontend/src/views/business/shipment/index.vue
Modify: shierfeng-fastapi-frontend/scripts/verify-business-contract.mjs

Interfaces:

Consumes: Tasks 3、7、8 的批次 service，Task 4/5 的旧 DELETE 固定拒绝，Task 10 的批次管理和纳入入口。
Produces: SQL 菜单移除八个旧删除权限并新增八个批次管理权限；Web 无物理删除按钮/调用；旧 DELETE 路由改用对应页面 :list 权限后返回统一业务拒绝。
Exact contract: 13 个 controller、86 条目标路由、85 条 interface-auth 依赖、61 个唯一权限、18 个页面权限、43 个按钮权限、41 条写路由。
Step 1: 先更新精确权限与无删除入口失败契约

test_api_permission_contract.py 做以下精确变化：

CONTROLLER_NAMES = (
    'strain', 'customer', 'site', 'line', 'warehouse', 'pallet',
    'inbound', 'filling', 'shipment', 'usage', 'return', 'report', 'test_batch',
)
TARGET_PREFIXES = ('/master', '/business', '/report', '/system/test-batch')
PERMISSION_LITERAL_PATTERN = re.compile(r'(?:master|business|report|system):[a-z]+:[a-z]+')
UNGUARDED_AUTHENTICATED_ROUTES = {('/system/test-batch/options', 'GET')}

# 八条兼容 DELETE 路由继续存在，但使用页面 list 权限进入固定拒绝逻辑。
('/master/strain/{strain_ids}', 'DELETE'): 'master:strain:list'
('/master/customer/{customer_ids}', 'DELETE'): 'master:customer:list'
('/master/site/{site_ids}', 'DELETE'): 'master:site:list'
('/master/line/{line_ids}', 'DELETE'): 'master:line:list'
('/master/warehouse/{warehouse_ids}', 'DELETE'): 'master:warehouse:list'
('/master/pallet/{pallet_ids}', 'DELETE'): 'master:pallet:list'
('/business/inbound/{inbound_ids}', 'DELETE'): 'business:inbound:list'
('/business/shipment/{shipment_ids}', 'DELETE'): 'business:shipment:list'

新增权限精确为 system:testbatch:list/query/add/edit/assign/close/adopt/purge，权限后缀集合中删除 remove，增加 adopt/close/purge。前端契约断言源代码不存在 delStrain/delCustomer/delSite/delLine/delWarehouse/delPallet/delInbound/delShipment 的导入、调用或 *:remove 权限。

控制器 AST 还要断言四个纳入/清理路由的 Log 均使用请求白名单和响应摘要模式，且白名单中只有 path_params.batch_id。

Step 2: 运行权限与前端契约并确认失败

Run:

cd shierfeng-fastapi-backend
PYTHONDONTWRITEBYTECODE=1 conda run -n ShiErFeng python -m unittest \
  tests.module_business.test_api_permission_contract \
  tests.module_business.test_sql_contract -v
cd ../shierfeng-fastapi-frontend
npm run test:business

Expected: FAIL，显示旧 remove 权限和删除 UI 仍存在。

Step 3: 一次注册完整批次 controller

test_batch_controller.py 按静态路径优先顺序注册 12 条路由：

GET  /system/test-batch/options
GET  /system/test-batch/list
POST /system/test-batch
PUT  /system/test-batch/{batch_id}
PUT  /system/test-batch/{batch_id}/users
PUT  /system/test-batch/{batch_id}/close
PUT  /system/test-batch/{batch_id}/reopen
POST /system/test-batch/{batch_id}/adoption/preview
POST /system/test-batch/{batch_id}/adoption/execute
POST /system/test-batch/{batch_id}/purge/preview
POST /system/test-batch/{batch_id}/purge
GET  /system/test-batch/{batch_id}

options 只有 PreAuthDependency 和 CurrentUserDependency；其他路由分别使用 system:testbatch:list/query/add/edit/assign/close/adopt/purge，并调用已有硬 admin service。纳入/清理接口使用安全日志白名单：

@Log(
    title='测试数据批次',
    business_type=BusinessType.CLEAN,
    request_log_mode='include',
    request_include_fields=(RequestLogFieldRoot.PATH_PARAMS.field('batch_id'),),
    response_log_mode='summary',
)

创建、编辑、授权、关闭和重新开启使用 request_log_mode='summary'、response_log_mode='summary'。确保操作日志不保存 previewToken、roots 全量 ID、参与用户列表或关键编号。

Step 4: 更新兼容 DELETE 路由的权限依赖

八条路由保留原 URL、method 和响应结构，只将 UserInterfaceAuthDependency('*:remove') 改为同页面 *:list。service 仍无条件拒绝，因此超级管理员、普通用户和旧客户端都无法物理删除正式或测试数据。

Step 5: 移除 Web 删除 API 包装和界面逻辑

删除六个主数据和入库/发货 API 模块中的 del* 导出；页面删除工具栏、行按钮、确认框、loading、canRemove 计算和 *:remove 权限。编辑、停用、单据确认以及 Task 10 的纳入操作保持不变；11 个纳入按钮此时统一增加 v-hasPermi="['system:testbatch:adopt']"。

Step 6: 更新全量 SQL 和增量菜单补丁

全量 SQL 不再插入菜单 ID 1068,1072,1076,1081,1085,1090,1095,1102。新增：

519   测试数据批次  parent=1  component=system/testBatch/index  perms=system:testbatch:list
1109  批次查询      parent=519 perms=system:testbatch:query
1110  批次新增      parent=519 perms=system:testbatch:add
1111  批次修改      parent=519 perms=system:testbatch:edit
1112  批次授权      parent=519 perms=system:testbatch:assign
1113  批次关闭      parent=519 perms=system:testbatch:close
1114  数据纳入      parent=519 perms=system:testbatch:adopt
1115  批次清理      parent=519 perms=system:testbatch:purge

两个增量补丁先删除八个旧菜单 ID 对应的 sys_role_menu，再删除 sys_menu，最后插入 519 与 1109-1115；不得给普通角色自动插入新菜单关联，超级管理员依靠现有 admin/通配权限访问。

Step 7: 运行精确权限、SQL、Web 契约和构建

Run:

cd shierfeng-fastapi-backend
PYTHONDONTWRITEBYTECODE=1 conda run -n ShiErFeng python -m unittest \
  tests.module_business.test_api_permission_contract \
  tests.module_business.test_sql_contract -v
cd ../shierfeng-fastapi-frontend
npm run test:business
npm run build:prod

Expected: PASS，计数严格等于 13/86/85/61/18/43/41；仓库中业务 Web 源码不再引用八个 remove 权限。

Step 8: 提交本任务

git add shierfeng-fastapi-backend/module_business/controller/test_batch_controller.py \
  shierfeng-fastapi-backend/module_business/controller/{strain,customer,site,line,warehouse,pallet,inbound,shipment}_controller.py \
  shierfeng-fastapi-backend/sql/shierfeng-fastapi.sql \
  shierfeng-fastapi-backend/sql/shierfeng-fastapi-pg.sql \
  shierfeng-fastapi-backend/sql/patches/20260922_test_data_batch_mysql.sql \
  shierfeng-fastapi-backend/sql/patches/20260922_test_data_batch_pg.sql \
  shierfeng-fastapi-backend/tests/module_business/test_api_permission_contract.py \
  shierfeng-fastapi-backend/tests/module_business/test_sql_contract.py \
  shierfeng-fastapi-frontend/src/api/master/{strain,customer,site,line,warehouse,pallet}.js \
  shierfeng-fastapi-frontend/src/api/business/{inbound,shipment}.js \
  shierfeng-fastapi-frontend/src/views/master/{strain,customer,site,line,warehouse,pallet}/index.vue \
  shierfeng-fastapi-frontend/src/views/business/{inbound,shipment}/index.vue \
  shierfeng-fastapi-frontend/scripts/verify-business-contract.mjs
git commit --only shierfeng-fastapi-backend/module_business/controller/test_batch_controller.py \
  shierfeng-fastapi-backend/module_business/controller/{strain,customer,site,line,warehouse,pallet,inbound,shipment}_controller.py \
  shierfeng-fastapi-backend/sql/shierfeng-fastapi.sql \
  shierfeng-fastapi-backend/sql/shierfeng-fastapi-pg.sql \
  shierfeng-fastapi-backend/sql/patches/20260922_test_data_batch_mysql.sql \
  shierfeng-fastapi-backend/sql/patches/20260922_test_data_batch_pg.sql \
  shierfeng-fastapi-backend/tests/module_business/test_api_permission_contract.py \
  shierfeng-fastapi-backend/tests/module_business/test_sql_contract.py \
  shierfeng-fastapi-frontend/src/api/master/{strain,customer,site,line,warehouse,pallet}.js \
  shierfeng-fastapi-frontend/src/api/business/{inbound,shipment}.js \
  shierfeng-fastapi-frontend/src/views/master/{strain,customer,site,line,warehouse,pallet}/index.vue \
  shierfeng-fastapi-frontend/src/views/business/{inbound,shipment}/index.vue \
  shierfeng-fastapi-frontend/scripts/verify-business-contract.mjs \
  -m "权限：移除业务物理删除入口"

Task 12: PDA 测试模式切换与请求上下文

Files:

Create: shierfeng-fastapi-app/src/api/system/testBatch.js
Create: shierfeng-fastapi-app/src/store/modules/testBatch.js
Create: shierfeng-fastapi-app/src/components/business/TestBatchModeSwitch.vue
Create: shierfeng-fastapi-app/src/components/business/TestBatchBanner.vue
Modify: shierfeng-fastapi-app/src/store/index.js
Modify: shierfeng-fastapi-app/src/store/modules/user.js
Modify: shierfeng-fastapi-app/src/main.ts
Modify: shierfeng-fastapi-app/src/utils/constant.js
Modify: shierfeng-fastapi-app/src/utils/request.js
Modify: shierfeng-fastapi-app/src/pages/work/index.vue
Modify: shierfeng-fastapi-app/src/pages/business/inbound/index.vue
Modify: shierfeng-fastapi-app/src/pages/business/filling/index.vue
Modify: shierfeng-fastapi-app/src/pages/business/shipment/index.vue
Modify: shierfeng-fastapi-app/src/pages/business/usage/index.vue
Modify: shierfeng-fastapi-app/src/pages/business/return/index.vue
Modify: shierfeng-fastapi-app/src/api/business/inbound.js
Modify: shierfeng-fastapi-app/src/api/business/shipment.js
Modify: shierfeng-fastapi-app/tests/business-contract.test.mjs

Interfaces:

Consumes: GET /system/test-batch/options、X-Test-Batch-Id、后端失效原因。
Produces: PDA useTestBatchStore、工作台模式切换、五个业务页持续可见的测试批次横幅。
Navigation contract: 模式切换使用 uni.reLaunch('/pages/work/index') 清空页面栈和未提交扫码/表单状态。
Step 1: 写 PDA 模式与删除入口失败契约

test('PDA 测试模式附加批次请求头并在五个业务页持续提示', async () => {
  const requestSource = await readFile(path.join(appRoot, 'src/utils/request.js'), 'utf8')
  assert.match(requestSource, /X-Test-Batch-Id/)
  assert.match(requestSource, /TEST_BATCH_UNAVAILABLE/)
  for (const page of ['inbound', 'filling', 'shipment', 'usage', 'return']) {
    const source = await readFile(path.join(appRoot, `src/pages/business/${page}/index.vue`), 'utf8')
    assert.match(source, /TestBatchBanner/)
  }
})

test('PDA 不再暴露入库和发货草稿物理删除', async () => {
  for (const file of ['src/api/business/inbound.js', 'src/api/business/shipment.js']) {
    const source = await readFile(path.join(appRoot, file), 'utf8')
    assert.doesNotMatch(source, /delInbound|delShipment|method:\s*["']delete["']/)
  }
})

Step 2: 运行 PDA 契约并确认失败

Run:

cd shierfeng-fastapi-app
npm run test:business

Expected: FAIL，报告模式组件、请求头和横幅缺失，旧删除仍存在。

Step 3: 实现 PDA store、持久化和统一请求处理

Store 用现有 storage 和 constant 保存当前批次 ID/名称；只从 options 接口刷新授权，不能把本地缓存视为授权。request.js 直接读取 storage，不导入 test batch store，避免 store -> api -> request -> store 循环，并在传输加密前设置：

const testBatchId = storage.get(constant.testBatchId)
if (getToken() && testBatchId) {
  config.header['X-Test-Batch-Id'] = String(testBatchId)
}

收到 TEST_BATCH_UNAVAILABLE 时清除批次 storage、触发 uni.$emit('test-batch-unavailable', message)、提示原因并 uni.reLaunch({ url: '/pages/work/index' })。main.ts 在 Pinia 安装后只注册一次监听器并调用 useTestBatchStore().leaveTestMode()；退出登录现有 storage.clean() 同时清理批次，用户 store 同步保存 admin，但 PDA 不提供管理、纳入或清理入口。

Step 4: 实现工作台切换和五页高可见横幅

工作台使用正式/测试分段控件与已授权 ACTIVE 批次选择。确认切换后调用：

testBatchStore.selectBatch(batch)
uni.reLaunch({ url: '/pages/work/index' })

TestBatchBanner 在测试模式显示红色边框、测试数据 和完整批次名；五个业务页放在首个操作控件上方。正式模式不占布局高度。

Step 5: 删除 PDA 入库/发货草稿删除能力

移除 delInbound/delShipment API、页面按钮、loading、事件处理和 business:*:remove 权限字符串；保留草稿编辑与确认。后端兼容 DELETE 路由仍按 Task 11 返回稳定拒绝提示。

Step 6: 运行 PDA 契约与 H5 构建

Run:

cd shierfeng-fastapi-app
npm run test:business
npm run build:h5

Expected: PASS；正式模式无批次头，测试模式五个业务页都有持续提示，切换会清空页面栈。

Step 7: 提交本任务

git add shierfeng-fastapi-app/src/api/system/testBatch.js \
  shierfeng-fastapi-app/src/store/modules/testBatch.js \
  shierfeng-fastapi-app/src/components/business/TestBatchModeSwitch.vue \
  shierfeng-fastapi-app/src/components/business/TestBatchBanner.vue \
  shierfeng-fastapi-app/src/store/index.js \
  shierfeng-fastapi-app/src/store/modules/user.js \
  shierfeng-fastapi-app/src/main.ts \
  shierfeng-fastapi-app/src/utils/constant.js \
  shierfeng-fastapi-app/src/utils/request.js \
  shierfeng-fastapi-app/src/pages/work/index.vue \
  shierfeng-fastapi-app/src/pages/business/{inbound,filling,shipment,usage,return}/index.vue \
  shierfeng-fastapi-app/src/api/business/{inbound,shipment}.js \
  shierfeng-fastapi-app/tests/business-contract.test.mjs
git commit --only shierfeng-fastapi-app/src/api/system/testBatch.js \
  shierfeng-fastapi-app/src/store/modules/testBatch.js \
  shierfeng-fastapi-app/src/components/business/TestBatchModeSwitch.vue \
  shierfeng-fastapi-app/src/components/business/TestBatchBanner.vue \
  shierfeng-fastapi-app/src/store/index.js \
  shierfeng-fastapi-app/src/store/modules/user.js \
  shierfeng-fastapi-app/src/main.ts \
  shierfeng-fastapi-app/src/utils/constant.js \
  shierfeng-fastapi-app/src/utils/request.js \
  shierfeng-fastapi-app/src/pages/work/index.vue \
  shierfeng-fastapi-app/src/pages/business/{inbound,filling,shipment,usage,return}/index.vue \
  shierfeng-fastapi-app/src/api/business/{inbound,shipment}.js \
  shierfeng-fastapi-app/tests/business-contract.test.mjs \
  -m "移动端：增加测试批次模式隔离"

Task 13: 全量回归、发布防护与运维说明

Files:

Create: docs/test-data-batch-operations.md
Modify: README.md
Modify: shierfeng-fastapi-app/README.md

Interfaces:

Consumes: Tasks 1-12 的代码、SQL、测试和构建产物。
Produces: 可审计的上线顺序、备份/恢复演练、只读结构盘点、压测与 TEST_BATCH_ONLINE_PURGE_MAX_ROWS 启用说明。
Does not produce: 自动迁移命令、生产数据库写入、默认在线清理阈值或绕过原子事务的分段脚本。
Step 1: 编写发布与清理运行手册

docs/test-data-batch-operations.md 必须包含以下固定阶段和停止条件：

1. 备份并在隔离实例完成恢复演练；恢复校验失败则停止。
2. 只读盘点目标库现有业务表、索引和外键，并确认即将新增的 2 张批次表及 16 个 `test_batch_id` 字段尚不存在；结构与补丁前置不符则停止。
3. 在 staging 应用对应方言补丁，验证 18 张业务表、16 个批次字段、索引、外键和 `on_delete`，且既有 `test_batch_id` 非 `NULL` 行数必须为 0。
4. 在已审批维护窗口内先对目标库应用同一补丁，立即重复结构和既有数据校验；任一检查失败则停止应用发布并按数据库变更回退预案处理。
5. 发布后端、Web、PDA；保持 `TEST_BATCH_ONLINE_PURGE_MAX_ROWS=0`。
6. 验证正式模式列表、扫码、首页和六报表与发布前基线一致。
7. 用 staging 数据分别压测 100、500、1000、逐级增加的闭包，记录锁等待、事务耗时和回滚耗时。
8. 取满足运维窗口和数据库锁等待预算的最大实测行数，配置正整数上限并重启后端。
9. 首次创建批次、首次纳入旧数据和首次清理分别走独立变更审批。

手册给出 MySQL information_schema 与 PostgreSQL pg_catalog 的只读查询，用于核对字段、索引、外键和删除规则；给出清理失败后的检查项，禁止手工补删半条链。

Step 2: 运行后端全量业务测试

Run:

cd shierfeng-fastapi-backend
PYTHONDONTWRITEBYTECODE=1 TEST_BATCH_ONLINE_PURGE_MAX_ROWS=10000 \
  conda run -n ShiErFeng python -m unittest discover \
  -s tests/module_business -p "test_*.py" -v

Expected: PASS；无 skipped、error 或 failure。

Step 3: 运行后端格式与静态检查

Run:

cd shierfeng-fastapi-backend
PYTHONDONTWRITEBYTECODE=1 conda run -n ShiErFeng python -m ruff format --check \
  module_business common/context.py middlewares/context_middleware.py \
  config/env.py tests/module_business
PYTHONDONTWRITEBYTECODE=1 conda run -n ShiErFeng python -m ruff check \
  module_business common/context.py middlewares/context_middleware.py \
  config/env.py tests/module_business

Expected: 两条命令均 exit 0。

Step 4: 运行 Web 与 PDA 契约和构建

Run:

cd shierfeng-fastapi-frontend
npm run test:business
npm run build:prod
cd ../shierfeng-fastapi-app
npm run test:business
npm run build:h5

Expected: 四条命令均 exit 0；构建日志无 unresolved import、模板编译或类型错误。

Step 5: 执行源码与 SQL 安全扫描

Run:

rg -n "master:(strain|customer|site|line|warehouse|pallet):remove|business:(inbound|shipment):remove" \
  shierfeng-fastapi-backend/module_business shierfeng-fastapi-frontend/src shierfeng-fastapi-app/src \
  shierfeng-fastapi-backend/sql
rg -ni "on[[:space:]]+delete[[:space:]]+cascade" \
  shierfeng-fastapi-backend/module_business shierfeng-fastapi-backend/sql
rg -n "test_batch_id[[:space:]]*=[[:space:]]*[^N]" \
  shierfeng-fastapi-backend/sql/patches/20260922_test_data_batch_*.sql
git diff --check

Expected: 前三条无匹配，git diff --check exit 0。若 SQL 中合法的列定义触发第三条，应改为扫描同时包含 UPDATE 与 SET test_batch_id 的行后重新执行，并在手册记录实际命令。

Step 6: 人工执行验收矩阵但不连接生产库

在本地或 staging 覆盖：正式/测试/跨批次列表；普通授权与未授权用户；五类完整业务链；关闭后只读；纳入预览变化；清理回滚；重复清理；首页与六报表；Web/PDA 授权失效回退。记录每项请求 ID、响应和数据库分类计数到变更单，不在仓库保存敏感业务明细。

Step 7: 更新 README 并提交文档

根 README 与 PDA README 只增加功能入口、请求头、默认关闭的在线清理配置和运行手册链接，不写生产凭据或固定环境地址。

git add docs/test-data-batch-operations.md README.md \
  shierfeng-fastapi-app/README.md
git commit --only docs/test-data-batch-operations.md README.md \
  shierfeng-fastapi-app/README.md \
  -m "文档：补充测试批次发布与清理手册"

Spec Coverage Map

设计要求	实施任务
角色、超级管理员硬校验、授权用户	Tasks 2-3
两张批次表、16 张业务表标记、双数据库 SQL	Task 1
正式/测试模式、请求伪造防护、状态锁	Tasks 2、5、9、12
六类主数据隔离与创建继承	Task 4
五类业务、库存、事件继承与跨批次拒绝	Task 5
选项、扫码、首页、六类报表隔离	Task 6
旧数据双向闭包、完整预览、原子纳入	Task 7
关闭后清理、外键顺序、回滚、幂等、审计	Task 8
Web 管理、模式提示、失效回退	Tasks 9-10
仅超级管理员可用“纳入测试批次”	Tasks 7、10-11
旧删除入口移除、兼容 API 稳定拒绝	Tasks 4-5、11-12
PDA 模式切换与五类现场业务	Task 12
备份恢复、结构盘点、压测阈值、发布顺序	Task 13

Completion Gate

13 个任务的测试命令均在各自提交前通过。
后端全量业务测试、ruff format/check、Web/PDA 契约和两个生产构建均通过。
MySQL/PostgreSQL 全量 SQL 与补丁契约一致，既有数据保持 test_batch_id = NULL。
源码不存在旧 remove 权限入口或业务 ON DELETE CASCADE。
未执行生产迁移、真实数据纳入或真实批次清理。
TEST_BATCH_ONLINE_PURGE_MAX_ROWS 在目标环境完成压测前保持 0。

105 KiB Raw Blame History Unescape Escape

测试数据批次隔离与级联清理实施计划

Global Constraints

Task 1: 数据模型、VO 与双数据库迁移契约

Task 2: 受信任批次上下文与超级管理员硬门禁

Task 3: 批次生命周期与用户授权服务

Task 4: 六类主数据的隔离、继承与旧删除封闭

Task 5: 五类业务流程的批次一致性与自动继承

Task 6: 报表、选项、扫码与首页数据隔离

Task 7: 旧数据双向关联闭包与纳入事务

Task 8: 清理预览、外键顺序、原子回滚与幂等

Task 9: Web 批次 Store、请求头和模式切换

Task 10: Web 批次管理页与“纳入测试批次”操作

Task 11: 移除物理删除入口并收紧权限契约

Task 12: PDA 测试模式切换与请求上下文

Task 13: 全量回归、发布防护与运维说明

Spec Coverage Map

Completion Gate

105 KiB

Raw Blame History