HEL-144 返工A：四件事（改名/登录端标识/结账设置真保存/提醒回改）+ 迁移升为 7 + 移除废弃 B-44 前端样式测试
Co-authored-by: multica-agent <github@multica.ai>
This commit is contained in:
leefer
@@ -0,0 +1,258 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user