腾讯WorkBuddy
85293b79df
- migration 6: manual_records, ledger_event_revisions chain, current projections, source claims, subject suggestions, eligible_position_events - ledger_events.py: bank-event reconciliation, reversal/adjustment/reopen, append-only revision chain and rebuildable current projection - subjects.py: fixed subject mirror, draft suggestion dictionary, explicit administrator subject confirmation with expected_revision + idempotency - manual_records.py: submit, approve new/link, return/exception/reverse, candidate hints, idempotent replay and concurrency-safe claims - positions.py: Decimal aggregation, both-perspective conservation asserts, cutoff window, unresolved gross buckets, keyset pagination, evidence visibility (visible/masked/missing) - server.py: admin + company intercompany APIs with tenant isolation (404 on cross-tenant reads, 403 on company writes) and auto reconcile wiring - admin/company portals: balance directory, pair drill-down drawer, evidence drawer, subject/manual audit queue, company balance summary - tests: ledger events, subjects, manual records, positions, HTTP API and migration persistence (233 total, all green)
307 lines
15 KiB
Python
307 lines
15 KiB
Python
"""B-44 ledger event layer tests: bank reconciliation, revision chain,
|
|
reversal/adjustment, projection rebuild and immutability boundaries."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from decimal import Decimal
|
|
import sqlite3
|
|
import unittest
|
|
|
|
from bank_importer import ledger_events, matching, subjects
|
|
from ledger_helpers import LedgerBase
|
|
|
|
|
|
class BankReconcileTests(LedgerBase):
|
|
def test_paired_eligible_event_creates_pending_subject_only(self) -> None:
|
|
self.pair(self.company_a, self.company_b, "100.00")
|
|
self.assertEqual(1, len(self.eligible()))
|
|
stats = ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
self.assertEqual(1, stats["created"])
|
|
events = self.ledger_events()
|
|
self.assertEqual(1, len(events))
|
|
revision = self.current(events[0]["id"])
|
|
self.assertEqual("pending_subject", revision["state"])
|
|
self.assertEqual(Decimal("100.00"), Decimal(revision["amount"]))
|
|
self.assertEqual("CNY", revision["currency"])
|
|
# Nothing confirmed yet -> no position event.
|
|
self.assertEqual([], self.position_events())
|
|
# The event is unresolved as subject_review.
|
|
from bank_importer import positions
|
|
unresolved = positions.unresolved_for_company(self.connection, self.company_a, "2026-12-31")
|
|
self.assertEqual("100.00", unresolved["by_reason"]["subject_review"]["gross_amount"])
|
|
|
|
def test_unlocked_single_never_enters_ledger(self) -> None:
|
|
self.add_row(
|
|
self.company_a, own_account="6222000000000001",
|
|
cp_account="6222000000000002", expense="100.00",
|
|
)
|
|
matching.reconcile_rows(self.connection, [1])
|
|
self.assertEqual([], self.eligible())
|
|
stats = ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
self.assertEqual(0, stats["created"])
|
|
self.assertEqual([], self.ledger_events())
|
|
|
|
def test_same_company_transfer_never_enters_ledger(self) -> None:
|
|
self.add_row(
|
|
self.company_a, own_account="6222000000000001",
|
|
cp_account="6222000000000001", expense="100.00",
|
|
)
|
|
self.add_row(
|
|
self.company_a, own_account="6222000000000001",
|
|
cp_account="6222000000000001", income="100.00",
|
|
)
|
|
matching.reconcile_rows(self.connection, [1, 2])
|
|
eligible = self.eligible()
|
|
self.assertEqual(0, len([e for e in eligible if e["payer_company_id"] == e["payee_company_id"]]))
|
|
stats = ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
self.assertEqual(0, stats["created"])
|
|
|
|
def test_reconcile_is_idempotent(self) -> None:
|
|
self.pair(self.company_a, self.company_b, "100.00")
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
events = self.ledger_events()
|
|
self.assertEqual(1, len(events))
|
|
before_revisions = self.connection.execute(
|
|
"SELECT COUNT(*) AS n FROM ledger_event_revisions"
|
|
).fetchone()["n"]
|
|
stats = ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
self.assertEqual(1, stats["unchanged"])
|
|
after_revisions = self.connection.execute(
|
|
"SELECT COUNT(*) AS n FROM ledger_event_revisions"
|
|
).fetchone()["n"]
|
|
self.assertEqual(before_revisions, after_revisions)
|
|
|
|
def test_reconcile_picks_up_newly_linked_rows_as_new_pending_event(self) -> None:
|
|
# A single observation locked as intercompany -> pending ledger event.
|
|
row_a = self.add_row(
|
|
self.company_a, own_account="6222000000000001",
|
|
cp_account="6222000000000002", expense="100.00",
|
|
)
|
|
matching.reconcile_rows(self.connection, [row_a])
|
|
single = self.connection.execute(
|
|
"SELECT * FROM current_transfer_decisions"
|
|
).fetchone()
|
|
event_id = single["event_id"]
|
|
decision = self.connection.execute(
|
|
"SELECT * FROM transfer_match_decisions WHERE id = ?",
|
|
(single["decision_id"],),
|
|
).fetchone()
|
|
matching.apply_manual_decision(
|
|
self.connection, event_id, "assign_participant",
|
|
reason="函证确认对方",
|
|
expected_revision=decision["revision"], request_key="assign-1",
|
|
actor=self.admin,
|
|
participant={"role": "payee", "company_id": self.company_b},
|
|
)
|
|
self.assertEqual(1, len(self.eligible()))
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
self.assertEqual(1, len(self.ledger_events()))
|
|
|
|
# The administrator links two fresh rows with a different amount onto a
|
|
# new canonical event; reconcile creates a second pending event.
|
|
row_a2 = self.add_row(
|
|
self.company_a, own_account="6222000000000001",
|
|
cp_account="6222000000000002", expense="150.00",
|
|
at="2026-02-01T10:00:00",
|
|
)
|
|
row_b2 = self.add_row(
|
|
self.company_b, own_account="6222000000000002",
|
|
cp_account="6222000000000001", income="150.00",
|
|
at="2026-02-01T11:00:00",
|
|
)
|
|
current_row = self.connection.execute(
|
|
"SELECT d.* FROM current_transfer_decisions c "
|
|
"JOIN transfer_match_decisions d ON d.id = c.decision_id "
|
|
"WHERE c.event_id = ?",
|
|
(event_id,),
|
|
).fetchone()
|
|
matching.apply_manual_decision(
|
|
self.connection, event_id, "link_rows",
|
|
reason="补录双边流水",
|
|
expected_revision=current_row["revision"], request_key="link-1",
|
|
actor=self.admin, source_row_ids=[row_a2, row_b2],
|
|
)
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
amounts = sorted(
|
|
row["amount"] for row in self.connection.execute(
|
|
"SELECT amount FROM ledger_event_revisions WHERE state = 'pending_subject'"
|
|
).fetchall()
|
|
)
|
|
self.assertEqual(["100.00", "150.00"], amounts)
|
|
|
|
def test_confirm_subject_enters_position_view(self) -> None:
|
|
self.pair(self.company_a, self.company_b, "100.00")
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
event_id = self.ledger_events()[0]["id"]
|
|
subjects.confirm_subject(
|
|
self.connection, event_id,
|
|
perspective_company_id=self.company_a, subject_code="other_receivable",
|
|
reason="借款建议其他应收", expected_revision=1, request_key="k1",
|
|
actor=self.admin,
|
|
)
|
|
positions = self.position_events()
|
|
self.assertEqual(1, len(positions))
|
|
self.assertEqual("other_receivable", positions[0]["subject_code"])
|
|
self.assertEqual(self.company_a, positions[0]["perspective_company_id"])
|
|
|
|
def test_confirm_stale_revision_conflicts(self) -> None:
|
|
self.pair(self.company_a, self.company_b, "100.00")
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
event_id = self.ledger_events()[0]["id"]
|
|
subjects.confirm_subject(
|
|
self.connection, event_id,
|
|
perspective_company_id=self.company_a, subject_code="receivable",
|
|
reason="确认应收", expected_revision=1, request_key="k1",
|
|
actor=self.admin,
|
|
)
|
|
with self.assertRaises(subjects.SubjectConflictError):
|
|
subjects.confirm_subject(
|
|
self.connection, event_id,
|
|
perspective_company_id=self.company_a, subject_code="receivable",
|
|
reason="重复确认", expected_revision=1, request_key="k2",
|
|
actor=self.admin,
|
|
)
|
|
|
|
def test_reversal_mirrors_direction_and_subject(self) -> None:
|
|
self.pair(self.company_a, self.company_b, "100.00")
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
event_id = self.ledger_events()[0]["id"]
|
|
subjects.confirm_subject(
|
|
self.connection, event_id,
|
|
perspective_company_id=self.company_a, subject_code="other_receivable",
|
|
reason="借款", expected_revision=1, request_key="k1", actor=self.admin,
|
|
)
|
|
reversal_id, _ = ledger_events.create_reversal(
|
|
self.connection, event_id, source_kind="adjustment",
|
|
reason="冲销错误确认", actor=self.admin,
|
|
)
|
|
reversal = self.current(reversal_id)
|
|
self.assertEqual("reversal", reversal["posting_kind"])
|
|
self.assertEqual(self.company_b, reversal["payer_company_id"])
|
|
self.assertEqual(self.company_a, reversal["payee_company_id"])
|
|
self.assertEqual("other_payable", reversal["subject_code"])
|
|
self.assertEqual(self.company_b, reversal["perspective_company_id"])
|
|
# Original event stays confirmed and still in the position view.
|
|
self.assertEqual("confirmed", self.current(event_id)["state"])
|
|
positions = self.position_events()
|
|
self.assertEqual(2, len(positions))
|
|
signed = sum(
|
|
Decimal(position["amount"]) * (
|
|
1 if position["payer_company_id"] == self.company_a else -1
|
|
)
|
|
for position in positions
|
|
)
|
|
self.assertEqual(Decimal("0"), signed)
|
|
|
|
def test_evidence_rows_stay_immutable_under_reversal(self) -> None:
|
|
self.pair(self.company_a, self.company_b, "100.00")
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
event_id = self.ledger_events()[0]["id"]
|
|
subjects.confirm_subject(
|
|
self.connection, event_id,
|
|
perspective_company_id=self.company_a, subject_code="receivable",
|
|
reason="确认", expected_revision=1, request_key="k1", actor=self.admin,
|
|
)
|
|
before = self.connection.execute(
|
|
"SELECT income, expense FROM source_rows ORDER BY id"
|
|
).fetchall()
|
|
ledger_events.create_reversal(
|
|
self.connection, event_id, source_kind="adjustment",
|
|
reason="冲销", actor=self.admin,
|
|
)
|
|
after = self.connection.execute(
|
|
"SELECT income, expense FROM source_rows ORDER BY id"
|
|
).fetchall()
|
|
self.assertEqual(before, after)
|
|
|
|
def test_reversal_idempotent_guard(self) -> None:
|
|
self.pair(self.company_a, self.company_b, "100.00")
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
event_id = self.ledger_events()[0]["id"]
|
|
subjects.confirm_subject(
|
|
self.connection, event_id,
|
|
perspective_company_id=self.company_a, subject_code="receivable",
|
|
reason="确认", expected_revision=1, request_key="k1", actor=self.admin,
|
|
)
|
|
# Simulate the B-43 event leaving the eligible set.
|
|
with self.connection:
|
|
self.connection.execute(
|
|
"UPDATE canonical_transfer_events SET lifecycle = 'superseded' WHERE id = ?",
|
|
(self.eligible()[0]["event_id"],),
|
|
)
|
|
stats = ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
self.assertEqual(1, stats["reversal"])
|
|
reversals = self.connection.execute(
|
|
"SELECT COUNT(*) AS n FROM ledger_event_revisions WHERE posting_kind = 'reversal'"
|
|
).fetchone()["n"]
|
|
self.assertEqual(1, reversals)
|
|
# Re-run does not create a second reversal.
|
|
stats = ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
self.assertEqual(0, stats["reversal"])
|
|
|
|
|
|
class ProjectionTests(LedgerBase):
|
|
def test_rebuild_current_projection(self) -> None:
|
|
self.pair(self.company_a, self.company_b, "100.00")
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
event_id = self.ledger_events()[0]["id"]
|
|
subjects.confirm_subject(
|
|
self.connection, event_id,
|
|
perspective_company_id=self.company_a, subject_code="receivable",
|
|
reason="确认", expected_revision=1, request_key="k1", actor=self.admin,
|
|
)
|
|
self.connection.execute("DELETE FROM current_ledger_event_revisions")
|
|
rebuilt = ledger_events.rebuild_current_ledger_projection(self.connection)
|
|
self.assertGreaterEqual(rebuilt, 1)
|
|
revision = self.current(event_id)
|
|
self.assertEqual("confirmed", revision["state"])
|
|
self.assertEqual("receivable", revision["subject_code"])
|
|
|
|
|
|
class SubjectSuggestionTests(LedgerBase):
|
|
def test_mirror_mapping_is_symmetric(self) -> None:
|
|
for subject, mirror in (
|
|
("receivable", "payable"), ("payable", "receivable"),
|
|
("other_receivable", "other_payable"), ("other_payable", "other_receivable"),
|
|
):
|
|
self.assertEqual(mirror, subjects.mirror_subject(subject))
|
|
self.assertEqual(subject, subjects.mirror_subject(mirror))
|
|
|
|
def test_loan_keyword_suggests_payer_other_receivable(self) -> None:
|
|
self.pair(self.company_a, self.company_b, "100.00", summary="借款", purpose="往来款")
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
event_id = self.ledger_events()[0]["id"]
|
|
suggestions = subjects.compute_suggestions(self.connection, event_id)
|
|
self.assertEqual(1, len(suggestions))
|
|
self.assertEqual(self.company_a, suggestions[0]["suggested_perspective_company_id"])
|
|
self.assertEqual("other_receivable", suggestions[0]["suggested_subject_code"])
|
|
self.assertFalse(suggestions[0]["evidence"]["approved"])
|
|
|
|
def test_repay_keyword_suggests_payee(self) -> None:
|
|
self.pair(self.company_a, self.company_b, "100.00", summary="归还往来款", purpose="还款")
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
event_id = self.ledger_events()[0]["id"]
|
|
suggestions = subjects.compute_suggestions(self.connection, event_id)
|
|
self.assertEqual(1, len(suggestions))
|
|
self.assertEqual(self.company_b, suggestions[0]["suggested_perspective_company_id"])
|
|
|
|
def test_trade_keyword_never_suggests(self) -> None:
|
|
# Trade vocabulary needs group approval; no dictionary hit -> review.
|
|
self.pair(self.company_a, self.company_b, "100.00", summary="货款", purpose="采购货款")
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
event_id = self.ledger_events()[0]["id"]
|
|
self.assertEqual([], subjects.compute_suggestions(self.connection, event_id))
|
|
|
|
def test_name_alone_never_confirms(self) -> None:
|
|
# A name match on the counterparty cannot decide the subject.
|
|
self.pair(self.company_a, self.company_b, "100.00", summary="某客户往来", purpose=None)
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
event_id = self.ledger_events()[0]["id"]
|
|
current = self.current(event_id)
|
|
self.assertEqual("pending_subject", current["state"])
|
|
self.assertEqual([], subjects.compute_suggestions(self.connection, event_id))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|