259 lines
9.7 KiB
Python
259 lines
9.7 KiB
Python
"""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)
|
|
|
|
|
|
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()
|