"""开放平台与第三方 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_")