119 lines
3.8 KiB
Python
119 lines
3.8 KiB
Python
"""商品中心集成测试：CRUD + 多租户隔离。
|
|
|
|
使用同步 APIClient（走 WSGI handler + adrf view），因为：
|
|
- DRF/adrf 的 auth/permissions/pagination 都是同步触发的
|
|
- 测试 client 同步请求 adrf 的 async view 时 Django 自动用 async_to_sync 包装
|
|
- 比 AsyncClient 稳定，async_to_sync 不会触发 RuntimeError
|
|
"""
|
|
|
|
import pytest
|
|
from model_bakery import baker
|
|
from rest_framework.test import APIClient
|
|
from rest_framework_simplejwt.tokens import RefreshToken
|
|
|
|
from apps.catalog.models import Product, Unit, UnitConversion
|
|
|
|
|
|
@pytest.fixture
|
|
def auth_client(db, user, tenant):
|
|
client = APIClient()
|
|
refresh = RefreshToken.for_user(user)
|
|
client.credentials(
|
|
HTTP_AUTHORIZATION=f"Bearer {refresh.access_token}",
|
|
HTTP_X_TENANT_ID=tenant.code,
|
|
)
|
|
return client
|
|
|
|
|
|
@pytest.fixture
|
|
def anon(db):
|
|
return APIClient()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_create_product(auth_client, tenant, org):
|
|
payload = {
|
|
"code": "P001",
|
|
"name": "测试商品",
|
|
"spec": "500ml",
|
|
"barcode": "6901234567890",
|
|
"cost_price": "10.0000",
|
|
"sale_price": "15.0000",
|
|
"status": "active",
|
|
}
|
|
resp = auth_client.post("/api/v1/catalog/products/", payload, format="json")
|
|
assert resp.status_code == 201, resp.content
|
|
data = resp.json()
|
|
assert data["code"] == "P001"
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_list_products(auth_client, tenant):
|
|
baker.make(Product, tenant=tenant, code="P001", name="商品1")
|
|
baker.make(Product, tenant=tenant, code="P002", name="商品2")
|
|
resp = auth_client.get("/api/v1/catalog/products/")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["count"] == 2
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_tenant_isolation(auth_client, tenant, other_tenant):
|
|
baker.make(Product, tenant=other_tenant, code="OTHER_P1")
|
|
baker.make(Product, tenant=tenant, code="MY_P1")
|
|
|
|
resp = auth_client.get("/api/v1/catalog/products/")
|
|
codes = {p["code"] for p in resp.json()["results"]}
|
|
assert "MY_P1" in codes
|
|
assert "OTHER_P1" not in codes
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_soft_delete_excluded(auth_client, tenant):
|
|
baker.make(Product, tenant=tenant, code="DEL_P", is_deleted=True)
|
|
resp = auth_client.get("/api/v1/catalog/products/")
|
|
codes = {x["code"] for x in resp.json()["results"]}
|
|
assert "DEL_P" not in codes
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_search(auth_client, tenant):
|
|
baker.make(Product, tenant=tenant, code="P001", name="苹果")
|
|
baker.make(Product, tenant=tenant, code="P002", name="香蕉")
|
|
resp = auth_client.get("/api/v1/catalog/products/?search=苹果")
|
|
assert resp.json()["count"] == 1
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_anon_cannot_access(anon, tenant):
|
|
anon.credentials(HTTP_X_TENANT_ID=tenant.code)
|
|
resp = anon.get("/api/v1/catalog/products/")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_product_units_expose_sale_and_cost_price_per_source_unit(
|
|
auth_client, tenant, org
|
|
):
|
|
base = baker.make(Unit, tenant=tenant, code="bottle", name="瓶", is_base=True)
|
|
box = baker.make(Unit, tenant=tenant, code="box", name="箱", is_base=False)
|
|
product = baker.make(
|
|
Product,
|
|
tenant=tenant,
|
|
code="PRICE-UNIT",
|
|
name="单价换算测试",
|
|
base_unit=base,
|
|
cost_price="3.0000",
|
|
sale_price="5.0000",
|
|
)
|
|
UnitConversion.objects.create(
|
|
tenant=tenant, product=product, unit=box, rate="24"
|
|
)
|
|
|
|
response = auth_client.get(f"/api/v1/catalog/products/{product.id}/units/")
|
|
|
|
assert response.status_code == 200
|
|
by_unit = {item["unit_code"]: item for item in response.json()["units"]}
|
|
assert by_unit["bottle"]["cost_price"] == "3.0000"
|
|
assert by_unit["box"]["cost_price"] == "72.0000"
|
|
assert by_unit["box"]["price"] == "120.0000"
|