"""Monthly close, reopen approval, snapshot hash and locked-period writes.""" from __future__ import annotations from datetime import date import hashlib import json from pathlib import Path import sys import unittest sys.path.insert(0, str(Path(__file__).resolve().parent)) from bank_importer import period_close from ledger_helpers import LedgerBase class PeriodCloseTests(LedgerBase): TODAY = date(2026, 8, 30) MONTH = "2026-07" def _cover_month(self) -> None: self.add_row( self.company_a, own_account="6222000000000001", expense="100.00", at="2026-07-10T10:00:00", ) self.add_row( self.company_b, own_account="6222000000000002", income="80.00", at="2026-07-12T10:00:00", ) def _close(self, month: str = MONTH): period_close.ensure_pending_tasks(self.connection, today=self.TODAY) return period_close.execute_close( self.connection, month, self.admin, confirm=True, ) def test_execute_requires_confirm_checkbox(self) -> None: self._cover_month() with self.assertRaises(period_close.PeriodCloseError) as ctx: period_close.execute_close( self.connection, self.MONTH, self.admin, confirm=False, ) self.assertIn("勾选", str(ctx.exception)) def test_blockers_reject_close(self) -> None: period_close.ensure_pending_tasks(self.connection, today=self.TODAY) evaluation = period_close.evaluate_preconditions(self.connection, self.MONTH) self.assertFalse(evaluation["ready"]) with self.assertRaises(period_close.PeriodCloseError): period_close.execute_close( self.connection, self.MONTH, self.admin, confirm=True, ) def test_close_snapshot_hash_stable_and_idempotent(self) -> None: self._cover_month() period_close.ensure_pending_tasks(self.connection, today=self.TODAY) before = period_close.build_snapshot(self.connection, self.MONTH) digest_before = hashlib.sha256( json.dumps(before, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") ).hexdigest() first = period_close.execute_close( self.connection, self.MONTH, self.admin, confirm=True, ) self.assertEqual("closed", first["status"]) self.assertTrue(first["report_no"].startswith("MR-202607-")) digest = first["snapshot_hash"] self.assertEqual(64, len(digest)) self.assertEqual(digest_before, digest) run = self.connection.execute( "SELECT snapshot_json, snapshot_hash FROM period_close_runs WHERE year_month = ?", (self.MONTH,), ).fetchone() self.assertEqual(digest, run["snapshot_hash"]) self.assertEqual( digest, hashlib.sha256(run["snapshot_json"].encode("utf-8")).hexdigest(), ) self.assertEqual( digest, hashlib.sha256(run["snapshot_json"].encode("utf-8")).hexdigest(), ) with self.assertRaises(period_close.PeriodConflictError): period_close.execute_close( self.connection, self.MONTH, self.admin, confirm=True, ) def test_locked_month_rejects_writes(self) -> None: self._cover_month() self._close() with self.assertRaises(period_close.PeriodLockedError) as ctx: period_close.assert_date_writable(self.connection, "2026-07-15") self.assertIn("2026-07", str(ctx.exception)) self.assertIn("已结账锁定", str(ctx.exception)) period_close.assert_date_writable(self.connection, "2026-08-01") def test_late_arrivals_do_not_rewrite_snapshot(self) -> None: self._cover_month() closed = self._close() digest = closed["snapshot_hash"] 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([], writable) self.assertEqual([(late_id, "2026-07")], locked) n = period_close.record_late_arrivals(self.connection, locked, self.admin) self.assertEqual(1, n) again = period_close.close_payload(self.connection, self.MONTH) self.assertEqual(digest, again["snapshot_hash"]) def test_snapshot_row_cannot_be_updated(self) -> None: self._cover_month() self._close() with self.assertRaises(Exception): with self.connection: self.connection.execute( "UPDATE period_close_runs SET snapshot_json = '{}' WHERE year_month = ?", (self.MONTH,), ) def test_reopen_reject_then_approve_and_reclose_version_chain(self) -> None: self._cover_month() first = self._close() with self.assertRaises(period_close.PeriodCloseError): period_close.request_reopen( self.connection, self.MONTH, self.admin, reason="太短", ) req = period_close.request_reopen( self.connection, self.MONTH, self.admin, reason="补录金牛煤业七月运输费并核对金额", companies_note="甲公司 ↔ 乙公司", window_days=3, ) self.assertEqual("pending", req["status"]) rejected = period_close.decide_reopen( self.connection, req["id"], self.admin, approve=False, comment="证据不足", ) self.assertEqual("rejected", rejected["status"]) self.assertTrue(period_close.is_month_locked(self.connection, self.MONTH)) req2 = period_close.request_reopen( self.connection, self.MONTH, self.admin, reason="已补齐银行回单，申请重开更正科目", ) approved = period_close.decide_reopen( self.connection, req2["id"], self.admin, approve=True, comment="同意", ) self.assertEqual("approved", approved["status"]) self.assertFalse(period_close.is_month_locked(self.connection, self.MONTH)) period_close.assert_date_writable(self.connection, "2026-07-15") second = period_close.execute_close( self.connection, self.MONTH, self.admin, confirm=True, ) self.assertEqual("closed", second["status"]) self.assertNotEqual(first["report_no"], second["report_no"]) versions = self.connection.execute( "SELECT version, report_no FROM period_close_runs WHERE year_month = ? ORDER BY version", (self.MONTH,), ).fetchall() self.assertGreaterEqual(len(versions), 2) self.assertEqual(1, versions[0]["version"]) self.assertEqual(2, versions[-1]["version"]) def test_wal_on_file_database(self) -> None: mode = self.connection.execute("PRAGMA journal_mode").fetchone()[0] self.assertEqual("wal", str(mode).lower()) if __name__ == "__main__": unittest.main()