372 lines
14 KiB
Python
372 lines
14 KiB
Python
"""批次 B2/B3 测试：AI 开单（抽取→匹配→配额）+ AI 经营问答（白名单取数→降级）。"""
|
||
|
||
import pytest
|
||
from datetime import date, timedelta
|
||
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
|
||
from apps.finance.models import Receivable
|
||
from apps.ai import orders as ai_orders
|
||
from apps.ai import ask as ai_ask
|
||
from apps.ai import usage as ai_usage
|
||
from apps.ai.models import AiUsage
|
||
|
||
|
||
@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 cola(db, tenant, base_unit, box_unit):
|
||
p = baker.make(Product, tenant=tenant, code="P001", name="可乐",
|
||
barcode="6901234567890", sale_price=Decimal("3"),
|
||
base_unit=base_unit)
|
||
UnitConversion.objects.create(tenant=tenant, product=p, unit=box_unit,
|
||
rate=Decimal("24"))
|
||
return p
|
||
|
||
|
||
@pytest.fixture
|
||
def milk(db, tenant, base_unit):
|
||
return baker.make(Product, tenant=tenant, code="P002", name="鲜牛奶",
|
||
sale_price=Decimal("15"), base_unit=base_unit)
|
||
|
||
|
||
@pytest.fixture
|
||
def warehouse(db, tenant):
|
||
return baker.make(Warehouse, tenant=tenant, code="WH01", name="主仓")
|
||
|
||
|
||
@pytest.fixture
|
||
def customer(db, tenant):
|
||
return baker.make(Customer, tenant=tenant, code="C001", name="张三便利店",
|
||
credit_limit=Decimal("10000"))
|
||
|
||
|
||
@pytest.fixture
|
||
def ai_quota(db, tenant):
|
||
"""把租户挂到一个自定义套餐上以控制 AI 配额（billing 装好后配额来源是套餐）。"""
|
||
from apps.billing.models import Plan, Subscription, seed_plans
|
||
|
||
seed_plans()
|
||
|
||
def _set(parse_order: int = 10, ask: int = 20):
|
||
plan, _created = Plan.objects.update_or_create(
|
||
code="test-quota", defaults={
|
||
"name": "测试套餐", "price_monthly": Decimal("0"),
|
||
"limits": {
|
||
"users": 99, "products": 99999, "bills_monthly": 0,
|
||
"ai_parse_order": parse_order, "ai_ask": ask,
|
||
"batch_managed": True, "finance_ledger": True,
|
||
"print_templates": 0, "storefront": False,
|
||
},
|
||
"sort_order": 99, "is_active": True,
|
||
},
|
||
)
|
||
sub = Subscription.objects.filter(tenant=tenant).first()
|
||
if sub is None:
|
||
Sub = Subscription
|
||
sub = Sub.objects.create(
|
||
tenant=tenant, plan=plan, status="active",
|
||
period_start=date.today(), period_end=date.today() + timedelta(days=30),
|
||
)
|
||
else:
|
||
sub.plan = plan
|
||
sub.status = "active"
|
||
sub.save(update_fields=["plan", "status", "updated_at"])
|
||
return plan
|
||
|
||
return _set
|
||
|
||
|
||
# ============================================================
|
||
# 抽取（无 LLM → 明确报错 / 规则兜底）
|
||
# ============================================================
|
||
|
||
|
||
def test_extract_raises_without_llm(db, settings):
|
||
"""无 KEY 时默认抛 LlmUnavailable（不做静默降级）。"""
|
||
settings.AI_PROVIDER = ""
|
||
settings.AI_API_KEY = ""
|
||
with pytest.raises(ai_orders.LlmUnavailable):
|
||
ai_orders.extract_items("可乐 2 箱")
|
||
|
||
|
||
def test_extract_rule_fallback_when_allowed(db, settings):
|
||
"""显式允许时走规则兜底：抓 "商品名 + 数量 + 单位"。"""
|
||
settings.AI_PROVIDER = ""
|
||
settings.AI_API_KEY = ""
|
||
items = ai_orders.extract_items("可乐 2箱\n鲜牛奶 3瓶", allow_rule_fallback=True)
|
||
assert len(items) == 2
|
||
assert items[0]["name"] == "可乐"
|
||
assert items[0]["qty"] == 2.0
|
||
assert items[0]["unit"] == "箱"
|
||
assert items[1]["name"] == "鲜牛奶"
|
||
|
||
|
||
def test_extract_rule_fallback_chinese_numbers(db, settings):
|
||
settings.AI_PROVIDER = ""
|
||
settings.AI_API_KEY = ""
|
||
items = ai_orders.extract_items("可乐 两箱，鲜牛奶 十瓶", allow_rule_fallback=True)
|
||
assert items[0]["qty"] == 2.0
|
||
assert items[1]["qty"] == 10.0
|
||
|
||
|
||
# ============================================================
|
||
# 商品匹配
|
||
# ============================================================
|
||
|
||
|
||
def test_match_by_barcode(db, tenant, cola):
|
||
out = ai_orders.match_products(tenant, [
|
||
{"name": "随便写的名字", "barcode": "6901234567890", "qty": 1, "unit": ""},
|
||
])
|
||
assert len(out["matched"]) == 1
|
||
assert out["matched"][0]["match_by"] == "barcode"
|
||
assert out["matched"][0]["product_code"] == "P001"
|
||
|
||
|
||
def test_match_by_code_exact(db, tenant, cola, milk):
|
||
out = ai_orders.match_products(tenant, [
|
||
{"name": "P002", "barcode": "", "qty": 2, "unit": ""},
|
||
])
|
||
assert out["matched"][0]["product_code"] == "P002"
|
||
assert out["matched"][0]["match_by"] == "code"
|
||
|
||
|
||
def test_match_by_name_exact_and_contains(db, tenant, cola, milk):
|
||
out = ai_orders.match_products(tenant, [
|
||
{"name": "可乐", "barcode": "", "qty": 1, "unit": ""},
|
||
{"name": "牛奶", "barcode": "", "qty": 1, "unit": ""}, # 包含匹配 鲜牛奶
|
||
])
|
||
assert len(out["matched"]) == 2
|
||
assert out["matched"][0]["match_by"] == "name_exact"
|
||
assert out["matched"][1]["match_by"] == "name_contains"
|
||
assert out["matched"][1]["product_name"] == "鲜牛奶"
|
||
|
||
|
||
def test_match_unmatched_reported(db, tenant, cola):
|
||
out = ai_orders.match_products(tenant, [
|
||
{"name": "完全不存在的东西XYZ", "barcode": "", "qty": 1, "unit": ""},
|
||
])
|
||
assert out["matched"] == []
|
||
assert len(out["unmatched"]) == 1
|
||
|
||
|
||
def test_match_resolves_unit_and_price(db, tenant, cola, box_unit):
|
||
def resolver(tenant_, product, item):
|
||
if item.get("unit") == "箱":
|
||
conv = UnitConversion.objects.get(product=product, unit=box_unit)
|
||
return {"unit_id": conv.unit_id, "unit_name": conv.unit.name,
|
||
"price": str(product.sale_price * conv.rate)}
|
||
return {"unit_id": None, "unit_name": "", "price": str(product.sale_price)}
|
||
|
||
out = ai_orders.match_products(
|
||
tenant,
|
||
[{"name": "可乐", "barcode": "", "qty": 2, "unit": "箱"}],
|
||
unit_resolver=resolver,
|
||
)
|
||
row = out["matched"][0]
|
||
assert row["unit_name"] == "箱"
|
||
assert Decimal(row["price"]) == Decimal("72") # 3 × 24
|
||
|
||
|
||
# ============================================================
|
||
# 配额
|
||
# ============================================================
|
||
|
||
|
||
def test_quota_check_and_exceed(db, tenant, ai_quota):
|
||
ai_quota(parse_order=2)
|
||
assert ai_usage.check_quota(tenant, ai_usage.KIND_PARSE_ORDER)["quota"] == 2
|
||
ai_usage.record_usage(tenant, ai_usage.KIND_PARSE_ORDER)
|
||
ai_usage.record_usage(tenant, ai_usage.KIND_PARSE_ORDER)
|
||
with pytest.raises(ai_usage.QuotaExceeded):
|
||
ai_usage.check_quota(tenant, ai_usage.KIND_PARSE_ORDER)
|
||
|
||
|
||
def test_quota_zero_means_unlimited(db, tenant, ai_quota):
|
||
ai_quota(parse_order=0)
|
||
for _ in range(5):
|
||
ai_usage.record_usage(tenant, ai_usage.KIND_PARSE_ORDER)
|
||
out = ai_usage.check_quota(tenant, ai_usage.KIND_PARSE_ORDER)
|
||
assert out["allowed"] is True
|
||
assert out["remaining"] is None
|
||
|
||
|
||
def test_usage_summary_counts_per_kind(db, tenant):
|
||
ai_usage.record_usage(tenant, ai_usage.KIND_PARSE_ORDER)
|
||
ai_usage.record_usage(tenant, ai_usage.KIND_ASK)
|
||
ai_usage.record_usage(tenant, ai_usage.KIND_ASK)
|
||
summary = ai_usage.usage_summary(tenant)
|
||
by_kind = {i["kind"]: i["used"] for i in summary["items"]}
|
||
assert by_kind[ai_usage.KIND_PARSE_ORDER] == 1
|
||
assert by_kind[ai_usage.KIND_ASK] == 2
|
||
|
||
|
||
# ============================================================
|
||
# API：AI 开单
|
||
# ============================================================
|
||
|
||
|
||
def test_parse_order_api_llm_unavailable_400(db, auth_client, settings, cola):
|
||
settings.AI_PROVIDER = ""
|
||
settings.AI_API_KEY = ""
|
||
resp = auth_client.post("/api/v1/ai/parse-order/", {"text": "可乐 2 箱"}, format="json")
|
||
assert resp.status_code == 400, resp.content
|
||
assert resp.json()["code"] == "llm_unavailable"
|
||
|
||
|
||
def test_parse_order_api_rule_fallback(db, auth_client, settings, cola, milk):
|
||
"""显式开兜底 → 走规则抽取 + 匹配，全链路跑通（不依赖 LLM）。"""
|
||
settings.AI_PROVIDER = ""
|
||
settings.AI_API_KEY = ""
|
||
resp = auth_client.post("/api/v1/ai/parse-order/", {
|
||
"text": "可乐 2箱\n鲜牛奶 3瓶",
|
||
"allow_rule_fallback": True,
|
||
}, format="json")
|
||
assert resp.status_code == 200, resp.content
|
||
body = resp.json()
|
||
assert len(body["matched"]) == 2
|
||
assert body["matched"][0]["product_code"] == "P001"
|
||
assert body["matched"][1]["product_code"] == "P002"
|
||
# 用量已登记
|
||
assert AiUsage.objects.filter(kind=ai_usage.KIND_PARSE_ORDER).count() == 1
|
||
|
||
|
||
def test_parse_order_api_quota_403(db, auth_client, tenant, settings, ai_quota, cola):
|
||
settings.AI_PROVIDER = ""
|
||
settings.AI_API_KEY = ""
|
||
ai_quota(parse_order=1)
|
||
ai_usage.record_usage(tenant, ai_usage.KIND_PARSE_ORDER)
|
||
|
||
resp = auth_client.post("/api/v1/ai/parse-order/", {
|
||
"text": "可乐 1箱", "allow_rule_fallback": True,
|
||
}, format="json")
|
||
assert resp.status_code == 403, resp.content
|
||
assert resp.json()["code"] == "quota_exceeded"
|
||
|
||
|
||
def test_parse_order_empty_text_400(db, auth_client):
|
||
resp = auth_client.post("/api/v1/ai/parse-order/", {"text": " "}, format="json")
|
||
assert resp.status_code == 400
|
||
|
||
|
||
def test_usage_api(db, auth_client):
|
||
resp = auth_client.get("/api/v1/ai/usage/")
|
||
assert resp.status_code == 200, resp.content
|
||
body = resp.json()
|
||
assert body["llm_available"] is False
|
||
assert len(body["items"]) == 3
|
||
|
||
|
||
# ============================================================
|
||
# B3 经营问答
|
||
# ============================================================
|
||
|
||
|
||
def test_detect_intents():
|
||
assert ai_ask.detect_intents("这个月销售额多少") == ["sales"]
|
||
assert set(ai_ask.detect_intents("应收和库存怎么样")) == {"receivable", "inventory"}
|
||
assert ai_ask.detect_intents("今天天气如何") == ["overview"]
|
||
|
||
|
||
def test_gather_tools_readonly_and_scoped(db, tenant, other_tenant, cola, warehouse):
|
||
"""工具取数只读且带租户隔离：别的租户数据不进结果。"""
|
||
out = ai_ask.gather(tenant, "应收账款情况")
|
||
assert "receivable" in out["data"]
|
||
assert "应收未结合计" in out["data"]["receivable"]
|
||
# 另一租户有应收，但不得出现在本租户结果里
|
||
other_customer = baker.make(Customer, tenant=other_tenant, code="OC1", name="别家客户")
|
||
baker.make(Receivable, tenant=other_tenant, customer=other_customer,
|
||
bill_no="RC-OTHER", bill_date=date.today(),
|
||
total_amount=Decimal("9999"), status="open")
|
||
out2 = ai_ask.gather(tenant, "应收账款情况")
|
||
assert Decimal(str(out2["data"]["receivable"]["应收未结合计"])) == Decimal("0")
|
||
|
||
|
||
def test_answer_falls_back_without_llm(db, tenant, settings):
|
||
settings.AI_PROVIDER = ""
|
||
settings.AI_API_KEY = ""
|
||
out = ai_ask.answer(tenant, "本月销售额和应收怎么样")
|
||
assert out["llm_enhanced"] is False
|
||
assert out["answer"]
|
||
assert "本月销售额" in out["answer"] or "应收未结" in out["answer"]
|
||
|
||
|
||
def test_answer_with_data(db, tenant, customer, warehouse, cola):
|
||
"""有单据数据时，统计答案里能读到金额。"""
|
||
from apps.inventory import services as inv
|
||
from apps.sales import services as sales_services
|
||
|
||
inv.inbound(tenant=tenant, warehouse=warehouse, product=cola,
|
||
quantity=Decimal("100"), unit_cost=Decimal("2"))
|
||
bill = sales_services.create_sales_bill(
|
||
tenant=tenant, customer=customer, warehouse=warehouse,
|
||
lines=[{"product": cola, "quantity": 10, "unit_price": 3}],
|
||
)
|
||
sales_services.confirm_sales_bill(bill)
|
||
|
||
out = ai_ask.answer(tenant, "本月销售额多少")
|
||
assert "30.00" in out["answer"]
|
||
|
||
|
||
def test_ask_api_without_llm_still_answers(db, auth_client, tenant, settings):
|
||
settings.AI_PROVIDER = ""
|
||
settings.AI_API_KEY = ""
|
||
resp = auth_client.post("/api/v1/ai/ask/", {"question": "本月销售和应收情况"},
|
||
format="json")
|
||
assert resp.status_code == 200, resp.content
|
||
body = resp.json()
|
||
assert body["llm_enhanced"] is False
|
||
assert body["answer"]
|
||
assert "sales" in body["intents"] or "receivable" in body["intents"]
|
||
|
||
|
||
def test_ask_api_requires_question(db, auth_client):
|
||
resp = auth_client.post("/api/v1/ai/ask/", {}, format="json")
|
||
assert resp.status_code == 400
|
||
|
||
|
||
def test_ask_records_usage(db, auth_client, tenant, settings):
|
||
settings.AI_PROVIDER = ""
|
||
settings.AI_API_KEY = ""
|
||
auth_client.post("/api/v1/ai/ask/", {"question": "库存多少"}, format="json")
|
||
assert AiUsage.objects.filter(kind=ai_usage.KIND_ASK).count() == 1
|
||
|
||
|
||
def test_ask_quota_403(db, auth_client, tenant, settings, ai_quota):
|
||
ai_quota(ask=1)
|
||
ai_usage.record_usage(tenant, ai_usage.KIND_ASK)
|
||
resp = auth_client.post("/api/v1/ai/ask/", {"question": "销售额"}, format="json")
|
||
assert resp.status_code == 403
|
||
assert resp.json()["code"] == "quota_exceeded"
|
||
|
||
|
||
def test_tenant_isolation_usage(db, tenant, other_tenant):
|
||
ai_usage.record_usage(tenant, ai_usage.KIND_ASK)
|
||
assert ai_usage.monthly_count(tenant, ai_usage.KIND_ASK) == 1
|
||
assert ai_usage.monthly_count(other_tenant, ai_usage.KIND_ASK) == 0
|