115 lines
5.0 KiB
Python
115 lines
5.0 KiB
Python
"""报表与 BI 大盘集成测试。"""
|
|
|
|
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
|
|
from apps.inventory.models import Warehouse, Stock
|
|
from apps.partner.models import Customer, Supplier
|
|
from apps.sales.models import SalesBill, SalesBillLine
|
|
from apps.purchase.models import PurchaseBill
|
|
from apps.finance.models import Receivable, Payable
|
|
from apps.report import services as report_services
|
|
from apps.report.models import ReportDefinition
|
|
|
|
|
|
@pytest.fixture
|
|
def auth_client(db, user, tenant):
|
|
client = APIClient()
|
|
refresh = RefreshToken.for_user(user)
|
|
client.credentials(
|
|
HTTP_AUTHORIZATION=f"Bearer {refresh.access_token}",
|
|
HTTP_X_TENANT_ID=tenant.code,
|
|
)
|
|
return client
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_init_system_reports(tenant):
|
|
count = report_services.init_system_reports(tenant)
|
|
assert count == 4
|
|
assert ReportDefinition.objects.filter(tenant=tenant).count() == 4
|
|
# 幂等性
|
|
assert report_services.init_system_reports(tenant) == 0
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_dashboard_summary_and_valuation(tenant):
|
|
wh = baker.make(Warehouse, tenant=tenant, code="WH01", name="总仓")
|
|
p1 = baker.make(Product, tenant=tenant, code="P01", name="商品1")
|
|
p2 = baker.make(Product, tenant=tenant, code="P02", name="商品2")
|
|
|
|
# 创建库存并设置成本
|
|
baker.make(Stock, tenant=tenant, warehouse=wh, product=p1, on_hand=Decimal("100"), avg_cost=Decimal("10"))
|
|
baker.make(Stock, tenant=tenant, warehouse=wh, product=p2, on_hand=Decimal("50"), avg_cost=Decimal("20"))
|
|
|
|
# 创建销售与采购
|
|
cust = baker.make(Customer, tenant=tenant, code="C01", name="客户1")
|
|
sb = baker.make(SalesBill, tenant=tenant, customer=cust, bill_date=date.today(), total_amount=Decimal("5000"), state="confirmed")
|
|
sup = baker.make(Supplier, tenant=tenant, code="S01", name="供应商1")
|
|
pb = baker.make(PurchaseBill, tenant=tenant, supplier=sup, bill_date=date.today(), total_amount=Decimal("3000"), state="confirmed")
|
|
|
|
# 应收应付
|
|
baker.make(Receivable, tenant=tenant, customer=cust, total_amount=Decimal("5000"), paid_amount=Decimal("2000"), status="partial", bill_date=date.today())
|
|
baker.make(Payable, tenant=tenant, supplier=sup, total_amount=Decimal("3000"), paid_amount=Decimal("1000"), status="partial", bill_date=date.today())
|
|
|
|
summary = report_services.get_dashboard_summary(tenant)
|
|
assert summary["month_sales"]["amount"] == Decimal("5000")
|
|
assert summary["month_purchase"]["amount"] == Decimal("3000")
|
|
assert summary["inventory"]["sku_count"] == 2
|
|
assert summary["inventory"]["total_quantity"] == Decimal("150")
|
|
# 估值 = 100*10 + 50*20 = 2000
|
|
assert summary["inventory"]["valuation"] == Decimal("2000")
|
|
# 待收 3000，待付 2000
|
|
assert summary["finance"]["receivable_balance"] == Decimal("3000")
|
|
assert summary["finance"]["payable_balance"] == Decimal("2000")
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_sales_rank_analytics(tenant):
|
|
cust1 = baker.make(Customer, tenant=tenant, code="C1", name="客户1")
|
|
cust2 = baker.make(Customer, tenant=tenant, code="C2", name="客户2")
|
|
prod1 = baker.make(Product, tenant=tenant, code="P1", name="雪碧")
|
|
prod2 = baker.make(Product, tenant=tenant, code="P2", name="可乐")
|
|
|
|
bill1 = baker.make(SalesBill, tenant=tenant, customer=cust1, state="confirmed", bill_date=date.today())
|
|
baker.make(SalesBillLine, tenant=tenant, bill=bill1, product=prod1, quantity=Decimal("10"), amount=Decimal("300"))
|
|
baker.make(SalesBillLine, tenant=tenant, bill=bill1, product=prod2, quantity=Decimal("5"), amount=Decimal("200"))
|
|
|
|
bill2 = baker.make(SalesBill, tenant=tenant, customer=cust2, state="confirmed", bill_date=date.today())
|
|
baker.make(SalesBillLine, tenant=tenant, bill=bill2, product=prod1, quantity=Decimal("20"), amount=Decimal("600"))
|
|
|
|
# 按商品排行
|
|
prod_rank = report_services.get_sales_rank(tenant, rank_by="product")
|
|
assert len(prod_rank) == 2
|
|
assert prod_rank[0]["product_code"] == "P1"
|
|
assert prod_rank[0]["total_amount"] == Decimal("900")
|
|
|
|
# 按客户排行
|
|
cust_rank = report_services.get_sales_rank(tenant, rank_by="customer")
|
|
assert len(cust_rank) == 2
|
|
assert cust_rank[0]["customer_code"] == "C2"
|
|
assert cust_rank[0]["total_amount"] == Decimal("600")
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_report_api_endpoints(auth_client, tenant):
|
|
# 初始化报表
|
|
r1 = auth_client.post("/api/v1/report/definitions/init-system/")
|
|
assert r1.status_code == 200
|
|
assert r1.json()["ok"] is True
|
|
|
|
# 查询大盘总览 API
|
|
r2 = auth_client.get("/api/v1/report/dashboard/summary/")
|
|
assert r2.status_code == 200
|
|
assert "inventory" in r2.json()
|
|
|
|
# 查询销售排行 API
|
|
r3 = auth_client.get("/api/v1/report/dashboard/sales-rank/?rank_by=product")
|
|
assert r3.status_code == 200
|
|
assert "results" in r3.json()
|