"""Integration tests for system settings persistence and reminder flow.""" from __future__ import annotations from http.client import HTTPConnection from http.cookies import SimpleCookie import json import os from pathlib import Path import tempfile import threading import unittest from bank_importer.db import connect, migrate from bank_importer import auth, settings import server BOOTSTRAP_PASSWORD = "BootAdmin123" ADMIN_PASSWORD = "AdminPass123" class Client: def __init__(self, host: str, port: int) -> None: self.host = host self.port = port self.cookies: dict[str, str] = {} def request(self, method, path, body=None, headers=None): connection = HTTPConnection(self.host, self.port) request_headers = dict(headers or {}) if self.cookies: request_headers["Cookie"] = "; ".join( f"{k}={v}" for k, v in self.cookies.items() ) connection.request(method, path, body=body, headers=request_headers) response = connection.getresponse() data = response.read() set_cookie = dict(response.getheaders()).get("Set-Cookie") if set_cookie: cookie = SimpleCookie() cookie.load(set_cookie) for key, morsel in cookie.items(): if morsel.value: self.cookies[key] = morsel.value else: self.cookies.pop(key, None) status = response.status connection.close() return status, data def get(self, path): return self.request("GET", path) def post_json(self, path, payload): return self.request( "POST", path, body=json.dumps(payload).encode("utf-8"), headers={"Content-Type": "application/json"}, ) def as_json(data: bytes): return json.loads(data.decode("utf-8")) class SettingsAndRemindersTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.temp_dir = tempfile.TemporaryDirectory() root = Path(cls.temp_dir.name) cls.db_path = root / "app.db" cls.storage = root / "files" cls._old_db_path = server.DB_PATH cls._old_storage = server.STORAGE_DIR server.DB_PATH = cls.db_path server.STORAGE_DIR = cls.storage os.environ["APP_BOOTSTRAP_ADMIN_PASSWORD"] = BOOTSTRAP_PASSWORD connection = connect(cls.db_path) migrate(connection) server.ensure_bootstrap_admin(connection) # One active account for company 1 so pending items are generated. with connection: connection.execute( "INSERT INTO companies (name, credit_code, status, created_at, updated_at) " "VALUES ('甲公司', NULL, 'active', 't', 't')" ) company_id = connection.execute( "SELECT id FROM companies WHERE name = '甲公司'" ).fetchone()["id"] connection.execute( "INSERT INTO bank_accounts (company_id, account_number, bank_name, status, created_at, updated_at) " "VALUES (?, '11112222', '工行', 'active', 't', 't')", (company_id,), ) cls.company_id = company_id connection.close() class QuietHandler(server.AppHandler): def log_message(self, *args) -> None: pass cls.httpd = server.ThreadingHTTPServer(("127.0.0.1", 0), QuietHandler) cls.port = cls.httpd.server_address[1] cls.thread = threading.Thread(target=cls.httpd.serve_forever, daemon=True) cls.thread.start() cls.admin = Client("127.0.0.1", cls.port) status, data = cls.admin.post_json( "/api/login", {"username": "group-admin", "password": BOOTSTRAP_PASSWORD, "portal": "admin"}, ) assert status == 200, data status, data = cls.admin.post_json( "/api/password/change", {"old_password": BOOTSTRAP_PASSWORD, "new_password": ADMIN_PASSWORD}, ) assert status == 200, data @classmethod def tearDownClass(cls) -> None: cls.httpd.shutdown() cls.httpd.server_close() server.DB_PATH = cls._old_db_path server.STORAGE_DIR = cls._old_storage os.environ.pop("APP_BOOTSTRAP_ADMIN_PASSWORD", None) cls.temp_dir.cleanup() def test_settings_roundtrip_persists(self) -> None: status, data = self.admin.post_json( "/api/admin/settings", {"closing_day": "1", "start_date": "2026-01-05", "auto_remind": "0", "remind_days": "5"}, ) self.assertEqual(200, status, data) updated = as_json(data)["settings"] self.assertEqual("1", updated["closing_day"]) self.assertEqual("2026-01-05", updated["start_date"]) self.assertEqual("0", updated["auto_remind"]) self.assertEqual("5", updated["remind_days"]) status, data = self.admin.get("/api/admin/settings") self.assertEqual(200, status) self.assertEqual("1", as_json(data)["settings"]["closing_day"]) def test_invalid_closing_day_rejected(self) -> None: for bad in ("0", "29", "abc"): status, data = self.admin.post_json("/api/admin/settings", {"closing_day": bad}) self.assertEqual(400, status, (bad, data)) def test_setting_change_is_audited(self) -> None: status, data = self.admin.post_json("/api/admin/settings", {"closing_day": "7"}) self.assertEqual(200, status) connection = connect(self.db_path) try: rows = connection.execute( "SELECT key, before_value, after_value, actor_username " "FROM system_setting_changes WHERE key = 'closing_day' " "ORDER BY id DESC LIMIT 1" ).fetchall() finally: connection.close() self.assertTrue(rows) self.assertEqual("7", rows[0]["after_value"]) self.assertNotEqual(rows[0]["before_value"], rows[0]["after_value"]) self.assertEqual("group-admin", rows[0]["actor_username"]) def test_reminder_pending_and_send(self) -> None: status, data = self.admin.get( f"/api/admin/reminders/pending?company_id={self.company_id}" ) self.assertEqual(200, status, data) items = as_json(data)["items"] self.assertTrue(items) status, data = self.admin.post_json( "/api/admin/reminders/send", {"company_id": self.company_id} ) self.assertEqual(200, status, data) payload = as_json(data) self.assertGreaterEqual(len(payload["reminders"]), 1) self.assertTrue(payload["deadline"]) status, data = self.admin.get("/api/admin/reminders") self.assertEqual(200, status) history = as_json(data)["reminders"] self.assertGreaterEqual(len(history), 1) self.assertEqual(self.company_id, history[0]["company_id"]) def test_company_user_forbidden_on_settings_and_reminders(self) -> None: # A company user must not be able to read or write admin settings. status, data = self.admin.post_json( "/api/admin/users", {"username": "cashier-x", "company_id": self.company_id} ) self.assertEqual(200, status, data) initial = as_json(data)["initial_password"] cashier = Client("127.0.0.1", self.port) status, data = cashier.post_json( "/api/login", {"username": "cashier-x", "password": initial, "portal": "company"}, ) self.assertEqual(200, status) status, data = cashier.post_json( "/api/password/change", {"old_password": initial, "new_password": "Changed123"}, ) self.assertEqual(200, status) for path in ("/api/admin/settings", "/api/admin/reminders"): status, _ = cashier.get(path) self.assertEqual(403, status, path) # Company session must get 403 on admin write endpoints (settings / # reminders / review decisions), not just GETs. write_calls = ( ("POST /api/admin/settings", lambda: cashier.post_json( "/api/admin/settings", {"closing_day": "10", "start_date": "2026-01-01", "auto_remind": "0", "remind_days": "3"}, )), ("POST /api/admin/reminders/send", lambda: cashier.post_json( "/api/admin/reminders/send", {"company_id": self.company_id} )), ("POST /api/admin/accounts/1/review", lambda: cashier.post_json( "/api/admin/accounts/1/review", {"decision": "approve"} )), ("POST /api/admin/transfer-events/1/decisions", lambda: cashier.post_json( "/api/admin/transfer-events/1/decisions", {"action": "confirm_single", "reason": "越权"}, )), ("POST /api/admin/manual-records/1/decisions", lambda: cashier.post_json( "/api/admin/manual-records/1/decisions", {"action": "approve_new", "reason": "越权"}, )), ("POST /api/admin/intercompany/events/1/subject-decisions", lambda: cashier.post_json( "/api/admin/intercompany/events/1/subject-decisions", {"subject": "other_receivable", "reason": "越权"}, )), ("POST /api/admin/intercompany/events/1/adjustments", lambda: cashier.post_json( "/api/admin/intercompany/events/1/adjustments", {"action": "reverse", "reason": "越权"}, )), ) for label, call in write_calls: status, data = call() self.assertEqual(403, status, f"{label}: {data}") # Settings must be unchanged after the forbidden write attempt. status, data = self.admin.get("/api/admin/settings") self.assertEqual(200, status) self.assertEqual("5", as_json(data)["settings"]["closing_day"]) class SettingsModuleTests(unittest.TestCase): """Unit tests for the settings module on a fresh in-memory database.""" def setUp(self) -> None: self.connection = connect(":memory:") self.addCleanup(self.connection.close) migrate(self.connection) def test_defaults_applied_when_no_row_exists(self) -> None: values = settings.get_settings(self.connection) self.assertEqual("5", values["closing_day"]) self.assertEqual("2026-01-01", values["start_date"]) self.assertEqual("1", values["auto_remind"]) self.assertEqual("3", values["remind_days"]) def test_validate_rejects_bad_values(self) -> None: for bad in ({"closing_day": "0"}, {"closing_day": "29"}, {"closing_day": "abc"}): _, error = settings.validate_settings(bad) self.assertIsNotNone(error) _, error = settings.validate_settings({"start_date": "2026-13-40"}) self.assertIsNotNone(error) _, error = settings.validate_settings({"auto_remind": "2"}) self.assertIsNotNone(error) def test_update_writes_value_and_trail(self) -> None: user_id = auth.create_user(self.connection, "group-admin", "AdminPass123", "admin") user = self.connection.execute( "SELECT id, username FROM users WHERE id = ?", (user_id,) ).fetchone() updated = settings.update_settings( self.connection, {"closing_day": "1", "auto_remind": "0"}, user ) self.assertEqual("1", updated["closing_day"]) self.assertEqual("0", updated["auto_remind"]) changes = self.connection.execute( "SELECT key, before_value, after_value FROM system_setting_changes ORDER BY id" ).fetchall() self.assertEqual(2, len(changes)) self.assertEqual("5", changes[0]["before_value"]) self.assertEqual("1", changes[0]["after_value"]) if __name__ == "__main__": unittest.main()