108bf87225
补齐源快照、资料预演、正式导入及工资核算的依赖链路。 新增版本化输入、逐段 FIFO、生成锁和结构化当月调整， 并提供默认关闭的周期采集、有限重试及 Windows 托管。 原始客户资料、私有配置及首页改动不纳入本次提交。 BREAKING CHANGE: 升级前需应用新增迁移，并通过 PAYROLL_MAKER_USER_ID 绑定实际工资制作人，否则业务只读。 旧单批次核算默认关闭；历史兼容需显式启用 PAYROLL_ALLOW_LEGACY_CALCULATION。生产启用和真实金额待验收。
335 lines
13 KiB
Python
335 lines
13 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"
|
||
_CUSTOMERS = "/api/v1/base/customers"
|
||
_BUSINESSES = "/api/v1/business/businesses"
|
||
|
||
|
||
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()
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def make_business_for_product(client: TestClient):
|
||
"""造「客户 → 业务 → 明细挂指定产品」的最小引用链（删除保护场景）。
|
||
|
||
business_items.product_id 是 RESTRICT 外键，产品删除保护以此为引用源；
|
||
客户/业务均按 t_ 前缀与系统编号由 conftest 兜底清理。
|
||
"""
|
||
|
||
def _factory(headers: dict, product_id: int) -> dict:
|
||
customer = client.post(
|
||
_CUSTOMERS,
|
||
json={
|
||
"company_name": f"t_引用公司_{uuid.uuid4().hex[:8]}",
|
||
"owner_name": "张老板",
|
||
"customer_type": "personal",
|
||
},
|
||
headers=headers,
|
||
)
|
||
assert customer.status_code == 200, customer.text
|
||
business = client.post(
|
||
_BUSINESSES,
|
||
json={
|
||
"customer_id": customer.json()["data"]["id"],
|
||
"business_date": "2026-01-01",
|
||
"items": [
|
||
{
|
||
"product_id": product_id,
|
||
"product_name": f"t_引用产品_{uuid.uuid4().hex[:6]}",
|
||
"product_category": PRODUCT_CATEGORY_MUNICIPAL_WATER,
|
||
"quantity": "1.00",
|
||
"sale_price": "10.00",
|
||
}
|
||
],
|
||
},
|
||
headers=headers,
|
||
)
|
||
assert business.status_code == 200, business.text
|
||
return business.json()["data"]
|
||
|
||
return _factory
|
||
|
||
|
||
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, writer: dict) -> None:
|
||
data = _create(client, writer["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, writer: dict) -> None:
|
||
data = _create(client, writer["headers"])
|
||
detail = client.get(f"{_PRODUCTS}/{data['id']}", headers=writer["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=writer["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, writer: dict) -> None:
|
||
name = _pname()
|
||
first = _create(client, writer["headers"], product_name=name, spec_model="DN100")
|
||
second = client.post(
|
||
_PRODUCTS,
|
||
json=_payload(name, spec_model="DN200"),
|
||
headers=writer["headers"],
|
||
)
|
||
assert second.status_code == 200, second.text
|
||
assert second.json()["data"]["id"] != first["id"]
|
||
|
||
def test_删除产品成功(self, client: TestClient, writer: dict) -> None:
|
||
data = _create(client, writer["headers"])
|
||
resp = client.delete(f"{_PRODUCTS}/{data['id']}", headers=writer["headers"])
|
||
assert resp.status_code == 200, resp.text
|
||
detail = client.get(f"{_PRODUCTS}/{data['id']}", headers=writer["headers"])
|
||
assert detail.status_code == 404
|
||
assert "产品不存在" in detail.json()["message"]
|
||
|
||
|
||
class TestDeleteProtection:
|
||
"""删除保护：被业务/发货明细引用的产品禁删（customer 先例同构）。"""
|
||
|
||
def test_被业务明细引用_400且产品仍在(
|
||
self, client: TestClient, writer: dict, make_business_for_product
|
||
) -> None:
|
||
product = _create(client, writer["headers"])
|
||
business = make_business_for_product(writer["headers"], product["id"])
|
||
resp = client.delete(f"{_PRODUCTS}/{product['id']}", headers=writer["headers"])
|
||
assert resp.status_code == 400, resp.text
|
||
assert "引用" in resp.json()["message"]
|
||
# 产品仍在（引导改走停用而非物理删除）
|
||
still = client.get(f"{_PRODUCTS}/{product['id']}", headers=writer["headers"])
|
||
assert still.status_code == 200
|
||
# 业务明细行确实引用了该产品（护栏：确认保护不是空转）
|
||
detail = client.get(f"{_BUSINESSES}/{business['id']}", headers=writer["headers"]).json()[
|
||
"data"
|
||
]
|
||
assert detail["items"][0]["product_id"] == product["id"]
|
||
|
||
def test_被发货明细引用_400(
|
||
self, client: TestClient, writer: dict, make_business_for_product
|
||
) -> None:
|
||
product = _create(client, writer["headers"])
|
||
make_business_for_product(writer["headers"], product["id"])
|
||
# 在业务明细引用之外再挂一笔发货明细（同样引用 products.id）
|
||
businesses = client.get("/api/v1/business/businesses", headers=writer["headers"]).json()[
|
||
"data"
|
||
]
|
||
business_id = businesses["list"][0]["id"]
|
||
shipment = client.post(
|
||
"/api/v1/business/shipments",
|
||
json={
|
||
"business_id": business_id,
|
||
"ship_date": "2026-01-01",
|
||
"items": [
|
||
{
|
||
"product_id": product["id"],
|
||
"product_name": product["product_name"],
|
||
"quantity": "1.00",
|
||
"sale_price": "10.00",
|
||
}
|
||
],
|
||
},
|
||
headers=writer["headers"],
|
||
)
|
||
assert shipment.status_code == 200, shipment.text
|
||
# 此前业务明细引用已令删除被拒，此处再证发货引用同一防线
|
||
resp = client.delete(f"{_PRODUCTS}/{product['id']}", headers=writer["headers"])
|
||
assert resp.status_code == 400, resp.text
|
||
|
||
|
||
class TestValidationRules:
|
||
"""类别值域与组合唯一 —— 核算阈值锚点不脏、业务标识不重复。"""
|
||
|
||
def test_重名同规格_400(self, client: TestClient, writer: dict) -> None:
|
||
name = _pname()
|
||
_create(client, writer["headers"], product_name=name)
|
||
dup = client.post(
|
||
_PRODUCTS,
|
||
json=_payload(name, spec_model="DN100"),
|
||
headers=writer["headers"],
|
||
)
|
||
assert dup.status_code == 400, dup.text
|
||
assert "已存在同名称同规格的产品" in dup.json()["message"]
|
||
|
||
def test_规格缺省归一空串且重名同空规格仍拦截(self, client: TestClient, writer: dict) -> None:
|
||
"""组合唯一键对 NULL 不去重：缺省规格若归 NULL 会放行重名同规格数据。"""
|
||
name = _pname()
|
||
first = _create(client, writer["headers"], product_name=name, spec_model=None)
|
||
assert first["spec_model"] == ""
|
||
dup = client.post(
|
||
_PRODUCTS,
|
||
json=_payload(name, spec_model=None),
|
||
headers=writer["headers"],
|
||
)
|
||
assert dup.status_code == 400, dup.text
|
||
assert "已存在同名称同规格的产品" in dup.json()["message"]
|
||
|
||
def test_产品类别非法_400(self, client: TestClient, writer: dict) -> None:
|
||
resp = client.post(
|
||
_PRODUCTS,
|
||
json=_payload(_pname(), product_category="not_a_category"),
|
||
headers=writer["headers"],
|
||
)
|
||
assert resp.status_code == 400, resp.text
|
||
assert "产品类别不合法" in resp.json()["message"]
|
||
|
||
def test_销售单价为负数_400(self, client: TestClient, writer: dict) -> None:
|
||
resp = client.post(
|
||
_PRODUCTS,
|
||
json=_payload(_pname(), sale_price="-1"),
|
||
headers=writer["headers"],
|
||
)
|
||
assert resp.status_code == 400, resp.text
|
||
|
||
def test_销售单价非数字_400(self, client: TestClient, writer: dict) -> None:
|
||
resp = client.post(
|
||
_PRODUCTS,
|
||
json=_payload(_pname(), sale_price="abc"),
|
||
headers=writer["headers"],
|
||
)
|
||
assert resp.status_code == 400, resp.text
|
||
|
||
def test_编辑改类别为合法值成功(self, client: TestClient, writer: dict) -> None:
|
||
data = _create(client, writer["headers"])
|
||
updated = client.put(
|
||
f"{_PRODUCTS}/{data['id']}",
|
||
json=_payload(
|
||
data["product_name"],
|
||
product_category=PRODUCT_CATEGORY_RAIL_TRANSIT,
|
||
sale_price="15.8",
|
||
),
|
||
headers=writer["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, writer: dict) -> None:
|
||
headers = writer["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, writer: dict) -> None:
|
||
data = _create(client, writer["headers"])
|
||
logs = [
|
||
log
|
||
for log in _logs_of(writer["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, writer: dict) -> None:
|
||
name = _pname()
|
||
_create(client, writer["headers"], product_name=name)
|
||
dup = client.post(
|
||
_PRODUCTS,
|
||
json=_payload(name, spec_model="DN100"),
|
||
headers=writer["headers"],
|
||
)
|
||
assert dup.status_code == 400, dup.text
|
||
failed = [
|
||
log
|
||
for log in _logs_of(writer["username"])
|
||
if log.module == MODULE_PRODUCT and log.success == 0
|
||
]
|
||
assert failed, "失败操作未留痕"
|
||
assert "已存在同名称同规格的产品" in (failed[0].error_message or "")
|