HEL-270: 修复闭期补录未提交即回滚丢失(补回归测试)

Co-authored-by: multica-agent <github@multica.ai>
This commit is contained in:
总工
2026-08-30 22:29:52 +08:00
co-authored by multica-agent
parent e857a0e46d
commit 267ebd37f5
3 changed files with 66 additions and 34 deletions
+27
View File
@@ -110,6 +110,33 @@ class PeriodCloseTests(LedgerBase):
again = period_close.close_payload(self.connection, self.MONTH)
self.assertEqual(digest, again["snapshot_hash"])
def test_late_arrivals_survive_connection_close(self) -> None:
"""Regression: the reconcile handler closes its connection right after
recording locked rows; uncommitted inserts used to vanish silently."""
from bank_importer.db import connect as db_connect
self._cover_month()
self._close()
late_id = self.add_row(
self.company_a, own_account="6222000000000001",
expense="12.00", at="2026-07-28T11:00:00",
)
writable, locked = period_close.split_writable_row_ids(self.connection, [late_id])
self.assertEqual(1, period_close.record_late_arrivals(self.connection, locked, self.admin))
self.connection.close()
fresh = db_connect(self.db_path)
try:
rows = fresh.execute(
"SELECT source_row_id FROM period_late_arrivals"
).fetchall()
audits = fresh.execute(
"SELECT action FROM period_audit_events WHERE action = 'late_arrival'"
).fetchall()
finally:
fresh.close()
self.assertEqual([late_id], [r["source_row_id"] for r in rows])
self.assertEqual(1, len(audits))
def test_snapshot_row_cannot_be_updated(self) -> None:
self._cover_month()
self._close()