test(base): 基础数据管理模块集成用例（48 例）与审计盲区修复
- 覆盖：401/引用校验/删除保护/子表全量替换/类别与组合唯一/归属变更留痕/审计双记录 - 缺陷修复：personnel/product/customer 三 service 的前置校验移入审计上下文， 校验失败（400）同样留痕（成功与失败都记的既定原则） - conftest 兜底清理扩展到 M2 五表（留痕无外键，按 operator_name + 前缀双轨清理） Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
liuwenxinandClaude Code committed
1 parent
15de81dd2c
commit
ca88ee7a0b
12 files changed
+1090
-29
No files matched your search
@@ -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 "")
|
||||
Reference in new issue
Block a user