183 lines
6.4 KiB
Python
183 lines
6.4 KiB
Python
"""P0 #6 账龄分析 + 客户对账单 + 公开分享链接集成测试。"""
|
||
|
||
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
|
||
from apps.inventory.models import Warehouse
|
||
from apps.partner.models import Customer
|
||
from apps.finance import services as fin_services
|
||
from apps.finance.models import Receivable, Receipt, StatementShare
|
||
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 anon_client(db):
|
||
return APIClient()
|
||
|
||
|
||
@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="商品A",
|
||
sale_price=Decimal("100"))
|
||
|
||
|
||
@pytest.fixture
|
||
def customer(db, tenant):
|
||
return baker.make(Customer, tenant=tenant, code="C001", name="账龄客户")
|
||
|
||
|
||
def _recv(tenant, customer, *, days_ago, total, paid=0, ref="RC-X"):
|
||
return baker.make(
|
||
Receivable, tenant=tenant, customer=customer, bill_no=ref,
|
||
source_type="sale", source_ref=ref, bill_date=date.today() - timedelta(days=days_ago),
|
||
total_amount=Decimal(str(total)), paid_amount=Decimal(str(paid)),
|
||
status="partial" if paid else "open",
|
||
)
|
||
|
||
|
||
# ---------- 账龄 ----------
|
||
|
||
|
||
def test_aging_buckets(tenant, customer):
|
||
_recv(tenant, customer, days_ago=10, total=100, ref="RC-A") # 0-30
|
||
_recv(tenant, customer, days_ago=45, total=200, paid=50, ref="RC-B") # 31-60 → 150
|
||
_recv(tenant, customer, days_ago=100, total=300, ref="RC-C") # 91-180
|
||
_recv(tenant, customer, days_ago=400, total=400, ref="RC-D") # 365+
|
||
data = fin_services.receivable_aging(tenant)
|
||
amounts = {b["bucket"]: b["amount"] for b in data["buckets"]}
|
||
assert amounts["0-30"] == "100.0000"
|
||
assert amounts["31-60"] == "150.0000"
|
||
assert amounts["61-90"] == "0.0000"
|
||
assert amounts["91-180"] == "300.0000"
|
||
assert amounts["181-365"] == "0.0000"
|
||
assert amounts["365+"] == "400.0000"
|
||
assert data["total"] == "950.0000"
|
||
|
||
|
||
def test_aging_paid_receivable_excluded(tenant, customer):
|
||
_recv(tenant, customer, days_ago=5, total=999, paid=999, ref="RC-P")
|
||
data = fin_services.receivable_aging(tenant)
|
||
assert data["total"] == "0.0000"
|
||
|
||
|
||
def test_aging_per_customer(tenant, customer):
|
||
other = baker.make(Customer, tenant=tenant, code="C002", name="别家")
|
||
_recv(tenant, customer, days_ago=10, total=100, ref="RC-A")
|
||
_recv(tenant, other, days_ago=10, total=500, ref="RC-B")
|
||
data = fin_services.receivable_aging(tenant, customer=customer)
|
||
assert data["total"] == "100.0000"
|
||
|
||
|
||
# ---------- 对账单 ----------
|
||
|
||
|
||
def test_statement_opening_lines_closing(tenant, customer, warehouse, product):
|
||
"""期初（区间前应收未结）→ 区间内新应收（借）→ 区间内收款（贷）→ 期末平衡。"""
|
||
# 期初：30 天前应收 1000，已收 300
|
||
_recv(tenant, customer, days_ago=40, total=1000, paid=300, ref="RC-OPEN")
|
||
# 区间内：今天销售开单 500
|
||
from apps.inventory import services as inv_services
|
||
inv_services.inbound(
|
||
tenant=tenant, warehouse=warehouse, product=product,
|
||
quantity=Decimal("10"), unit_cost=Decimal("50"),
|
||
)
|
||
bill = sales_services.create_sales_bill(
|
||
tenant=tenant, customer=customer, warehouse=warehouse,
|
||
lines=[{"product": product, "quantity": 5, "unit_price": 100}],
|
||
)
|
||
sales_services.confirm_sales_bill(bill) # 应收 RC… 500
|
||
|
||
# 区间内收款 200，核销到期初应收
|
||
receipt = baker.make(
|
||
Receipt, tenant=tenant, customer=customer, bill_no="SK001",
|
||
bill_date=date.today(), amount=Decimal("200"), status="draft",
|
||
method="银行转账",
|
||
)
|
||
recv_open = Receivable.objects.get(bill_no="RC-OPEN")
|
||
fin_services.allocate_receipt(receipt, [(recv_open, Decimal("200"))])
|
||
|
||
data = fin_services.customer_statement(
|
||
tenant, customer, date_from=date.today(), date_to=date.today()
|
||
)
|
||
assert data["opening_balance"] == "700.0000" # 1000-300
|
||
assert data["closing_balance"] == "1000.0000" # 700 + 500 - 200
|
||
types = [(l["type"], l["ref"]) for l in data["lines"]]
|
||
# 应收行 ref 是应收单号（RC…，source_ref 才是销售单号）
|
||
rc_refs = [l["ref"] for l in data["lines"] if l["type"] == "receivable"]
|
||
assert len(rc_refs) == 1 and rc_refs[0].startswith("RC")
|
||
assert ("receipt", "SK001") in types
|
||
# 余额滚动校验
|
||
last = data["lines"][-1]
|
||
assert last["balance"] == "1000.0000"
|
||
|
||
|
||
# ---------- 分享链接 ----------
|
||
|
||
|
||
def test_share_link_flow(db, auth_client, anon_client, tenant, customer):
|
||
# 生成分享
|
||
resp = auth_client.post(
|
||
"/api/v1/finance/statements/share/",
|
||
{
|
||
"customer_id": customer.id,
|
||
"date_from": date.today().isoformat(),
|
||
"date_to": date.today().isoformat(),
|
||
"expires_days": 1,
|
||
},
|
||
format="json",
|
||
)
|
||
assert resp.status_code == 201, resp.content
|
||
token = resp.json()["token"]
|
||
url = resp.json()["url"]
|
||
|
||
# 匿名可访问（无需租户头/JWT）
|
||
resp2 = anon_client.get(url)
|
||
assert resp2.status_code == 200, resp2.content
|
||
body = resp2.json()
|
||
assert body["customer"]["code"] == "C001"
|
||
assert body["share"]["token"] == token
|
||
|
||
# 吊销后 404
|
||
share = StatementShare.objects.get(token=token)
|
||
share.revoked = True
|
||
share.save()
|
||
resp3 = anon_client.get(url)
|
||
assert resp3.status_code == 404
|
||
|
||
|
||
def test_share_link_expired_404(db, auth_client, anon_client, tenant, customer):
|
||
from django.utils import timezone
|
||
share = baker.make(
|
||
StatementShare, tenant=tenant, customer=customer,
|
||
date_from=date.today(), date_to=date.today(),
|
||
expires_at=timezone.now() - timedelta(hours=1),
|
||
)
|
||
resp = anon_client.get(f"/api/v1/open/statements/{share.token}/")
|
||
assert resp.status_code == 404
|
||
|
||
|
||
def test_share_list_requires_auth(db, anon_client):
|
||
resp = anon_client.get("/api/v1/finance/statements/share/")
|
||
assert resp.status_code in (401, 403)
|