92 lines
3.7 KiB
Python
92 lines
3.7 KiB
Python
"""P0-4 · 重复编码 400 回归（28 模型一处修复）。
|
||
|
||
根因：DRF 的 `get_unique_together_validators()` 要求 unique_together 字段
|
||
全部出现在 `Meta.fields` 中；本项目 `tenant` 由 `BaseTenantViewSet.acreate()`
|
||
注入、不在 fields 里 → 校验器被静默丢弃 → 唯一冲突直达 DB → 500。
|
||
|
||
修复落点：`apps/core/viewset.py` 的 `acreate()` —— 租户感知的唯一性预检
|
||
（友好 400）+ `IntegrityError` 并发兜底（转 400）。
|
||
|
||
注意：会触发 DB 异常的回归测试一律用
|
||
`@pytest.mark.django_db(transaction=True)`，否则测试自带的外层 atomic
|
||
被污染，后续 ORM 全变 `TransactionManagementError`（测试假象，非生产问题）。
|
||
"""
|
||
|
||
import pytest
|
||
from rest_framework.test import APIClient
|
||
from rest_framework_simplejwt.tokens import RefreshToken
|
||
|
||
|
||
pytestmark = pytest.mark.django_db(transaction=True)
|
||
|
||
|
||
def _jwt_client(user, tenant_code):
|
||
c = APIClient()
|
||
c.credentials(
|
||
HTTP_AUTHORIZATION=f"Bearer {RefreshToken.for_user(user).access_token}",
|
||
HTTP_X_TENANT_ID=tenant_code,
|
||
)
|
||
return c
|
||
|
||
|
||
# (url, payload) —— 覆盖主数据 8 类高频手输编码
|
||
DUPLICATE_CASES = [
|
||
("/api/v1/catalog/categories/", {"code": "DUP01", "name": "分类A"}),
|
||
("/api/v1/catalog/brands/", {"code": "DUP01", "name": "品牌A"}),
|
||
("/api/v1/catalog/units/", {"code": "DUP01", "name": "单位A"}),
|
||
("/api/v1/catalog/products/", {"code": "DUP01", "name": "商品A"}),
|
||
("/api/v1/partner/price-levels/", {"code": "DUP01", "name": "等级A"}),
|
||
("/api/v1/partner/customers/", {"code": "DUP01", "name": "客户A"}),
|
||
("/api/v1/partner/suppliers/", {"code": "DUP01", "name": "供应商A"}),
|
||
("/api/v1/inventory/warehouses/", {"code": "DUP01", "name": "仓库A"}),
|
||
]
|
||
|
||
|
||
@pytest.mark.parametrize("url,payload", DUPLICATE_CASES)
|
||
def test_duplicate_code_is_400_not_500(db, tenant, user, url, payload):
|
||
"""重复编码 → 400 + 字段级错误，不含 exc_type/堆栈。"""
|
||
c = _jwt_client(user, tenant.code)
|
||
r1 = c.post(url, {**payload, "name": payload["name"] + "一"}, format="json")
|
||
assert r1.status_code == 201, (url, r1.content)
|
||
r2 = c.post(url, {**payload, "name": payload["name"] + "二"}, format="json")
|
||
assert r2.status_code == 400, (url, r2.content)
|
||
body = r2.json()
|
||
assert "exc_type" not in body
|
||
assert "traceback" not in str(body).lower()
|
||
|
||
|
||
def test_duplicate_code_is_tenant_scoped(db, tenant, other_tenant, user):
|
||
"""跨租户同编码允许：唯一性是租户内的，不是全局的。"""
|
||
from apps.core.models import TenantMembership
|
||
|
||
TenantMembership.objects.get_or_create(
|
||
user=user, tenant=other_tenant,
|
||
defaults={"role": "member", "is_active": True},
|
||
)
|
||
c1 = _jwt_client(user, tenant.code)
|
||
c2 = _jwt_client(user, other_tenant.code)
|
||
r1 = c1.post(
|
||
"/api/v1/catalog/products/",
|
||
{"code": "SHARED01", "name": "甲租户商品"}, format="json",
|
||
)
|
||
assert r1.status_code == 201, r1.content
|
||
r2 = c2.post(
|
||
"/api/v1/catalog/products/",
|
||
{"code": "SHARED01", "name": "乙租户商品"}, format="json",
|
||
)
|
||
assert r2.status_code == 201, r2.content
|
||
|
||
|
||
def test_unknown_error_has_no_exc_type(db, tenant, user):
|
||
"""兜底 500 不泄露 exc_type/原始消息（P0-4 附带收尾，已在 exceptions 落地）。"""
|
||
from rest_framework.test import APIRequestFactory
|
||
from apps.core.exceptions import api_exception_handler
|
||
|
||
factory = APIRequestFactory()
|
||
req = factory.get("/api/v1/catalog/products/")
|
||
resp = api_exception_handler(RuntimeError("boom-secret"), {"request": req, "view": None})
|
||
assert resp.status_code == 500
|
||
assert resp.data["code"] == "server_error"
|
||
assert "exc_type" not in resp.data
|
||
assert "boom-secret" not in str(resp.data)
|