526 lines
20 KiB
Python
526 lines
20 KiB
Python
"""批次 D1 · B2B 订货商城测试。
|
||
|
||
覆盖：商品白名单可见性、商城价复用取价引擎、多单位下单、额度拦截、
|
||
SalesOrder 草稿流转、驳回、会话 token 安全、内部授权管理、billing 功能开关。
|
||
"""
|
||
|
||
import pytest
|
||
from datetime import date
|
||
from decimal import Decimal
|
||
from model_bakery import baker
|
||
from rest_framework.test import APIClient
|
||
from rest_framework_simplejwt.tokens import RefreshToken
|
||
|
||
from apps.catalog.models import Product, Unit, UnitConversion
|
||
from apps.inventory.models import Warehouse
|
||
from apps.partner.models import Customer, CustomerProductPrice
|
||
from apps.storefront.models import (
|
||
CustomerProductAuth, StorefrontAccount, StorefrontOrder,
|
||
issue_session_token, verify_session_token,
|
||
)
|
||
from apps.storefront import services as sf_services
|
||
|
||
|
||
@pytest.fixture
|
||
def client(db):
|
||
"""覆盖 pytest-django 默认的 Django 测试客户端。
|
||
|
||
默认 client 不认 `format="json"`，会把 JSON 结构按表单编码（导致 lines 变成
|
||
字符串列表）。商城用例都用 DRF 的 APIClient。
|
||
"""
|
||
return APIClient()
|
||
|
||
|
||
@pytest.fixture
|
||
def pro_plan(db, tenant):
|
||
"""默认给测试租户专业版：商城是专业版功能（billing 功能开关）。"""
|
||
from apps.billing.models import Plan, Subscription, seed_plans
|
||
|
||
seed_plans()
|
||
pro = Plan.objects.get(code="pro")
|
||
Subscription.objects.filter(tenant=tenant).delete()
|
||
Subscription.objects.create(
|
||
tenant=tenant, plan=pro, status="active",
|
||
period_start=date.today(), period_end=date.today(),
|
||
)
|
||
return pro
|
||
|
||
|
||
@pytest.fixture
|
||
def auth_client(db, user, tenant):
|
||
c = APIClient()
|
||
refresh = RefreshToken.for_user(user)
|
||
c.credentials(
|
||
HTTP_AUTHORIZATION=f"Bearer {refresh.access_token}",
|
||
HTTP_X_TENANT_ID=tenant.code,
|
||
)
|
||
return c
|
||
|
||
|
||
@pytest.fixture
|
||
def base_unit(db, tenant):
|
||
return baker.make(Unit, tenant=tenant, code="bottle", name="瓶", is_base=True)
|
||
|
||
|
||
@pytest.fixture
|
||
def box_unit(db, tenant):
|
||
return baker.make(Unit, tenant=tenant, code="box", name="箱", is_base=False)
|
||
|
||
|
||
@pytest.fixture
|
||
def customer(db, tenant):
|
||
return baker.make(Customer, tenant=tenant, code="C001", name="张三便利店",
|
||
phone="13900000001", credit_limit=Decimal("10000"))
|
||
|
||
|
||
@pytest.fixture
|
||
def warehouse(db, tenant):
|
||
return baker.make(Warehouse, tenant=tenant, code="WH01", name="主仓",
|
||
is_default=True, is_active=True)
|
||
|
||
|
||
@pytest.fixture
|
||
def products(db, tenant, base_unit, box_unit):
|
||
out = {}
|
||
for i, (code, name, price) in enumerate([
|
||
("P001", "可乐", "3.00"), ("P002", "雪碧", "3.00"),
|
||
("P003", "鲜牛奶", "15.00"),
|
||
], start=1):
|
||
p = baker.make(Product, tenant=tenant, code=code, name=name,
|
||
sale_price=Decimal(price), base_unit=base_unit,
|
||
status="active")
|
||
UnitConversion.objects.create(tenant=tenant, product=p, unit=box_unit,
|
||
rate=Decimal("24"))
|
||
out[code] = p
|
||
# 目录外商品（不授权）
|
||
out["SECRET"] = baker.make(Product, tenant=tenant, code="P999", name="未授权商品",
|
||
sale_price=Decimal("99"), base_unit=base_unit,
|
||
status="active")
|
||
return out
|
||
|
||
|
||
@pytest.fixture
|
||
def account(db, tenant, customer):
|
||
a = baker.make(StorefrontAccount, tenant=tenant, customer=customer,
|
||
phone="13900000001", display_name="张三便利店")
|
||
a.set_password("store12345")
|
||
a.save(update_fields=["password_hash"])
|
||
return a
|
||
|
||
|
||
@pytest.fixture
|
||
def granted(db, tenant, customer, products):
|
||
"""授权 P001/P002 给客户（P003、P999 不授权）。"""
|
||
for code in ("P001", "P002"):
|
||
CustomerProductAuth.objects.create(
|
||
tenant=tenant, customer=customer, product=products[code], is_active=True
|
||
)
|
||
return True
|
||
|
||
|
||
def _login(client, phone="13900000001", password="store12345", tenant="test"):
|
||
return client.post("/api/v1/storefront/login/",
|
||
{"phone": phone, "password": password, "tenant_code": tenant},
|
||
format="json")
|
||
|
||
|
||
# ---------- 会话 token ----------
|
||
|
||
|
||
def test_session_token_roundtrip(db, account):
|
||
token = issue_session_token(account)
|
||
got = verify_session_token(token)
|
||
assert got is not None
|
||
assert got.id == account.id
|
||
|
||
|
||
def test_session_token_rejects_tampering(db, account):
|
||
token = issue_session_token(account)
|
||
bad = token[:-1] + ("0" if token[-1] != "0" else "1")
|
||
assert verify_session_token(bad) is None
|
||
|
||
|
||
def test_session_token_rejects_expired(db, account, monkeypatch):
|
||
token = issue_session_token(account, ttl_hours=-1)
|
||
assert verify_session_token(token) is None
|
||
|
||
|
||
def test_password_hashing(db, account):
|
||
assert account.check_password("store12345") is True
|
||
assert account.check_password("wrong") is False
|
||
assert "store12345" not in account.password_hash
|
||
|
||
|
||
# ---------- 登录 ----------
|
||
|
||
|
||
def test_login_success(db, client, pro_plan, account):
|
||
resp = _login(client)
|
||
assert resp.status_code == 200, resp.content
|
||
body = resp.json()
|
||
assert body["token"]
|
||
assert body["customer"]["code"] == "C001"
|
||
|
||
|
||
def test_login_wrong_password(db, client, pro_plan, account):
|
||
resp = _login(client, password="nope")
|
||
assert resp.status_code == 401
|
||
assert resp.json()["code"] == "invalid_credentials"
|
||
|
||
|
||
def test_login_unknown_phone(db, client, pro_plan, account):
|
||
resp = _login(client, phone="13000000000")
|
||
assert resp.status_code == 401
|
||
|
||
|
||
# ---------- 可见性（白名单） ----------
|
||
|
||
|
||
def test_catalog_hides_unauthorized_products(db, client, pro_plan, tenant, account, products, granted):
|
||
token = _login(client).json()["token"]
|
||
resp = client.get("/api/v1/storefront/catalog/", HTTP_X_TENANT_ID=tenant.code,
|
||
HTTP_AUTHORIZATION=f"Storefront {token}")
|
||
assert resp.status_code == 200, resp.content
|
||
codes = {r["code"] for r in resp.json()["results"]}
|
||
assert codes == {"P001", "P002"}
|
||
assert "P999" not in codes # 未授权商品不可见
|
||
assert "P003" not in codes
|
||
|
||
|
||
def test_catalog_requires_login(db, client, pro_plan, tenant, granted):
|
||
resp = client.get("/api/v1/storefront/catalog/", HTTP_X_TENANT_ID=tenant.code)
|
||
assert resp.status_code == 401
|
||
|
||
|
||
def test_catalog_search(db, client, pro_plan, tenant, account, granted):
|
||
token = _login(client).json()["token"]
|
||
resp = client.get("/api/v1/storefront/catalog/?search=可乐", HTTP_X_TENANT_ID=tenant.code,
|
||
HTTP_AUTHORIZATION=f"Storefront {token}")
|
||
codes = [r["code"] for r in resp.json()["results"]]
|
||
assert codes == ["P001"]
|
||
|
||
|
||
# ---------- 商城价（复用取价引擎） ----------
|
||
|
||
|
||
def test_catalog_price_uses_customer_specific_price(db, client, pro_plan, tenant, account,
|
||
customer, products, granted):
|
||
CustomerProductPrice.objects.create(
|
||
tenant=tenant, customer=customer, product=products["P001"], price=Decimal("2.50")
|
||
)
|
||
token = _login(client).json()["token"]
|
||
resp = client.get("/api/v1/storefront/catalog/", HTTP_X_TENANT_ID=tenant.code,
|
||
HTTP_AUTHORIZATION=f"Storefront {token}")
|
||
row = next(r for r in resp.json()["results"] if r["code"] == "P001")
|
||
assert Decimal(row["price"]) == Decimal("2.50")
|
||
assert row["price_source"] == "customer"
|
||
|
||
|
||
def test_catalog_price_falls_back_to_default(db, client, pro_plan, tenant, account, products, granted):
|
||
token = _login(client).json()["token"]
|
||
resp = client.get("/api/v1/storefront/catalog/", HTTP_X_TENANT_ID=tenant.code,
|
||
HTTP_AUTHORIZATION=f"Storefront {token}")
|
||
row = next(r for r in resp.json()["results"] if r["code"] == "P002")
|
||
assert Decimal(row["price"]) == Decimal("3.00")
|
||
assert row["price_source"] == "default"
|
||
|
||
|
||
def test_catalog_lists_units_with_prices(db, client, pro_plan, tenant, account, products, granted):
|
||
token = _login(client).json()["token"]
|
||
resp = client.get("/api/v1/storefront/catalog/", HTTP_X_TENANT_ID=tenant.code,
|
||
HTTP_AUTHORIZATION=f"Storefront {token}")
|
||
row = next(r for r in resp.json()["results"] if r["code"] == "P001")
|
||
units = {u["unit_name"]: u for u in row["units"]}
|
||
assert Decimal(units["箱"]["rate"]) == Decimal("24")
|
||
assert Decimal(units["瓶"]["price"]) == Decimal("3.00")
|
||
assert Decimal(units["箱"]["price"]) == Decimal("72.00")
|
||
|
||
|
||
# ---------- 下单 ----------
|
||
|
||
|
||
def test_submit_order_creates_storefront_order(db, client, pro_plan, tenant, account,
|
||
products, granted):
|
||
token = _login(client).json()["token"]
|
||
resp = client.post("/api/v1/storefront/orders/", {
|
||
"lines": [{"product_id": products["P001"].id, "quantity": 10}],
|
||
"remark": "明天送",
|
||
}, format="json", HTTP_X_TENANT_ID=tenant.code,
|
||
HTTP_AUTHORIZATION=f"Storefront {token}")
|
||
assert resp.status_code == 201, resp.content
|
||
body = resp.json()
|
||
assert body["status"] == "submitted"
|
||
assert Decimal(body["total_amount"]) == Decimal("30.00")
|
||
|
||
order = StorefrontOrder.objects.get(pk=body["id"])
|
||
assert order.lines.count() == 1
|
||
assert order.lines.first().quantity == Decimal("10")
|
||
|
||
|
||
def test_submit_order_with_box_unit(db, client, pro_plan, tenant, account, products, granted):
|
||
"""按箱下单：数量折基本单位，单价按箱价。"""
|
||
box = UnitConversion.objects.get(product=products["P001"]).unit
|
||
token = _login(client).json()["token"]
|
||
resp = client.post("/api/v1/storefront/orders/", {
|
||
"lines": [{"product_id": products["P001"].id, "quantity": 2,
|
||
"source_unit": box.id}],
|
||
}, format="json", HTTP_X_TENANT_ID=tenant.code,
|
||
HTTP_AUTHORIZATION=f"Storefront {token}")
|
||
assert resp.status_code == 201, resp.content
|
||
order = StorefrontOrder.objects.get(pk=resp.json()["id"])
|
||
line = order.lines.first()
|
||
assert line.quantity == Decimal("48") # 2 × 24
|
||
assert line.source_quantity == Decimal("2")
|
||
assert Decimal(line.unit_price) == Decimal("72") # 3 × 24
|
||
assert Decimal(order.total_amount) == Decimal("144")
|
||
|
||
|
||
def test_submit_order_rejects_unauthorized_product(db, client, pro_plan, tenant, account,
|
||
products, granted):
|
||
token = _login(client).json()["token"]
|
||
resp = client.post("/api/v1/storefront/orders/", {
|
||
"lines": [{"product_id": products["SECRET"].id, "quantity": 1}],
|
||
}, format="json", HTTP_X_TENANT_ID=tenant.code,
|
||
HTTP_AUTHORIZATION=f"Storefront {token}")
|
||
assert resp.status_code == 403
|
||
assert resp.json()["code"] == "product_not_authorized"
|
||
assert StorefrontOrder.objects.count() == 0
|
||
|
||
|
||
def test_submit_order_blocks_over_credit(db, client, pro_plan, tenant, account, products, granted):
|
||
"""额度不足：提交即拦（402），不生成订单。"""
|
||
from apps.finance.models import Receivable
|
||
|
||
baker.make(Receivable, tenant=tenant, customer=account.customer, bill_no="RC-OLD",
|
||
bill_date=date.today(), total_amount=Decimal("9900"), status="open")
|
||
token = _login(client).json()["token"]
|
||
resp = client.post("/api/v1/storefront/orders/", {
|
||
"lines": [{"product_id": products["P001"].id, "quantity": 100}], # 300 元
|
||
}, format="json", HTTP_X_TENANT_ID=tenant.code,
|
||
HTTP_AUTHORIZATION=f"Storefront {token}")
|
||
assert resp.status_code == 402, resp.content
|
||
assert resp.json()["code"] == "credit_limit_exceeded"
|
||
assert StorefrontOrder.objects.count() == 0
|
||
|
||
|
||
def test_submit_order_rejects_zero_qty(db, client, pro_plan, tenant, account, products, granted):
|
||
token = _login(client).json()["token"]
|
||
resp = client.post("/api/v1/storefront/orders/", {
|
||
"lines": [{"product_id": products["P001"].id, "quantity": 0}],
|
||
}, format="json", HTTP_X_TENANT_ID=tenant.code,
|
||
HTTP_AUTHORIZATION=f"Storefront {token}")
|
||
assert resp.status_code == 400
|
||
|
||
|
||
def test_my_orders_list(db, client, pro_plan, tenant, account, products, granted):
|
||
token = _login(client).json()["token"]
|
||
client.post("/api/v1/storefront/orders/", {
|
||
"lines": [{"product_id": products["P001"].id, "quantity": 5}],
|
||
}, format="json", HTTP_X_TENANT_ID=tenant.code,
|
||
HTTP_AUTHORIZATION=f"Storefront {token}")
|
||
resp = client.get("/api/v1/storefront/orders/", HTTP_X_TENANT_ID=tenant.code,
|
||
HTTP_AUTHORIZATION=f"Storefront {token}")
|
||
assert resp.status_code == 200
|
||
assert len(resp.json()["results"]) == 1
|
||
|
||
|
||
# ---------- 内部端流转 ----------
|
||
|
||
|
||
def test_admin_confirm_creates_sales_order_draft(db, auth_client, pro_plan, tenant, account,
|
||
products, granted, warehouse):
|
||
order = sf_services.submit_order(
|
||
account.customer, account=account,
|
||
lines=[{"product_id": products["P001"].id, "quantity": 10}],
|
||
)
|
||
resp = auth_client.post(f"/api/v1/storefront/admin/orders/{order.id}/confirm/",
|
||
{}, format="json")
|
||
assert resp.status_code == 200, resp.content
|
||
body = resp.json()
|
||
assert body["ok"] is True
|
||
assert body["sales_order_id"]
|
||
|
||
from apps.sales.models import SalesOrder
|
||
|
||
so = SalesOrder.objects.get(pk=body["sales_order_id"])
|
||
assert so.state == "draft" # 草稿，未过账
|
||
assert so.lines.count() == 1
|
||
order.refresh_from_db()
|
||
assert order.status == "confirmed"
|
||
|
||
|
||
def test_admin_confirm_picks_default_warehouse(db, auth_client, pro_plan, tenant, account,
|
||
products, granted, warehouse):
|
||
"""不传 warehouse 时自动选默认仓库。"""
|
||
order = sf_services.submit_order(
|
||
account.customer, lines=[{"product_id": products["P001"].id, "quantity": 1}],
|
||
)
|
||
resp = auth_client.post(f"/api/v1/storefront/admin/orders/{order.id}/confirm/",
|
||
{}, format="json")
|
||
assert resp.status_code == 200, resp.content
|
||
|
||
|
||
def test_admin_reject_order(db, auth_client, pro_plan, tenant, account, products, granted):
|
||
order = sf_services.submit_order(
|
||
account.customer, lines=[{"product_id": products["P001"].id, "quantity": 1}],
|
||
)
|
||
resp = auth_client.post(f"/api/v1/storefront/admin/orders/{order.id}/reject/",
|
||
{"reason": "库存不足"}, format="json")
|
||
assert resp.status_code == 200, resp.content
|
||
order.refresh_from_db()
|
||
assert order.status == "rejected"
|
||
assert "库存不足" in order.remark
|
||
|
||
|
||
def test_admin_orders_list_default_submitted(db, auth_client, pro_plan, tenant, account,
|
||
products, granted, warehouse):
|
||
o1 = sf_services.submit_order(
|
||
account.customer, lines=[{"product_id": products["P001"].id, "quantity": 1}],
|
||
)
|
||
o2 = sf_services.submit_order(
|
||
account.customer, lines=[{"product_id": products["P002"].id, "quantity": 1}],
|
||
)
|
||
sf_services.confirm_order(o2, warehouse=warehouse)
|
||
|
||
resp = auth_client.get("/api/v1/storefront/admin/orders/")
|
||
assert resp.status_code == 200
|
||
orders = resp.json()["results"]
|
||
assert len(orders) == 1
|
||
assert orders[0]["id"] == o1.id
|
||
|
||
|
||
def test_admin_orders_pagination(db, auth_client, pro_plan, tenant, account,
|
||
products, granted, warehouse):
|
||
"""P2-1：admin_orders 支持 page/page_size 翻页，count 为过滤后总数。"""
|
||
ids = []
|
||
for _ in range(3):
|
||
o = sf_services.submit_order(
|
||
account.customer,
|
||
lines=[{"product_id": products["P001"].id, "quantity": 1}],
|
||
)
|
||
ids.append(o.id)
|
||
p1 = auth_client.get(
|
||
"/api/v1/storefront/admin/orders/",
|
||
{"status": "submitted", "page": 1, "page_size": 2},
|
||
)
|
||
assert p1.status_code == 200, p1.content
|
||
b1 = p1.json()
|
||
assert b1["count"] == 3
|
||
assert len(b1["results"]) == 2
|
||
p2 = auth_client.get(
|
||
"/api/v1/storefront/admin/orders/",
|
||
{"status": "submitted", "page": 2, "page_size": 2},
|
||
)
|
||
assert p2.status_code == 200, p2.content
|
||
b2 = p2.json()
|
||
assert b2["count"] == 3
|
||
assert len(b2["results"]) == 1
|
||
# 两页无重叠
|
||
got = {r["id"] for r in b1["results"]} | {r["id"] for r in b2["results"]}
|
||
assert got == set(ids)
|
||
|
||
|
||
def test_admin_confirm_twice_rejected(db, auth_client, pro_plan, tenant, account,
|
||
products, granted, warehouse):
|
||
order = sf_services.submit_order(
|
||
account.customer, lines=[{"product_id": products["P001"].id, "quantity": 1}],
|
||
)
|
||
auth_client.post(f"/api/v1/storefront/admin/orders/{order.id}/confirm/", {}, format="json")
|
||
resp2 = auth_client.post(f"/api/v1/storefront/admin/orders/{order.id}/confirm/",
|
||
{}, format="json")
|
||
assert resp2.status_code == 400
|
||
|
||
|
||
# ---------- 授权管理 ----------
|
||
|
||
|
||
def test_admin_grant_and_revoke(db, auth_client, tenant, customer, products):
|
||
resp = auth_client.post("/api/v1/storefront/admin/auths/", {
|
||
"customer_id": customer.id,
|
||
"product_ids": [products["P001"].id, products["P002"].id],
|
||
}, format="json")
|
||
assert resp.status_code == 201, resp.content
|
||
assert resp.json()["created"] == 2
|
||
|
||
listing = auth_client.get(
|
||
f"/api/v1/storefront/admin/auths/?customer_id={customer.id}"
|
||
).json()["results"]
|
||
assert len(listing) == 2
|
||
|
||
auth_id = listing[0]["id"]
|
||
resp2 = auth_client.delete(f"/api/v1/storefront/admin/auths/?id={auth_id}")
|
||
assert resp2.status_code == 200
|
||
# 撤销后不可见
|
||
assert not CustomerProductAuth.objects.filter(
|
||
pk=auth_id, is_active=True, is_deleted=False
|
||
).exists()
|
||
|
||
|
||
def test_admin_create_account(db, auth_client, tenant, customer):
|
||
resp = auth_client.post("/api/v1/storefront/admin/accounts/", {
|
||
"customer_id": customer.id, "phone": "13700000001",
|
||
"password": "abc12345",
|
||
}, format="json")
|
||
assert resp.status_code == 201, resp.content
|
||
acc = StorefrontAccount.objects.get(phone="13700000001")
|
||
assert acc.check_password("abc12345")
|
||
|
||
|
||
# ---------- 租户隔离 ----------
|
||
|
||
|
||
def test_cross_tenant_product_not_visible(db, client, pro_plan, tenant, other_tenant,
|
||
account, granted, products):
|
||
"""另一租户的商品不会出现在本租户商城目录里。"""
|
||
other_p = baker.make(Product, tenant=other_tenant, code="X001", name="别家商品",
|
||
sale_price=Decimal("1"), status="active")
|
||
CustomerProductAuth.objects.create(
|
||
tenant=other_tenant, customer=account.customer, product=other_p, is_active=True
|
||
) # 即便误建了跨租户授权
|
||
token = _login(client).json()["token"]
|
||
resp = client.get("/api/v1/storefront/catalog/", HTTP_X_TENANT_ID=tenant.code,
|
||
HTTP_AUTHORIZATION=f"Storefront {token}")
|
||
codes = {r["code"] for r in resp.json()["results"]}
|
||
assert "X001" not in codes
|
||
|
||
|
||
def test_catalog_excludes_inactive_products(db, client, pro_plan, tenant, account,
|
||
products, granted):
|
||
products["P001"].status = "inactive"
|
||
products["P001"].save()
|
||
token = _login(client).json()["token"]
|
||
resp = client.get("/api/v1/storefront/catalog/", HTTP_X_TENANT_ID=tenant.code,
|
||
HTTP_AUTHORIZATION=f"Storefront {token}")
|
||
codes = {r["code"] for r in resp.json()["results"]}
|
||
assert "P001" not in codes
|
||
|
||
|
||
# ---------- billing 功能开关 ----------
|
||
|
||
|
||
def test_storefront_blocked_on_free_plan(db, client, tenant, account, products, granted):
|
||
"""免费版无订货商城 → 登录被 billing 功能开关拒绝（403 + 升级引导）。"""
|
||
from apps.billing.models import Plan, Subscription, seed_plans
|
||
|
||
seed_plans()
|
||
free = Plan.objects.get(code="free")
|
||
Subscription.objects.create(tenant=tenant, plan=free, status="active",
|
||
period_start=date.today(),
|
||
period_end=date.today())
|
||
resp = _login(client)
|
||
assert resp.status_code == 403, resp.content
|
||
assert resp.json()["code"] == "quota_exceeded"
|
||
assert resp.json()["kind"] == "storefront"
|
||
|
||
|
||
def test_storefront_allowed_on_pro_plan(db, client, tenant, account, products, granted):
|
||
from apps.billing.models import Plan, Subscription, seed_plans
|
||
|
||
seed_plans()
|
||
pro = Plan.objects.get(code="pro")
|
||
Subscription.objects.create(tenant=tenant, plan=pro, status="active",
|
||
period_start=date.today(),
|
||
period_end=date.today())
|
||
resp = _login(client)
|
||
assert resp.status_code == 200, resp.content
|