101 lines
3.4 KiB
Python
101 lines
3.4 KiB
Python
"""开放平台与第三方 API Key 认证集成测试。"""
|
|
|
|
import pytest
|
|
from decimal import Decimal
|
|
from model_bakery import baker
|
|
from rest_framework.test import APIClient
|
|
from rest_framework_simplejwt.tokens import RefreshToken
|
|
|
|
from apps.catalog.models import Product
|
|
from apps.inventory.models import Warehouse, Stock
|
|
from apps.partner.models import Customer
|
|
from apps.sales.models import SalesBill
|
|
from apps.openapi.models import APIKey
|
|
|
|
|
|
@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.mark.django_db
|
|
def test_apikey_generation_and_verification(tenant, user):
|
|
key_obj, raw_key = APIKey.generate(
|
|
tenant=tenant,
|
|
name="分销商A客户端",
|
|
scopes=["products:read", "stocks:read"],
|
|
created_by=user,
|
|
)
|
|
assert raw_key.startswith("dh_")
|
|
assert "." in raw_key
|
|
assert key_obj.verify_key(raw_key) is True
|
|
assert key_obj.verify_key("dh_fake.fake_secret") is False
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_external_system_openapi_access(tenant):
|
|
prod = baker.make(Product, tenant=tenant, code="P_OPEN_01", name="开放商品", sale_price=Decimal("88.00"))
|
|
wh = baker.make(Warehouse, tenant=tenant, code="WH_OPEN_01", name="开放仓")
|
|
baker.make(Stock, tenant=tenant, warehouse=wh, product=prod, on_hand=Decimal("50.00"), locked=Decimal("5.00"))
|
|
cust = baker.make(Customer, tenant=tenant, code="C_OPEN_01", name="外部客户A")
|
|
|
|
key_obj, raw_key = APIKey.generate(
|
|
tenant=tenant,
|
|
name="外部系统密钥",
|
|
scopes=["products:read", "stocks:read", "orders:write"],
|
|
)
|
|
|
|
client = APIClient()
|
|
# 1. 未带 API Key 访问应被拒绝 (401 或 403)
|
|
r1 = client.get("/api/v1/openapi/v1/products/")
|
|
assert r1.status_code in (401, 403)
|
|
|
|
# 2. 携带 X-API-Key 访问商品列表
|
|
client.credentials(HTTP_X_API_KEY=raw_key)
|
|
r2 = client.get("/api/v1/openapi/v1/products/")
|
|
assert r2.status_code == 200, r2.content
|
|
assert r2.json()["count"] == 1
|
|
assert r2.json()["results"][0]["code"] == "P_OPEN_01"
|
|
|
|
# 3. 实时查询库存
|
|
r3 = client.get("/api/v1/openapi/v1/stocks/")
|
|
assert r3.status_code == 200
|
|
assert r3.json()["count"] == 1
|
|
stock_item = r3.json()["results"][0]
|
|
assert stock_item["on_hand"] == Decimal("50.00")
|
|
assert stock_item["available"] == Decimal("45.00")
|
|
|
|
# 4. 外部系统推送销售开单
|
|
order_payload = {
|
|
"customer_code": "C_OPEN_01",
|
|
"warehouse_code": "WH_OPEN_01",
|
|
"remark": "来自下游 ERP 自动同步",
|
|
"lines": [
|
|
{"product_code": "P_OPEN_01", "quantity": "3", "unit_price": "85.00"}
|
|
],
|
|
}
|
|
r4 = client.post("/api/v1/openapi/v1/orders/", order_payload, format="json")
|
|
assert r4.status_code == 201, r4.content
|
|
data = r4.json()
|
|
assert data["ok"] is True
|
|
assert data["total_amount"] == Decimal("255.00")
|
|
assert SalesBill.objects.filter(tenant=tenant, bill_no=data["bill_no"]).exists()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_apikey_management_api(auth_client, tenant):
|
|
# 管理员创建 Key
|
|
r = auth_client.post("/api/v1/openapi/keys/", {
|
|
"name": "测试密钥1",
|
|
"scopes": ["products:read"],
|
|
}, format="json")
|
|
assert r.status_code == 201
|
|
assert "raw_key" in r.json()
|
|
assert r.json()["prefix"].startswith("dh_")
|