64 lines
1.8 KiB
Python
64 lines
1.8 KiB
Python
"""快速验证 AsyncClient 是否能带 header 跑通。"""
|
|
|
|
import asyncio
|
|
import os
|
|
import django
|
|
|
|
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings.dev")
|
|
django.setup()
|
|
|
|
from django.test import AsyncClient
|
|
from rest_framework_simplejwt.tokens import RefreshToken
|
|
from django.contrib.auth import get_user_model
|
|
from apps.core.models import Tenant
|
|
from asgiref.sync import sync_to_async
|
|
|
|
|
|
async def main():
|
|
tenant, _ = await sync_to_async(Tenant.objects.get_or_create)(
|
|
code="default", defaults={"name": "默认", "is_active": True}
|
|
)
|
|
user = await sync_create_user()
|
|
token = await sync_make_token(user)
|
|
print(f"[setup] tenant={tenant.code} user={user.username} token-len={len(token)}")
|
|
|
|
client = AsyncClient()
|
|
r1 = await client.get("/api/v1/ping/")
|
|
print(f"[ping] {r1.status_code} {r1.content[:120]}")
|
|
|
|
r2 = await client.get(
|
|
"/api/v1/catalog/products/",
|
|
HTTP_AUTHORIZATION=f"Bearer {token}",
|
|
HTTP_X_TENANT_ID=tenant.code,
|
|
)
|
|
print(f"[catalog/list] {r2.status_code} {r2.content[:300]}")
|
|
|
|
import json
|
|
body = json.dumps({"code": "P001", "name": "测试商品", "status": "active"})
|
|
r3 = await client.post(
|
|
"/api/v1/catalog/products/",
|
|
data=body,
|
|
content_type="application/json",
|
|
HTTP_AUTHORIZATION=f"Bearer {token}",
|
|
HTTP_X_TENANT_ID=tenant.code,
|
|
)
|
|
print(f"[catalog/create] {r3.status_code} {r3.content[:300]}")
|
|
|
|
|
|
@sync_to_async
|
|
def sync_create_user():
|
|
U = get_user_model()
|
|
user, _ = U.objects.get_or_create(username="alice", defaults={"email": "a@a.com"})
|
|
user.set_password("alice12345")
|
|
user.save()
|
|
return user
|
|
|
|
|
|
@sync_to_async
|
|
def sync_make_token(user):
|
|
refresh = RefreshToken.for_user(user)
|
|
return str(refresh.access_token)
|
|
|
|
|
|
asyncio.run(main())
|