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