From f16e8aad34ce52362eeddfbe365202bb5c398596 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=85=BE=E8=AE=AFWorkBuddy?= Date: Wed, 19 Aug 2026 13:28:28 +0800 Subject: [PATCH] =?UTF-8?q?B-44:=20=E8=BF=94=E4=BF=AE=E2=80=94=E2=80=94?= =?UTF-8?q?=E5=A4=9A=E5=B8=81=E7=A7=8D=E5=AF=B9=E6=96=B9=E6=B1=87=E6=80=BB?= =?UTF-8?q?=E3=80=81=E9=95=9C=E5=83=8F=E7=A7=91=E7=9B=AE=E7=AD=9B=E9=80=89?= =?UTF-8?q?=E3=80=81=E5=86=B2=E9=94=80=E7=8B=AC=E7=AB=8B=E7=94=9F=E6=95=88?= =?UTF-8?q?=E6=97=A5=E3=80=81=E5=88=9B=E5=BB=BA=E6=9D=A5=E6=BA=90=E6=92=A4?= =?UTF-8?q?=E9=94=80=E8=BE=B9=E7=95=8C=E4=B8=8E=E5=B9=B6=E5=8F=91=E5=B9=82?= =?UTF-8?q?=E7=AD=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- server.py | 13 +- src/bank_importer/manual_records.py | 137 +++++++++------ src/bank_importer/positions.py | 9 +- src/bank_importer/subjects.py | 54 +++--- tests/test_manual_records.py | 259 ++++++++++++++++++++++++++++ tests/test_positions.py | 41 +++++ tests/test_positions_api.py | 106 +++++++++++- 7 files changed, 528 insertions(+), 91 deletions(-) diff --git a/server.py b/server.py index e05f88a..3ad02b8 100644 --- a/server.py +++ b/server.py @@ -2435,6 +2435,7 @@ class AppHandler(SimpleHTTPRequestHandler): actor=user, subject_code=data.get("subject_code"), target_ledger_event_id=data.get("target_ledger_event_id"), + effective_at=data.get("effective_at"), ) except manual_records.ManualConflictError as exc: self._send_json(409, {"status": "error", "message": str(exc)}) @@ -2495,7 +2496,9 @@ class AppHandler(SimpleHTTPRequestHandler): connection, from_=from_, cutoff=cutoff, currency=currency, company_id=company_id, ) - buckets: dict[int, dict[str, object]] = {} + # Buckets are keyed by ``(counterparty, currency)``: amounts never mix + # across currencies, so one counterparty with CNY and USD yields two rows. + buckets: dict[tuple[int, str], dict[str, object]] = {} for event in events: counterparty = ( event["payee_company_id"] @@ -2507,8 +2510,9 @@ class AppHandler(SimpleHTTPRequestHandler): if int(event["payer_company_id"]) == int(company_id) else event["payer_company_name"] ) + key = (int(counterparty), event["currency"]) bucket = buckets.setdefault( - counterparty, + key, { "counterparty_company_id": counterparty, "counterparty_company_name": name, @@ -2550,7 +2554,7 @@ class AppHandler(SimpleHTTPRequestHandler): else row["payer_company_name"] ) buckets.setdefault( - counterparty, + (int(counterparty), row["currency"]), { "counterparty_company_id": counterparty, "counterparty_company_name": name, @@ -2560,8 +2564,7 @@ class AppHandler(SimpleHTTPRequestHandler): }, ) items = [] - for counterparty, bucket in sorted(buckets.items()): - cur = bucket["currency"] + for (counterparty, cur), bucket in sorted(buckets.items()): signed = bucket["signed"] unresolved = positions.unresolved_for_company( connection, company_id, cutoff, currency=cur, diff --git a/src/bank_importer/manual_records.py b/src/bank_importer/manual_records.py index 975e19f..f8eaca4 100644 --- a/src/bank_importer/manual_records.py +++ b/src/bank_importer/manual_records.py @@ -10,7 +10,7 @@ is never edited. Idempotency keys and UNIQUE claims prevent double counting. from __future__ import annotations -from datetime import datetime +from datetime import datetime, timedelta, timezone from decimal import Decimal, InvalidOperation import json import sqlite3 @@ -60,6 +60,11 @@ def _validate_date(value: object, field: str) -> str: return text +def _business_today() -> str: + """Shanghai business date (the default reversal effective date).""" + return datetime.now(timezone(timedelta(hours=8))).date().isoformat() + + # --------------------------------------------------------------------------- # Submit # --------------------------------------------------------------------------- @@ -144,40 +149,56 @@ def submit( raise ManualInputError("supersedes_record_id 无效。") supersedes_id = int(supersedes_record_id) - existing = connection.execute( - "SELECT id FROM manual_records WHERE company_id = ? AND request_key = ?", - (int(company_id), request_key), - ).fetchone() - if existing is not None: - return _record_payload(connection, existing["id"], idempotent_replay=True) - now = utc_now() began = False if not connection.in_transaction: connection.execute("BEGIN IMMEDIATE") began = True try: - cursor = connection.execute( - """ - INSERT INTO manual_records ( - company_id, counterparty_company_id, occurred_at, direction, - amount, amount_scale, currency, funding_source, bank_account_id, - personal_transit_mapping_id, related_source_row_id, - requested_subject, summary, reason, evidence_json, request_key, - supersedes_record_id, submitted_by, created_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - """, - ( - int(company_id), int(counterparty_company_id), occurred_at, direction, - amount, _scale_of(amount), currency, funding_source, bank_account_id, - mapping_id, related_row["id"] if related_row is not None else None, - requested_subject, str(summary or "") or None, - str(reason or "") or None, - json.dumps(evidence, ensure_ascii=False) if evidence else None, - request_key, supersedes_id, - actor["id"], now, - ), - ) + # Re-check inside the write transaction: concurrent identical submits + # serialize here, so a replay is found before any INSERT. + existing = connection.execute( + "SELECT id FROM manual_records WHERE company_id = ? AND request_key = ?", + (int(company_id), request_key), + ).fetchone() + if existing is not None: + if began: + connection.commit() + return _record_payload(connection, existing["id"], idempotent_replay=True) + try: + cursor = connection.execute( + """ + INSERT INTO manual_records ( + company_id, counterparty_company_id, occurred_at, direction, + amount, amount_scale, currency, funding_source, bank_account_id, + personal_transit_mapping_id, related_source_row_id, + requested_subject, summary, reason, evidence_json, request_key, + supersedes_record_id, submitted_by, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + int(company_id), int(counterparty_company_id), occurred_at, direction, + amount, _scale_of(amount), currency, funding_source, bank_account_id, + mapping_id, related_row["id"] if related_row is not None else None, + requested_subject, str(summary or "") or None, + str(reason or "") or None, + json.dumps(evidence, ensure_ascii=False) if evidence else None, + request_key, supersedes_id, + actor["id"], now, + ), + ) + except sqlite3.IntegrityError: + # A concurrent identical submit won the race and committed first; + # surface the existing record idempotently instead of a UNIQUE 500. + if began: + connection.rollback() + existing = connection.execute( + "SELECT id FROM manual_records WHERE company_id = ? AND request_key = ?", + (int(company_id), request_key), + ).fetchone() + if existing is not None: + return _record_payload(connection, existing["id"], idempotent_replay=True) + raise record_id = int(cursor.lastrowid) _append_decision( connection, record_id, state="pending", action="submit", @@ -279,14 +300,16 @@ def decide( actor: sqlite3.Row, subject_code: object = None, target_ledger_event_id: object = None, + effective_at: object = None, ) -> dict[str, object]: """Apply an administrator decision to a manual record. ``approve_new`` creates a confirmed ledger event; ``approve_link`` joins an existing ledger event without adding a second economic impact; ``return`` and ``exception`` never produce a balance; ``reverse`` creates an opposite - reversal event (or detaches a linked claim). Replays return the earlier - outcome via ``idempotency_key``. + reversal event (or detaches a linked claim) with an independent business + effective date — explicit ``effective_at`` or the approval business day. + Replays return the earlier outcome via ``idempotency_key``. """ reason = (reason or "").strip() if not reason: @@ -350,7 +373,10 @@ def decide( else: # reverse if current["state"] != "approved": raise ManualConflictError("只有已批准记录可以冲销。") - outcome = _reverse(connection, record, current, actor, request_key, reason) + outcome = _reverse( + connection, record, current, actor, request_key, reason, + effective_at=effective_at, + ) _store_audit(connection, record, current, action, outcome, reason, actor) except Exception: @@ -455,6 +481,7 @@ def _reverse( actor: sqlite3.Row, request_key: str | None, reason: str, + effective_at: object = None, ) -> dict[str, object]: claim = manual_source_claim(connection, record["id"]) if claim is None: @@ -464,25 +491,30 @@ def _reverse( if revision is None: raise ManualConflictError("关联的往来事件没有当前修订。") - if revision["source_kind"] == "manual" and not _event_has_other_sources( - connection, event_id, record["id"] - ): - # The manual record is the only economic source: reverse it with a new - # opposite event at an independent effective date. + if _is_manual_creation_source(revision, record["id"]): + # This record created the event (approve_new): it added the economic + # impact, so reversing it must always produce an equal-amount reversal + # event — even when other manual evidence was later linked onto the + # same event. The original impact must not survive in balances. + if effective_at is not None and str(effective_at).strip(): + effective_at = _validate_date(effective_at, "冲销生效日") + else: + effective_at = _business_today() create_reversal( connection, event_id, source_kind="manual", source_revision_token=str(record["id"]), + effective_at=effective_at, reason=reason, actor=actor, idempotency_key=request_key, rule_version="manual-record-v1", ) else: - # The manual was linked evidence on a bank event: it never added a - # second impact, so reversing detaches the claim without a reversal - # event; the bank impact stays. + # The manual was linked evidence (approve_link) on an event it never + # created: it added no second impact, so reversing detaches the claim + # and the underlying economic impact stays. connection.execute( "DELETE FROM ledger_event_manual_sources WHERE manual_record_id = ?", (record["id"],), @@ -495,17 +527,20 @@ def _reverse( return {"decision_id": decision_id, "ledger_event_id": None} -def _event_has_other_sources(connection, event_id: int, manual_record_id: int) -> bool: - bank = connection.execute( - "SELECT 1 FROM ledger_event_bank_sources WHERE ledger_event_id = ? LIMIT 1", - (event_id,), - ).fetchone() - other_manual = connection.execute( - "SELECT 1 FROM ledger_event_manual_sources " - "WHERE ledger_event_id = ? AND manual_record_id != ? LIMIT 1", - (event_id, manual_record_id), - ).fetchone() - return bank is not None or other_manual is not None +def _is_manual_creation_source(revision: sqlite3.Row, manual_record_id: int) -> bool: + """True when ``manual_record_id`` created the event via ``approve_new``. + + The creation source is recorded in the event's immutable revision + ``evidence_json``; a linked evidence record is never the creation source + and carries no second economic impact. + """ + if revision["source_kind"] != "manual": + return False + try: + evidence = json.loads(revision["evidence_json"] or "{}") + except (TypeError, ValueError): + return False + return evidence.get("manual_record_id") == manual_record_id def _store_audit( diff --git a/src/bank_importer/positions.py b/src/bank_importer/positions.py index 1d216bb..3016a3e 100644 --- a/src/bank_importer/positions.py +++ b/src/bank_importer/positions.py @@ -133,10 +133,11 @@ def _event_filters( if subject: if subject not in SUBJECTS: raise PositionInputError("科目筛选无效。") - conditions.append( - "(p.subject_code = ? OR (p.subject_code = ? AND p.subject_code = ?))" - ) - params.extend([subject, MIRROR[subject], subject]) + # The stored subject lives on one company's perspective; the viewer on + # the other side sees its mirror. Match both so mirror events are never + # dropped from the filter. + conditions.append("(p.subject_code = ? OR p.subject_code = ?)") + params.extend([subject, MIRROR[subject]]) if posting_kind: conditions.append("p.posting_kind = ?") params.append(posting_kind) diff --git a/src/bank_importer/subjects.py b/src/bank_importer/subjects.py index 3f290d4..2312af8 100644 --- a/src/bank_importer/subjects.py +++ b/src/bank_importer/subjects.py @@ -209,36 +209,42 @@ def confirm_subject( raise SubjectInputError("必须填写科目确认依据。") if subject_code not in SUBJECTS: raise SubjectInputError("科目必须是应收/应付/其他应收/其他应付之一。") - subject_code = subject_code - - if request_key: - existing = connection.execute( - """ - SELECT * FROM ledger_event_revisions - WHERE ledger_event_id = ? AND idempotency_key = ? - ORDER BY id LIMIT 1 - """, - (ledger_event_id, request_key), - ).fetchone() - if existing is not None: - return _revision_payload(connection, existing) - - current = current_revision(connection, ledger_event_id) - if current is None: - raise SubjectConflictError("该事件不存在或没有当前修订。") - if current["state"] != "pending_subject": - raise SubjectConflictError("只有待确认科目的事件可以确认科目。") - if expected_revision is not None and int(expected_revision) != current["revision"]: - raise SubjectConflictError("事件已发生变更，请刷新后重试。") - participants = {current["payer_company_id"], current["payee_company_id"]} - if perspective_company_id not in participants: - raise SubjectInputError("视角公司必须是事件参与方。") began = False if not connection.in_transaction: connection.execute("BEGIN IMMEDIATE") began = True try: + if request_key: + existing = connection.execute( + """ + SELECT * FROM ledger_event_revisions + WHERE ledger_event_id = ? AND idempotency_key = ? + ORDER BY id LIMIT 1 + """, + (ledger_event_id, request_key), + ).fetchone() + if existing is not None: + if began: + connection.commit() + return _revision_payload(connection, existing) + + current = current_revision(connection, ledger_event_id) + if current is None: + raise SubjectConflictError("该事件不存在或没有当前修订。") + if current["state"] != "pending_subject": + raise SubjectConflictError("只有待确认科目的事件可以确认科目。") + # ``expected_revision`` may be the revision row id (what the API/UI + # sends as ``ledger_revision_id``) or the per-event sequence number; + # both identify the exact revision the client saw. + if expected_revision is not None and int(expected_revision) not in ( + current["id"], current["revision"], + ): + raise SubjectConflictError("事件已发生变更，请刷新后重试。") + participants = {current["payer_company_id"], current["payee_company_id"]} + if perspective_company_id not in participants: + raise SubjectInputError("视角公司必须是事件参与方。") + revision_id = append_revision( connection, ledger_event_id, diff --git a/tests/test_manual_records.py b/tests/test_manual_records.py index 425bd33..da69324 100644 --- a/tests/test_manual_records.py +++ b/tests/test_manual_records.py @@ -262,6 +262,146 @@ class DecisionTests(LedgerBase): request_key="dec-x", actor=self.admin, ) + def test_reverse_uses_explicit_effective_date_across_cutoff(self) -> None: + # Reversing with an independent effective date must keep the original + # impact for cutoffs before it and net it to zero only on/after it — + # never rewrite the historical period retroactively. + from bank_importer import positions + approved = self.decide("approve_new") + event_id = approved["ledger_event_id"] + self.assertEqual("2026-01-10", self.current(event_id)["effective_at"][:10]) + + manual_records.decide( + self.connection, self.record["id"], "reverse", + reason="误录冲销", expected_decision_id=approved["decision_id"], + request_key="dec-rev-date", actor=self.admin, + effective_at="2026-06-15", + ) + reversal = self.connection.execute( + "SELECT * FROM ledger_event_revisions WHERE posting_kind = 'reversal'" + ).fetchone() + self.assertEqual("2026-06-15", reversal["effective_at"][:10]) + + 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("-50.00", item["result"]["signed_amount"]) + for item in after["items"]: + if item["company_id"] == self.company_a: + self.assertEqual("0.00", item["result"]["signed_amount"]) + + def test_reverse_defaults_to_approval_business_day(self) -> None: + from datetime import datetime, timedelta, timezone + from bank_importer import positions + approved = self.decide("approve_new") + manual_records.decide( + self.connection, self.record["id"], "reverse", + reason="误录冲销", expected_decision_id=approved["decision_id"], + request_key="dec-rev-default", actor=self.admin, + ) + today = datetime.now(timezone(timedelta(hours=8))).date().isoformat() + reversal = self.connection.execute( + "SELECT * FROM ledger_event_revisions WHERE posting_kind = 'reversal'" + ).fetchone() + self.assertEqual(today, reversal["effective_at"][:10]) + + # The day before the approval business day keeps the original impact. + before = positions.company_balances( + self.connection, from_="2026-01-01", + cutoff=(datetime.fromisoformat(today) - timedelta(days=1)).date().isoformat(), + ) + for item in before["items"]: + if item["company_id"] == self.company_a: + self.assertEqual("-50.00", item["result"]["signed_amount"]) + + def test_reverse_of_creation_source_with_later_linked_evidence(self) -> None: + # approve_new created the event; a second manual record later links + # onto it. Reversing the creation source must still create an + # equal-amount reversal — the original economic impact must not survive + # just because other evidence was attached later. + approved = self.decide("approve_new") + event_id = approved["ledger_event_id"] + + linked = manual_records.submit( + self.connection, company_id=self.company_a, + counterparty_company_id=self.company_b, + occurred_at="2026-01-12T09:00:00", direction="incoming", + amount="50.00", currency="CNY", funding_source="other", + requested_subject="receivable", request_key="mr-link-ev", + actor=self.admin, + ) + manual_records.decide( + self.connection, linked["id"], "approve_link", + reason="同源补充证据", expected_decision_id=linked["decision_id"], + request_key="dec-link-ev", actor=self.admin, + target_ledger_event_id=event_id, + ) + self.assertEqual(1, len(self.position_events())) + + reversed_outcome = manual_records.decide( + self.connection, self.record["id"], "reverse", + reason="误录冲销", expected_decision_id=approved["decision_id"], + request_key="dec-rev-create", actor=self.admin, + ) + self.assertEqual("reversed", reversed_outcome["state"]) + rows = self.connection.execute( + "SELECT * FROM eligible_position_events ORDER BY ledger_event_id" + ).fetchall() + self.assertEqual(2, len(rows)) + original, reversal = rows + self.assertEqual("normal", original["posting_kind"]) + self.assertEqual("reversal", reversal["posting_kind"]) + self.assertEqual(Decimal(original["amount"]), Decimal(reversal["amount"])) + # The later linked evidence stays attached to the event. + links = self.connection.execute( + "SELECT * FROM ledger_event_manual_sources ORDER BY manual_record_id" + ).fetchall() + self.assertEqual(2, len(links)) + + def test_reverse_of_linked_record_after_creation_reversal_detaches_only(self) -> None: + # M1 creates the event, M2 links; after M1's reversal created the + # offset, reversing the linked M2 must only detach, never add a second + # reversal event. + approved = self.decide("approve_new") + event_id = approved["ledger_event_id"] + linked = manual_records.submit( + self.connection, company_id=self.company_a, + counterparty_company_id=self.company_b, + occurred_at="2026-01-12T09:00:00", direction="incoming", + amount="50.00", currency="CNY", funding_source="other", + requested_subject="receivable", request_key="mr-link-ev2", + actor=self.admin, + ) + linked_approved = manual_records.decide( + self.connection, linked["id"], "approve_link", + reason="同源补充证据", expected_decision_id=linked["decision_id"], + request_key="dec-link-ev2", actor=self.admin, + target_ledger_event_id=event_id, + ) + manual_records.decide( + self.connection, self.record["id"], "reverse", + reason="误录冲销", expected_decision_id=approved["decision_id"], + request_key="dec-rev-create2", actor=self.admin, + ) + manual_records.decide( + self.connection, linked["id"], "reverse", + reason="撤销关联证据", expected_decision_id=linked_approved["decision_id"], + request_key="dec-rev-link2", actor=self.admin, + ) + reversals = self.connection.execute( + "SELECT COUNT(*) AS n FROM ledger_event_revisions WHERE posting_kind = 'reversal'" + ).fetchone()["n"] + self.assertEqual(1, reversals) + links = self.connection.execute( + "SELECT * FROM ledger_event_manual_sources" + ).fetchall() + self.assertEqual(1, len(links)) + class DedupAndConcurrencyTests(LedgerBase): def test_duplicate_manual_record_pairing_candidate(self) -> None: @@ -343,6 +483,125 @@ class DedupAndConcurrencyTests(LedgerBase): ).fetchone()["n"] self.assertEqual(1, events) + def test_concurrent_submit_same_request_key_is_idempotent(self) -> None: + import sqlite3 + + results: list[dict] = [] + errors: list[Exception] = [] + + def worker(thread_key: str) -> None: + try: + connection = sqlite3.connect(self.db_path, timeout=30) + connection.row_factory = sqlite3.Row + try: + payload = manual_records.submit( + 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-conc-submit", actor=self.admin, + ) + results.append(payload) + finally: + connection.close() + except Exception as exc: # pragma: no cover + errors.append(exc) + + threads = [ + threading.Thread(target=worker, args=(f"t{i}",)) for i in range(4) + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + self.assertEqual([], errors) + rows = self.connection.execute( + "SELECT COUNT(*) AS n FROM manual_records" + ).fetchone()["n"] + self.assertEqual(1, rows) + self.assertEqual(1, len({payload["id"] for payload in results})) + decisions = self.connection.execute( + "SELECT COUNT(*) AS n FROM manual_record_decisions" + ).fetchone()["n"] + self.assertEqual(1, decisions) + + def test_concurrent_confirm_subject_same_key_creates_single_revision(self) -> None: + import sqlite3 + + self.pair(self.company_a, self.company_b, "100.00") + from bank_importer import ledger_events + ledger_events.reconcile_bank_events(self.connection, actor=self.admin) + event_id = self.ledger_events()[0]["id"] + + outcomes: list[dict] = [] + errors: list[Exception] = [] + + def worker(thread_key: str) -> None: + try: + connection = sqlite3.connect(self.db_path, timeout=30) + connection.row_factory = sqlite3.Row + try: + payload = subjects.confirm_subject( + connection, event_id, + perspective_company_id=self.company_a, + subject_code="other_receivable", reason="并发确认", + expected_revision=1, request_key="conc-subj-key", + actor=self.admin, + ) + outcomes.append(payload) + finally: + connection.close() + except Exception as exc: # pragma: no cover + errors.append(exc) + + threads = [ + threading.Thread(target=worker, args=(f"t{i}",)) for i in range(4) + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + self.assertEqual([], errors) + confirmed = self.connection.execute( + "SELECT COUNT(*) AS n FROM ledger_event_revisions WHERE state = 'confirmed'" + ).fetchone()["n"] + self.assertEqual(1, confirmed) + self.assertEqual(1, len({payload["revision_id"] for payload in outcomes})) + + def test_repeated_confirm_same_key_appends_no_noise_revision(self) -> None: + self.pair(self.company_a, self.company_b, "100.00") + from bank_importer import ledger_events + ledger_events.reconcile_bank_events(self.connection, actor=self.admin) + event_id = self.ledger_events()[0]["id"] + first = subjects.confirm_subject( + self.connection, event_id, + perspective_company_id=self.company_a, subject_code="other_receivable", + reason="确认", expected_revision=1, request_key="noise-key", + actor=self.admin, + ) + second = subjects.confirm_subject( + self.connection, event_id, + perspective_company_id=self.company_a, subject_code="other_receivable", + reason="重复确认", expected_revision=1, request_key="noise-key", + actor=self.admin, + ) + self.assertEqual(first["revision_id"], second["revision_id"]) + revisions = self.connection.execute( + "SELECT COUNT(*) AS n FROM ledger_event_revisions WHERE ledger_event_id = ?", + (event_id,), + ).fetchone()["n"] + # One pending_subject + one confirmed; the replay appended nothing. + self.assertEqual(2, revisions) + confirmed = self.connection.execute( + "SELECT COUNT(*) AS n FROM ledger_event_revisions " + "WHERE ledger_event_id = ? AND state = 'confirmed'", + (event_id,), + ).fetchone()["n"] + self.assertEqual(1, confirmed) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_positions.py b/tests/test_positions.py index 847b74e..54cbf56 100644 --- a/tests/test_positions.py +++ b/tests/test_positions.py @@ -267,6 +267,47 @@ class UnresolvedTests(LedgerBase): 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" diff --git a/tests/test_positions_api.py b/tests/test_positions_api.py index e872ffc..6ce2cfc 100644 --- a/tests/test_positions_api.py +++ b/tests/test_positions_api.py @@ -49,12 +49,14 @@ def workbook_bytes(rows) -> bytes: return buffer.getvalue() -def outgoing(own: str, cp: str, amount: str, at: str = "2026-01-05 10:00:00"): - return [own, "测试公司", at, amount, "", "50000.00", "RMB", "对方", cp, "某银行", "借款", ""] +def outgoing(own: str, cp: str, amount: str, at: str = "2026-01-05 10:00:00", + currency: str = "RMB"): + return [own, "测试公司", at, amount, "", "50000.00", currency, "对方", cp, "某银行", "借款", ""] -def incoming(own: str, cp: str, amount: str, at: str = "2026-01-05 11:00:00"): - return [own, "测试公司", at, "", amount, "50000.00", "RMB", "对方", cp, "某银行", "借款", ""] +def incoming(own: str, cp: str, amount: str, at: str = "2026-01-05 11:00:00", + currency: str = "RMB"): + return [own, "测试公司", at, "", amount, "50000.00", currency, "对方", cp, "某银行", "借款", ""] class IntercompanyApiTests(unittest.TestCase): @@ -167,15 +169,16 @@ class IntercompanyApiTests(unittest.TestCase): assert status == 200, data return batch_id - def _fresh_pair(self, amount: str = "100.00", at: str = "2026-01-05 10:00:00") -> dict: + def _fresh_pair(self, amount: str = "100.00", at: str = "2026-01-05 10:00:00", + currency: str = "RMB") -> dict: """Upload+confirm a new A<->B pair; returns the pending review item.""" self._upload_and_confirm( self.cashier_a, self.company_a, - [outgoing(ACCOUNT_A, ACCOUNT_B, amount, at)], + [outgoing(ACCOUNT_A, ACCOUNT_B, amount, at, currency)], ) self._upload_and_confirm( self.cashier_b, self.company_b, - [incoming(ACCOUNT_B, ACCOUNT_A, amount, at.replace("10:", "11:"))], + [incoming(ACCOUNT_B, ACCOUNT_A, amount, at.replace("10:", "11:"), currency)], ) status, _, data = self.admin.get( "/api/admin/subject-reviews?from=2026-01-01&cutoff=2026-12-31" @@ -458,6 +461,95 @@ class IntercompanyApiTests(unittest.TestCase): ) self.assertEqual(404, status, data) + def test_company_counterparty_summary_splits_by_currency(self) -> None: + # A->B 100 CNY and A->B 50 USD must render two rows, never a mixed + # "CNY 150.00" bucket. + cny = self._fresh_pair("100.00", currency="CNY") + usd = self._fresh_pair("50.00", at="2026-01-20 10:00:00", currency="USD") + for pending in (cny, usd): + self._confirm(pending["ledger_event_id"]) + + status, _, data = self.cashier_a.get( + "/api/company/intercompany/balances?from=2026-01-01&cutoff=2026-12-31" + ) + self.assertEqual(200, status, data) + counterparties = as_json(data)["counterparties"] + rows = [row for row in counterparties if row["counterparty_company_id"] == self.company_b] + self.assertEqual(2, len(rows), counterparties) + by_currency = {row["currency"]: row for row in rows} + self.assertEqual({"CNY", "USD"}, set(by_currency)) + self.assertEqual("100.00", by_currency["CNY"]["result"]["signed_amount"]) + self.assertEqual("50.00", by_currency["USD"]["result"]["signed_amount"]) + for row in rows: + self.assertNotEqual("150.00", row["result"]["signed_amount"]) + self.assertEqual(1, row["event_count"]) + + def test_company_events_subject_filter_matches_mirror(self) -> None: + # A confirms subject "receivable" (stored from A's perspective); B must + # still find the event when filtering by the mirror subject "payable". + pending = self._fresh_pair("100.00") + status, _, data = self.admin.post_json( + f"/api/admin/intercompany/events/{pending['ledger_event_id']}/subject-decisions", + { + "perspective_company_id": self.company_a, + "subject_code": "receivable", "reason": "确认应收", + "expected_revision": 1, "request_key": "mirror-subj", + }, + ) + self.assertEqual(200, status, data) + + status, _, data = self.cashier_b.get( + "/api/company/intercompany/events?from=2026-01-01&cutoff=2026-12-31&subject=payable" + ) + self.assertEqual(200, status, data) + items = as_json(data)["items"] + self.assertTrue(any( + item["ledger_event_id"] == pending["ledger_event_id"] for item in items + ), data) + + def test_manual_reverse_via_api_uses_explicit_effective_date(self) -> None: + status, _, data = self.cashier_a.post_json( + "/api/company/manual-records", + { + "counterparty_company_id": self.company_b, + "occurred_at": "2026-02-01T09:00:00", + "direction": "incoming", "amount": "20.00", "currency": "CNY", + "funding_source": "other", "requested_subject": "other_receivable", + "request_key": "mr-rev-api", "summary": "还款", + }, + ) + self.assertEqual(200, status, data) + record = as_json(data)["record"] + status, _, data = self.admin.post_json( + f"/api/admin/manual-records/{record['id']}/decisions", + {"action": "approve_new", "reason": "确认入账", + "expected_decision_id": record["decision_id"], "request_key": "dec-rev-api"}, + ) + self.assertEqual(200, status, data) + event_id = as_json(data)["decision"]["ledger_event_id"] + + status, _, data = self.admin.get( + f"/api/admin/intercompany/events/{event_id}" + ) + self.assertEqual(200, status, data) + revision_id = as_json(data)["event"]["ledger_revision_id"] + + status, _, data = self.admin.post_json( + f"/api/admin/manual-records/{record['id']}/decisions", + {"action": "reverse", "reason": "误录冲销", "effective_at": "2026-06-15", + "expected_decision_id": None, "request_key": "dec-rev-api-2"}, + ) + self.assertEqual(200, status, data) + + status, _, data = self.admin.get( + "/api/admin/intercompany/events?from=2026-01-01&cutoff=2026-12-31" + ) + self.assertEqual(200, status, data) + reversals = [item for item in as_json(data)["items"] if item["posting_kind"] == "reversal"] + self.assertEqual(1, len(reversals)) + self.assertEqual("2026-06-15", reversals[0]["effective_at"][:10]) + self.assertEqual(event_id, reversals[0]["reverses_ledger_event_id"]) + if __name__ == "__main__": unittest.main()