"""完整财务深度总账及业财一体化集成测试: 1. 会计科目初始化与期间管理 2. 记账凭证录入、借贷平衡校验、状态流转(草稿/已过账/已作废) 3. 业财一体化:销售应收、销售成本、采购应付、收付款核销自动生成记账凭证 4. 期末损益结转与结账/反结账 5. 财务三大报表(科目余额表试算平衡、资产负债表等式平衡、利润表)与明细账 6. 财务 REST API 接口测试(初始化科目、录入凭证、过账、结账、报表查询) """ 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.partner.models import Customer, Supplier from apps.inventory.models import Warehouse from apps.catalog.models import Product from apps.finance import services as fs from apps.finance.models import ( Account, Period, Voucher, VoucherEntry, Receivable, Payable, Receipt, Payment, Allocation, ) @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.fixture def customer(tenant): return baker.make(Customer, tenant=tenant, code="C001", name="测试客户A") @pytest.fixture def supplier(tenant): return baker.make(Supplier, tenant=tenant, code="S001", name="测试供应商A") # ============================================================ # 1. 会计科目与期间 # ============================================================ @pytest.mark.django_db def test_init_chart_of_accounts(tenant): count1 = fs.init_chart_of_accounts(tenant) assert count1 == 22 assert Account.objects.filter(tenant=tenant).count() == 22 # 幂等性:重复执行不重复创建 count2 = fs.init_chart_of_accounts(tenant) assert count2 == 0 assert Account.objects.filter(tenant=tenant).count() == 22 # 验证关键科目存在 cash = fs.get_account(tenant, "1001") assert cash.name == "现金" assert cash.category == "asset" assert cash.balance_type == "debit" ar = fs.get_account(tenant, "1122") assert ar.aux_customer is True ap = fs.get_account(tenant, "2202") assert ap.aux_supplier is True @pytest.mark.django_db def test_accounting_periods(tenant): today = date(2026, 9, 15) period = fs.period_of(tenant, today) assert period.code == "2026-09" assert period.start_date == date(2026, 9, 1) assert period.end_date == date(2026, 9, 30) assert period.status == "open" # 跨月批量初始化 n = fs.ensure_periods(tenant, date(2026, 1, 1), date(2026, 3, 1)) assert n == 3 assert Period.objects.filter(tenant=tenant, code="2026-01").exists() # ============================================================ # 2. 凭证服务与借贷校验 # ============================================================ @pytest.mark.django_db def test_voucher_create_and_balance_validation(tenant, customer): fs.init_chart_of_accounts(tenant) today = date(2026, 9, 10) # 1. 借贷不平衡时必须抛出异常 with pytest.raises(fs.VoucherError, match="借贷不平衡"): fs.create_voucher( tenant=tenant, voucher_date=today, entries=[ {"account_code": "1001", "debit": Decimal("100"), "credit": 0}, {"account_code": "1002", "debit": 0, "credit": Decimal("90")}, ], ) # 2. 借贷同时有值必须抛异常 with pytest.raises(fs.VoucherError, match="借贷不能同时有值"): fs.create_voucher( tenant=tenant, voucher_date=today, entries=[ {"account_code": "1001", "debit": Decimal("100"), "credit": Decimal("50")}, {"account_code": "1002", "debit": 0, "credit": Decimal("50")}, ], ) # 3. 辅助核算缺失时必须抛异常 with pytest.raises(fs.VoucherError, match="需要关联客户辅助核算"): fs.create_voucher( tenant=tenant, voucher_date=today, entries=[ {"account_code": "1122", "debit": Decimal("500"), "credit": 0, "customer_id": None}, {"account_code": "6001", "debit": 0, "credit": Decimal("500")}, ], ) # 4. 正常凭证创建 voucher = fs.create_voucher( tenant=tenant, voucher_date=today, summary="提现备用金", entries=[ {"account_code": "1001", "debit": Decimal("500"), "credit": 0}, {"account_code": "1002", "debit": 0, "credit": Decimal("500")}, ], ) assert voucher.status == "draft" assert voucher.bill_no.startswith("V202609") assert voucher.total_debit == Decimal("500") assert voucher.total_credit == Decimal("500") assert voucher.entries.count() == 2 @pytest.mark.django_db def test_voucher_posting_and_cancel_lifecycle(tenant, user): fs.init_chart_of_accounts(tenant) today = date(2026, 9, 10) voucher = fs.create_voucher( tenant=tenant, voucher_date=today, summary="日常报销", entries=[ {"account_code": "6602", "debit": Decimal("200"), "credit": 0}, {"account_code": "1002", "debit": 0, "credit": Decimal("200")}, ], ) assert voucher.status == "draft" # 过账 posted_v = fs.post_voucher(voucher, user=user) assert posted_v.status == "posted" assert posted_v.posted_by == user assert posted_v.posted_at is not None # 重复过账抛异常 with pytest.raises(fs.VoucherError, match="只能对草稿凭证过账"): fs.post_voucher(posted_v) # 作废 cancelled_v = fs.cancel_voucher(posted_v, user=user) assert cancelled_v.status == "cancelled" # ============================================================ # 3. 业财一体化自动凭证 # ============================================================ @pytest.mark.django_db def test_auto_voucher_from_sales_and_cogs(tenant, customer): fs.init_chart_of_accounts(tenant) today = date(2026, 9, 10) # 1. 产生销售应收 -> 自动生成凭证 recv = fs.create_receivable_from_sale( tenant=tenant, customer=customer, total_amount=Decimal("1200.00"), source_ref="XS202609100001", bill_date=today, ) # 验证自动生成的凭证存在且已过账 v_sale = Voucher.objects.get(tenant=tenant, source_type="sale", source_ref=recv.bill_no) assert v_sale.status == "posted" assert v_sale.total_debit == Decimal("1200.00") # 分录 1:借 1122 应收账款 (挂客户) e1 = v_sale.entries.get(account__code="1122") assert e1.debit == Decimal("1200.00") assert e1.customer == customer # 分录 2:贷 6001 主营业务收入 e2 = v_sale.entries.get(account__code="6001") assert e2.credit == Decimal("1200.00") # 2. 产生销售成本凭证 v_cogs = fs.voucher_for_sale_cogs( tenant=tenant, sales_bill_no="XS202609100001", bill_date=today, cost_amount=Decimal("800.00"), auto_post=True, ) assert v_cogs.status == "posted" assert v_cogs.total_debit == Decimal("800.00") assert v_cogs.entries.get(account__code="6401").debit == Decimal("800.00") assert v_cogs.entries.get(account__code="1403").credit == Decimal("800.00") @pytest.mark.django_db def test_auto_voucher_from_purchase_and_payment(tenant, supplier): fs.init_chart_of_accounts(tenant) today = date(2026, 9, 10) # 1. 采购应付 -> 自动生成采购凭证 pay = fs.create_payable_from_purchase( tenant=tenant, supplier=supplier, total_amount=Decimal("2000.00"), source_ref="PB202609100001", bill_date=today, ) v_pay = Voucher.objects.get(tenant=tenant, source_type="purchase", source_ref=pay.bill_no) assert v_pay.status == "posted" assert v_pay.entries.get(account__code="1403").debit == Decimal("2000.00") e_ap = v_pay.entries.get(account__code="2202") assert e_ap.credit == Decimal("2000.00") assert e_ap.supplier == supplier # 2. 付款核销 -> 自动生成付款凭证 payment = baker.make( Payment, tenant=tenant, supplier=supplier, bill_no="PMT-101", bill_date=today, amount=Decimal("2000.00"), ) fs.allocate_payment(payment, [(pay, Decimal("2000.00"))]) v_pmt = Voucher.objects.get(tenant=tenant, source_type="payment", source_ref=payment.bill_no) assert v_pmt.status == "posted" assert v_pmt.entries.get(account__code="2202").debit == Decimal("2000.00") assert v_pmt.entries.get(account__code="1002").credit == Decimal("2000.00") @pytest.mark.django_db def test_auto_voucher_from_receipt(tenant, customer): fs.init_chart_of_accounts(tenant) today = date(2026, 9, 10) recv = fs.create_receivable_from_sale( tenant=tenant, customer=customer, total_amount=Decimal("1500.00"), source_ref="XS-002", bill_date=today, ) receipt = baker.make( Receipt, tenant=tenant, customer=customer, bill_no="RCT-101", bill_date=today, amount=Decimal("1500.00"), method="银行转账", ) fs.allocate_receipt(receipt, [(recv, Decimal("1500.00"))]) v_rct = Voucher.objects.get(tenant=tenant, source_type="receipt", source_ref=receipt.bill_no) assert v_rct.status == "posted" assert v_rct.entries.get(account__code="1002").debit == Decimal("1500.00") assert v_rct.entries.get(account__code="1122").credit == Decimal("1500.00") # ============================================================ # 4. 期末损益结转与结账/反结账 # ============================================================ @pytest.mark.django_db def test_period_close_and_reopen(tenant, customer, user): fs.init_chart_of_accounts(tenant) p = fs.period_of(tenant, date(2026, 9, 1)) # 制造一笔收入:1000 元(借:应收 1000 贷:主营业务收入 1000) fs.create_voucher( tenant=tenant, voucher_date=date(2026, 9, 5), entries=[ {"account_code": "1122", "debit": Decimal("1000"), "credit": 0, "customer_id": customer.id}, {"account_code": "6001", "debit": 0, "credit": Decimal("1000")}, ], auto_post=True, ) # 制造两笔费用:主营业务成本 600 元,管理费用 100 元 fs.create_voucher( tenant=tenant, voucher_date=date(2026, 9, 6), entries=[ {"account_code": "6401", "debit": Decimal("600"), "credit": 0}, {"account_code": "1403", "debit": 0, "credit": Decimal("600")}, ], auto_post=True, ) fs.create_voucher( tenant=tenant, voucher_date=date(2026, 9, 7), entries=[ {"account_code": "6602", "debit": Decimal("100"), "credit": 0}, {"account_code": "1002", "debit": 0, "credit": Decimal("100")}, ], auto_post=True, ) # 此时未结账:净利润应为 1000 - 600 - 100 = 300 p_stmt = fs.income_statement(tenant, p) assert p_stmt["total_revenue"] == Decimal("1000") assert p_stmt["total_expense"] == Decimal("700") assert p_stmt["net_profit"] == Decimal("300") # 执行期末结账 close_voucher = fs.close_period(p, user=user) assert close_voucher is not None assert close_voucher.status == "posted" p.refresh_from_db() assert p.status == "closed" assert p.closed_by == user # 结转凭证分录检查: # 借:6001 主营业务收入 1000 # 贷:6401 主营业务成本 600 # 贷:6602 管理费用 100 # 贷:4103 本年利润 300 (净利润) assert close_voucher.entries.get(account__code="6001").debit == Decimal("1000") assert close_voucher.entries.get(account__code="6401").credit == Decimal("600") assert close_voucher.entries.get(account__code="6602").credit == Decimal("100") profit_entry = close_voucher.entries.get(account__code="4103") assert profit_entry.credit == Decimal("300") # 结账后,已结账期间不允许创建凭证 with pytest.raises(fs.VoucherError, match="已结账,禁止"): fs.create_voucher( tenant=tenant, voucher_date=date(2026, 9, 15), entries=[ {"account_code": "1001", "debit": Decimal("10"), "credit": 0}, {"account_code": "1002", "debit": 0, "credit": Decimal("10")}, ], ) # 反结账 fs.reopen_period(p, user=user) p.refresh_from_db() assert p.status == "open" assert p.closed_at is None # 原结账凭证应被自动作废 close_voucher.refresh_from_db() assert close_voucher.status == "cancelled" # ============================================================ # 5. 财务报表平衡与明细账 # ============================================================ @pytest.mark.django_db def test_financial_reports_and_balance_invariants(tenant, customer, supplier): fs.init_chart_of_accounts(tenant) p = fs.period_of(tenant, date(2026, 9, 1)) # 1. 股东实收资本注资 100,000 元(借:银行存款 100000,贷:实收资本 100000) fs.create_voucher( tenant=tenant, voucher_date=date(2026, 9, 1), entries=[ {"account_code": "1002", "debit": Decimal("100000"), "credit": 0}, {"account_code": "4001", "debit": 0, "credit": Decimal("100000")}, ], auto_post=True, ) # 2. 采购商品 20,000 元(借:库存商品 20000,贷:应付账款 20000) fs.create_voucher( tenant=tenant, voucher_date=date(2026, 9, 2), entries=[ {"account_code": "1403", "debit": Decimal("20000"), "credit": 0}, {"account_code": "2202", "debit": 0, "credit": Decimal("20000"), "supplier_id": supplier.id}, ], auto_post=True, ) # 3. 销售商品 30,000 元(借:应收账款 30000,贷:主营业务收入 30000) fs.create_voucher( tenant=tenant, voucher_date=date(2026, 9, 3), entries=[ {"account_code": "1122", "debit": Decimal("30000"), "credit": 0, "customer_id": customer.id}, {"account_code": "6001", "debit": 0, "credit": Decimal("30000")}, ], auto_post=True, ) # 4. 结转销售成本 15,000 元(借:主营业务成本 15000,贷:库存商品 15000) fs.create_voucher( tenant=tenant, voucher_date=date(2026, 9, 3), entries=[ {"account_code": "6401", "debit": Decimal("15000"), "credit": 0}, {"account_code": "1403", "debit": 0, "credit": Decimal("15000")}, ], auto_post=True, ) # 5. 收到客户货款 30,000 元(借:银行存款 30000,贷:应收账款 30000) fs.create_voucher( tenant=tenant, voucher_date=date(2026, 9, 5), entries=[ {"account_code": "1002", "debit": Decimal("30000"), "credit": 0}, {"account_code": "1122", "debit": 0, "credit": Decimal("30000"), "customer_id": customer.id}, ], auto_post=True, ) # 6. 支付供应商货款 10,000 元(借:应付账款 10000,贷:银行存款 10000) fs.create_voucher( tenant=tenant, voucher_date=date(2026, 9, 6), entries=[ {"account_code": "2202", "debit": Decimal("10000"), "credit": 0, "supplier_id": supplier.id}, {"account_code": "1002", "debit": 0, "credit": Decimal("10000")}, ], auto_post=True, ) # === 校验 1:科目余额表(试算平衡)=== tb = fs.trial_balance(tenant, p) assert tb["is_balanced"] is True assert tb["totals"]["period_debit"] == tb["totals"]["period_credit"] assert tb["totals"]["ending_debit"] == tb["totals"]["ending_credit"] # === 校验 2:资产负债表等式 === # 资产类: # 银行存款 = 100000 (注资) + 30000 (收客户) - 10000 (付供应商) = 120000 # 应收账款 = 30000 - 30000 = 0 # 库存商品 = 20000 (入库) - 15000 (出库) = 5000 # 资产总计 = 125000 # 负债类: # 应付账款 = 20000 - 10000 = 10000 # 所有者权益类: # 实收资本 = 100000 # 未结转本期利润 = 收入 30000 - 成本 15000 = 15000 # 负债及所有者权益总计 = 10000 + 100000 + 15000 = 125000 bs = fs.balance_sheet(tenant, p) assert bs["asset_total"] == Decimal("125000") assert bs["liability_total"] == Decimal("10000") assert bs["equity_total"] == Decimal("100000") assert bs["unclosed_profit"] == Decimal("15000") assert bs["balanced"] is True # === 校验 3:利润表 === stmt = fs.income_statement(tenant, p) assert stmt["total_revenue"] == Decimal("30000") assert stmt["cogs"] == Decimal("15000") assert stmt["gross_profit"] == Decimal("15000") assert stmt["net_profit"] == Decimal("15000") # === 校验 4:银行存款明细账 === bank_acc = fs.get_account(tenant, "1002") ledger = fs.account_ledger(tenant, bank_acc, p) assert ledger["account_code"] == "1002" assert len(ledger["lines"]) == 3 # 注资、收客户、付供应商 assert ledger["ending_balance"] == Decimal("120000") # ============================================================ # 6. REST API 接口测试 # ============================================================ @pytest.mark.django_db def test_receipt_allocation_api(auth_client, tenant, customer, other_tenant): """收款核销 REST action 调用既有服务并隔离租户/客户。""" recv = fs.create_receivable_from_sale( tenant=tenant, customer=customer, total_amount=Decimal("400.00") ) receipt = baker.make( Receipt, tenant=tenant, customer=customer, bill_no="RCT-API-001", bill_date=date.today(), amount=Decimal("400.00"), ) response = auth_client.post( f"/api/v1/finance/receipts/{receipt.id}/allocate/", {"allocations": [{"receivable": recv.id, "amount": "400.00"}]}, format="json", ) assert response.status_code == 200, response.content assert response.json()["ok"] is True assert response.json()["receipt"]["status"] == "posted" assert response.json()["allocations"][0]["receivable"] == recv.id recv.refresh_from_db() assert recv.status == "paid" assert Allocation.objects.filter(receipt=receipt, receivable=recv).count() == 1 other_customer = baker.make(Customer, tenant=other_tenant, code="OTHER-C1") foreign_recv = baker.make( Receivable, tenant=other_tenant, customer=other_customer, bill_no="RC-OTHER-001", bill_date=date.today(), total_amount=Decimal("400.00"), ) second_receipt = baker.make( Receipt, tenant=tenant, customer=customer, bill_no="RCT-API-002", bill_date=date.today(), amount=Decimal("400.00"), ) response = auth_client.post( f"/api/v1/finance/receipts/{second_receipt.id}/allocate/", {"allocations": [{"receivable": foreign_recv.id, "amount": "400.00"}]}, format="json", ) assert response.status_code == 400, response.content assert "allocations" in response.json() assert Allocation.objects.filter(receipt=second_receipt).count() == 0 same_tenant_customer = baker.make(Customer, tenant=tenant, code="C-SAME-TENANT") mismatched_recv = baker.make( Receivable, tenant=tenant, customer=same_tenant_customer, bill_no="RC-API-MISMATCH", bill_date=date.today(), total_amount=Decimal("400.00"), ) third_receipt = baker.make( Receipt, tenant=tenant, customer=customer, bill_no="RCT-API-003", bill_date=date.today(), amount=Decimal("400.00"), ) response = auth_client.post( f"/api/v1/finance/receipts/{third_receipt.id}/allocate/", {"allocations": [{"receivable": mismatched_recv.id, "amount": "400.00"}]}, format="json", ) assert response.status_code == 400, response.content assert "客户" in response.json().get("detail", "") assert Allocation.objects.filter(receipt=third_receipt).count() == 0 @pytest.mark.django_db def test_payment_allocation_api(auth_client, tenant, supplier, other_tenant): """付款核销 REST action 校验供应商一致性并返回创建的核销明细。""" payable = fs.create_payable_from_purchase( tenant=tenant, supplier=supplier, total_amount=Decimal("250.00") ) payment = baker.make( Payment, tenant=tenant, supplier=supplier, bill_no="PMT-API-001", bill_date=date.today(), amount=Decimal("250.00"), ) response = auth_client.post( f"/api/v1/finance/payments/{payment.id}/allocate/", {"allocations": [{"payable_id": payable.id, "amount": "250.00"}]}, format="json", ) assert response.status_code == 200, response.content assert response.json()["ok"] is True assert response.json()["payment"]["status"] == "posted" assert response.json()["allocations"][0]["payable"] == payable.id payable.refresh_from_db() assert payable.status == "paid" assert Allocation.objects.filter(payment=payment, payable=payable).count() == 1 other_supplier = baker.make(Supplier, tenant=tenant, code="S-OTHER") foreign_payable = baker.make( Payable, tenant=tenant, supplier=other_supplier, bill_no="PY-OTHER-001", bill_date=date.today(), total_amount=Decimal("250.00"), ) second_payment = baker.make( Payment, tenant=tenant, supplier=supplier, bill_no="PMT-API-002", bill_date=date.today(), amount=Decimal("250.00"), ) response = auth_client.post( f"/api/v1/finance/payments/{second_payment.id}/allocate/", {"allocations": [{"payable": foreign_payable.id, "amount": "250.00"}]}, format="json", ) assert response.status_code == 400, response.content assert "供应商" in response.json().get("detail", "") assert Allocation.objects.filter(payment=second_payment).count() == 0 @pytest.mark.django_db def test_finance_api_endpoints(auth_client, tenant, customer): # 1. 初始化科目接口 resp = auth_client.post("/api/v1/finance/accounts/init-chart/") assert resp.status_code == 200, resp.content assert resp.json()["ok"] is True assert resp.json()["created_count"] == 22 # 2. 查询科目列表 resp = auth_client.get("/api/v1/finance/accounts/?search=银行存款") assert resp.status_code == 200 results = resp.json()["results"] assert len(results) == 1 assert results[0]["code"] == "1002" # 3. 创建凭证接口 (带有借贷分录) payload = { "voucher_date": "2026-09-08", "summary": "API手工凭证", "auto_post": False, "entries": [ {"account_code": "1001", "debit": "600.00", "credit": "0", "summary": "现金收入"}, {"account_code": "1122", "debit": "0", "credit": "600.00", "summary": "冲销应收", "customer_id": customer.id}, ], } resp = auth_client.post("/api/v1/finance/vouchers/", payload, format="json") assert resp.status_code == 201, resp.content v_data = resp.json() v_id = v_data["id"] assert v_data["bill_no"].startswith("V202609") assert v_data["status"] == "draft" # 4. 凭证过账接口 resp = auth_client.post(f"/api/v1/finance/vouchers/{v_id}/post/") assert resp.status_code == 200 assert resp.json()["status"] == "posted" # 5. 科目余额表报表接口 resp = auth_client.get("/api/v1/finance/reports/trial-balance/?period=2026-09") assert resp.status_code == 200 tb_data = resp.json() assert tb_data["is_balanced"] is True assert len(tb_data["rows"]) > 0 # 6. 资产负债表报表接口 resp = auth_client.get("/api/v1/finance/reports/balance-sheet/?period=2026-09") assert resp.status_code == 200 bs_data = resp.json() assert bs_data["balanced"] is True # 现金借方+600 与 应收借方-600 相互抵消,总资产变动为 0 assert Decimal(str(bs_data["asset_total"])) == Decimal("0") assert len(bs_data["assets"]) == 2 # 7. 期间结账接口 period = Period.objects.get(tenant=tenant, code="2026-09") resp = auth_client.post(f"/api/v1/finance/periods/{period.id}/close/") assert resp.status_code == 200 assert resp.json()["status"] == "closed" # 8. 期间反结账接口 resp = auth_client.post(f"/api/v1/finance/periods/{period.id}/reopen/") assert resp.status_code == 200 assert resp.json()["status"] == "open"