48 lines
1.6 KiB
Python
48 lines
1.6 KiB
Python
"""同步跑通验证：用 sync APIClient 验证 token/headers/租户/CRUD。"""
|
|
|
|
import os
|
|
import django
|
|
|
|
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings.dev")
|
|
django.setup()
|
|
|
|
import json
|
|
from rest_framework.test import APIClient
|
|
from rest_framework_simplejwt.tokens import RefreshToken
|
|
from django.contrib.auth import get_user_model
|
|
from apps.core.models import Tenant
|
|
|
|
|
|
def main():
|
|
tenant, _ = Tenant.objects.get_or_create(
|
|
code="default", defaults={"name": "默认", "is_active": True}
|
|
)
|
|
U = get_user_model()
|
|
user, _ = U.objects.get_or_create(username="alice", defaults={"email": "a@a.com"})
|
|
user.set_password("alice12345")
|
|
user.save()
|
|
token = str(RefreshToken.for_user(user).access_token)
|
|
print(f"[setup] tenant={tenant.code} user={user.username} token-len={len(token)}")
|
|
|
|
client = APIClient()
|
|
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}", HTTP_X_TENANT_ID=tenant.code)
|
|
|
|
r1 = client.get("/api/v1/ping/")
|
|
print(f"[ping] {r1.status_code} body={r1.content[:200].decode()}")
|
|
|
|
r2 = client.get("/api/v1/catalog/products/")
|
|
print(f"[catalog/list] {r2.status_code} body={r2.content[:400].decode(errors='replace')}")
|
|
|
|
r3 = client.post(
|
|
"/api/v1/catalog/products/",
|
|
data=json.dumps({"code": "P001", "name": "测试商品", "status": "active"}),
|
|
content_type="application/json",
|
|
)
|
|
print(f"[catalog/create] {r3.status_code} body={r3.content[:400].decode(errors='replace')}")
|
|
|
|
r4 = client.get("/api/v1/catalog/products/?search=P001")
|
|
print(f"[catalog/search] {r4.status_code} body={r4.content[:400].decode(errors='replace')}")
|
|
|
|
|
|
main()
|