371 lines
16 KiB
Python
371 lines
16 KiB
Python
"""边界输入与畸形请求测试（迭代深挖）。
|
|
|
|
针对外部输入面做模糊测试：负数/零/超大数/非法类型/超长串/注入串。
|
|
目标不是"功能正确"，而是**不崩、不 500、不留脏数据**（返回 4xx 即合格）。
|
|
"""
|
|
|
|
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.inventory.models import Warehouse, Stock
|
|
from apps.partner.models import Customer
|
|
from apps.sales import services as sales_services
|
|
|
|
|
|
@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("0"))
|
|
|
|
|
|
# ============================================================
|
|
# 建单接口：畸形 lines
|
|
# ============================================================
|
|
|
|
|
|
def _create_bill(client, customer, warehouse, lines):
|
|
return client.post("/api/v1/sales/bills/create-bill/", {
|
|
"customer": customer.id, "warehouse": warehouse.id, "lines": lines,
|
|
}, format="json")
|
|
|
|
|
|
@pytest.mark.parametrize("bad_qty", ["0", "-1", "-999", "abc", "", None])
|
|
def test_create_bill_rejects_bad_quantity(db, auth_client, customer, warehouse,
|
|
product, bad_qty):
|
|
"""非法数量必须 4xx（不能 500、不能落库）。"""
|
|
resp = _create_bill(auth_client, customer, warehouse,
|
|
[{"product": product.id, "quantity": bad_qty,
|
|
"unit_price": "10"}])
|
|
assert 400 <= resp.status_code < 500, f"qty={bad_qty!r} → {resp.status_code}"
|
|
|
|
|
|
@pytest.mark.parametrize("bad_price", ["-1", "abc"])
|
|
def test_create_bill_rejects_bad_price(db, auth_client, customer, warehouse,
|
|
product, bad_price):
|
|
resp = _create_bill(auth_client, customer, warehouse,
|
|
[{"product": product.id, "quantity": 1,
|
|
"unit_price": bad_price}])
|
|
assert 400 <= resp.status_code < 500, f"price={bad_price!r} → {resp.status_code}"
|
|
|
|
|
|
def test_create_bill_rejects_huge_quantity(db, auth_client, customer, warehouse, product):
|
|
"""超大数量不该导致 500（库存校验会拒绝，或数量格式非法）。"""
|
|
resp = _create_bill(auth_client, customer, warehouse,
|
|
[{"product": product.id, "quantity": "999999999999999999999",
|
|
"unit_price": "10"}])
|
|
assert resp.status_code < 500, f"超大数量 → {resp.status_code}"
|
|
|
|
|
|
def test_create_bill_rejects_nonexistent_product(db, auth_client, customer, warehouse):
|
|
resp = _create_bill(auth_client, customer, warehouse,
|
|
[{"product": 999999, "quantity": 1, "unit_price": "10"}])
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_create_bill_rejects_nonexistent_customer(db, auth_client, warehouse, product):
|
|
resp = auth_client.post("/api/v1/sales/bills/create-bill/", {
|
|
"customer": 999999, "warehouse": warehouse.id,
|
|
"lines": [{"product": product.id, "quantity": 1, "unit_price": "10"}],
|
|
}, format="json")
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_create_bill_rejects_empty_lines(db, auth_client, customer, warehouse):
|
|
resp = _create_bill(auth_client, customer, warehouse, [])
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_create_bill_rejects_malformed_lines(db, auth_client, customer, warehouse):
|
|
"""lines 传字符串/数字/非 dict 元素等畸形结构：一律 4xx，不能 500。"""
|
|
for bad in ["not-a-list", 123, {"a": 1}, [None], ["string"], [123], [{}]]:
|
|
resp = _create_bill(auth_client, customer, warehouse, bad)
|
|
assert 400 <= resp.status_code < 500, f"lines={bad!r} → {resp.status_code}"
|
|
|
|
|
|
def test_create_bill_rejects_invalid_round_to(db, auth_client, customer, warehouse, product):
|
|
resp = auth_client.post("/api/v1/sales/bills/create-bill/", {
|
|
"customer": customer.id, "warehouse": warehouse.id, "round_to": "abc",
|
|
"lines": [{"product": product.id, "quantity": 1, "unit_price": "10"}],
|
|
}, format="json")
|
|
assert resp.status_code < 500
|
|
|
|
|
|
# ============================================================
|
|
# 超长 / 注入串
|
|
# ============================================================
|
|
|
|
|
|
def test_create_bill_handles_long_remark(db, auth_client, customer, warehouse, product):
|
|
"""超长备注不该 500（数据库列是 Text，但要有合理处理）。"""
|
|
resp = auth_client.post("/api/v1/sales/bills/create-bill/", {
|
|
"customer": customer.id, "warehouse": warehouse.id,
|
|
"remark": "x" * 10000,
|
|
"lines": [{"product": product.id, "quantity": 1, "unit_price": "10"}],
|
|
}, format="json")
|
|
assert resp.status_code < 500
|
|
|
|
|
|
def test_product_code_with_sql_meta(db, auth_client, tenant):
|
|
"""SQL 元字符进商品编码：应被当普通文本（Django ORM 参数化）。"""
|
|
resp = auth_client.post("/api/v1/catalog/products/", {
|
|
"code": "P'; DROP TABLE catalog_product; --",
|
|
"name": "注入测试",
|
|
}, format="json")
|
|
assert resp.status_code in (201, 400)
|
|
# 表还在
|
|
assert Product.objects.count() >= 0
|
|
|
|
|
|
def test_product_name_with_html(db, auth_client):
|
|
"""HTML/脚本进商品名：存储原样，渲染时转义（打印已测）。"""
|
|
resp = auth_client.post("/api/v1/catalog/products/", {
|
|
"code": "XSS01", "name": "<script>alert(1)</script>",
|
|
}, format="json")
|
|
assert resp.status_code in (201, 400)
|
|
|
|
|
|
def test_search_with_special_chars(db, auth_client, product):
|
|
"""搜索框输入 SQL/正则元字符不应 500。"""
|
|
for q in ["%", "_", "'", '"', "\\", ".*", "(a+)+", "'; DROP TABLE--"]:
|
|
resp = auth_client.get("/api/v1/catalog/products/", {"search": q})
|
|
assert resp.status_code == 200, f"search={q!r} → {resp.status_code}"
|
|
|
|
|
|
# ============================================================
|
|
# AI 接口边界
|
|
# ============================================================
|
|
|
|
|
|
def test_ai_parse_order_empty_text(db, auth_client):
|
|
assert auth_client.post("/api/v1/ai/parse-order/", {"text": ""},
|
|
format="json").status_code == 400
|
|
assert auth_client.post("/api/v1/ai/parse-order/", {},
|
|
format="json").status_code == 400
|
|
|
|
|
|
def test_ai_parse_order_huge_text(db, auth_client, settings):
|
|
"""超长文本不该打爆服务（应有上限或降级）。"""
|
|
settings.AI_PROVIDER = ""
|
|
settings.AI_API_KEY = ""
|
|
resp = auth_client.post("/api/v1/ai/parse-order/",
|
|
{"text": "可乐 2 箱\n" * 5000}, format="json")
|
|
assert resp.status_code < 500
|
|
|
|
|
|
def test_ai_ask_empty_question(db, auth_client):
|
|
assert auth_client.post("/api/v1/ai/ask/", {"question": ""},
|
|
format="json").status_code == 400
|
|
|
|
|
|
def test_ai_ask_special_chars(db, auth_client, settings):
|
|
settings.AI_PROVIDER = ""
|
|
settings.AI_API_KEY = ""
|
|
resp = auth_client.post("/api/v1/ai/ask/",
|
|
{"question": "'; DROP TABLE ai_usage; -- 销售额"},
|
|
format="json")
|
|
assert resp.status_code == 200
|
|
|
|
|
|
# ============================================================
|
|
# 分页与查询参数
|
|
# ============================================================
|
|
|
|
|
|
@pytest.mark.parametrize("params", [
|
|
{"page": "0"}, {"page": "-1"}, {"page": "abc"}, {"page": "999999999"},
|
|
{"page_size": "0"}, {"page_size": "-5"}, {"page_size": "99999"},
|
|
{"page_size": "abc"},
|
|
])
|
|
def test_pagination_handles_bad_params(db, auth_client, product, params):
|
|
"""异常分页参数不该 500。"""
|
|
resp = auth_client.get("/api/v1/catalog/products/", params)
|
|
assert resp.status_code == 200, f"params={params} → {resp.status_code}"
|
|
|
|
|
|
def test_pagination_page_size_capped(db, auth_client, tenant):
|
|
"""page_size 应有上限（防止一次拉全表）。"""
|
|
for i in range(5):
|
|
baker.make(Product, tenant=tenant, code=f"P{i:03d}", name=f"商品{i}")
|
|
resp = auth_client.get("/api/v1/catalog/products/", {"page_size": "99999"})
|
|
assert resp.status_code == 200
|
|
assert len(resp.json()["results"]) <= 200 # max_page_size
|
|
|
|
|
|
# ============================================================
|
|
# 商城下单边界
|
|
# ============================================================
|
|
|
|
|
|
def test_storefront_order_bad_payload(db, client, tenant):
|
|
"""商城未登录 + 畸形 payload：应是 401，不是 500。"""
|
|
for payload in [{}, {"lines": []}, {"lines": "x"}, {"lines": [{"product_id": -1}]}]:
|
|
resp = client.post("/api/v1/storefront/orders/", payload,
|
|
format="json", HTTP_X_TENANT_ID=tenant.code)
|
|
assert resp.status_code < 500, f"{payload} → {resp.status_code}"
|
|
|
|
|
|
def test_storefront_login_bad_payload(db, client, tenant):
|
|
for payload in [{}, {"phone": ""}, {"password": ""},
|
|
{"phone": "x" * 500, "password": "y"},
|
|
{"phone": 123, "password": 456}]:
|
|
resp = client.post("/api/v1/storefront/login/", payload,
|
|
format="json", HTTP_X_TENANT_ID=tenant.code)
|
|
assert resp.status_code < 500, f"{payload} → {resp.status_code}"
|
|
|
|
|
|
# ============================================================
|
|
# 极小金额单据（迭代第 2 轮发现）
|
|
# ============================================================
|
|
|
|
|
|
def test_create_bill_rejects_tiny_amount(db, auth_client, customer, warehouse, product):
|
|
"""0.0001 数量会产生 0.0016 元的荒谬单据——后端必须拒绝。
|
|
|
|
背景：前端 el-input-number 的 `:min` 会把越界的 0 自动「纠正」为 min，
|
|
用户看到的仍是 0 但提交的是 min 值。后端不能依赖前端校验。
|
|
"""
|
|
resp = _create_bill(auth_client, customer, warehouse,
|
|
[{"product": product.id, "quantity": "0.0001",
|
|
"unit_price": "15.5"}])
|
|
assert resp.status_code == 400, f"0.0001 数量应被拒 → {resp.status_code}"
|
|
assert resp.json()["code"] in ("invalid_quantity", "invalid_line")
|
|
|
|
|
|
def test_create_bill_rejects_zero_amount(db, auth_client, customer, warehouse, product):
|
|
"""单价 0 + 数量正常 → 合计为 0，应拒绝（避免空金额单据）。"""
|
|
resp = _create_bill(auth_client, customer, warehouse,
|
|
[{"product": product.id, "quantity": "5", "unit_price": "0"}])
|
|
assert resp.status_code == 400, f"0 金额应被拒 → {resp.status_code}"
|
|
|
|
|
|
def test_purchase_bill_rejects_zero_amount(db, auth_client, tenant, warehouse):
|
|
from apps.partner.models import Supplier
|
|
|
|
supplier = baker.make(Supplier, tenant=tenant, code="S900", name="供应商")
|
|
prod = baker.make(Product, tenant=tenant, code="PZ01", name="测试品")
|
|
resp = auth_client.post("/api/v1/purchase/bills/create-bill/", {
|
|
"supplier": supplier.id, "warehouse": warehouse.id,
|
|
"lines": [{"product": prod.id, "quantity": 5, "unit_price": "0"}],
|
|
}, format="json")
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_quantity_upper_bound_enforced(db, auth_client, customer, warehouse, product):
|
|
"""数量上限 10 亿（防止天文数字撑爆计算）。"""
|
|
resp = _create_bill(auth_client, customer, warehouse,
|
|
[{"product": product.id, "quantity": "1000000001",
|
|
"unit_price": "10"}])
|
|
assert resp.status_code == 400
|
|
assert resp.json()["code"] == "invalid_quantity"
|
|
|
|
|
|
def test_normal_bill_still_works(db, auth_client, customer, warehouse, product):
|
|
"""正常单据不受影响（回归保护）。"""
|
|
from apps.inventory import services as inv
|
|
|
|
inv.inbound(tenant=tenant_of(auth_client), warehouse=warehouse, product=product,
|
|
quantity=Decimal("100"), unit_cost=Decimal("5"))
|
|
resp = _create_bill(auth_client, customer, warehouse,
|
|
[{"product": product.id, "quantity": "3", "unit_price": "10"}])
|
|
assert resp.status_code == 201, resp.content
|
|
|
|
|
|
def tenant_of(client):
|
|
from apps.core.models import Tenant
|
|
|
|
return Tenant.objects.get(code="test")
|
|
|
|
|
|
# ============================================================
|
|
# 0 价拦截（迭代第 4 轮）
|
|
# ============================================================
|
|
|
|
|
|
def test_zero_price_rejected_sales(db, auth_client, customer, warehouse, product):
|
|
"""0 元销售单必须被拒（白送应走专门流程，不该产生 0 金额单据）。"""
|
|
resp = _create_bill(auth_client, customer, warehouse,
|
|
[{"product": product.id, "quantity": 5, "unit_price": "0"}])
|
|
assert resp.status_code == 400, resp.content
|
|
assert resp.json()["code"] == "invalid_price"
|
|
|
|
|
|
def test_zero_price_rejected_purchase(db, auth_client, tenant, warehouse):
|
|
"""0 元进货必须被拒：会污染加权平均成本（把真实成本拉低 → 毛利虚高）。"""
|
|
from apps.partner.models import Supplier
|
|
|
|
supplier = baker.make(Supplier, tenant=tenant, code="S-ZERO", name="供应商")
|
|
prod = baker.make(Product, tenant=tenant, code="PZERO", name="测试品")
|
|
resp = auth_client.post("/api/v1/purchase/bills/create-bill/", {
|
|
"supplier": supplier.id, "warehouse": warehouse.id,
|
|
"lines": [{"product": prod.id, "quantity": 10, "unit_price": "0"}],
|
|
}, format="json")
|
|
assert resp.status_code == 400, resp.content
|
|
assert resp.json()["code"] == "invalid_price"
|
|
|
|
|
|
def test_negative_price_rejected_sales(db, auth_client, customer, warehouse, product):
|
|
resp = _create_bill(auth_client, customer, warehouse,
|
|
[{"product": product.id, "quantity": 5, "unit_price": "-10"}])
|
|
assert resp.status_code == 400
|
|
assert resp.json()["code"] == "invalid_price"
|
|
|
|
|
|
def test_positive_price_still_works(db, auth_client, customer, warehouse, product):
|
|
"""回归保护：正常价格不受影响。"""
|
|
from apps.inventory import services as inv
|
|
from apps.core.models import Tenant
|
|
|
|
t = Tenant.objects.get(code="test")
|
|
inv.inbound(tenant=t, warehouse=warehouse, product=product,
|
|
quantity=Decimal("100"), unit_cost=Decimal("5"))
|
|
resp = _create_bill(auth_client, customer, warehouse,
|
|
[{"product": product.id, "quantity": 3, "unit_price": "0.01"}])
|
|
assert resp.status_code == 201, resp.content
|
|
|
|
|
|
def test_weighted_cost_not_polluted_by_zero_cost(db, tenant, warehouse, product):
|
|
"""单元级：0 成本入库不改变加权平均成本（因为业务层已拒绝）。
|
|
|
|
这里直接验证 inbound 的行为：unit_cost=0 时不参与加权（保持原成本）。
|
|
"""
|
|
from apps.inventory import services as inv
|
|
from apps.inventory.models import Stock
|
|
|
|
inv.inbound(tenant=tenant, warehouse=warehouse, product=product,
|
|
quantity=Decimal("10"), unit_cost=Decimal("5"))
|
|
st = Stock.objects.get(product=product)
|
|
assert st.avg_cost == Decimal("5.0000")
|
|
|
|
# 0 成本入库（理论上业务层已拦，这里测服务层兜底行为）
|
|
inv.inbound(tenant=tenant, warehouse=warehouse, product=product,
|
|
quantity=Decimal("10"), unit_cost=Decimal("0"))
|
|
st.refresh_from_db()
|
|
# 成本不应被 0 拉低
|
|
assert st.avg_cost == Decimal("5.0000"), f"0 成本污染了加权成本：{st.avg_cost}"
|