诚实性修复: API限流Retry-After(wait=60s)+断言 / purge定时接线(04:00 cron+actor+测试) / Channel表单校验(ref+rules+validate) / README PG数字513

2026-09-12 02:20:50 +08:00
parent b3f3095d53
commit 67540e8d30
40 changed files with 1805 additions and 31 deletions
@@ -0,0 +1,149 @@
+"""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