"""Tests for automatic reminder detection, delivery, isolation and audit trail.""" from __future__ import annotations from datetime import date, timedelta from pathlib import Path import json import sqlite3 import unittest from unittest.mock import patch from bank_importer import auth, reminders from bank_importer.db import connect, migrate, utc_now class ReminderTestCase(unittest.TestCase): def setUp(self) -> None: self.connection = connect(":memory:") self.addCleanup(self.connection.close) migrate(self.connection) now = utc_now() with self.connection: self.connection.execute( "INSERT INTO companies (name, created_at, updated_at) VALUES ('甲公司', ?, ?)", (now, now), ) self.connection.execute( "INSERT INTO companies (name, created_at, updated_at) VALUES ('乙公司', ?, ?)", (now, now), ) self.company_a = int( self.connection.execute("SELECT id FROM companies WHERE name = '甲公司'").fetchone()["id"] ) self.company_b = int( self.connection.execute("SELECT id FROM companies WHERE name = '乙公司'").fetchone()["id"] ) self.admin_id = auth.create_user(self.connection, "admin1", "AdminPass123", "admin") self.user_a = auth.create_user( self.connection, "cashier-a", "CashierA123", "company", company_id=self.company_a ) self.user_b = auth.create_user( self.connection, "cashier-b", "CashierB123", "company", company_id=self.company_b ) self.admin = self.connection.execute( "SELECT * FROM users WHERE id = ?", (self.admin_id,) ).fetchone() self.actor_a = self.connection.execute( "SELECT * FROM users WHERE id = ?", (self.user_a,) ).fetchone() def _insert_batch( self, company_id: int, *, period_start: str, period_end: str, ) -> None: now = utc_now() with self.connection: self.connection.execute( """ INSERT INTO source_files (sha256, original_filename, size_bytes, storage_path, created_at) VALUES (?, 'f.xls', 1, '/tmp/f.xls', ?) """, (f"sha-{company_id}-{period_end}", now), ) file_id = int(self.connection.execute("SELECT last_insert_rowid()").fetchone()[0]) self.connection.execute( """ INSERT INTO import_batches (source_file_id, status, company_id, created_at, updated_at) VALUES (?, 'parsed', ?, ?, ?) """, (file_id, company_id, now, now), ) batch_id = int(self.connection.execute("SELECT last_insert_rowid()").fetchone()[0]) self.connection.execute( """ INSERT INTO sheet_batches ( import_batch_id, sheet_name, bank_name, template_id, template_version, header_row, period_start, period_end, transaction_count, created_at ) VALUES (?, 's1', '工行', 'tpl', 1, 1, ?, ?, 1, ?) """, (batch_id, period_start, period_end, now), ) def _insert_pending_account(self, company_id: int) -> None: now = utc_now() with self.connection: self.connection.execute( """ INSERT INTO bank_accounts ( company_id, account_number, bank_name, status, created_at, updated_at ) VALUES (?, ?, '工行', 'pending', ?, ?) """, (company_id, f"622{company_id:012d}", now, now), ) class RuleDetectionTests(ReminderTestCase): def test_unsubmitted_detected_when_no_current_month_batch(self) -> None: today = date.today() if today.day < 5: self.skipTest("monthly start day gate not reached today") findings = reminders.scan_findings(self.connection) keys = {item.company_id: item for item in findings if item.rule_key == reminders.RULE_UNSUBMITTED} self.assertIn(self.company_a, keys) self.assertIn(self.company_b, keys) def test_unsubmitted_not_reported_when_batch_exists(self) -> None: today = date.today() if today.day < 5: self.skipTest("monthly start day gate not reached today") period = f"{today.year:04d}-{today.month:02d}" self._insert_batch(self.company_a, period_start=f"{period}-01", period_end=f"{period}-15") findings = reminders.scan_findings(self.connection) for item in findings: if item.rule_key == reminders.RULE_UNSUBMITTED: self.assertNotEqual(self.company_a, item.company_id) def test_gap_detected_after_threshold(self) -> None: old_end = (date.today() - timedelta(days=10)).isoformat() self._insert_batch(self.company_a, period_start="2026-01-01", period_end=old_end) findings = reminders.scan_findings(self.connection) gap = [item for item in findings if item.rule_key == reminders.RULE_GAP and item.company_id == self.company_a] self.assertEqual(1, len(gap)) def test_gap_not_reported_for_recent_batch(self) -> None: recent = (date.today() - timedelta(days=1)).isoformat() self._insert_batch(self.company_a, period_start="2026-07-01", period_end=recent) findings = reminders.scan_findings(self.connection) gap = [item for item in findings if item.rule_key == reminders.RULE_GAP and item.company_id == self.company_a] self.assertEqual(0, len(gap)) def test_pending_review_detected_with_pending_account(self) -> None: self._insert_pending_account(self.company_a) findings = reminders.scan_findings(self.connection) pending = [ item for item in findings if item.rule_key == reminders.RULE_PENDING and item.company_id == self.company_a ] self.assertEqual(1, len(pending)) self.assertGreater(pending[0].rule_params["pending_count"], 0) def test_pending_review_not_reported_when_clean(self) -> None: findings = reminders.scan_findings(self.connection) pending = [ item for item in findings if item.rule_key == reminders.RULE_PENDING and item.company_id == self.company_a ] self.assertEqual(0, len(pending)) class DeliveryAndDedupTests(ReminderTestCase): def test_send_creates_reminder_and_event(self) -> None: today = date.today() if today.day < 5: self.skipTest("monthly start day gate not reached today") findings = reminders.scan_findings(self.connection) self.assertTrue(findings) key = findings[0].dedupe_key reminder_id = reminders.deliver_finding(self.connection, key, actor=self.admin) self.assertIsNotNone(reminder_id) row = self.connection.execute( "SELECT send_count, status FROM reminders WHERE id = ?", (reminder_id,) ).fetchone() self.assertEqual(1, row["send_count"]) self.assertEqual("open", row["status"]) events = self.connection.execute( "SELECT event_type FROM reminder_events WHERE reminder_id = ?", (reminder_id,) ).fetchall() self.assertEqual(["sent"], [item["event_type"] for item in events]) def test_repeat_send_increments_count_without_new_row(self) -> None: today = date.today() if today.day < 5: self.skipTest("monthly start day gate not reached today") findings = reminders.scan_findings(self.connection) key = findings[0].dedupe_key first = reminders.deliver_finding(self.connection, key, actor=self.admin) second = reminders.deliver_finding(self.connection, key, actor=self.admin) self.assertEqual(first, second) row = self.connection.execute( "SELECT send_count FROM reminders WHERE dedupe_key = ?", (key,) ).fetchone() self.assertEqual(2, row["send_count"]) count = self.connection.execute("SELECT COUNT(*) FROM reminders WHERE dedupe_key = ?", (key,)).fetchone()[0] self.assertEqual(1, count) events = self.connection.execute( "SELECT COUNT(*) FROM reminder_events WHERE reminder_id = ?", (first,) ).fetchone()[0] self.assertEqual(2, events) def test_resolved_reminder_reopens_same_row(self) -> None: self._insert_pending_account(self.company_a) finding = next( item for item in reminders.scan_findings(self.connection) if item.rule_key == reminders.RULE_PENDING and item.company_id == self.company_a ) first = reminders.deliver_finding(self.connection, finding.dedupe_key, actor=self.admin) self.assertIsNotNone(first) reminders.update_reminder_status( self.connection, first, "resolved", company_id=self.company_a, actor=self.actor_a ) second = reminders.deliver_finding(self.connection, finding.dedupe_key, actor=self.admin) self.assertEqual(first, second) row = self.connection.execute( "SELECT send_count, status, dedupe_key FROM reminders WHERE id = ?", (first,), ).fetchone() self.assertEqual(2, row["send_count"]) self.assertEqual("open", row["status"]) count = self.connection.execute( "SELECT COUNT(*) FROM reminders WHERE dedupe_key = ?", (finding.dedupe_key,) ).fetchone()[0] self.assertEqual(1, count) events = [ item["event_type"] for item in self.connection.execute( "SELECT event_type FROM reminder_events WHERE reminder_id = ? ORDER BY id", (first,), ).fetchall() ] self.assertEqual(["sent", "resolved", "sent"], events) def test_deliver_many_continues_after_one_failure(self) -> None: self._insert_pending_account(self.company_a) finding = next( item for item in reminders.scan_findings(self.connection) if item.rule_key == reminders.RULE_PENDING and item.company_id == self.company_a ) original = reminders.deliver_finding def flaky(connection, key, **kwargs): if key == "boom": raise sqlite3.IntegrityError("UNIQUE constraint failed: reminders.dedupe_key") return original(connection, key, **kwargs) with patch.object(reminders, "deliver_finding", side_effect=flaky): sent = reminders.deliver_many( self.connection, ["boom", finding.dedupe_key], actor=self.admin ) self.assertEqual(1, len(sent)) self.assertEqual( 1, self.connection.execute("SELECT COUNT(*) FROM reminders").fetchone()[0], ) def test_manual_reminder_each_send_is_separate(self) -> None: first = reminders.send_manual( self.connection, company_id=self.company_a, display_type="其他", content="请尽快处理", deadline="2026-09-01", actor=self.admin, ) second = reminders.send_manual( self.connection, company_id=self.company_a, display_type="其他", content="再次提醒", deadline=None, actor=self.admin, ) self.assertNotEqual(first, second) class IsolationTests(ReminderTestCase): def setUp(self) -> None: super().setUp() reminders.send_manual( self.connection, company_id=self.company_a, display_type="流水未提交", content="甲公司专属", deadline=None, actor=self.admin, ) reminders.send_manual( self.connection, company_id=self.company_b, display_type="流水未提交", content="乙公司专属", deadline=None, actor=self.admin, ) def test_company_sees_only_own_reminders(self) -> None: items_a = reminders.list_company_reminders(self.connection, self.company_a) items_b = reminders.list_company_reminders(self.connection, self.company_b) self.assertEqual(1, len(items_a)) self.assertEqual(1, len(items_b)) self.assertIn("甲公司", items_a[0]["title"]) self.assertIn("乙公司", items_b[0]["title"]) def test_company_cannot_update_other_company_reminder(self) -> None: other_id = reminders.list_company_reminders(self.connection, self.company_b)[0]["id"] ok = reminders.update_reminder_status( self.connection, other_id, "acknowledged", company_id=self.company_a, actor=self.actor_a ) self.assertFalse(ok) class AuditTrailTests(ReminderTestCase): def test_scan_writes_audit_log(self) -> None: reminders.run_scan(self.connection, actor=self.admin, ip="127.0.0.1") row = self.connection.execute( "SELECT action, detail FROM audit_log WHERE action = 'reminder_scan'" ).fetchone() self.assertIsNotNone(row) detail = json.loads(row["detail"]) self.assertIn("total", detail) def test_events_are_append_only(self) -> None: reminder_id = reminders.send_manual( self.connection, company_id=self.company_a, display_type="测试", content="内容", deadline=None, actor=self.admin, ) with self.assertRaises(Exception): with self.connection: self.connection.execute( "UPDATE reminder_events SET detail = 'tampered' WHERE reminder_id = ?", (reminder_id,), ) with self.assertRaises(Exception): with self.connection: self.connection.execute( "DELETE FROM reminder_events WHERE reminder_id = ?", (reminder_id,) ) def test_reminders_cannot_be_deleted(self) -> None: reminder_id = reminders.send_manual( self.connection, company_id=self.company_a, display_type="测试", content="内容", deadline=None, actor=self.admin, ) with self.assertRaises(Exception): with self.connection: self.connection.execute("DELETE FROM reminders WHERE id = ?", (reminder_id,)) class StatusFlowTests(ReminderTestCase): def test_acknowledge_and_resolve(self) -> None: reminder_id = reminders.send_manual( self.connection, company_id=self.company_a, display_type="待确认", content="请处理", deadline=None, actor=self.admin, ) ok = reminders.update_reminder_status( self.connection, reminder_id, "acknowledged", company_id=self.company_a, actor=self.actor_a ) self.assertTrue(ok) row = self.connection.execute( "SELECT status FROM reminders WHERE id = ?", (reminder_id,) ).fetchone() self.assertEqual("acknowledged", row["status"]) ok = reminders.update_reminder_status( self.connection, reminder_id, "resolved", company_id=self.company_a, actor=self.actor_a ) self.assertTrue(ok) events = self.connection.execute( "SELECT event_type FROM reminder_events WHERE reminder_id = ? ORDER BY id", (reminder_id,), ).fetchall() self.assertEqual( ["sent", "acknowledged", "resolved"], [item["event_type"] for item in events], ) class SettingsTests(ReminderTestCase): def test_default_settings_and_update(self) -> None: settings = reminders.get_settings(self.connection) self.assertEqual("5", settings["monthly_start_day"]) self.assertEqual("5", settings["gap_days"]) updated = reminders.update_settings( self.connection, {"gap_days": "7", "monthly_start_day": "6"} ) self.assertEqual("7", updated["gap_days"]) self.assertEqual("6", updated["monthly_start_day"]) def test_unknown_setting_rejected(self) -> None: with self.assertRaises(ValueError): reminders.update_settings(self.connection, {"unknown_key": "1"}) def test_invalid_monthly_start_day_rejected(self) -> None: with self.assertRaises(ValueError): reminders.update_settings(self.connection, {"monthly_start_day": "0"}) with self.assertRaises(ValueError): reminders.update_settings(self.connection, {"monthly_start_day": "29"}) self.assertEqual("5", reminders.get_settings(self.connection)["monthly_start_day"]) def test_invalid_gap_days_rejected(self) -> None: with self.assertRaises(ValueError): reminders.update_settings(self.connection, {"gap_days": "0"}) with self.assertRaises(ValueError): reminders.update_settings(self.connection, {"gap_days": "abc"}) self.assertEqual("5", reminders.get_settings(self.connection)["gap_days"]) def test_invalid_scan_time_rejected(self) -> None: with self.assertRaises(ValueError): reminders.update_settings(self.connection, {"scan_time": "25:00"}) with self.assertRaises(ValueError): reminders.update_settings(self.connection, {"scan_time": "8:0"}) self.assertEqual("08:00", reminders.get_settings(self.connection)["scan_time"]) def test_corrupt_settings_do_not_break_scan(self) -> None: with self.connection: self.connection.execute( "UPDATE reminder_settings SET value = 'not-a-number' WHERE key = 'monthly_start_day'" ) self.connection.execute( "UPDATE reminder_settings SET value = '0' WHERE key = 'gap_days'" ) findings = reminders.scan_findings(self.connection) self.assertIsInstance(findings, list) class NoticeListDelegationTests(unittest.TestCase): def test_async_notice_go_handle_uses_event_delegation(self) -> None: source = (Path(__file__).resolve().parents[1] / "web" / "app.js").read_text( encoding="utf-8" ) init_body = source.split("function initNotifications()", 1)[1].split("\nfunction ", 1)[0] self.assertIn('list.addEventListener("click"', init_body) self.assertIn('closest("[data-view-link]")', init_body) self.assertIn("showView(viewLink.dataset.viewLink)", init_body) self.assertIn('document.addEventListener("click"', source) self.assertNotIn( '$$("[data-view-link]").forEach((button) => button.addEventListener("click"', source, ) def test_delegated_lookup_finds_button_inserted_after_init(self) -> None: """Simulate #notice-list after async replaceChildren: click target is the new button.""" list_root = {"id": "notice-list", "parent": None, "attrs": {}} side = {"id": "lr-side", "parent": list_root, "attrs": {}} button = { "id": "go", "parent": side, "attrs": {"data-view-link": "reconcile"}, } list_root["children"] = [side] side["children"] = [button] def closest(node, attr): current = node while current is not None: if attr in current.get("attrs", {}): return current current = current.get("parent") return None clicked = closest(button, "data-view-link") self.assertIsNotNone(clicked) self.assertEqual("reconcile", clicked["attrs"]["data-view-link"]) self.assertIs(list_root, clicked["parent"]["parent"]) class MigrationTests(unittest.TestCase): def test_v6_migration_applies_and_rolls_back(self) -> None: connection = connect(":memory:") self.addCleanup(connection.close) migrate(connection) versions = connection.execute( "SELECT version FROM schema_migrations ORDER BY version" ).fetchall() self.assertEqual(10, versions[-1]["version"]) connection.execute("DELETE FROM schema_migrations WHERE version = 10") connection.executescript( """ DROP TRIGGER IF EXISTS reminders_no_delete; DROP TRIGGER IF EXISTS reminder_events_no_delete; DROP TRIGGER IF EXISTS reminder_events_no_update; DROP TABLE IF EXISTS reminder_events; DROP TABLE IF EXISTS reminders; DROP TABLE IF EXISTS reminder_settings; """ ) row = connection.execute( "SELECT name FROM sqlite_master WHERE name = 'reminders'" ).fetchone() self.assertIsNone(row)