ca88ee7a0b
- 覆盖：401/引用校验/删除保护/子表全量替换/类别与组合唯一/归属变更留痕/审计双记录 - 缺陷修复：personnel/product/customer 三 service 的前置校验移入审计上下文， 校验失败（400）同样留痕（成功与失败都记的既定原则） - conftest 兜底清理扩展到 M2 五表（留痕无外键，按 operator_name + 前缀双轨清理） Co-Authored-By: Claude Code <noreply@anthropic.com>
239 lines
9.2 KiB
Python
239 lines
9.2 KiB
Python
"""产品管理接口集成测试（需要 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 "")
|