"""产品管理接口集成测试（需要 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 "")