caiwuzongzhang/tests/test_settings_reminders.py

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