446 lines
17 KiB
Python
446 lines
17 KiB
Python
"""全局异常处理器测试（迭代第 2 轮）。
|
|
|
|
验证：业务异常 → 正确的 HTTP 语义与 `code`；未知异常 → 结构化 500（不泄漏栈）。
|
|
"""
|
|
|
|
import pytest
|
|
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
|
|
from apps.core.exceptions import api_exception_handler
|
|
from apps.core.services import (
|
|
BelowMinPrice, InvalidLinePrice, InvalidLineQuantity,
|
|
)
|
|
from apps.inventory.models import Warehouse
|
|
from apps.inventory.services import InsufficientStock
|
|
from apps.partner.models import Customer
|
|
from apps.sales import services as sales_services
|
|
from apps.sales.services import CreditLimitExceeded
|
|
|
|
|
|
@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 warehouse(db, tenant):
|
|
return baker.make(Warehouse, tenant=tenant, code="WH01", name="主仓")
|
|
|
|
|
|
@pytest.fixture
|
|
def product(db, tenant):
|
|
return baker.make(Product, tenant=tenant, code="P001", name="可乐",
|
|
sale_price=Decimal("10"))
|
|
|
|
|
|
@pytest.fixture
|
|
def customer(db, tenant):
|
|
return baker.make(Customer, tenant=tenant, code="C001", name="张三",
|
|
credit_limit=Decimal("10000"))
|
|
|
|
|
|
# ============================================================
|
|
# 单元级：处理器映射正确性
|
|
# ============================================================
|
|
|
|
|
|
def _call(exc):
|
|
return api_exception_handler(exc, {"view": None, "request": None})
|
|
|
|
|
|
def test_handler_maps_invalid_quantity():
|
|
exc = InvalidLineQuantity("P001", -5, "数量必须大于 0")
|
|
resp = _call(exc)
|
|
assert resp.status_code == 400
|
|
assert resp.data["code"] == "invalid_quantity"
|
|
assert resp.data["product_code"] == "P001"
|
|
|
|
|
|
def test_handler_maps_invalid_price():
|
|
exc = InvalidLinePrice("P001", -1, "单价不能为负")
|
|
resp = _call(exc)
|
|
assert resp.status_code == 400
|
|
assert resp.data["code"] == "invalid_price"
|
|
|
|
|
|
def test_handler_maps_below_min_price():
|
|
exc = BelowMinPrice("P001", Decimal("5"), Decimal("8"))
|
|
resp = _call(exc)
|
|
assert resp.status_code == 400
|
|
assert resp.data["code"] == "below_min_price"
|
|
assert resp.data["unit_price"] == "5"
|
|
assert resp.data["min_price"] == "8"
|
|
|
|
|
|
def test_handler_maps_insufficient_stock():
|
|
resp = _call(InsufficientStock("no stock"))
|
|
assert resp.status_code == 400
|
|
assert resp.data["code"] == "insufficient_stock"
|
|
|
|
|
|
def test_handler_maps_credit_limit():
|
|
exc = CreditLimitExceeded("C001", Decimal("900"), Decimal("1000"), Decimal("200"))
|
|
resp = _call(exc)
|
|
assert resp.status_code == 402
|
|
assert resp.data["code"] == "credit_limit_exceeded"
|
|
assert resp.data["customer_code"] == "C001"
|
|
|
|
|
|
def test_handler_returns_structured_500_for_unknown(db):
|
|
"""未知异常 → 结构化 500，且**不把栈信息/原始消息**给客户端。"""
|
|
resp = _call(RuntimeError("内部数据库连接串 postgres://user:pw@host/db 泄露了"))
|
|
assert resp.status_code == 500
|
|
assert resp.data["code"] == "server_error"
|
|
assert "exc_type" not in resp.data
|
|
body = str(resp.data)
|
|
assert "postgres://" not in body # 原始消息不外泄
|
|
assert "Traceback" not in body
|
|
|
|
|
|
def test_handler_passes_through_drf_exceptions(db, rf):
|
|
"""DRF 内置异常（ValidationError）由 DRF 处理，不被我们的兜底覆盖。"""
|
|
from rest_framework.exceptions import ValidationError
|
|
|
|
resp = _call(ValidationError({"field": "必填"}))
|
|
assert resp.status_code == 400
|
|
assert "field" in resp.data
|
|
|
|
|
|
# ============================================================
|
|
# 集成级：真实 API 路径
|
|
# ============================================================
|
|
|
|
|
|
def test_api_invalid_quantity_returns_code(db, auth_client, customer, warehouse, product):
|
|
resp = auth_client.post("/api/v1/sales/bills/create-bill/", {
|
|
"customer": customer.id, "warehouse": warehouse.id,
|
|
"lines": [{"product": product.id, "quantity": "-5", "unit_price": "10"}],
|
|
}, format="json")
|
|
assert resp.status_code == 400
|
|
assert resp.json()["code"] == "invalid_quantity"
|
|
|
|
|
|
def test_api_insufficient_stock_returns_code(db, auth_client, customer, warehouse, product):
|
|
"""库存不足：经由全局处理器返回 400 + code（此前视图里有重复分支）。"""
|
|
bill = auth_client.post("/api/v1/sales/bills/create-bill/", {
|
|
"customer": customer.id, "warehouse": warehouse.id,
|
|
"lines": [{"product": product.id, "quantity": "999", "unit_price": "10"}],
|
|
}, format="json").json()
|
|
resp = auth_client.post(f"/api/v1/sales/bills/{bill['id']}/confirm/", {},
|
|
format="json")
|
|
assert resp.status_code == 400
|
|
assert resp.json()["code"] == "insufficient_stock"
|
|
|
|
|
|
def test_api_credit_limit_returns_402(db, auth_client, tenant, warehouse, product):
|
|
"""信用超限：402 + code（前端据此弹强制过账确认框）。"""
|
|
from apps.finance.models import Receivable
|
|
from apps.inventory import services as inv
|
|
|
|
limited = baker.make(Customer, tenant=tenant, code="C999", name="受限",
|
|
credit_limit=Decimal("100"))
|
|
baker.make(Receivable, tenant=tenant, customer=limited, bill_no="RC-1",
|
|
total_amount=Decimal("90"), status="open")
|
|
inv.inbound(tenant=tenant, warehouse=warehouse, product=product,
|
|
quantity=Decimal("100"), unit_cost=Decimal("2"))
|
|
|
|
bill = auth_client.post("/api/v1/sales/bills/create-bill/", {
|
|
"customer": limited.id, "warehouse": warehouse.id,
|
|
"lines": [{"product": product.id, "quantity": 50, "unit_price": "10"}],
|
|
}, format="json").json()
|
|
resp = auth_client.post(f"/api/v1/sales/bills/{bill['id']}/confirm/", {},
|
|
format="json")
|
|
assert resp.status_code == 402
|
|
assert resp.json()["code"] == "credit_limit_exceeded"
|
|
|
|
|
|
def test_api_quota_exceeded_returns_403(db, auth_client, tenant, settings):
|
|
"""AI 配额超限：403 + 升级引导（此前视图里有重复分支）。"""
|
|
from apps.ai import usage as ai_usage
|
|
from apps.billing.models import Plan, Subscription, seed_plans
|
|
|
|
seed_plans()
|
|
plan, _ = Plan.objects.update_or_create(
|
|
code="quota-test", defaults={
|
|
"name": "配额测试", "price_monthly": Decimal("0"),
|
|
"limits": {"users": 9, "products": 9, "bills_monthly": 0,
|
|
"ai_parse_order": 1, "ai_ask": 1,
|
|
"batch_managed": True, "finance_ledger": True,
|
|
"print_templates": 0, "storefront": False},
|
|
"sort_order": 99, "is_active": True,
|
|
},
|
|
)
|
|
Subscription.objects.update_or_create(
|
|
tenant=tenant, defaults={"plan": plan, "status": "active"},
|
|
)
|
|
ai_usage.record_usage(tenant, ai_usage.KIND_ASK)
|
|
|
|
settings.AI_PROVIDER = ""
|
|
settings.AI_API_KEY = ""
|
|
resp = auth_client.post("/api/v1/ai/ask/", {"question": "销售额"}, format="json")
|
|
assert resp.status_code == 403
|
|
body = resp.json()
|
|
assert body["code"] == "quota_exceeded"
|
|
assert body["upgrade_url"] == "/#/pricing"
|
|
|
|
|
|
def test_api_storefront_error_returns_400(db, client, tenant, settings):
|
|
"""商城业务错误（空明细）→ 400 + code。"""
|
|
from apps.billing.models import Plan, Subscription, seed_plans
|
|
|
|
seed_plans()
|
|
pro = Plan.objects.get(code="pro")
|
|
Subscription.objects.update_or_create(
|
|
tenant=tenant, defaults={"plan": pro, "status": "active"},
|
|
)
|
|
resp = client.post("/api/v1/storefront/orders/", {"lines": []},
|
|
format="json", HTTP_X_TENANT_ID=tenant.code)
|
|
assert resp.status_code < 500
|
|
assert resp.json().get("code") in ("invalid_order", "unauthorized")
|
|
|
|
|
|
# ============================================================
|
|
# 审计日志（迭代第 3 轮）
|
|
# ============================================================
|
|
|
|
|
|
def test_audit_log_written_on_sales_post(db, tenant, warehouse, customer, product):
|
|
"""销售过账必须留审计（谁、何时、哪张单、金额多少）。"""
|
|
from apps.core.audit import query_logs
|
|
from apps.core.models import AuditLog
|
|
from apps.inventory import services as inv
|
|
|
|
inv.inbound(tenant=tenant, warehouse=warehouse, product=product,
|
|
quantity=Decimal("100"), unit_cost=Decimal("5"))
|
|
bill = sales_services.create_sales_bill(
|
|
tenant=tenant, customer=customer, warehouse=warehouse,
|
|
lines=[{"product": product, "quantity": 3, "unit_price": 10}],
|
|
)
|
|
sales_services.confirm_sales_bill(bill)
|
|
|
|
logs = query_logs(tenant, target_type="SalesBill", target_id=bill.bill_no)
|
|
assert logs, "过账未写审计日志"
|
|
entry = logs[0]
|
|
assert entry.detail["business_action"] == "post"
|
|
assert entry.detail["bill_no"] == bill.bill_no
|
|
assert entry.detail["amount"] == str(bill.total_amount)
|
|
|
|
|
|
def test_audit_log_written_on_purchase_post(db, tenant, warehouse, product):
|
|
"""采购过账同样留痕。"""
|
|
from apps.core.audit import query_logs
|
|
from apps.partner.models import Supplier
|
|
from apps.purchase import services as purchase_services
|
|
|
|
supplier = baker.make(Supplier, tenant=tenant, code="S001", name="上游厂")
|
|
bill = purchase_services.create_purchase_bill(
|
|
tenant=tenant, supplier=supplier, warehouse=warehouse,
|
|
lines=[{"product": product, "quantity": 5, "unit_price": 3}],
|
|
)
|
|
purchase_services.confirm_purchase_bill(bill)
|
|
|
|
logs = query_logs(tenant, target_type="PurchaseBill", target_id=bill.bill_no)
|
|
assert logs and logs[0].detail["business_action"] == "post"
|
|
|
|
|
|
def test_audit_log_written_on_credit_force_release(db, tenant, warehouse, product):
|
|
"""信用超限强制放行 → 必须留痕（高风险动作，事后要能追责）。"""
|
|
from apps.core.audit import query_logs
|
|
from apps.finance.models import Receivable
|
|
from apps.inventory import services as inv
|
|
|
|
limited = baker.make(Customer, tenant=tenant, code="C777", name="受限客户",
|
|
credit_limit=Decimal("100"))
|
|
baker.make(Receivable, tenant=tenant, customer=limited, bill_no="RC-9",
|
|
total_amount=Decimal("90"), status="open")
|
|
inv.inbound(tenant=tenant, warehouse=warehouse, product=product,
|
|
quantity=Decimal("100"), unit_cost=Decimal("2"))
|
|
|
|
bill = sales_services.create_sales_bill(
|
|
tenant=tenant, customer=limited, warehouse=warehouse,
|
|
lines=[{"product": product, "quantity": 5, "unit_price": 10}],
|
|
)
|
|
sales_services.confirm_sales_bill(bill, force=True)
|
|
|
|
logs = query_logs(tenant, target_type="credit_limit", target_id=bill.bill_no)
|
|
assert logs, "强制放行未留审计"
|
|
assert logs[0].detail["business_action"] == "force"
|
|
assert logs[0].detail["customer_code"] == "C777"
|
|
|
|
|
|
def test_audit_log_written_on_below_min_price_release(db, tenant, warehouse, product):
|
|
"""低于最低售价放行 → 留痕（放行可能亏本）。"""
|
|
from apps.core.audit import query_logs
|
|
from apps.core.services import check_min_price
|
|
|
|
product.min_sale_price = Decimal("8")
|
|
product.save()
|
|
|
|
check_min_price(product, unit_price=Decimal("5"), allow_below_min=True)
|
|
|
|
logs = query_logs(tenant, target_type="below_min_price", target_id=product.code)
|
|
assert logs, "低价放行未留审计"
|
|
assert logs[0].detail["min_price"] == "8"
|
|
|
|
|
|
def test_audit_log_failure_does_not_break_business(db, tenant, monkeypatch):
|
|
"""审计写失败**不能**影响主业务（旁路设计）。"""
|
|
from apps.core import audit as audit_log
|
|
|
|
def boom(**kwargs):
|
|
raise RuntimeError("审计服务不可用")
|
|
|
|
monkeypatch.setattr(audit_log.AuditLog.objects, "create", boom)
|
|
# 不应抛异常
|
|
result = audit_log.log(tenant=tenant, action="post", target_type="Test",
|
|
target_id="X1")
|
|
assert result is None # 失败返回 None
|
|
|
|
|
|
def test_audit_records_request_context(db, tenant, rf, django_user_model):
|
|
"""给了 request 就自动带 user / ip / user_agent。"""
|
|
from apps.core import audit as audit_log
|
|
from apps.core.models import AuditLog
|
|
|
|
user = django_user_model.objects.create_user("auditor", password="x")
|
|
req = rf.post("/api/v1/sales/bills/1/confirm/", REMOTE_ADDR="10.1.2.3",
|
|
HTTP_USER_AGENT="TestAgent/1.0")
|
|
req.user = user
|
|
|
|
audit_log.log(tenant=tenant, action="post", target_type="SalesBill",
|
|
target_id="XS-1", request=req)
|
|
|
|
entry = AuditLog.objects.filter(tenant=tenant, target_id="XS-1").first()
|
|
assert entry is not None
|
|
assert entry.user_id == user.id
|
|
assert str(entry.ip) == "10.1.2.3"
|
|
assert entry.user_agent == "TestAgent/1.0"
|
|
|
|
|
|
def test_audit_query_filters_by_target(db, tenant, warehouse, customer, product):
|
|
"""按单据号查轨迹（"这张单都发生了什么"）。"""
|
|
from apps.core.audit import query_logs
|
|
from apps.inventory import services as inv
|
|
|
|
inv.inbound(tenant=tenant, warehouse=warehouse, product=product,
|
|
quantity=Decimal("100"), unit_cost=Decimal("5"))
|
|
b1 = sales_services.create_sales_bill(
|
|
tenant=tenant, customer=customer, warehouse=warehouse,
|
|
lines=[{"product": product, "quantity": 1, "unit_price": 10}],
|
|
)
|
|
b2 = sales_services.create_sales_bill(
|
|
tenant=tenant, customer=customer, warehouse=warehouse,
|
|
lines=[{"product": product, "quantity": 2, "unit_price": 10}],
|
|
)
|
|
sales_services.confirm_sales_bill(b1)
|
|
sales_services.confirm_sales_bill(b2)
|
|
|
|
logs = query_logs(tenant, target_type="SalesBill", target_id=b1.bill_no)
|
|
assert len(logs) == 1
|
|
assert logs[0].target_id == b1.bill_no
|
|
assert logs[0].target_id != b2.bill_no
|
|
|
|
|
|
def test_audit_tenant_isolation(db, tenant, other_tenant, warehouse, customer, product):
|
|
"""审计日志按租户隔离。"""
|
|
from apps.core.audit import log, query_logs
|
|
from apps.inventory import services as inv
|
|
|
|
inv.inbound(tenant=tenant, warehouse=warehouse, product=product,
|
|
quantity=Decimal("10"), unit_cost=Decimal("1"))
|
|
bill = sales_services.create_sales_bill(
|
|
tenant=tenant, customer=customer, warehouse=warehouse,
|
|
lines=[{"product": product, "quantity": 1, "unit_price": 10}],
|
|
)
|
|
sales_services.confirm_sales_bill(bill)
|
|
|
|
assert query_logs(tenant, target_type="SalesBill")
|
|
assert query_logs(other_tenant, target_type="SalesBill") == []
|
|
|
|
|
|
# ============================================================
|
|
# 请求上下文透传（审计取 IP/UA）
|
|
# ============================================================
|
|
|
|
|
|
def test_service_layer_gets_request_via_contextvar(db, tenant, warehouse,
|
|
customer, product, rf,
|
|
django_user_model):
|
|
"""服务层没有 request 参数，但审计仍能记录 IP/UA（经 contextvar 透传）。"""
|
|
from apps.core import audit as audit_log
|
|
from apps.core.context import clear_current_request, set_current_request
|
|
from apps.core.models import AuditLog
|
|
from apps.inventory import services as inv
|
|
|
|
user = django_user_model.objects.create_user("ctx_user", password="x")
|
|
inv.inbound(tenant=tenant, warehouse=warehouse, product=product,
|
|
quantity=Decimal("100"), unit_cost=Decimal("5"))
|
|
|
|
req = rf.post("/api/v1/sales/bills/1/confirm/", REMOTE_ADDR="192.168.9.9",
|
|
HTTP_USER_AGENT="CtxAgent/2.0")
|
|
req.user = user
|
|
set_current_request(req)
|
|
try:
|
|
bill = sales_services.create_sales_bill(
|
|
tenant=tenant, customer=customer, warehouse=warehouse,
|
|
lines=[{"product": product, "quantity": 1, "unit_price": 10}],
|
|
)
|
|
# 注意：这里没有传 request/user 给服务层
|
|
sales_services.confirm_sales_bill(bill)
|
|
finally:
|
|
clear_current_request()
|
|
|
|
entry = AuditLog.objects.filter(target_id=bill.bill_no).first()
|
|
assert entry is not None
|
|
assert str(entry.ip) == "192.168.9.9" # 从 contextvar 拿到
|
|
assert entry.user_agent == "CtxAgent/2.0"
|
|
assert entry.user_id == user.id
|
|
|
|
|
|
def test_contextvar_cleared_between_requests(db, tenant, rf):
|
|
"""请求结束必须清除（防止跨请求串号）。"""
|
|
from apps.core.audit import log
|
|
from apps.core.context import clear_current_request, current_request, set_current_request
|
|
from apps.core.models import AuditLog
|
|
import django.contrib.auth as auth
|
|
|
|
assert current_request() is None
|
|
|
|
req = rf.get("/x/", REMOTE_ADDR="1.1.1.1")
|
|
set_current_request(req)
|
|
assert current_request() is req
|
|
|
|
clear_current_request()
|
|
assert current_request() is None
|
|
|
|
# 清除后再写审计：不应带上被清除请求的 IP
|
|
log(tenant=tenant, action="post", target_type="T", target_id="NOCTX")
|
|
entry = AuditLog.objects.filter(target_id="NOCTX").first()
|
|
assert entry is not None
|
|
assert entry.ip is None
|
|
|
|
|
|
def test_audit_tolerates_missing_request(db, tenant):
|
|
"""无请求上下文（如后台任务）时审计仍能写入，只是没有 IP/UA。"""
|
|
from apps.core.audit import log
|
|
from apps.core.models import AuditLog
|
|
|
|
log(tenant=tenant, action="post", target_type="CronJob", target_id="JOB-1",
|
|
detail={"note": "定时任务"})
|
|
|
|
entry = AuditLog.objects.filter(target_id="JOB-1").first()
|
|
assert entry is not None
|
|
assert entry.ip is None
|
|
assert entry.user_id is None
|
|
assert entry.detail["note"] == "定时任务"
|