306 lines
12 KiB
Python
306 lines
12 KiB
Python
"""审计日志查询 API 测试（迭代第 4 轮）。
|
||
|
||
验证：列表筛选、动作标签与风险分级、统计汇总、单据轨迹、租户隔离、
|
||
以及**审计日志不可篡改**（无写接口）。
|
||
"""
|
||
|
||
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.audit import log, log_bill_posted, log_force_release
|
||
from apps.core.models import AuditLog
|
||
from apps.inventory.models import Warehouse
|
||
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 customer(db, tenant):
|
||
return baker.make(Customer, tenant=tenant, code="C001", name="张三便利店",
|
||
credit_limit=Decimal("99999"))
|
||
|
||
|
||
@pytest.fixture
|
||
def product(db, tenant):
|
||
return baker.make(Product, tenant=tenant, code="P001", name="可乐",
|
||
sale_price=Decimal("10"))
|
||
|
||
|
||
def _make_posted_bill(tenant, warehouse, customer, product):
|
||
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": 2, "unit_price": 10}],
|
||
)
|
||
sales_services.confirm_sales_bill(bill)
|
||
return bill
|
||
|
||
|
||
# ============================================================
|
||
# 列表
|
||
# ============================================================
|
||
|
||
|
||
def test_list_returns_audit_entries(db, auth_client, tenant, warehouse,
|
||
customer, product):
|
||
bill = _make_posted_bill(tenant, warehouse, customer, product)
|
||
|
||
resp = auth_client.get("/api/v1/core/audit-logs/")
|
||
assert resp.status_code == 200, resp.content
|
||
body = resp.json()
|
||
assert body["count"] >= 1
|
||
entry = next(e for e in body["results"] if e["target_id"] == bill.bill_no)
|
||
assert entry["action"] == "post"
|
||
assert entry["action_label"] == "过账"
|
||
assert entry["target_type"] == "SalesBill"
|
||
assert entry["detail"]["amount"] == str(bill.total_amount)
|
||
|
||
|
||
def test_list_includes_action_label_and_risk(db, auth_client, tenant,
|
||
warehouse, customer, product):
|
||
bill = _make_posted_bill(tenant, warehouse, customer, product)
|
||
log_force_release(tenant=tenant, kind="credit_limit", target=bill.bill_no,
|
||
detail={"note": "测试放行"})
|
||
|
||
body = auth_client.get("/api/v1/core/audit-logs/").json()
|
||
force_entry = next(e for e in body["results"] if e["action"] == "force")
|
||
assert force_entry["action_label"] == "管控放行"
|
||
assert force_entry["risk"] == "high" # 高风险管理动作
|
||
|
||
post_entry = next(e for e in body["results"] if e["action"] == "post")
|
||
assert post_entry["risk"] == "info"
|
||
|
||
|
||
def test_list_filter_by_target(db, auth_client, tenant, warehouse, customer, product):
|
||
b1 = _make_posted_bill(tenant, warehouse, customer, product)
|
||
b2 = sales_services.create_sales_bill(
|
||
tenant=tenant, customer=customer, warehouse=warehouse,
|
||
lines=[{"product": product, "quantity": 1, "unit_price": 10}],
|
||
)
|
||
sales_services.confirm_sales_bill(b2)
|
||
|
||
body = auth_client.get(
|
||
f"/api/v1/core/audit-logs/?target_type=SalesBill&target_id={b1.bill_no}"
|
||
).json()
|
||
assert body["count"] == 1
|
||
assert body["results"][0]["target_id"] == b1.bill_no
|
||
|
||
|
||
def test_list_filter_by_action(db, auth_client, tenant, warehouse, customer, product):
|
||
bill = _make_posted_bill(tenant, warehouse, customer, product)
|
||
log_force_release(tenant=tenant, kind="credit_limit", target="X1", detail={})
|
||
|
||
body = auth_client.get("/api/v1/core/audit-logs/?action=force").json()
|
||
assert body["count"] >= 1
|
||
assert all(e["action"] == "force" for e in body["results"])
|
||
|
||
|
||
def test_list_limit_capped(db, auth_client, tenant):
|
||
for i in range(20):
|
||
log(tenant=tenant, action="post", target_type="T", target_id=f"ID{i}")
|
||
|
||
body = auth_client.get("/api/v1/core/audit-logs/?limit=99999").json()
|
||
assert body["count"] <= 500 # 上限保护
|
||
|
||
|
||
def test_list_ordered_desc(db, auth_client, tenant):
|
||
"""最新的在前。
|
||
|
||
注意：同一秒内创建的记录 created_at 相同，仅按时间排序不稳定，
|
||
因此查询用 `-created_at, -id` 双键（见 audit_views）。
|
||
"""
|
||
for i in range(5):
|
||
log(tenant=tenant, action="post", target_type="T", target_id=f"SEQ{i}")
|
||
|
||
body = auth_client.get("/api/v1/core/audit-logs/").json()
|
||
seq_ids = [e["target_id"] for e in body["results"] if e["target_id"].startswith("SEQ")]
|
||
assert seq_ids[0] == "SEQ4" # 最新在前
|
||
assert seq_ids == ["SEQ4", "SEQ3", "SEQ2", "SEQ1", "SEQ0"]
|
||
|
||
|
||
# ============================================================
|
||
# 汇总
|
||
# ============================================================
|
||
|
||
|
||
def test_summary_counts_by_action(db, auth_client, tenant, warehouse,
|
||
customer, product):
|
||
bill = _make_posted_bill(tenant, warehouse, customer, product)
|
||
log_force_release(tenant=tenant, kind="credit_limit", target="A", detail={})
|
||
log_force_release(tenant=tenant, kind="below_min_price", target="B", detail={})
|
||
|
||
body = auth_client.get("/api/v1/core/audit-logs/summary/").json()
|
||
assert body["total"] >= 3
|
||
items = {i["action"]: i for i in body["items"]}
|
||
assert items["post"]["count"] >= 1
|
||
assert items["force"]["count"] == 2
|
||
assert body["high_risk_count"] == 2 # 两次放行都是高风险
|
||
|
||
|
||
def test_summary_days_filter(db, auth_client, tenant):
|
||
log(tenant=tenant, action="post", target_type="T", target_id="RECENT")
|
||
|
||
body = auth_client.get("/api/v1/core/audit-logs/summary/?days=1").json()
|
||
assert body["days"] == 1
|
||
assert body["total"] >= 1
|
||
|
||
|
||
# ============================================================
|
||
# 单据轨迹
|
||
# ============================================================
|
||
|
||
|
||
def test_timeline_for_bill(db, auth_client, tenant, warehouse, customer, product):
|
||
"""单据轨迹：过账 + 该单的强制放行（放行的 target_type 是管控类型）。"""
|
||
bill = _make_posted_bill(tenant, warehouse, customer, product)
|
||
# 放行记录挂在管控类型下（credit_limit），但 detail 里带 bill_no
|
||
log_force_release(tenant=tenant, kind="credit_limit", target=bill.bill_no,
|
||
detail={"bill_no": bill.bill_no, "outstanding": "900"})
|
||
|
||
# 单据自身的轨迹（含 detail.bill_no 匹配到的放行记录）
|
||
body = auth_client.get(
|
||
f"/api/v1/core/audit-logs/timeline/?target_type=SalesBill&target_id={bill.bill_no}"
|
||
).json()
|
||
assert body["target"]["id"] == bill.bill_no
|
||
assert body["count"] >= 1
|
||
assert any(t["action"] == "post" for t in body["timeline"])
|
||
|
||
# 按管控类型查：应包含放行记录本身；
|
||
# 由于轨迹会双向串联（detail.bill_no 也匹配），过账记录同样会出现——
|
||
# 这是有意的：无论从哪个入口查，都能看到这张单的完整动作序列。
|
||
body2 = auth_client.get(
|
||
f"/api/v1/core/audit-logs/timeline/?target_type=credit_limit&target_id={bill.bill_no}"
|
||
).json()
|
||
assert body2["count"] >= 1
|
||
actions2 = [t["action"] for t in body2["timeline"]]
|
||
assert "force" in actions2
|
||
|
||
|
||
def test_timeline_requires_params(db, auth_client):
|
||
assert auth_client.get("/api/v1/core/audit-logs/timeline/").status_code == 400
|
||
assert auth_client.get(
|
||
"/api/v1/core/audit-logs/timeline/?target_type=SalesBill"
|
||
).status_code == 400
|
||
|
||
|
||
def test_timeline_includes_user_and_ip(db, auth_client, tenant):
|
||
log(tenant=tenant, action="post", target_type="X", target_id="U1",
|
||
detail={"who": "test"})
|
||
|
||
body = auth_client.get(
|
||
"/api/v1/core/audit-logs/timeline/?target_type=X&target_id=U1"
|
||
).json()
|
||
assert body["count"] == 1
|
||
assert body["timeline"][0]["user"] in ("系统", None) or isinstance(
|
||
body["timeline"][0]["user"], str)
|
||
|
||
|
||
# ============================================================
|
||
# 隔离与不可篡改
|
||
# ============================================================
|
||
|
||
|
||
def test_tenant_isolation(db, auth_client, tenant, other_tenant):
|
||
log(tenant=tenant, action="post", target_type="T", target_id="MINE")
|
||
log(tenant=other_tenant, action="post", target_type="T", target_id="THEIRS")
|
||
|
||
body = auth_client.get("/api/v1/core/audit-logs/?limit=500").json()
|
||
ids = [e["target_id"] for e in body["results"]]
|
||
assert "MINE" in ids
|
||
assert "THEIRS" not in ids
|
||
|
||
|
||
def test_audit_logs_are_read_only(db, auth_client, tenant):
|
||
"""审计日志没有写接口：不能 POST/PUT/DELETE（证据链完整）。"""
|
||
log(tenant=tenant, action="post", target_type="T", target_id="RO1")
|
||
entry = AuditLog.objects.filter(tenant=tenant, target_id="RO1").first()
|
||
|
||
assert auth_client.post("/api/v1/core/audit-logs/", {}, format="json").status_code in (403, 404, 405)
|
||
assert auth_client.delete(f"/api/v1/core/audit-logs/{entry.id}/").status_code in (403, 404, 405)
|
||
assert auth_client.put(f"/api/v1/core/audit-logs/{entry.id}/", {},
|
||
format="json").status_code in (403, 404, 405)
|
||
|
||
# 记录仍在
|
||
assert AuditLog.objects.filter(pk=entry.pk).exists()
|
||
|
||
|
||
def test_requires_auth(db, client, tenant):
|
||
resp = client.get("/api/v1/core/audit-logs/", HTTP_X_TENANT_ID=tenant.code)
|
||
assert resp.status_code in (401, 403)
|
||
|
||
|
||
def test_requires_tenant(db, user, tenant):
|
||
"""未知租户必须被拒绝。
|
||
|
||
注意：APIClient.credentials() 设的 header 是**持久**的，单次请求传
|
||
HTTP_X_TENANT_ID 不会覆盖它——必须新建 client 才能测出真实行为。
|
||
"""
|
||
c = APIClient()
|
||
c.credentials(
|
||
HTTP_AUTHORIZATION=f"Bearer {RefreshToken.for_user(user).access_token}",
|
||
HTTP_X_TENANT_ID="no-such-tenant",
|
||
)
|
||
resp = c.get("/api/v1/core/audit-logs/")
|
||
assert resp.status_code == 400, resp.content
|
||
assert "tenant" in str(resp.json())
|
||
|
||
|
||
def test_timeline_merges_related_actions(db, auth_client, tenant, warehouse, product):
|
||
"""轨迹要覆盖"这张单的所有动作"，包括挂在其他 target_type 下的放行记录。
|
||
|
||
背景：强制放行的 target_type 是 credit_limit（管控类型），
|
||
若轨迹只按 target_type 过滤，用户看到的单据轨迹会缺掉最关键的那次放行。
|
||
靠 detail.bill_no 把两者串起来。
|
||
"""
|
||
from apps.finance.models import Receivable
|
||
from apps.inventory import services as inv
|
||
|
||
limited = baker.make(Customer, tenant=tenant, code="CLIM", name="受限客户",
|
||
credit_limit=Decimal("100"))
|
||
baker.make(Receivable, tenant=tenant, customer=limited, bill_no="RC-X",
|
||
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)
|
||
|
||
body = auth_client.get(
|
||
f"/api/v1/core/audit-logs/timeline/?target_type=SalesBill&target_id={bill.bill_no}"
|
||
).json()
|
||
|
||
actions = [t["action"] for t in body["timeline"]]
|
||
assert "post" in actions, f"缺少过账记录：{actions}"
|
||
assert "force" in actions, f"缺少放行记录（轨迹未串起同一单据的动作）：{actions}"
|
||
|
||
# 顺序反映**真实执行顺序**：confirm_sales_bill 先做额度校验（可能放行），
|
||
# 再做库存扣减、生成应收，最后才写"过账"审计。
|
||
# 因此放行在前、过账在后是正确的——测试记录这个事实，避免后人误以为是 bug。
|
||
assert actions == ["force", "post"], f"执行顺序变了：{actions}"
|