"""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)