diff --git a/backend/app/api/v1/base/personnel.py b/backend/app/api/v1/base/personnel.py index 9b305c2..2210514 100644 --- a/backend/app/api/v1/base/personnel.py +++ b/backend/app/api/v1/base/personnel.py @@ -137,5 +137,7 @@ def replace_plans( meta: RequestMetaDep, ) -> ApiResponse[None]: """全量替换拓展计划，空列表即清空。""" - personnel_service.replace_plans(session, personnel_id, payload, operator=current_user, meta=meta) + personnel_service.replace_plans( + session, personnel_id, payload, operator=current_user, meta=meta + ) return ok(message="已保存") diff --git a/backend/app/api/v1/base/products.py b/backend/app/api/v1/base/products.py index 7b12e8e..fce3f27 100644 --- a/backend/app/api/v1/base/products.py +++ b/backend/app/api/v1/base/products.py @@ -97,9 +97,7 @@ def delete_product( return ok(message="产品已删除") -@router.put( - "/{product_id}/status", response_model=ApiResponse[ProductOut], summary="启用/停用产品" -) +@router.put("/{product_id}/status", response_model=ApiResponse[ProductOut], summary="启用/停用产品") def set_status( session: DbSession, current_user: CurrentUser, diff --git a/backend/app/repositories/product_repository.py b/backend/app/repositories/product_repository.py index a574538..b399362 100644 --- a/backend/app/repositories/product_repository.py +++ b/backend/app/repositories/product_repository.py @@ -21,9 +21,7 @@ def get_by_id(session: Session, product_id: int, *, for_update: bool = False) -> return session.get(Product, product_id) -def get_by_name_spec( - session: Session, product_name: str, spec_model: str -) -> Product | None: +def get_by_name_spec(session: Session, product_name: str, spec_model: str) -> Product | None: """按 (名称, 规格型号) 组合查询。 组合是 uk_products_name_spec 唯一键的键值，先查用于给出即时业务提示； diff --git a/backend/app/schemas/personnel.py b/backend/app/schemas/personnel.py index 12032cc..b2cfa3f 100644 --- a/backend/app/schemas/personnel.py +++ b/backend/app/schemas/personnel.py @@ -70,7 +70,9 @@ class PersonnelCreateIn(BaseModel): phone: str | None = Field(default=None, max_length=20, description="联系电话") hire_date: date | None = Field(default=None, description="入职日期，ISO 格式") is_probation: int = Field(default=0, ge=0, le=1, description="是否试用期：1是 0否") - salary_standard: str | None = Field(default=None, description="月工资标准（元），字符串形式的数字") + salary_standard: str | None = Field( + default=None, description="月工资标准（元），字符串形式的数字" + ) remark: str | None = Field(default=None, max_length=255) @field_validator("salary_standard") diff --git a/backend/app/schemas/product.py b/backend/app/schemas/product.py index f109341..67e6639 100644 --- a/backend/app/schemas/product.py +++ b/backend/app/schemas/product.py @@ -58,7 +58,9 @@ class ProductCreateIn(BaseModel): model_config = ConfigDict(str_strip_whitespace=True) product_name: str = Field(min_length=1, max_length=128, description="产品名称") - spec_model: str | None = Field(default=None, max_length=128, description="规格型号，空归一为空串") + spec_model: str | None = Field( + default=None, max_length=128, description="规格型号，空归一为空串" + ) unit: str | None = Field(default=None, max_length=16, description="计量单位") stock_quantity: str | None = Field(default=None, description="数量，非负数字字符串") sale_price: str | None = Field(default=None, description="销售单价（元），非负数字字符串") diff --git a/backend/app/services/customer_service.py b/backend/app/services/customer_service.py index c4c89ad..c4da6e7 100644 --- a/backend/app/services/customer_service.py +++ b/backend/app/services/customer_service.py @@ -166,10 +166,6 @@ def create_customer( :raises BadRequestError: 客户分类不合法 / 归属人员不存在或已离职 / 「公司名称+负责人」已存在（含并发兜底） """ - _ensure_customer_type(payload.customer_type) - owner_name: str | None = None - if payload.owner_personnel_id is not None: - owner_name = _ensure_owner_personnel(session, payload.owner_personnel_id).personnel_name with audit_service.audit( session, operator=operator, @@ -181,6 +177,12 @@ def create_customer( ), meta=meta, ): + # 值域与归属校验在审计上下文内：失败（400）也要留痕 —— + # 与 personnel_service._validate_refs 移入上下文的口径一致 + _ensure_customer_type(payload.customer_type) + owner_name: str | None = None + if payload.owner_personnel_id is not None: + owner_name = _ensure_owner_personnel(session, payload.owner_personnel_id).personnel_name if ( customer_repository.get_by_company_owner( session, payload.company_name, payload.owner_name @@ -233,7 +235,6 @@ def update_customer( :raises NotFoundError: 客户不存在 :raises BadRequestError: 客户分类不合法 / 「公司名称+负责人」已存在（含并发兜底） """ - _ensure_customer_type(payload.customer_type) # 只读预取在审计上下文之前：目标不存在（NotFound 失败）时也能留下可读记录， # 摘要与回包都要用客户行；归属姓名本端点不变更，一并带出 row = customer_repository.get_with_owner_name(session, customer_id) @@ -249,6 +250,8 @@ def update_customer( ), meta=meta, ): + # 值域校验在审计上下文内：分类非法的 400 也要留痕（同 create 口径） + _ensure_customer_type(payload.customer_type) if row is None: raise NotFoundError("客户不存在") customer, owner_name = row diff --git a/backend/app/services/personnel_service.py b/backend/app/services/personnel_service.py index f58e91d..2a06778 100644 --- a/backend/app/services/personnel_service.py +++ b/backend/app/services/personnel_service.py @@ -67,7 +67,10 @@ def list_personnel(session: Session, query: PersonnelQuery) -> PageResult[Person business_role_id=query.business_role_id, ) return PageResult[PersonnelOut]( - records=[_to_out(personnel, role_name, manager_name) for personnel, role_name, manager_name in rows], + records=[ + _to_out(personnel, role_name, manager_name) + for personnel, role_name, manager_name in rows + ], total=total, page_num=query.page_num, page_size=query.page_size, @@ -107,7 +110,6 @@ def create_personnel( :raises BadRequestError: 岗位不存在或已停用 / 直属经理不合法 """ - role, manager_name = _validate_refs(session, payload.business_role_id, payload.manager_id, None) with audit_service.audit( session, operator=operator, @@ -119,6 +121,12 @@ def create_personnel( ), meta=meta, ): + # 引用校验放在审计上下文内：失败同样要留痕（与 update_personnel 口径 + # 一致）——「谁在何时试图把人挂到停用岗位/离职经理上」是审计对象， + # 校验留在上下文外会让这类失败操作无据可查 + role, manager_name = _validate_refs( + session, payload.business_role_id, payload.manager_id, None + ) personnel = Personnel( personnel_name=payload.personnel_name, business_role_id=payload.business_role_id, @@ -221,10 +229,9 @@ def delete_personnel( raise NotFoundError("人员不存在") ids = [personnel_id] - if ( - personnel_repository.count_team_members(session, ids) - or personnel_repository.count_customer_ownerships(session, ids) - ): + if personnel_repository.count_team_members( + session, ids + ) or personnel_repository.count_customer_ownerships(session, ids): raise BadRequestError(_DELETE_BLOCKED) name = personnel.personnel_name # 固化：对象随删除失效 @@ -302,7 +309,9 @@ def replace_regions( if row is None: raise NotFoundError("人员不存在") rows = [ - PersonnelRegion(personnel_id=personnel_id, region_name=item.region_name, remark=item.remark) + PersonnelRegion( + personnel_id=personnel_id, region_name=item.region_name, remark=item.remark + ) for item in payload.items ] personnel_repository.replace_regions(session, personnel_id, rows) diff --git a/backend/app/services/product_service.py b/backend/app/services/product_service.py index 2088906..fbb3cce 100644 --- a/backend/app/services/product_service.py +++ b/backend/app/services/product_service.py @@ -146,7 +146,6 @@ def create_product( :raises BadRequestError: 产品类别不合法 / 已存在同名称同规格的产品 """ - _ensure_category(payload.product_category) with audit_service.audit( session, operator=operator, @@ -158,8 +157,13 @@ def create_product( ), meta=meta, ): + # 值域校验在审计上下文内：类别非法的 400 也要留痕（成功与失败都记） + _ensure_category(payload.product_category) spec_model = payload.spec_model or "" # 归一空串：组合唯一键对 NULL 不去重 - if product_repository.get_by_name_spec(session, payload.product_name, spec_model) is not None: + if ( + product_repository.get_by_name_spec(session, payload.product_name, spec_model) + is not None + ): raise BadRequestError("已存在同名称同规格的产品") product = Product( @@ -182,9 +186,7 @@ def create_product( session.rollback() _raise_if_duplicate(exc) - logger.info( - "新增产品 | product_id=%s | category=%s", product.id, payload.product_category - ) + logger.info("新增产品 | product_id=%s | category=%s", product.id, payload.product_category) return out @@ -202,7 +204,6 @@ def update_product( :raises BadRequestError: 产品类别不合法 / 已存在同名称同规格的产品 """ product = product_repository.get_by_id(session, product_id) - _ensure_category(payload.product_category) with audit_service.audit( session, operator=operator, @@ -214,6 +215,8 @@ def update_product( ), meta=meta, ): + # 值域校验在审计上下文内：类别非法的 400 也要留痕（同 create 口径） + _ensure_category(payload.product_category) if product is None: raise NotFoundError("产品不存在") diff --git a/backend/tests/integration/conftest.py b/backend/tests/integration/conftest.py index be9fe07..6323f94 100644 --- a/backend/tests/integration/conftest.py +++ b/backend/tests/integration/conftest.py @@ -7,6 +7,9 @@ - 测试账号/角色统一 ``t_`` 前缀；工厂夹具清理自己创建的对象，另有 autouse 兜底按前缀全量清理（用例经**接口**创建的数据不在工厂名下， 失败中断时会残留并影响后续用例的 total 断言） +- M2 基础数据（人员姓名/产品名/公司名/测试岗位编码）同用 ``t_`` 前缀， + 兜底清理按前缀删除；归属留痕对客户/人员 ID 无外键固化、不随主表删除 + 消失，必须显式清理，否则残留会影响后续用例对留痕条数的断言 - 操作日志的清理同时按 username 与 user_id —— 账号被删除后其 ``user_id`` 会被 SET NULL，只能靠冗余的 username 追溯，清理逻辑必须覆盖两种状态 """ @@ -14,12 +17,24 @@ import uuid import pytest -from sqlalchemy import delete, select +from sqlalchemy import delete, select, update from app.core.constants import ROLE_ADMIN, ROLE_USER from app.core.db import get_session_factory from app.core.security import create_access_token, hash_password -from app.models import OperationLog, Role, User, UserRole +from app.models import ( + BusinessRole, + Customer, + OperationLog, + OwnershipChangeLog, + Personnel, + PersonnelPlan, + PersonnelRegion, + Product, + Role, + User, + UserRole, +) TEST_PASSWORD = "TestPwd@2026" # 模块级预计算一次：bcrypt（rounds=12）单次约 0.2–0.4 秒，逐用例重复哈希 @@ -53,6 +68,45 @@ def _purge_test_data() -> None: if role_ids: session.execute(delete(UserRole).where(UserRole.role_id.in_(role_ids))) session.execute(delete(Role).where(Role.id.in_(role_ids))) + + # ---- M2 基础数据：客户/归属留痕/人员/产品/测试岗位 ---- + # 外键均为 RESTRICT，删除顺序不能倒：留痕无外键先按 customer_id 清， + # 再删客户（子表）→ 人员（先置空 manager_id 自引用、清子表）→ 产品 + personnel_ids = list( + session.execute( + select(Personnel.id).where(Personnel.personnel_name.like("t\\_%")) + ).scalars() + ) + customer_ids = list( + session.execute( + select(Customer.id).where(Customer.company_name.like("t\\_%")) + ).scalars() + ) + if customer_ids: + session.execute( + delete(OwnershipChangeLog).where(OwnershipChangeLog.customer_id.in_(customer_ids)) + ) + session.execute(delete(Customer).where(Customer.id.in_(customer_ids))) + # 客户被物理删除后其 customer_id 已无从收集，按操作人用户名兜底一条 + session.execute( + delete(OwnershipChangeLog).where(OwnershipChangeLog.operator_name.like("t\\_%")) + ) + if personnel_ids: + session.execute( + update(Personnel) + .where(Personnel.manager_id.in_(personnel_ids)) + .values(manager_id=None) + ) + session.execute( + delete(PersonnelRegion).where(PersonnelRegion.personnel_id.in_(personnel_ids)) + ) + session.execute( + delete(PersonnelPlan).where(PersonnelPlan.personnel_id.in_(personnel_ids)) + ) + session.execute(delete(Personnel).where(Personnel.id.in_(personnel_ids))) + session.execute(delete(Product).where(Product.product_name.like("t\\_%"))) + # 人员已先删（business_role_id 外键 RESTRICT），直建的测试岗位此时可安全删 + session.execute(delete(BusinessRole).where(BusinessRole.role_code.like("t\\_%"))) session.commit() diff --git a/backend/tests/integration/test_customers.py b/backend/tests/integration/test_customers.py new file mode 100644 index 0000000..f290f9f --- /dev/null +++ b/backend/tests/integration/test_customers.py @@ -0,0 +1,352 @@ +"""客户管理接口集成测试（需要 MySQL 与 Redis）。 + +守的是客户归属的一致性 —— 归属变更必须「主表更新 + 留痕新增」同事务 +完成，拆开就会出现「主表已换人、留痕缺失」的静默不一致，事后追溯无据； +编辑端点若能触归属，就会出现「改了归属却无留痕」的旁路，破坏留痕的 +append-only 完整性；「公司名称+负责人」组合唯一缺失会放行重复客户。 + +运行前置：``docker compose -f docker/docker-compose.dev.yml up -d`` + +数据卫生：公司名与归属人员姓名一律 ``t_`` 前缀，由 conftest 兜底清理。 +""" + +import uuid + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import select + +from app.core.constants import ( + ACTION_CREATE, + BUSINESS_ROLE_SALESPERSON, + CUSTOMER_TYPE_ENTERPRISE, + CUSTOMER_TYPE_PERSONAL, + MODULE_CUSTOMER, +) +from app.core.db import get_session_factory +from app.models import BusinessRole, OperationLog, OwnershipChangeLog +from app.models.customer import OWNER_TYPE_CUSTOMER + +pytestmark = pytest.mark.integration + +_PERSONNEL = "/api/v1/base/personnel" +_CUSTOMERS = "/api/v1/base/customers" + + +def _company() -> str: + return f"t_公司_{uuid.uuid4().hex[:8]}" + + +def _role_id() -> int: + """种子岗位（营销员）：归属人员的合法岗位引用。""" + factory = get_session_factory() + with factory() as session: + return int( + session.execute( + select(BusinessRole.id).where( + BusinessRole.role_code == BUSINESS_ROLE_SALESPERSON, + BusinessRole.status == 1, + ) + ).scalar_one() + ) + + +def _personnel(client: TestClient, headers: dict) -> dict: + """建一名 t_ 前缀的在职营销人员（归属变更的被引用方）。""" + resp = client.post( + _PERSONNEL, + json={ + "personnel_name": f"t_人员_{uuid.uuid4().hex[:8]}", + "business_role_id": _role_id(), + }, + headers=headers, + ) + assert resp.status_code == 200, resp.text + return resp.json()["data"] + + +def _cpayload(company: str, **overrides: object) -> dict: + body: dict = { + "company_name": company, + "owner_name": "张老板", + "customer_type": CUSTOMER_TYPE_PERSONAL, + "address": "本溪市平山区", + } + body.update(overrides) + return body + + +def _create_customer(client: TestClient, headers: dict, **overrides: object) -> dict: + resp = client.post(_CUSTOMERS, json=_cpayload(_company(), **overrides), headers=headers) + assert resp.status_code == 200, resp.text + return resp.json()["data"] + + +def _logs_of(username: str) -> list[OperationLog]: + factory = get_session_factory() + with factory() as session: + return list( + session.execute( + select(OperationLog) + .where(OperationLog.username == username) + .order_by(OperationLog.id.desc()) + ).scalars() + ) + + +def _latest_log(customer_id: int) -> OwnershipChangeLog | None: + """取该客户最新一条归属留痕（按 id 倒序）。""" + factory = get_session_factory() + with factory() as session: + return ( + session.execute( + select(OwnershipChangeLog) + .where(OwnershipChangeLog.customer_id == customer_id) + .order_by(OwnershipChangeLog.id.desc()) + ) + .scalars() + .first() + ) + + +class TestAuth: + """鉴权：登录后接口，裸访问一律 401。""" + + def test_未登录访问客户列表_401(self, client: TestClient) -> None: + resp = client.get(_CUSTOMERS) + assert resp.status_code == 401, resp.text + + +class TestCrudFlow: + """正向链路与组合唯一 ——「公司名称+负责人」是业务唯一标识。""" + + def test_新增客户带初始归属_回显归属人员姓名(self, client: TestClient, normal: dict) -> None: + owner = _personnel(client, normal["headers"]) + data = _create_customer(client, normal["headers"], owner_personnel_id=owner["id"]) + assert data["owner_personnel_id"] == owner["id"] + assert data["owner_personnel_name"] == owner["personnel_name"] + assert data["customer_type"] == CUSTOMER_TYPE_PERSONAL + + def test_同公司名不同负责人创建成功(self, client: TestClient, normal: dict) -> None: + company = _company() + first = _create_customer(client, normal["headers"], company_name=company) + second = client.post( + _CUSTOMERS, + json=_cpayload(company, owner_name="李老板"), + headers=normal["headers"], + ) + assert second.status_code == 200, second.text + assert second.json()["data"]["id"] != first["id"] + + def test_重名同负责人_400(self, client: TestClient, normal: dict) -> None: + company = _company() + _create_customer(client, normal["headers"], company_name=company) + dup = client.post( + _CUSTOMERS, + json=_cpayload(company, owner_name="张老板"), + headers=normal["headers"], + ) + assert dup.status_code == 400, dup.text + assert "该公司名称与负责人的客户已存在" in dup.json()["message"] + + def test_客户分类非法_400(self, client: TestClient, normal: dict) -> None: + resp = client.post( + _CUSTOMERS, + json=_cpayload(_company(), customer_type="not_a_type"), + headers=normal["headers"], + ) + assert resp.status_code == 400, resp.text + assert "客户分类不合法" in resp.json()["message"] + + def test_编辑其他字段不触归属_塞归属字段也被忽略( + self, client: TestClient, normal: dict + ) -> None: + """归属只有 /ownership 一条写路径 —— 编辑端点防「改归属不留痕」旁路。""" + headers = normal["headers"] + owner = _personnel(client, headers) + stranger = _personnel(client, headers) + customer = _create_customer(client, headers, owner_personnel_id=owner["id"]) + + updated = client.put( + f"{_CUSTOMERS}/{customer['id']}", + json={ + "company_name": customer["company_name"], + "owner_name": customer["owner_name"], + "customer_type": CUSTOMER_TYPE_ENTERPRISE, + "address": "沈阳市沈河区", + # 试图经编辑端点改归属：schema 未声明该字段，必须被忽略 + "ownerPersonnelId": stranger["id"], + }, + headers=headers, + ) + assert updated.status_code == 200, updated.text + body = updated.json()["data"] + assert body["customer_type"] == CUSTOMER_TYPE_ENTERPRISE + assert body["address"] == "沈阳市沈河区" + assert body["owner_personnel_id"] == owner["id"], "编辑端点不得改动归属" + assert body["owner_personnel_name"] == owner["personnel_name"] + + detail = client.get(f"{_CUSTOMERS}/{customer['id']}", headers=headers) + assert detail.json()["data"]["owner_personnel_id"] == owner["id"] + assert _latest_log(int(customer["id"])) is None, "编辑不应产生归属留痕" + + +class TestOwnership: + """归属变更：主表更新与留痕同事务；校验失败不留脏数据。""" + + def test_归属变更_主表更新且留痕新增(self, client: TestClient, normal: dict) -> None: + headers = normal["headers"] + old = _personnel(client, headers) + new = _personnel(client, headers) + customer = _create_customer(client, headers, owner_personnel_id=old["id"]) + + resp = client.put( + f"{_CUSTOMERS}/{customer['id']}/ownership", + json={"new_owner_personnel_id": new["id"], "reason": "区域调整"}, + headers=headers, + ) + assert resp.status_code == 200, resp.text + body = resp.json()["data"] + assert body["owner_personnel_id"] == new["id"] + assert body["owner_personnel_name"] == new["personnel_name"] + + log = _latest_log(int(customer["id"])) + assert log is not None, "归属变更未写留痕" + assert log.owner_type == OWNER_TYPE_CUSTOMER + assert log.customer_id == int(customer["id"]) + assert log.old_owner_personnel_id == int(old["id"]) + assert log.new_owner_personnel_id == int(new["id"]) + assert log.change_reason == "区域调整" + assert log.operator_name == normal["username"] + + def test_变更后归属人员已离职_400(self, client: TestClient, normal: dict) -> None: + headers = normal["headers"] + target = _personnel(client, headers) + client.put(f"{_PERSONNEL}/{target['id']}/status", json={"status": 0}, headers=headers) + customer = _create_customer(client, headers) + resp = client.put( + f"{_CUSTOMERS}/{customer['id']}/ownership", + json={"new_owner_personnel_id": target["id"]}, + headers=headers, + ) + assert resp.status_code == 400, resp.text + assert "归属人员不存在或已离职" in resp.json()["message"] + assert _latest_log(int(customer["id"])) is None, "失败变更不应留痕" + + def test_新旧归属相同_400(self, client: TestClient, normal: dict) -> None: + headers = normal["headers"] + owner = _personnel(client, headers) + customer = _create_customer(client, headers, owner_personnel_id=owner["id"]) + resp = client.put( + f"{_CUSTOMERS}/{customer['id']}/ownership", + json={"new_owner_personnel_id": owner["id"]}, + headers=headers, + ) + assert resp.status_code == 400, resp.text + assert "归属未变化" in resp.json()["message"] + + def test_双方均为空_400(self, client: TestClient, normal: dict) -> None: + """无归属客户再清一次归属也属「未变化」——不产生空转留痕。""" + customer = _create_customer(client, normal["headers"]) + resp = client.put( + f"{_CUSTOMERS}/{customer['id']}/ownership", + json={"new_owner_personnel_id": None}, + headers=normal["headers"], + ) + assert resp.status_code == 400, resp.text + assert "归属未变化" in resp.json()["message"] + + def test_变更为空即清空归属_主表与留痕一致(self, client: TestClient, normal: dict) -> None: + headers = normal["headers"] + owner = _personnel(client, headers) + customer = _create_customer(client, headers, owner_personnel_id=owner["id"]) + resp = client.put( + f"{_CUSTOMERS}/{customer['id']}/ownership", + json={"new_owner_personnel_id": None, "reason": "暂无负责人"}, + headers=headers, + ) + assert resp.status_code == 200, resp.text + body = resp.json()["data"] + assert body["owner_personnel_id"] is None + assert body["owner_personnel_name"] is None + + log = _latest_log(int(customer["id"])) + assert log is not None + assert log.old_owner_personnel_id == int(owner["id"]) + assert log.new_owner_personnel_id is None + assert log.change_reason == "暂无负责人" + + +class TestOwnershipLogs: + """留痕分页：新旧归属人姓名由联表带出，倒序可追溯变更轨迹。""" + + def test_留痕分页带新旧归属人姓名并倒序(self, client: TestClient, normal: dict) -> None: + headers = normal["headers"] + first = _personnel(client, headers) + second = _personnel(client, headers) + customer = _create_customer(client, headers, owner_personnel_id=first["id"]) + + r1 = client.put( + f"{_CUSTOMERS}/{customer['id']}/ownership", + json={"new_owner_personnel_id": second["id"], "reason": "第一次调整"}, + headers=headers, + ) + assert r1.status_code == 200, r1.text + r2 = client.put( + f"{_CUSTOMERS}/{customer['id']}/ownership", + json={"new_owner_personnel_id": None, "reason": "清空归属"}, + headers=headers, + ) + assert r2.status_code == 200, r2.text + + resp = client.get(f"{_CUSTOMERS}/{customer['id']}/ownership-logs", headers=headers) + assert resp.status_code == 200, resp.text + body = resp.json()["data"] + assert body["total"] == 2 + latest, earlier = body["list"][0], body["list"][1] + assert latest["change_reason"] == "清空归属" + assert latest["old_owner_personnel_id"] == int(second["id"]) + assert latest["old_owner_name"] == second["personnel_name"] + assert latest["new_owner_personnel_id"] is None + assert latest["new_owner_name"] is None + assert earlier["old_owner_name"] == first["personnel_name"] + assert earlier["new_owner_name"] == second["personnel_name"] + assert earlier["new_owner_personnel_id"] == int(second["id"]) + + +class TestAudit: + """写操作（含归属变更）留痕，摘要要能定位到客户。""" + + def test_新增客户留审计(self, client: TestClient, normal: dict) -> None: + data = _create_customer(client, normal["headers"]) + logs = [ + log + for log in _logs_of(normal["username"]) + if log.module == MODULE_CUSTOMER and log.action == ACTION_CREATE + ] + assert logs, "新增客户未留操作日志" + latest = logs[0] + assert latest.success == 1 + assert data["company_name"] in latest.summary + + def test_归属变更审计摘要含客户名(self, client: TestClient, normal: dict) -> None: + headers = normal["headers"] + old = _personnel(client, headers) + new = _personnel(client, headers) + customer = _create_customer(client, headers, owner_personnel_id=old["id"]) + resp = client.put( + f"{_CUSTOMERS}/{customer['id']}/ownership", + json={"new_owner_personnel_id": new["id"], "reason": "区域调整"}, + headers=headers, + ) + assert resp.status_code == 200, resp.text + + logs = [ + log + for log in _logs_of(normal["username"]) + if log.module == MODULE_CUSTOMER and log.success == 1 + ] + assert logs, "归属变更未留操作日志" + latest = logs[0] + assert customer["company_name"] in latest.summary, "摘要应含客户名" + assert "→" in latest.summary, "摘要应体现新旧归属走向" diff --git a/backend/tests/integration/test_personnel.py b/backend/tests/integration/test_personnel.py new file mode 100644 index 0000000..7e2d4a2 --- /dev/null +++ b/backend/tests/integration/test_personnel.py @@ -0,0 +1,400 @@ +"""人员管理接口集成测试（需要 MySQL 与 Redis）。 + +守的是人员档案的引用完整性 —— 岗位与直属经理是 M3 核算取公式模板与 +经理聚合展开的锚点，悬空引用或环引用会让核算取不到公式、聚合死循环； +删除保护缺失会留下悬空的 manager_id 与客户归属，历史核算无法回溯到人； +辐射区域与拓展计划若把「全量替换」做成了追加，编辑一次就翻倍一份脏数据。 + +运行前置：``docker compose -f docker/docker-compose.dev.yml up -d`` + +数据卫生：人员姓名一律 ``t_`` 前缀，由 conftest 兜底清理；仅「删除」 +语义的用例自验 DELETE 端点。 +""" + +import uuid + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import select + +from app.core.constants import ( + ACTION_CREATE, + BUSINESS_ROLE_SALESPERSON, + MODULE_PERSONNEL, +) +from app.core.db import get_session_factory +from app.models import BusinessRole, OperationLog + +pytestmark = pytest.mark.integration + +_PERSONNEL = "/api/v1/base/personnel" +_CUSTOMERS = "/api/v1/base/customers" + + +def _uname() -> str: + return f"t_人员_{uuid.uuid4().hex[:8]}" + + +def _role_id() -> int: + """种子岗位（营销员，内置且启用）：人员归属岗位的合法引用。""" + factory = get_session_factory() + with factory() as session: + return int( + session.execute( + select(BusinessRole.id).where( + BusinessRole.role_code == BUSINESS_ROLE_SALESPERSON, + BusinessRole.status == 1, + ) + ).scalar_one() + ) + + +def _payload(name: str, **overrides: object) -> dict: + body: dict = { + "personnel_name": name, + "business_role_id": _role_id(), + "salary_standard": "3500.50", + } + body.update(overrides) + return body + + +def _create(client: TestClient, headers: dict, **overrides: object) -> dict: + resp = client.post(_PERSONNEL, json=_payload(_uname(), **overrides), headers=headers) + assert resp.status_code == 200, resp.text + return resp.json()["data"] + + +def _logs_of(username: str) -> list[OperationLog]: + """取某操作人的审计行（按 id 倒序，最新在前）。""" + factory = get_session_factory() + with factory() as session: + return list( + session.execute( + select(OperationLog) + .where(OperationLog.username == username) + .order_by(OperationLog.id.desc()) + ).scalars() + ) + + +class TestAuth: + """鉴权：基础数据是业务操作不要求管理员，但仍是登录后接口。""" + + def test_未登录访问人员列表_401(self, client: TestClient) -> None: + resp = client.get(_PERSONNEL) + assert resp.status_code == 401, resp.text + + +class TestCrudFlow: + """正向链路：建 → 查（列表/详情）→ 改 → 状态切换 → 删。""" + + def test_新增人员默认在职且工资标准字符串直出(self, client: TestClient, normal: dict) -> None: + data = _create(client, normal["headers"]) + assert data["status"] == 1, "新增人员应默认在职" + assert data["is_probation"] == 0 + # 金额红线：str 进 str 出，不得经 float + assert data["salary_standard"] == "3500.50" + + detail = client.get(f"{_PERSONNEL}/{data['id']}", headers=normal["headers"]) + assert detail.status_code == 200, detail.text + body = detail.json()["data"] + assert body["salary_standard"] == "3500.50" + assert body["regions"] == [] + assert body["plans"] == [] + + def test_新增人员携带直属经理_回显经理与岗位名(self, client: TestClient, normal: dict) -> None: + headers = normal["headers"] + manager = _create(client, headers) + member = _create(client, headers, manager_id=manager["id"]) + assert member["manager_id"] == manager["id"] + assert member["manager_name"] == manager["personnel_name"] + assert member["role_name"] == manager["role_name"] + + def test_列表按姓名关键词可查到新增人员(self, client: TestClient, normal: dict) -> None: + data = _create(client, normal["headers"]) + listed = client.get( + _PERSONNEL, params={"keyword": data["personnel_name"]}, headers=normal["headers"] + ) + assert listed.status_code == 200, listed.text + rows = listed.json()["data"]["list"] + assert [r["id"] for r in rows] == [data["id"]] + + def test_编辑人员全量替换生效(self, client: TestClient, normal: dict) -> None: + data = _create(client, normal["headers"]) + updated = client.put( + f"{_PERSONNEL}/{data['id']}", + json=_payload(f"t_人员_改_{uuid.uuid4().hex[:6]}", phone="13900001111", is_probation=1), + headers=normal["headers"], + ) + assert updated.status_code == 200, updated.text + body = updated.json()["data"] + assert body["phone"] == "13900001111" + assert body["is_probation"] == 1 + + def test_在职离职双向切换(self, client: TestClient, normal: dict) -> None: + headers = normal["headers"] + data = _create(client, headers) + off = client.put(f"{_PERSONNEL}/{data['id']}/status", json={"status": 0}, headers=headers) + assert off.status_code == 200, off.text + assert off.json()["data"]["status"] == 0 + on = client.put(f"{_PERSONNEL}/{data['id']}/status", json={"status": 1}, headers=headers) + assert on.status_code == 200, on.text + assert on.json()["data"]["status"] == 1 + + def test_删除无引用人员成功(self, client: TestClient, normal: dict) -> None: + data = _create(client, normal["headers"]) + resp = client.delete(f"{_PERSONNEL}/{data['id']}", headers=normal["headers"]) + assert resp.status_code == 200, resp.text + detail = client.get(f"{_PERSONNEL}/{data['id']}", headers=normal["headers"]) + assert detail.status_code == 404 + assert "人员不存在" in detail.json()["message"] + + +class TestRefRules: + """岗位/经理引用校验 —— 核算锚点不悬空、不环引用。""" + + def test_编辑时直属经理是自己_400(self, client: TestClient, normal: dict) -> None: + data = _create(client, normal["headers"]) + resp = client.put( + f"{_PERSONNEL}/{data['id']}", + json=_payload(data["personnel_name"], manager_id=data["id"]), + headers=normal["headers"], + ) + assert resp.status_code == 400, resp.text + assert "直属经理不能是自己" in resp.json()["message"] + + def test_直属经理不存在_400(self, client: TestClient, normal: dict) -> None: + resp = client.post( + _PERSONNEL, + json=_payload(_uname(), manager_id=999999999), + headers=normal["headers"], + ) + assert resp.status_code == 400, resp.text + assert "直属经理不存在或已离职" in resp.json()["message"] + + def test_直属经理已离职_400(self, client: TestClient, normal: dict) -> None: + headers = normal["headers"] + manager = _create(client, headers) + quit_resp = client.put( + f"{_PERSONNEL}/{manager['id']}/status", json={"status": 0}, headers=headers + ) + assert quit_resp.status_code == 200, quit_resp.text + resp = client.post( + _PERSONNEL, + json=_payload(_uname(), manager_id=manager["id"]), + headers=headers, + ) + assert resp.status_code == 400, resp.text + assert "直属经理不存在或已离职" in resp.json()["message"] + + def test_岗位不存在_400(self, client: TestClient, normal: dict) -> None: + resp = client.post( + _PERSONNEL, + json=_payload(_uname(), business_role_id=999999999), + headers=normal["headers"], + ) + assert resp.status_code == 400, resp.text + assert "岗位不存在或已停用" in resp.json()["message"] + + def test_岗位已停用_400(self, client: TestClient, normal: dict) -> None: + """停用岗位不能挂人：岗位决定计提模板，挂上去核算会取不到公式。""" + code = f"t_br_{uuid.uuid4().hex[:8]}" + factory = get_session_factory() + with factory() as session: + session.add( + BusinessRole( + role_code=code, + role_name="已停用测试岗位", + calc_template="TPL_SALESPERSON_DECAY", + status=0, + ) + ) + session.commit() + role_id = int( + session.execute( + select(BusinessRole.id).where(BusinessRole.role_code == code) + ).scalar_one() + ) + resp = client.post( + _PERSONNEL, + json=_payload(_uname(), business_role_id=role_id), + headers=normal["headers"], + ) + assert resp.status_code == 400, resp.text + assert "岗位不存在或已停用" in resp.json()["message"] + + +class TestDeleteProtection: + """删除保护：被团队成员/客户归属引用的人员禁删 —— 历史要能回溯到人。""" + + def test_被团队成员引用禁删_解除后可删(self, client: TestClient, normal: dict) -> None: + headers = normal["headers"] + manager = _create(client, headers) + member = _create(client, headers, manager_id=manager["id"]) # 团队成员占用 + blocked = client.delete(f"{_PERSONNEL}/{manager['id']}", headers=headers) + assert blocked.status_code == 400, blocked.text + assert "存在团队成员或客户归属引用，请改为离职而非删除" in blocked.json()["message"] + + client.delete(f"{_PERSONNEL}/{member['id']}", headers=headers) + freed = client.delete(f"{_PERSONNEL}/{manager['id']}", headers=headers) + assert freed.status_code == 200, freed.text + + def test_被客户归属引用禁删_离职不放宽_解除后可删( + self, client: TestClient, normal: dict + ) -> None: + headers = normal["headers"] + owner = _create(client, headers) + created = client.post( + _CUSTOMERS, + json={ + "company_name": f"t_公司_{uuid.uuid4().hex[:8]}", + "owner_name": "张老板", + "customer_type": "personal", + "owner_personnel_id": owner["id"], + }, + headers=headers, + ) + assert created.status_code == 200, created.text + customer = created.json()["data"] + + blocked = client.delete(f"{_PERSONNEL}/{owner['id']}", headers=headers) + assert blocked.status_code == 400, blocked.text + assert "存在团队成员或客户归属引用，请改为离职而非删除" in blocked.json()["message"] + + # 保护与在职状态无关：离职并不能让物理删除放行 + client.put(f"{_PERSONNEL}/{owner['id']}/status", json={"status": 0}, headers=headers) + still = client.delete(f"{_PERSONNEL}/{owner['id']}", headers=headers) + assert still.status_code == 400, still.text + + # 删客户解除引用后方可删人 + assert client.delete(f"{_CUSTOMERS}/{customer['id']}", headers=headers).status_code == 200 + assert client.delete(f"{_PERSONNEL}/{owner['id']}", headers=headers).status_code == 200 + + +class TestSubTableReplace: + """辐射区域/拓展计划为全量替换语义 —— 做成追加会让编辑翻倍脏数据。""" + + def test_辐射区域设置后详情回显一致_再次替换生效( + self, client: TestClient, normal: dict + ) -> None: + headers = normal["headers"] + data = _create(client, headers) + resp = client.put( + f"{_PERSONNEL}/{data['id']}/regions", + json={ + "items": [ + {"region_name": "本溪市平山区", "remark": "核心区域"}, + {"region_name": "本溪市溪湖区"}, + ] + }, + headers=headers, + ) + assert resp.status_code == 200, resp.text + detail = client.get(f"{_PERSONNEL}/{data['id']}", headers=headers).json()["data"] + assert [r["region_name"] for r in detail["regions"]] == ["本溪市平山区", "本溪市溪湖区"] + assert detail["regions"][0]["remark"] == "核心区域" + + # 再替换为 1 条：是覆盖而非追加 + resp2 = client.put( + f"{_PERSONNEL}/{data['id']}/regions", + json={"items": [{"region_name": "沈阳市沈河区"}]}, + headers=headers, + ) + assert resp2.status_code == 200, resp2.text + detail2 = client.get(f"{_PERSONNEL}/{data['id']}", headers=headers).json()["data"] + assert [r["region_name"] for r in detail2["regions"]] == ["沈阳市沈河区"] + + def test_辐射区域替换为空列表即清空(self, client: TestClient, normal: dict) -> None: + headers = normal["headers"] + data = _create(client, headers) + client.put( + f"{_PERSONNEL}/{data['id']}/regions", + json={"items": [{"region_name": "本溪市平山区"}]}, + headers=headers, + ) + cleared = client.put( + f"{_PERSONNEL}/{data['id']}/regions", json={"items": []}, headers=headers + ) + assert cleared.status_code == 200, cleared.text + detail = client.get(f"{_PERSONNEL}/{data['id']}", headers=headers).json()["data"] + assert detail["regions"] == [] + + def test_拓展计划设置后详情回显一致_再次替换生效( + self, client: TestClient, normal: dict + ) -> None: + headers = normal["headers"] + data = _create(client, headers) + resp = client.put( + f"{_PERSONNEL}/{data['id']}/plans", + json={ + "items": [ + { + "plan_period": "2026Q1", + "plan_content": "拓展本溪市政客户", + "target_desc": "新增3户", + }, + {"plan_period": "2026Q2", "plan_content": "维护存量客户"}, + ] + }, + headers=headers, + ) + assert resp.status_code == 200, resp.text + detail = client.get(f"{_PERSONNEL}/{data['id']}", headers=headers).json()["data"] + assert [p["plan_period"] for p in detail["plans"]] == ["2026Q1", "2026Q2"] + assert detail["plans"][0]["target_desc"] == "新增3户" + + resp2 = client.put( + f"{_PERSONNEL}/{data['id']}/plans", + json={"items": [{"plan_period": "2026Q3", "plan_content": "开拓轨交市场"}]}, + headers=headers, + ) + assert resp2.status_code == 200, resp2.text + detail2 = client.get(f"{_PERSONNEL}/{data['id']}", headers=headers).json()["data"] + assert [p["plan_period"] for p in detail2["plans"]] == ["2026Q3"] + + def test_拓展计划替换为空列表即清空(self, client: TestClient, normal: dict) -> None: + headers = normal["headers"] + data = _create(client, headers) + client.put( + f"{_PERSONNEL}/{data['id']}/plans", + json={"items": [{"plan_period": "2026Q1", "plan_content": "拓展本溪市政客户"}]}, + headers=headers, + ) + cleared = client.put( + f"{_PERSONNEL}/{data['id']}/plans", json={"items": []}, headers=headers + ) + assert cleared.status_code == 200, cleared.text + detail = client.get(f"{_PERSONNEL}/{data['id']}", headers=headers).json()["data"] + assert detail["plans"] == [] + + +class TestAudit: + """写操作（含校验失败的）都在 operation_logs 留痕 —— 审计是验收项。""" + + def test_新增人员成功留审计(self, client: TestClient, normal: dict) -> None: + data = _create(client, normal["headers"]) + logs = [ + log + for log in _logs_of(normal["username"]) + if log.module == MODULE_PERSONNEL and log.action == ACTION_CREATE + ] + assert logs, "新增人员未留操作日志" + latest = logs[0] + assert latest.success == 1 + assert data["personnel_name"] in latest.summary + + def test_校验失败也留审计(self, client: TestClient, normal: dict) -> None: + resp = client.post( + _PERSONNEL, + json=_payload(_uname(), manager_id=999999999), + headers=normal["headers"], + ) + assert resp.status_code == 400, resp.text + failed = [ + log + for log in _logs_of(normal["username"]) + if log.module == MODULE_PERSONNEL and log.success == 0 + ] + assert failed, "失败操作未留痕" + assert "直属经理不存在或已离职" in (failed[0].error_message or "") diff --git a/backend/tests/integration/test_products.py b/backend/tests/integration/test_products.py new file mode 100644 index 0000000..4789fbd --- /dev/null +++ b/backend/tests/integration/test_products.py @@ -0,0 +1,238 @@ +"""产品管理接口集成测试（需要 MySQL 与 Redis）。 + +守的是产品主数据的两条硬约束 —— product_category 是折算阈值的行级锚点， +脏类别会让核算按类别回溯参数取不到阈值；「名称+规格」组合唯一缺失会 +放行重复产品，业务数据不知道该挂在哪一条上。数量/单价以字符串进出， +断言同时验证 str 直出（金额红线：不经 float）。 + +运行前置：``docker compose -f docker/docker-compose.dev.yml up -d`` + +数据卫生：产品名一律 ``t_`` 前缀，由 conftest 兜底清理；删除语义用例 +自验 DELETE 端点。 +""" + +import uuid +from decimal import Decimal + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import select + +from app.core.constants import ( + ACTION_CREATE, + MODULE_PRODUCT, + PRODUCT_CATEGORY_MUNICIPAL_WATER, + PRODUCT_CATEGORY_RAIL_TRANSIT, +) +from app.core.db import get_session_factory +from app.models import OperationLog + +pytestmark = pytest.mark.integration + +_PRODUCTS = "/api/v1/base/products" + + +def _pname() -> str: + return f"t_产品_{uuid.uuid4().hex[:8]}" + + +def _payload(name: str, **overrides: object) -> dict: + body: dict = { + "product_name": name, + "spec_model": "DN100", + "unit": "台", + "stock_quantity": "100", + "sale_price": "12.5", + "product_category": PRODUCT_CATEGORY_MUNICIPAL_WATER, + } + body.update(overrides) + return body + + +def _create(client: TestClient, headers: dict, **overrides: object) -> dict: + resp = client.post(_PRODUCTS, json=_payload(_pname(), **overrides), headers=headers) + assert resp.status_code == 200, resp.text + return resp.json()["data"] + + +def _logs_of(username: str) -> list[OperationLog]: + factory = get_session_factory() + with factory() as session: + return list( + session.execute( + select(OperationLog) + .where(OperationLog.username == username) + .order_by(OperationLog.id.desc()) + ).scalars() + ) + + +class TestAuth: + """鉴权：登录后接口，裸访问一律 401。""" + + def test_未登录访问产品列表_401(self, client: TestClient) -> None: + resp = client.get(_PRODUCTS) + assert resp.status_code == 401, resp.text + + +class TestCrudFlow: + """正向链路：建 → 查 → 改（含换类别）→ 删。""" + + def test_新增产品默认启用且数量单价字符串直出(self, client: TestClient, normal: dict) -> None: + data = _create(client, normal["headers"]) + assert data["status"] == 1, "新产品应默认启用" + assert data["spec_model"] == "DN100" + # 金额红线：str 进 str 出，且数值与入参精确相等（Decimal 比较，不经 float） + assert isinstance(data["sale_price"], str) + assert Decimal(data["sale_price"]) == Decimal("12.5") + assert isinstance(data["stock_quantity"], str) + assert Decimal(data["stock_quantity"]) == Decimal("100") + + def test_详情与列表均能查到新增产品(self, client: TestClient, normal: dict) -> None: + data = _create(client, normal["headers"]) + detail = client.get(f"{_PRODUCTS}/{data['id']}", headers=normal["headers"]) + assert detail.status_code == 200, detail.text + assert detail.json()["data"]["id"] == data["id"] + + listed = client.get( + _PRODUCTS, params={"keyword": data["product_name"]}, headers=normal["headers"] + ) + assert listed.status_code == 200, listed.text + assert [r["id"] for r in listed.json()["data"]["list"]] == [data["id"]] + + def test_同名称不同规格创建成功(self, client: TestClient, normal: dict) -> None: + name = _pname() + first = _create(client, normal["headers"], product_name=name, spec_model="DN100") + second = client.post( + _PRODUCTS, + json=_payload(name, spec_model="DN200"), + headers=normal["headers"], + ) + assert second.status_code == 200, second.text + assert second.json()["data"]["id"] != first["id"] + + def test_删除产品成功(self, client: TestClient, normal: dict) -> None: + data = _create(client, normal["headers"]) + resp = client.delete(f"{_PRODUCTS}/{data['id']}", headers=normal["headers"]) + assert resp.status_code == 200, resp.text + detail = client.get(f"{_PRODUCTS}/{data['id']}", headers=normal["headers"]) + assert detail.status_code == 404 + assert "产品不存在" in detail.json()["message"] + + +class TestValidationRules: + """类别值域与组合唯一 —— 核算阈值锚点不脏、业务标识不重复。""" + + def test_重名同规格_400(self, client: TestClient, normal: dict) -> None: + name = _pname() + _create(client, normal["headers"], product_name=name) + dup = client.post( + _PRODUCTS, + json=_payload(name, spec_model="DN100"), + headers=normal["headers"], + ) + assert dup.status_code == 400, dup.text + assert "已存在同名称同规格的产品" in dup.json()["message"] + + def test_规格缺省归一空串且重名同空规格仍拦截(self, client: TestClient, normal: dict) -> None: + """组合唯一键对 NULL 不去重：缺省规格若归 NULL 会放行重名同规格数据。""" + name = _pname() + first = _create(client, normal["headers"], product_name=name, spec_model=None) + assert first["spec_model"] == "" + dup = client.post( + _PRODUCTS, + json=_payload(name, spec_model=None), + headers=normal["headers"], + ) + assert dup.status_code == 400, dup.text + assert "已存在同名称同规格的产品" in dup.json()["message"] + + def test_产品类别非法_400(self, client: TestClient, normal: dict) -> None: + resp = client.post( + _PRODUCTS, + json=_payload(_pname(), product_category="not_a_category"), + headers=normal["headers"], + ) + assert resp.status_code == 400, resp.text + assert "产品类别不合法" in resp.json()["message"] + + def test_销售单价为负数_400(self, client: TestClient, normal: dict) -> None: + resp = client.post( + _PRODUCTS, + json=_payload(_pname(), sale_price="-1"), + headers=normal["headers"], + ) + assert resp.status_code == 400, resp.text + + def test_销售单价非数字_400(self, client: TestClient, normal: dict) -> None: + resp = client.post( + _PRODUCTS, + json=_payload(_pname(), sale_price="abc"), + headers=normal["headers"], + ) + assert resp.status_code == 400, resp.text + + def test_编辑改类别为合法值成功(self, client: TestClient, normal: dict) -> None: + data = _create(client, normal["headers"]) + updated = client.put( + f"{_PRODUCTS}/{data['id']}", + json=_payload( + data["product_name"], + product_category=PRODUCT_CATEGORY_RAIL_TRANSIT, + sale_price="15.8", + ), + headers=normal["headers"], + ) + assert updated.status_code == 200, updated.text + body = updated.json()["data"] + assert body["product_category"] == PRODUCT_CATEGORY_RAIL_TRANSIT + assert Decimal(body["sale_price"]) == Decimal("15.8") + + +class TestStatusSwitch: + """启用/停用独立于编辑端点 —— 编辑表单不携带状态，防误触停用。""" + + def test_停用再启用切换(self, client: TestClient, normal: dict) -> None: + headers = normal["headers"] + data = _create(client, headers) + off = client.put(f"{_PRODUCTS}/{data['id']}/status", json={"status": 0}, headers=headers) + assert off.status_code == 200, off.text + assert off.json()["message"] == "已停用" + assert off.json()["data"]["status"] == 0 + on = client.put(f"{_PRODUCTS}/{data['id']}/status", json={"status": 1}, headers=headers) + assert on.status_code == 200, on.text + assert on.json()["message"] == "已启用" + assert on.json()["data"]["status"] == 1 + + +class TestAudit: + """写操作（含失败）留痕 —— 审计是验收项。""" + + def test_新增产品成功留审计(self, client: TestClient, normal: dict) -> None: + data = _create(client, normal["headers"]) + logs = [ + log + for log in _logs_of(normal["username"]) + if log.module == MODULE_PRODUCT and log.action == ACTION_CREATE + ] + assert logs, "新增产品未留操作日志" + latest = logs[0] + assert latest.success == 1 + assert data["product_name"] in latest.summary + + def test_重名失败留审计(self, client: TestClient, normal: dict) -> None: + name = _pname() + _create(client, normal["headers"], product_name=name) + dup = client.post( + _PRODUCTS, + json=_payload(name, spec_model="DN100"), + headers=normal["headers"], + ) + assert dup.status_code == 400, dup.text + failed = [ + log + for log in _logs_of(normal["username"]) + if log.module == MODULE_PRODUCT and log.success == 0 + ] + assert failed, "失败操作未留痕" + assert "已存在同名称同规格的产品" in (failed[0].error_message or "")