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