431 lines
20 KiB
Python
431 lines
20 KiB
Python
"""B-44 position aggregation tests: conservation, repayments, reversals across
|
|
cutoffs, currency isolation, Decimal precision, unresolved gross, normal
|
|
balance exceptions and keyset pagination."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from decimal import Decimal, getcontext
|
|
import unittest
|
|
|
|
from bank_importer import ledger_events, manual_records, positions, subjects
|
|
from ledger_helpers import LedgerBase
|
|
|
|
|
|
def confirm(connection, event_id: int, perspective: int, subject: str, key: str, actor):
|
|
return subjects.confirm_subject(
|
|
connection, event_id,
|
|
perspective_company_id=perspective, subject_code=subject,
|
|
reason="审核确认", expected_revision=1, request_key=key, actor=actor,
|
|
)
|
|
|
|
|
|
class ConservationTests(LedgerBase):
|
|
def setUp(self) -> None:
|
|
super().setUp()
|
|
getcontext().prec = 50
|
|
|
|
def _confirm_all(self, subject: str = "other_receivable"):
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
for event in self.ledger_events():
|
|
revision = self.current(event["id"])
|
|
confirm(
|
|
self.connection, event["id"],
|
|
revision["payer_company_id"], subject, f"subj-{event['id']}",
|
|
self.admin,
|
|
)
|
|
|
|
def test_both_perspectives_mirror_exactly(self) -> None:
|
|
self.pair(self.company_a, self.company_b, "100.00")
|
|
self.pair(self.company_a, self.company_b, "30.00", at="2026-01-10T10:00:00")
|
|
self.pair(self.company_a, self.company_b, "70.00", at="2026-02-05T10:00:00")
|
|
self._confirm_all()
|
|
pair = positions.pair_detail(
|
|
self.connection, self.company_a, self.company_b,
|
|
from_="2026-01-01", cutoff="2026-07-31",
|
|
)
|
|
item = pair["items"][0]
|
|
self.assertTrue(item["conservation"]["opposite"])
|
|
self.assertTrue(item["conservation"]["abs_equal"])
|
|
self.assertEqual(Decimal("200.00"), Decimal(item["a"]["result"]["signed_amount"]))
|
|
self.assertEqual(Decimal("-200.00"), Decimal(item["b"]["result"]["signed_amount"]))
|
|
|
|
def test_repayment_reduces_the_original_balance(self) -> None:
|
|
# A lends B 100 (A pays) then B repays 40 (B pays).
|
|
self.pair(self.company_a, self.company_b, "100.00")
|
|
self.pair(self.company_b, self.company_a, "40.00", at="2026-03-01T10:00:00",
|
|
summary="还款", purpose="归还借款")
|
|
self._confirm_all()
|
|
balances = positions.company_balances(
|
|
self.connection, from_="2026-01-01", cutoff="2026-07-31"
|
|
)
|
|
for item in balances["items"]:
|
|
if item["company_id"] == self.company_a:
|
|
self.assertEqual(Decimal("60.00"), Decimal(item["result"]["signed_amount"]))
|
|
self.assertEqual(Decimal("100.00"), Decimal(item["period"]["debit"]))
|
|
self.assertEqual(Decimal("40.00"), Decimal(item["period"]["credit"]))
|
|
else:
|
|
self.assertEqual(Decimal("-60.00"), Decimal(item["result"]["signed_amount"]))
|
|
self.assertEqual(Decimal("40.00"), Decimal(item["period"]["debit"]))
|
|
self.assertEqual(Decimal("100.00"), Decimal(item["period"]["credit"]))
|
|
|
|
def test_over_repayment_flags_normal_balance_exception(self) -> None:
|
|
# A lends B 100, then B repays 150 while A books the repayment against
|
|
# its receivable: A's receivable and B's payable both go negative.
|
|
self.pair(self.company_a, self.company_b, "100.00")
|
|
self.pair(self.company_b, self.company_a, "150.00", at="2026-03-01T10:00:00",
|
|
summary="还款", purpose="归还借款")
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
for event in self.ledger_events():
|
|
revision = self.current(event["id"])
|
|
# Book every event from A's perspective on the receivable side so
|
|
# the over-repayment produces abnormal balances.
|
|
confirm(
|
|
self.connection, event["id"], self.company_a,
|
|
"other_receivable", f"k-{event['id']}", self.admin,
|
|
)
|
|
pair = positions.pair_detail(
|
|
self.connection, self.company_a, self.company_b,
|
|
from_="2026-01-01", cutoff="2026-07-31",
|
|
)
|
|
item = pair["items"][0]
|
|
self.assertTrue(item["normal_balance_exception"])
|
|
self.assertIn("other_receivable", item["normal_balance_exception"])
|
|
self.assertIn("other_payable", item["normal_balance_exception"])
|
|
|
|
|
|
class CutoffAndReversalTests(LedgerBase):
|
|
def _loan(self) -> int:
|
|
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"]
|
|
confirm(self.connection, event_id, self.company_a, "other_receivable", "k1", self.admin)
|
|
return event_id
|
|
|
|
def test_cutoff_before_reversal_keeps_original_impact(self) -> None:
|
|
event_id = self._loan()
|
|
# Reversal dated after the original event.
|
|
ledger_events.create_reversal(
|
|
self.connection, event_id, source_kind="adjustment",
|
|
effective_at="2026-06-15T00:00:00", reason="冲销", actor=self.admin,
|
|
)
|
|
before = positions.company_balances(
|
|
self.connection, from_="2026-01-01", cutoff="2026-05-31"
|
|
)
|
|
after = positions.company_balances(
|
|
self.connection, from_="2026-01-01", cutoff="2026-07-31"
|
|
)
|
|
for item in before["items"]:
|
|
if item["company_id"] == self.company_a:
|
|
self.assertEqual(Decimal("100.00"), Decimal(item["result"]["signed_amount"]))
|
|
for item in after["items"]:
|
|
if item["company_id"] == self.company_a:
|
|
self.assertEqual(Decimal("0"), Decimal(item["result"]["signed_amount"]))
|
|
|
|
def test_from_gt_cutoff_is_rejected(self) -> None:
|
|
with self.assertRaises(positions.PositionInputError):
|
|
positions.validate_window("2026-08-01", "2026-07-31")
|
|
|
|
def test_events_before_from_are_excluded_from_period(self) -> None:
|
|
self._loan()
|
|
balances = positions.company_balances(
|
|
self.connection, from_="2026-06-01", cutoff="2026-07-31"
|
|
)
|
|
self.assertEqual([], balances["items"])
|
|
|
|
def test_reversal_uses_approved_effective_date(self) -> None:
|
|
event_id = self._loan()
|
|
ledger_events.create_reversal(
|
|
self.connection, event_id, source_kind="adjustment",
|
|
effective_at="2026-08-01T00:00:00", reason="未来冲销", actor=self.admin,
|
|
)
|
|
balances = positions.company_balances(
|
|
self.connection, from_="2026-01-01", cutoff="2026-07-31"
|
|
)
|
|
for item in balances["items"]:
|
|
if item["company_id"] == self.company_a:
|
|
# The future reversal does not rewrite the earlier cutoff.
|
|
self.assertEqual(Decimal("100.00"), Decimal(item["result"]["signed_amount"]))
|
|
|
|
|
|
class CurrencyAndPrecisionTests(LedgerBase):
|
|
def test_currencies_never_mix(self) -> None:
|
|
self.pair(self.company_a, self.company_b, "100.00", currency="CNY")
|
|
self.pair(self.company_a, self.company_b, "50.00", at="2026-01-20T10:00:00",
|
|
currency="USD")
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
events = self.ledger_events()
|
|
for event in events:
|
|
revision = self.current(event["id"])
|
|
confirm(self.connection, event["id"], revision["payer_company_id"],
|
|
"other_receivable", f"k-{event['id']}", self.admin)
|
|
balances = positions.company_balances(
|
|
self.connection, from_="2026-01-01", cutoff="2026-07-31"
|
|
)
|
|
by_company = {}
|
|
for item in balances["items"]:
|
|
by_company.setdefault(item["company_id"], {})[item["currency"]] = item["result"]
|
|
self.assertEqual(Decimal("100.00"), Decimal(by_company[self.company_a]["CNY"]["signed_amount"]))
|
|
self.assertEqual(Decimal("50.00"), Decimal(by_company[self.company_a]["USD"]["signed_amount"]))
|
|
|
|
def test_decimal_precision_is_preserved(self) -> None:
|
|
self.pair(self.company_a, self.company_b, "0.01")
|
|
self.pair(self.company_a, self.company_b, "12345.6789", at="2026-01-11T10:00:00")
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
for event in self.ledger_events():
|
|
revision = self.current(event["id"])
|
|
confirm(self.connection, event["id"], revision["payer_company_id"],
|
|
"other_receivable", f"k-{event['id']}", self.admin)
|
|
balances = positions.company_balances(
|
|
self.connection, from_="2026-01-01", cutoff="2026-07-31"
|
|
)
|
|
for item in balances["items"]:
|
|
if item["company_id"] == self.company_a:
|
|
self.assertEqual(
|
|
Decimal("12345.6889"), Decimal(item["result"]["signed_amount"])
|
|
)
|
|
# The stored string keeps every decimal place.
|
|
self.assertIn("12345.6889", item["result"]["signed_amount"])
|
|
|
|
def test_manual_record_preserves_amount_scale(self) -> None:
|
|
record = manual_records.submit(
|
|
self.connection, company_id=self.company_a,
|
|
counterparty_company_id=self.company_b,
|
|
occurred_at="2026-02-01T09:00:00", direction="incoming",
|
|
amount="12.3400", currency="CNY", funding_source="other",
|
|
requested_subject="receivable", request_key="mr-prec",
|
|
actor=self.admin,
|
|
)
|
|
manual_records.decide(
|
|
self.connection, record["id"], "approve_new",
|
|
reason="确认", expected_decision_id=record["decision_id"],
|
|
request_key="dec-prec", actor=self.admin,
|
|
)
|
|
stored = self.connection.execute(
|
|
"SELECT amount, amount_scale FROM eligible_position_events"
|
|
).fetchone()
|
|
self.assertEqual("12.3400", stored["amount"])
|
|
self.assertEqual(4, stored["amount_scale"])
|
|
|
|
|
|
class UnresolvedTests(LedgerBase):
|
|
def test_unresolved_gross_never_nets(self) -> None:
|
|
# Two pending_subject events of opposite economic signs must sum their
|
|
# absolute values, never cancel.
|
|
self.pair(self.company_a, self.company_b, "100.00")
|
|
self.pair(self.company_a, self.company_b, "40.00", at="2026-01-20T10:00:00",
|
|
summary="还款", purpose="归还借款")
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
unresolved = positions.unresolved_for_company(
|
|
self.connection, self.company_a, "2026-07-31"
|
|
)
|
|
self.assertEqual(2, unresolved["count"])
|
|
self.assertEqual(
|
|
Decimal("140.00"), Decimal(unresolved["gross_amount"])
|
|
)
|
|
self.assertEqual(
|
|
Decimal("140.00"),
|
|
Decimal(unresolved["by_reason"]["subject_review"]["gross_amount"]),
|
|
)
|
|
|
|
def test_unmatched_single_reason_bucket(self) -> None:
|
|
self.add_row(
|
|
self.company_a, own_account="6222000000000001",
|
|
cp_account="6222000000000002", expense="88.00",
|
|
)
|
|
from bank_importer import matching
|
|
matching.reconcile_rows(self.connection, [1])
|
|
unresolved = positions.unresolved_for_company(
|
|
self.connection, self.company_a, "2026-07-31"
|
|
)
|
|
self.assertEqual(1, unresolved["count"])
|
|
self.assertEqual(
|
|
Decimal("88.00"),
|
|
Decimal(unresolved["by_reason"]["unmatched_single"]["gross_amount"]),
|
|
)
|
|
|
|
def test_pending_manual_only_counts_for_submitting_company(self) -> None:
|
|
manual_records.submit(
|
|
self.connection, company_id=self.company_a,
|
|
counterparty_company_id=self.company_b,
|
|
occurred_at="2026-01-10T09:00:00", direction="incoming",
|
|
amount="50.00", currency="CNY", funding_source="other",
|
|
requested_subject="receivable", request_key="mr-a",
|
|
actor=self.admin,
|
|
)
|
|
a_unresolved = positions.unresolved_for_company(
|
|
self.connection, self.company_a, "2026-07-31"
|
|
)
|
|
b_unresolved = positions.unresolved_for_company(
|
|
self.connection, self.company_b, "2026-07-31"
|
|
)
|
|
self.assertEqual(1, a_unresolved["count"])
|
|
self.assertEqual(
|
|
Decimal("50.00"), Decimal(a_unresolved["by_reason"]["manual_pending"]["gross_amount"])
|
|
)
|
|
# The counterparty never sees the unapproved declaration.
|
|
self.assertEqual(0, b_unresolved["count"])
|
|
|
|
|
|
class PaginationTests(LedgerBase):
|
|
def test_subject_filter_matches_stored_and_mirror_subjects(self) -> None:
|
|
# A books "receivable" from its own perspective; B sees the mirror
|
|
# "payable". Filtering by B's mirror subject must still find the event.
|
|
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"]
|
|
confirm(self.connection, event_id, self.company_a, "receivable", "k-mirror",
|
|
self.admin)
|
|
|
|
stored = positions.list_events(
|
|
self.connection, from_="2026-01-01", cutoff="2026-07-31",
|
|
subject="receivable",
|
|
)
|
|
self.assertTrue(any(
|
|
item["ledger_event_id"] == event_id for item in stored["items"]
|
|
))
|
|
# Company B filters by its own perspective: the stored "receivable" is
|
|
# the mirror of "payable", so it must be included.
|
|
mirror = positions.list_events(
|
|
self.connection, from_="2026-01-01", cutoff="2026-07-31",
|
|
subject="payable", viewer_company_id=self.company_b,
|
|
)
|
|
self.assertTrue(any(
|
|
item["ledger_event_id"] == event_id for item in mirror["items"]
|
|
), mirror)
|
|
for item in mirror["items"]:
|
|
if item["ledger_event_id"] == event_id:
|
|
self.assertEqual("payable", item["own_subject"])
|
|
|
|
def test_subject_filter_without_match_returns_empty(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"]
|
|
confirm(self.connection, event_id, self.company_a, "receivable", "k-none",
|
|
self.admin)
|
|
other = positions.list_events(
|
|
self.connection, from_="2026-01-01", cutoff="2026-07-31",
|
|
subject="other_receivable",
|
|
)
|
|
self.assertEqual([], other["items"])
|
|
|
|
def test_pagination_is_stable_and_complete(self) -> None:
|
|
for index in range(7):
|
|
at = f"2026-01-{(index % 28) + 1:02d}T10:00:00"
|
|
self.pair(self.company_a, self.company_b, f"{index + 1}.00", at=at)
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
for event in self.ledger_events():
|
|
revision = self.current(event["id"])
|
|
confirm(self.connection, event["id"], revision["payer_company_id"],
|
|
"other_receivable", f"k-{event['id']}", self.admin)
|
|
|
|
seen: list[int] = []
|
|
cursor = None
|
|
pages = 0
|
|
while True:
|
|
page = positions.list_events(
|
|
self.connection, from_="2026-01-01", cutoff="2026-07-31",
|
|
limit=3, cursor=cursor,
|
|
)
|
|
pages += 1
|
|
seen.extend(item["ledger_event_id"] for item in page["items"])
|
|
cursor = page["next_cursor"]
|
|
if not page["has_more"]:
|
|
break
|
|
self.assertEqual(7, len(seen))
|
|
self.assertEqual(len(set(seen)), len(seen))
|
|
self.assertGreater(pages, 2)
|
|
|
|
def test_directory_pagination_complete(self) -> None:
|
|
for index in range(6):
|
|
at = f"2026-01-{(index % 28) + 1:02d}T10:00:00"
|
|
self.pair(self.company_a, self.company_b, f"{index + 1}.00", at=at)
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
for event in self.ledger_events():
|
|
revision = self.current(event["id"])
|
|
confirm(self.connection, event["id"], revision["payer_company_id"],
|
|
"other_receivable", f"k-{event['id']}", self.admin)
|
|
seen: set[tuple[int, str]] = set()
|
|
cursor = None
|
|
while True:
|
|
page = positions.company_balances(
|
|
self.connection, from_="2026-01-01", cutoff="2026-07-31",
|
|
limit=1, cursor=cursor,
|
|
)
|
|
seen.update((item["company_id"], item["currency"]) for item in page["items"])
|
|
cursor = page["next_cursor"]
|
|
if not page["has_more"]:
|
|
break
|
|
self.assertEqual(
|
|
{(self.company_a, "CNY"), (self.company_b, "CNY")}, seen
|
|
)
|
|
|
|
def test_event_payload_account_chip_and_repayment_flag(self) -> None:
|
|
self.pair(
|
|
self.company_a, self.company_b, "40.00",
|
|
summary="归还往来款", purpose="还款",
|
|
)
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
event_id = int(self.ledger_events()[0]["id"])
|
|
detail = positions.event_detail(self.connection, event_id)
|
|
event = detail["event"]
|
|
self.assertEqual("visible", event["payer_account"]["visibility"])
|
|
self.assertEqual("测试 0001", event["payer_account"]["label"])
|
|
self.assertEqual("visible", event["payee_account"]["visibility"])
|
|
self.assertEqual("测试 0002", event["payee_account"]["label"])
|
|
self.assertEqual("归还往来款", event["summary"])
|
|
self.assertTrue(event["is_repayment"])
|
|
|
|
company_view = positions.event_payload(
|
|
self.connection,
|
|
self.connection.execute(
|
|
positions._DETAIL_SELECT + " WHERE p.ledger_event_id = ?",
|
|
(event_id,),
|
|
).fetchone(),
|
|
viewer_company_id=self.company_a,
|
|
)
|
|
self.assertEqual("visible", company_view["payer_account"]["visibility"])
|
|
self.assertEqual("masked", company_view["payee_account"]["visibility"])
|
|
self.assertEqual("按对方授权不可见", company_view["payee_account"]["label"])
|
|
|
|
def test_park_subject_exception_hides_from_review_queue(self) -> None:
|
|
self.pair(self.company_a, self.company_b, "100.00")
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
event_id = int(self.ledger_events()[0]["id"])
|
|
current = self.current(event_id)
|
|
subjects.park_subject(
|
|
self.connection, event_id,
|
|
disposition="exception", reason="转异常核查",
|
|
expected_revision=current["id"], request_key="park-exc-1",
|
|
actor=self.admin,
|
|
)
|
|
queue = positions.subject_review_queue(
|
|
self.connection, from_="2026-01-01", cutoff="2026-07-31",
|
|
)
|
|
self.assertFalse(
|
|
any(item["ledger_event_id"] == event_id for item in queue["items"])
|
|
)
|
|
revision = self.current(event_id)
|
|
self.assertEqual("pending_subject", revision["state"])
|
|
|
|
def test_park_subject_return_stays_in_review_queue(self) -> None:
|
|
self.pair(self.company_a, self.company_b, "80.00")
|
|
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
|
|
event_id = int(self.ledger_events()[0]["id"])
|
|
current = self.current(event_id)
|
|
subjects.park_subject(
|
|
self.connection, event_id,
|
|
disposition="return", reason="退回补充摘要",
|
|
expected_revision=current["id"], request_key="park-ret-1",
|
|
actor=self.admin,
|
|
)
|
|
queue = positions.subject_review_queue(
|
|
self.connection, from_="2026-01-01", cutoff="2026-07-31",
|
|
)
|
|
self.assertTrue(
|
|
any(item["ledger_event_id"] == event_id for item in queue["items"])
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|