96 lines
2.7 KiB
Python
96 lines
2.7 KiB
Python
"""共享 fixtures：tenant/org/user/async_api_client + postgres 用例自动跳过。"""
|
||
|
||
import pytest
|
||
import pytest_asyncio
|
||
from model_bakery import baker
|
||
from django.test import AsyncClient
|
||
|
||
from apps.core.models import Tenant, Org, TenantMembership
|
||
|
||
|
||
@pytest.fixture
|
||
def tenant(db):
|
||
return baker.make(Tenant, code="test", name="测试租户", is_active=True)
|
||
|
||
|
||
@pytest.fixture
|
||
def other_tenant(db):
|
||
return baker.make(Tenant, code="other", name="其他租户", is_active=True)
|
||
|
||
|
||
@pytest.fixture
|
||
def org(db, tenant):
|
||
return baker.make(
|
||
Org, tenant=tenant, code="hq", name="总部", org_path="001", is_active=True
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def user(db, tenant, django_user_model):
|
||
account = django_user_model.objects.create_user(
|
||
username="alice", password="alice12345"
|
||
)
|
||
TenantMembership.objects.create(user=account, tenant=tenant, role="member")
|
||
return account
|
||
|
||
|
||
@pytest.fixture
|
||
def auth_token(user):
|
||
from rest_framework_simplejwt.tokens import RefreshToken
|
||
|
||
refresh = RefreshToken.for_user(user)
|
||
return str(refresh.access_token)
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def api_client(user, tenant):
|
||
"""async 客户端 + JWT + 租户头。"""
|
||
from rest_framework_simplejwt.tokens import RefreshToken
|
||
|
||
refresh = RefreshToken.for_user(user)
|
||
client = AsyncClient()
|
||
# 用 META 风格注入 headers
|
||
return client
|
||
|
||
|
||
@pytest.fixture
|
||
def api_headers(user, tenant):
|
||
"""返回 dict 形式的 headers，调用方传给 AsyncClient 请求。"""
|
||
from rest_framework_simplejwt.tokens import RefreshToken
|
||
|
||
refresh = RefreshToken.for_user(user)
|
||
return {
|
||
"HTTP_AUTHORIZATION": f"Bearer {refresh.access_token}",
|
||
"HTTP_X_TENANT_ID": tenant.code,
|
||
}
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def anon_client():
|
||
return AsyncClient()
|
||
|
||
|
||
# ============================================================
|
||
# PostgreSQL 专用用例：SQLite 下自动跳过
|
||
# ============================================================
|
||
|
||
def pytest_collection_modifyitems(config, items):
|
||
"""标记 `@pytest.mark.postgres` 的用例只在 PG 后端执行。
|
||
|
||
SQLite 是数据库级写锁，多线程并发会报 `database table is locked`，
|
||
这类用例在 SQLite 上跑没有意义（会假失败）。
|
||
"""
|
||
from django.conf import settings
|
||
|
||
engine = settings.DATABASES["default"]["ENGINE"]
|
||
if "postgresql" in engine:
|
||
return # PG 后端：全部执行
|
||
|
||
skip_pg = pytest.mark.skip(
|
||
reason=f"需要 PostgreSQL（当前 {engine.split('.')[-1]}）；"
|
||
"用 DJANGO_SETTINGS_MODULE=config.settings.pgtest 运行"
|
||
)
|
||
for item in items:
|
||
if "postgres" in item.keywords:
|
||
item.add_marker(skip_pg)
|