150 lines
5.8 KiB
Python
150 lines
5.8 KiB
Python
"""P0-1 · 租户 membership 授权语义回归（权限类此前零测试覆盖）。
|
||
|
||
语义合同：
|
||
- 未知租户 code → 交给视图层报 400「无法识别租户」，权限层不掩盖为 403
|
||
- 已知租户无 membership → 403 + code=tenant_membership_required
|
||
- is_active=False 的 membership → 403
|
||
- 超级用户无 membership → 放行（bootstrap 契约）
|
||
- API Key：不带头或带同租户头 → 放行；显式带其他租户头 → 403
|
||
- demo token 进入 demo 租户只读可读 → 200
|
||
"""
|
||
|
||
import pytest
|
||
from decimal import Decimal
|
||
from model_bakery import baker
|
||
from django.core.management import call_command
|
||
from rest_framework.test import APIClient
|
||
from rest_framework_simplejwt.tokens import RefreshToken
|
||
|
||
from apps.catalog.models import Product
|
||
from apps.core.models import TenantMembership
|
||
from apps.inventory.models import Warehouse
|
||
from apps.openapi.models import APIKey
|
||
|
||
|
||
def _jwt_client(user, tenant_code):
|
||
c = APIClient()
|
||
c.credentials(
|
||
HTTP_AUTHORIZATION=f"Bearer {RefreshToken.for_user(user).access_token}",
|
||
HTTP_X_TENANT_ID=tenant_code,
|
||
)
|
||
return c
|
||
|
||
|
||
def test_no_membership_valid_tenant_is_403(db, tenant, django_user_model):
|
||
stranger = django_user_model.objects.create_user("stranger", password="x")
|
||
c = _jwt_client(stranger, tenant.code)
|
||
resp = c.get("/api/v1/catalog/products/")
|
||
assert resp.status_code == 403, resp.content
|
||
assert resp.json()["code"] == "tenant_membership_required"
|
||
|
||
|
||
def test_membership_of_a_requesting_b_is_403(db, tenant, other_tenant, user):
|
||
c = _jwt_client(user, other_tenant.code)
|
||
resp = c.get("/api/v1/catalog/products/")
|
||
assert resp.status_code == 403, resp.content
|
||
assert resp.json()["code"] == "tenant_membership_required"
|
||
|
||
|
||
def test_inactive_membership_is_403(db, tenant, django_user_model):
|
||
u = django_user_model.objects.create_user("sleepy", password="x")
|
||
TenantMembership.objects.create(
|
||
user=u, tenant=tenant, role="member", is_active=False
|
||
)
|
||
c = _jwt_client(u, tenant.code)
|
||
resp = c.get("/api/v1/catalog/products/")
|
||
assert resp.status_code == 403, resp.content
|
||
|
||
|
||
def test_unknown_tenant_is_400_not_403(db, user):
|
||
c = _jwt_client(user, "no-such-tenant")
|
||
resp = c.get("/api/v1/catalog/products/")
|
||
assert resp.status_code == 400, resp.content
|
||
assert "tenant" in str(resp.json())
|
||
|
||
|
||
def test_superuser_without_membership_passes(db, tenant, django_user_model):
|
||
root = django_user_model.objects.create_superuser(
|
||
"rootop", password="x", email="r@x.com"
|
||
)
|
||
c = _jwt_client(root, tenant.code)
|
||
resp = c.get("/api/v1/catalog/products/")
|
||
assert resp.status_code == 200, resp.content
|
||
|
||
|
||
def test_apikey_branch_unit_and_integration_reality(db, tenant, other_tenant):
|
||
"""API Key 租户绑定分支：单元级验证权限类分支，集成级记录真实接线。
|
||
|
||
实测接线现实（写进测试，代替纸面假设）：
|
||
- `TenantMembershipPermission` 的 APIKey 分支目前没有视图同时接线
|
||
（BaseTenantViewSet 系用 JWT/Session；Open* 系用 require_scope），
|
||
所以分支按单元级直接调用验证；
|
||
- 集成级：catalog 系不接受 APIKey（401，要 JWT），Open* 系按 key 绑定租户
|
||
出数据（显式带其他租户头被忽略，不泄露对方数据）。
|
||
"""
|
||
from django.contrib.auth import get_user_model
|
||
from rest_framework.test import APIRequestFactory
|
||
|
||
from apps.core.permissions import TenantMembershipPermission
|
||
|
||
User = get_user_model()
|
||
u = User.objects.create_user("keyowner", password="x")
|
||
key_obj, _raw = APIKey.generate(
|
||
tenant=tenant, name="m1", scopes=["products:read"], created_by=u,
|
||
)
|
||
perm = TenantMembershipPermission()
|
||
factory = APIRequestFactory()
|
||
|
||
def _req(header=None):
|
||
req = factory.get("/api/v1/catalog/products/")
|
||
req.user = u
|
||
req.auth = key_obj
|
||
req.META = {}
|
||
if header is not None:
|
||
req.META["HTTP_X_TENANT_ID"] = header
|
||
return req
|
||
|
||
assert perm.has_permission(_req(), None) is True
|
||
assert perm.has_permission(_req(tenant.code), None) is True
|
||
assert perm.has_permission(_req(other_tenant.code), None) is False
|
||
|
||
# 集成级现实：catalog + APIKey 头 → 401（该系只认 JWT）
|
||
c3 = APIClient()
|
||
c3.credentials(HTTP_X_API_KEY=_raw, HTTP_X_TENANT_ID=other_tenant.code)
|
||
r3 = c3.get("/api/v1/catalog/products/")
|
||
assert r3.status_code == 401, r3.content
|
||
|
||
|
||
def test_demo_token_reads_demo_tenant(db):
|
||
call_command("seed_demo")
|
||
token = APIClient().post("/api/v1/demo/enter/", {}, format="json").json()["access"]
|
||
c = APIClient()
|
||
c.credentials(HTTP_AUTHORIZATION=f"Bearer {token}", HTTP_X_TENANT_ID="demo")
|
||
resp = c.get("/api/v1/sales/bills/")
|
||
assert resp.status_code == 200, resp.content
|
||
|
||
|
||
def test_seed_initial_data_membership_idempotent(db, django_user_model):
|
||
"""seed_initial_data 幂等：普通用户只得 default、超管得全部租户。"""
|
||
from apps.core.models import Tenant
|
||
|
||
call_command("seed_initial_data")
|
||
call_command("seed_initial_data")
|
||
|
||
plain = django_user_model.objects.create_user("plain1", password="x")
|
||
root = django_user_model.objects.create_superuser(
|
||
"root1", password="x", email="r1@x.com"
|
||
)
|
||
call_command("seed_initial_data")
|
||
|
||
default = Tenant.objects.get(code="default")
|
||
demo = Tenant.objects.get(code="demo")
|
||
assert TenantMembership.objects.filter(user=plain, tenant=default).exists()
|
||
assert not TenantMembership.objects.filter(user=plain, tenant=demo).exists()
|
||
assert TenantMembership.objects.filter(user=root, tenant=default).exists()
|
||
assert TenantMembership.objects.filter(user=root, tenant=demo).exists()
|
||
|
||
before = TenantMembership.objects.count()
|
||
call_command("seed_initial_data")
|
||
assert TenantMembership.objects.count() == before
|