"""批次 C2 · 演示账套与免注册体验测试。 覆盖：seed_demo 幂等重建、数据构成（批次/多单位/账龄六桶/风险客户）、 免注册 token 签发、演示租户只读保护（含手工 curl 场景）、只读不误伤正常租户。 """ import pytest from datetime import date from decimal import Decimal from django.core.management import call_command from rest_framework.test import APIClient @pytest.fixture def demo_db(db): """跑一次 seed_demo（测试库内）。""" call_command("seed_demo") return True @pytest.fixture def demo_client(demo_db): return APIClient() # ---------- seed_demo 数据构成 ---------- def test_seed_demo_creates_tenant_and_user(demo_db): from django.contrib.auth import get_user_model from apps.core.models import Tenant tenant = Tenant.objects.get(code="demo") assert tenant.is_active is True user = get_user_model().objects.get(username="demo") assert user.is_staff is False assert user.is_superuser is False assert user.check_password("demo12345") def test_seed_demo_idempotent(demo_db): """重复执行不产生重复单据。""" from apps.purchase.models import PurchaseBill from apps.sales.models import SalesBill from apps.finance.models import Receivable before = (SalesBill.objects.filter(tenant__code="demo").count(), PurchaseBill.objects.filter(tenant__code="demo").count(), Receivable.objects.filter(tenant__code="demo").count()) call_command("seed_demo") after = (SalesBill.objects.filter(tenant__code="demo").count(), PurchaseBill.objects.filter(tenant__code="demo").count(), Receivable.objects.filter(tenant__code="demo").count()) assert before == after, f"重复执行产生了新数据：{before} → {after}" def test_seed_demo_products_feature_matrix(demo_db): """商品覆盖：多单位、批次效期、称重（kg 基本单位）、最低售价。""" from apps.catalog.models import Product, UnitConversion tenant_products = Product.objects.filter(tenant__code="demo") # 扩充后 22 个 SKU（覆盖 4 品类 / 批次 / 称重 / 多单位 / 3 档税率） assert tenant_products.count() >= 20 cola = tenant_products.get(code="P1001") assert cola.min_sale_price == Decimal("1.6000") assert UnitConversion.objects.filter(product=cola).count() >= 2 # 箱 + 件 milk = tenant_products.get(code="P2001") assert milk.is_batch_managed is True assert milk.shelf_life_days == 21 # 称重商品用 kg 作基本单位 bulk = tenant_products.get(code="P4003") assert bulk.base_unit.code == "kg" # 零食类用袋作基本单位（多单位体系里 bag 也是基本单位） snack = tenant_products.get(code="P3001") assert snack.base_unit.code == "bag" def test_seed_demo_batches_have_expiry(demo_db): from apps.inventory.models import StockBatch batches = StockBatch.objects.filter(tenant__code="demo", on_hand__gt=0) assert batches.count() >= 2 assert all(b.expiry_date is not None for b in batches) # 至少一个近效期（≤30 天）用于演示预警 today = date.today() assert any((b.expiry_date - today).days <= 30 for b in batches) def test_seed_demo_aging_covers_buckets(demo_db): """账龄六桶都要有分布（否则演示图只有一根柱）。""" from apps.core.models import Tenant from apps.finance.services import receivable_aging tenant = Tenant.objects.get(code="demo") aging = receivable_aging(tenant) non_zero = { b["bucket"] for b in aging["buckets"] if Decimal(b["amount"]) > 0 } assert non_zero == {"0-30", "31-60", "61-90", "91-180", "181-365", "365+"}, aging assert Decimal(aging["total"]) > 0 def test_seed_demo_has_risk_customer(demo_db): """风险客户（C1005 长期欠款未开单）应得到中高分数。""" from apps.core.models import Tenant from apps.ai import risk as ai_risk tenant = Tenant.objects.get(code="demo") ranking = ai_risk.risk_ranking(tenant, top_n=5) assert ranking, "演示账套应有风险客户" top = ranking[0] assert top["outstanding"] > 0 assert top["score"] > 0 def test_seed_demo_reset_clears_and_rebuilds(demo_db): """reset 后重建业务数据和商城数据，且剧本单 + 历史单都在。""" from apps.catalog.models import Product from apps.core.models import Tenant from apps.partner.models import Customer from apps.sales.models import SalesBill from apps.storefront.models import ( CustomerProductAuth, StorefrontAccount, StorefrontOrder, StorefrontOrderLine, ) tenant = Tenant.objects.get(code="demo") customer = Customer.objects.get(tenant=tenant, code="C1001") # Use a product outside the seeded first-four whitelist so the stale row # does not collide with the idempotent seed data's unique constraint. product = Product.objects.get(tenant=tenant, code="P1005") stale_account = StorefrontAccount.objects.create( tenant=tenant, customer=customer, phone="13999999999", display_name="待清理商城账号", ) stale_auth = CustomerProductAuth.objects.create( tenant=tenant, customer=customer, product=product, is_active=True, ) stale_order = StorefrontOrder.objects.create( tenant=tenant, customer=customer, account=stale_account, order_no="STale-DEMO-RESET", total_amount=Decimal("2.0000"), ) StorefrontOrderLine.objects.create( tenant=tenant, order=stale_order, product=product, quantity=Decimal("1"), unit_price=Decimal("2"), amount=Decimal("2"), ) call_command("seed_demo", "--reset") n = SalesBill.objects.filter(tenant=tenant).count() # 固定种子下应有 10 张剧本单 + 数十张历史单 assert n >= 60, f"重建后单据过少：{n}" assert not StorefrontAccount.objects.filter(pk=stale_account.pk).exists() assert not CustomerProductAuth.objects.filter(pk=stale_auth.pk).exists() assert not StorefrontOrder.objects.filter(pk=stale_order.pk).exists() assert StorefrontAccount.objects.filter(tenant=tenant).count() == customer_count(tenant) def customer_count(tenant): """Expected one storefront account per active demo customer.""" from apps.partner.models import Customer return Customer.objects.filter(tenant=tenant, is_active=True).count() # ---------- 免注册进入 ---------- def test_enter_demo_returns_token(demo_client): resp = demo_client.post("/api/v1/demo/enter/", {}, format="json") assert resp.status_code == 200, resp.content body = resp.json() assert body["tenant"] == "demo" assert body["read_only"] is True assert body["access"] def test_enter_demo_token_works_for_reads(demo_client): token = demo_client.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 assert resp.json()["count"] >= 6 def test_enter_demo_503_when_not_seeded(db): c = APIClient() resp = c.post("/api/v1/demo/enter/", {}, format="json") assert resp.status_code == 503 assert resp.json()["code"] == "demo_not_ready" # ---------- 只读保护 ---------- def test_demo_read_only_blocks_writes(demo_client): """演示租户任何写操作被 403（即使拿到有效 token）。""" token = demo_client.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.post("/api/v1/catalog/products/", {"code": "HACK01", "name": "不该被创建"}, format="json") assert resp.status_code == 403, resp.content assert resp.json()["code"] == "demo_read_only" resp2 = c.post("/api/v1/sales/bills/create-bill/", { "customer": 1, "warehouse": 1, "lines": [{"product": 1, "quantity": 1, "unit_price": 1}], }, format="json") assert resp2.status_code == 403 assert resp2.json()["code"] == "demo_read_only" def test_demo_read_only_allows_reads(demo_client): """读接口 + AI 问答（POST 但只读语义）在演示租户下放行。""" token = demo_client.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/report/dashboard/summary/") assert resp.status_code == 200, resp.content def test_demo_middleware_does_not_affect_other_tenants(db, tenant, django_user_model): """正常租户写操作不受演示只读拦截。""" from rest_framework_simplejwt.tokens import RefreshToken from apps.core.models import TenantMembership user = django_user_model.objects.create_user(username="normal", password="pass12345") # This case is about the demo read-only middleware, not authorization. # Give the user the membership the permission layer requires so the write # actually reaches the middleware check. TenantMembership.objects.create(user=user, tenant=tenant, role="member") c = APIClient() c.credentials(HTTP_AUTHORIZATION=f"Bearer {RefreshToken.for_user(user).access_token}", HTTP_X_TENANT_ID=tenant.code) resp = c.post("/api/v1/catalog/products/", {"code": "NORMAL01", "name": "正常商品"}, format="json") assert resp.status_code == 201, resp.content def test_demo_enter_is_public(db): """进入演示不需要 JWT。""" c = APIClient() # 未初始化时应是 503（而非 401）→ 证明鉴权放行 resp = c.post("/api/v1/demo/enter/", {}, format="json") assert resp.status_code in (200, 503) assert resp.status_code != 401 def test_seed_demo_rich_dataset(demo_db): """扩充后的数据量足以支撑报表与 AI 演示。""" from apps.core.models import Tenant from apps.inventory.models import StockBatch, StockMovement from apps.partner.models import Customer from apps.catalog.models import Product from apps.sales.models import SalesBill t = Tenant.objects.get(code="demo") assert Product.objects.filter(tenant=t).count() >= 20 assert Customer.objects.filter(tenant=t).count() >= 12 assert SalesBill.objects.filter(tenant=t).count() >= 60 assert StockMovement.objects.filter(tenant=t).count() >= 100 assert StockBatch.objects.filter(tenant=t, on_hand__gt=0).count() >= 5 def test_seed_demo_spans_multiple_months(demo_db): """历史单据必须跨多个月份（否则报表趋势是一条直线）。""" from django.db.models import Count from django.db.models.functions import TruncMonth from apps.core.models import Tenant from apps.sales.models import SalesBill t = Tenant.objects.get(code="demo") months = ( SalesBill.objects.filter(tenant=t, state="confirmed") .annotate(m=TruncMonth("bill_date")).values("m") .annotate(n=Count("id")).count() ) assert months >= 4, f"只有 {months} 个月份的数据，趋势图无意义" def test_seed_demo_risk_differentiation(demo_db): """AI 风险分必须有区分度：至少一个高风险 + 多数低风险。""" from apps.ai import risk as ai_risk from apps.core.models import Tenant from apps.partner.models import Customer t = Tenant.objects.get(code="demo") # 给风险户补上"近期慢回款"画像（seed 已生成，这里直接评估） scores = [] for c in Customer.objects.filter(tenant=t): d = ai_risk.score_customer(t, c) if d["outstanding"] > 0: scores.append(d["score"]) assert scores, "没有有待收余额的客户" assert max(scores) >= 60, f"最高风险分只有 {max(scores):.1f}，无高风险样本" assert max(scores) - min(scores) >= 30, "风险分缺乏区分度"