212b1f9b9b
Reuse resolved reminder rows instead of inserting duplicates, skip failed keys in batch send, reject invalid scan settings with 400, and bind 去处理 via event delegation after async render. Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>
591 lines
24 KiB
Python
591 lines
24 KiB
Python
"""HTTP integration tests for authentication, RBAC and tenant isolation.
|
|
|
|
Spins up a real ``ThreadingHTTPServer`` with a temp database/storage and
|
|
drives it with stdlib ``http.client`` (cookies handled by hand). The server
|
|
module reads ``APP_DB_PATH`` / ``APP_STORAGE_DIR`` from module globals at
|
|
request time, so tests patch them per class.
|
|
"""
|
|
|
|
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 import auth
|
|
from bank_importer.db import connect, migrate, utc_now
|
|
|
|
import server
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SAMPLES = ROOT / "流水模板"
|
|
CCB_SAMPLE = SAMPLES / "中国建设银行账户流水.xls"
|
|
CITIC_SAMPLE = SAMPLES / "中信银行账户流水.xlsx"
|
|
|
|
BOOTSTRAP_PASSWORD = "BootAdmin123"
|
|
ADMIN_PASSWORD = "AdminPass123"
|
|
CASHIER_A_PASSWORD = "CashierA123"
|
|
|
|
|
|
class Client:
|
|
"""Minimal HTTP client with a cookie jar."""
|
|
|
|
def __init__(self, host: str, port: int) -> None:
|
|
self.host = host
|
|
self.port = port
|
|
self.cookies: dict[str, str] = {}
|
|
|
|
def request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
body: bytes | None = None,
|
|
headers: dict[str, str] | None = None,
|
|
) -> tuple[int, dict[str, str], bytes]:
|
|
connection = HTTPConnection(self.host, self.port)
|
|
request_headers = dict(headers or {})
|
|
if self.cookies:
|
|
request_headers["Cookie"] = "; ".join(
|
|
f"{key}={value}" for key, value in self.cookies.items()
|
|
)
|
|
connection.request(method, path, body=body, headers=request_headers)
|
|
response = connection.getresponse()
|
|
data = response.read()
|
|
response_headers = {key.lower(): value for key, value in response.getheaders()}
|
|
set_cookie = response_headers.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, response_headers, data
|
|
|
|
def get(self, path: str) -> tuple[int, dict[str, str], bytes]:
|
|
return self.request("GET", path)
|
|
|
|
def post_json(self, path: str, payload: dict) -> tuple[int, dict, bytes]:
|
|
return self.request(
|
|
"POST",
|
|
path,
|
|
body=json.dumps(payload).encode("utf-8"),
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
|
|
def post_multipart(
|
|
self, path: str, fields: dict[str, str], filename: str, content: bytes
|
|
) -> tuple[int, dict, bytes]:
|
|
boundary = "----cwtestboundary7f3a9c1e"
|
|
parts: list[bytes] = []
|
|
for name, value in fields.items():
|
|
parts.append(
|
|
f'--{boundary}\r\nContent-Disposition: form-data; name="{name}"\r\n\r\n{value}\r\n'.encode()
|
|
)
|
|
parts.append(
|
|
f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="{filename}"\r\n'
|
|
"Content-Type: application/octet-stream\r\n\r\n".encode()
|
|
+ content
|
|
+ b"\r\n"
|
|
)
|
|
parts.append(f"--{boundary}--\r\n".encode())
|
|
return self.request(
|
|
"POST",
|
|
path,
|
|
body=b"".join(parts),
|
|
headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
|
|
)
|
|
|
|
|
|
def as_json(data: bytes) -> dict:
|
|
return json.loads(data.decode("utf-8"))
|
|
|
|
|
|
class ServerAuthMatrixTests(unittest.TestCase):
|
|
"""One live server; setUpClass builds the shared fixture via the API."""
|
|
|
|
@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)
|
|
generated = server.ensure_bootstrap_admin(connection)
|
|
assert generated is None, "env password set, nothing should be generated"
|
|
connection.close()
|
|
|
|
class QuietHandler(server.AppHandler):
|
|
def log_message(self, *args) -> None: # silence per-request logs
|
|
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.known_passwords = {BOOTSTRAP_PASSWORD, ADMIN_PASSWORD, CASHIER_A_PASSWORD}
|
|
|
|
# --- Admin bootstrap: must_change_password gate, then change. ---
|
|
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
|
|
assert as_json(data)["must_change_password"] is True
|
|
status, _, data = cls.admin.get("/api/batches")
|
|
assert status == 403, "must_change_password must block API access"
|
|
status, _, data = cls.admin.post_json(
|
|
"/api/password/change",
|
|
{"old_password": BOOTSTRAP_PASSWORD, "new_password": ADMIN_PASSWORD},
|
|
)
|
|
assert status == 200, data
|
|
status, _, _ = cls.admin.get("/api/batches")
|
|
assert status == 200
|
|
|
|
# --- Companies A and B. ---
|
|
status, _, data = cls.admin.post_json("/api/admin/companies", {"name": "甲公司"})
|
|
assert status == 200, data
|
|
cls.company_a = as_json(data)["company_id"]
|
|
status, _, data = cls.admin.post_json("/api/admin/companies", {"name": "乙公司"})
|
|
assert status == 200, data
|
|
cls.company_b = as_json(data)["company_id"]
|
|
|
|
# --- Company user A: random one-time initial password, forced change. ---
|
|
status, _, data = cls.admin.post_json(
|
|
"/api/admin/users", {"username": "cashier-a", "company_id": cls.company_a}
|
|
)
|
|
assert status == 200, data
|
|
payload = as_json(data)
|
|
# Security: the initial password is random and never equals the username.
|
|
assert payload["initial_password"] != "cashier-a"
|
|
assert len(payload["initial_password"]) >= 12
|
|
cls.initial_password_a = payload["initial_password"]
|
|
cls.known_passwords.add(cls.initial_password_a)
|
|
|
|
cls.cashier_a = Client("127.0.0.1", cls.port)
|
|
status, _, data = cls.cashier_a.post_json(
|
|
"/api/login",
|
|
{
|
|
"username": "cashier-a",
|
|
"password": cls.initial_password_a,
|
|
"portal": "company",
|
|
},
|
|
)
|
|
assert status == 200, data
|
|
assert as_json(data)["must_change_password"] is True
|
|
status, _, _ = cls.cashier_a.get("/api/batches")
|
|
assert status == 403, "must_change_password must block company API access"
|
|
status, _, data = cls.cashier_a.post_json(
|
|
"/api/password/change",
|
|
{"old_password": cls.initial_password_a, "new_password": CASHIER_A_PASSWORD},
|
|
)
|
|
assert status == 200, data
|
|
|
|
# --- B gets a batch (admin upload), A gets its own batch. ---
|
|
status, _, data = cls.admin.post_multipart(
|
|
"/api/parse",
|
|
{"company_id": str(cls.company_b)},
|
|
CCB_SAMPLE.name,
|
|
CCB_SAMPLE.read_bytes(),
|
|
)
|
|
assert status == 200, data
|
|
cls.b_batch_id = as_json(data)["batch_id"]
|
|
cls.confirm_all_sheets(cls.admin, cls.b_batch_id)
|
|
|
|
status, _, data = cls.cashier_a.post_multipart(
|
|
"/api/parse", {}, CITIC_SAMPLE.name, CITIC_SAMPLE.read_bytes()
|
|
)
|
|
assert status == 200, data
|
|
cls.a_batch_id = as_json(data)["batch_id"]
|
|
cls.confirm_all_sheets(cls.cashier_a, cls.a_batch_id)
|
|
|
|
@classmethod
|
|
def confirm_all_sheets(cls, client, batch_id: int) -> None:
|
|
status, _, data = client.get(f"/api/batches/{batch_id}/sheets")
|
|
assert status == 200, data
|
|
names = [
|
|
sheet["sheet_name"]
|
|
for sheet in as_json(data)["sheets"]
|
|
if sheet["outcome"] == "parsed"
|
|
]
|
|
if not names:
|
|
return
|
|
status, _, data = client.post_json(
|
|
f"/api/batches/{batch_id}/confirm", {"sheets": names}
|
|
)
|
|
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()
|
|
|
|
# ------------------------------------------------------------------
|
|
# Helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
def fresh_client(self) -> Client:
|
|
return Client("127.0.0.1", self.port)
|
|
|
|
def create_company_user(self, username: str) -> tuple[Client, str, int]:
|
|
status, _, data = self.admin.post_json(
|
|
"/api/admin/users", {"username": username, "company_id": self.company_a}
|
|
)
|
|
self.assertEqual(200, status, data)
|
|
payload = as_json(data)
|
|
initial = payload["initial_password"]
|
|
self.known_passwords.add(initial)
|
|
client = self.fresh_client()
|
|
status, _, data = client.post_json(
|
|
"/api/login",
|
|
{"username": username, "password": initial, "portal": "company"},
|
|
)
|
|
self.assertEqual(200, status, data)
|
|
new_password = "Changed123"
|
|
self.known_passwords.add(new_password)
|
|
status, _, data = client.post_json(
|
|
"/api/password/change",
|
|
{"old_password": initial, "new_password": new_password},
|
|
)
|
|
self.assertEqual(200, status, data)
|
|
return client, new_password, payload["user_id"]
|
|
|
|
# ------------------------------------------------------------------
|
|
# Unauthenticated matrix
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_unauthenticated_api_calls_return_401(self) -> None:
|
|
anon = self.fresh_client()
|
|
for method_check in (
|
|
lambda: anon.post_multipart("/api/parse", {}, "x.xls", b"data"),
|
|
lambda: anon.get("/api/batches"),
|
|
lambda: anon.get(f"/api/batches/{self.a_batch_id}/rows"),
|
|
lambda: anon.get("/api/export.csv"),
|
|
lambda: anon.get("/api/admin/users"),
|
|
lambda: anon.get("/api/admin/companies"),
|
|
lambda: anon.get("/api/admin/audit-log"),
|
|
lambda: anon.get("/api/admin/reminders"),
|
|
lambda: anon.get("/api/company/reminders"),
|
|
lambda: anon.get("/api/me"),
|
|
):
|
|
status, _, data = method_check()
|
|
self.assertEqual(401, status, data)
|
|
self.assertEqual("error", as_json(data)["status"])
|
|
|
|
def test_unauthenticated_portal_pages_redirect(self) -> None:
|
|
anon = self.fresh_client()
|
|
for page in ("/admin.html", "/company.html"):
|
|
status, headers, _ = anon.get(page)
|
|
self.assertEqual(302, status, page)
|
|
self.assertEqual("/", headers.get("location"))
|
|
|
|
def test_wrong_password_returns_generic_401(self) -> None:
|
|
anon = self.fresh_client()
|
|
status, _, data = anon.post_json(
|
|
"/api/login",
|
|
{"username": "group-admin", "password": "WrongPass1", "portal": "admin"},
|
|
)
|
|
self.assertEqual(401, status)
|
|
message = as_json(data)["message"]
|
|
self.assertNotIn("密码不正确", message.replace("账号或密码不正确", ""))
|
|
|
|
def test_rate_limit_after_five_failures(self) -> None:
|
|
anon = self.fresh_client()
|
|
for _ in range(5):
|
|
status, _, _ = anon.post_json(
|
|
"/api/login",
|
|
{"username": "ghost-user", "password": "WrongPass1", "portal": "admin"},
|
|
)
|
|
self.assertEqual(401, status)
|
|
status, _, _ = anon.post_json(
|
|
"/api/login",
|
|
{"username": "ghost-user", "password": "WrongPass1", "portal": "admin"},
|
|
)
|
|
self.assertEqual(429, status)
|
|
|
|
def test_portal_mismatch_returns_403(self) -> None:
|
|
anon = self.fresh_client()
|
|
status, _, data = anon.post_json(
|
|
"/api/login",
|
|
{"username": "group-admin", "password": ADMIN_PASSWORD, "portal": "company"},
|
|
)
|
|
self.assertEqual(403, status)
|
|
self.assertIn("端口", as_json(data)["message"])
|
|
|
|
def test_logout_revokes_session(self) -> None:
|
|
client, _, _ = self.create_company_user("cashier-logout")
|
|
status, _, _ = client.get("/api/me")
|
|
self.assertEqual(200, status)
|
|
status, _, _ = client.request("POST", "/api/logout")
|
|
self.assertEqual(200, status)
|
|
self.assertNotIn("cw_session", client.cookies)
|
|
status, _, _ = client.get("/api/me")
|
|
self.assertEqual(401, status)
|
|
|
|
def test_expired_session_returns_401(self) -> None:
|
|
client, _, user_id = self.create_company_user("cashier-expired")
|
|
connection = connect(self.db_path)
|
|
try:
|
|
with connection:
|
|
connection.execute(
|
|
"UPDATE sessions SET expires_at = ? WHERE user_id = ?",
|
|
("2000-01-01T00:00:00+00:00", user_id),
|
|
)
|
|
finally:
|
|
connection.close()
|
|
status, _, _ = client.get("/api/me")
|
|
self.assertEqual(401, status)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Admin and company lifecycle
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_duplicate_company_name_returns_409(self) -> None:
|
|
status, _, _ = self.admin.post_json("/api/admin/companies", {"name": "甲公司"})
|
|
self.assertEqual(409, status)
|
|
|
|
def test_me_returns_profile_without_password_material(self) -> None:
|
|
status, _, data = self.cashier_a.get("/api/me")
|
|
self.assertEqual(200, status)
|
|
payload = as_json(data)
|
|
self.assertEqual("cashier-a", payload["username"])
|
|
self.assertEqual("company", payload["role"])
|
|
self.assertEqual(self.company_a, payload["company_id"])
|
|
self.assertEqual("甲公司", payload["company_name"])
|
|
self.assertFalse(payload["must_change_password"])
|
|
|
|
def test_company_user_forbidden_on_all_admin_endpoints(self) -> None:
|
|
calls = (
|
|
lambda: self.cashier_a.get("/api/admin/companies"),
|
|
lambda: self.cashier_a.post_json("/api/admin/companies", {"name": "丙公司"}),
|
|
lambda: self.cashier_a.get("/api/admin/users"),
|
|
lambda: self.cashier_a.post_json(
|
|
"/api/admin/users", {"username": "x", "company_id": self.company_a}
|
|
),
|
|
lambda: self.cashier_a.request("POST", "/api/admin/users/1/disable"),
|
|
lambda: self.cashier_a.request("POST", "/api/admin/users/1/enable"),
|
|
lambda: self.cashier_a.request("POST", "/api/admin/users/1/reset-password"),
|
|
lambda: self.cashier_a.get("/api/admin/audit-log"),
|
|
lambda: self.cashier_a.get("/api/admin/reminders"),
|
|
lambda: self.cashier_a.get("/api/admin/reminders/pending"),
|
|
lambda: self.cashier_a.post_json("/api/admin/reminder-settings", {"gap_days": "3"}),
|
|
lambda: self.cashier_a.post_json("/api/admin/reminders/send", {"dedupe_keys": ["x"]}),
|
|
)
|
|
for call in calls:
|
|
status, _, data = call()
|
|
self.assertEqual(403, status, data)
|
|
|
|
def test_admin_upload_requires_company_id(self) -> None:
|
|
status, _, data = self.admin.post_multipart(
|
|
"/api/parse", {}, CCB_SAMPLE.name, CCB_SAMPLE.read_bytes()
|
|
)
|
|
self.assertEqual(400, status, data)
|
|
|
|
def test_admin_batches_filter_by_company(self) -> None:
|
|
status, _, data = self.admin.get(f"/api/batches?company_id={self.company_b}")
|
|
self.assertEqual(200, status)
|
|
batches = as_json(data)["batches"]
|
|
self.assertTrue(batches)
|
|
for batch in batches:
|
|
self.assertEqual(self.company_b, batch["company_id"])
|
|
self.assertEqual("乙公司", batch["company_name"])
|
|
|
|
# ------------------------------------------------------------------
|
|
# Tenant isolation
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_batches_scoped_to_own_company(self) -> None:
|
|
status, _, data = self.cashier_a.get("/api/batches")
|
|
self.assertEqual(200, status)
|
|
batches = as_json(data)["batches"]
|
|
self.assertTrue(batches)
|
|
for batch in batches:
|
|
self.assertEqual(self.company_a, batch["company_id"])
|
|
self.assertNotIn(self.b_batch_id, [batch["id"] for batch in batches])
|
|
|
|
def test_idor_batch_rows_of_other_company_return_404(self) -> None:
|
|
status, _, _ = self.cashier_a.get(f"/api/batches/{self.b_batch_id}/rows")
|
|
self.assertEqual(404, status)
|
|
status, _, data = self.cashier_a.get(f"/api/batches/{self.a_batch_id}/rows")
|
|
self.assertEqual(200, status)
|
|
self.assertTrue(as_json(data)["rows"])
|
|
|
|
def test_export_csv_forced_to_own_company(self) -> None:
|
|
status, _, _ = self.cashier_a.get(f"/api/export.csv?company_id={self.company_b}")
|
|
self.assertEqual(403, status)
|
|
|
|
status, headers, data = self.cashier_a.get("/api/export.csv")
|
|
self.assertEqual(200, status)
|
|
self.assertEqual("text/csv; charset=utf-8", headers.get("content-type"))
|
|
text = data.decode("utf-8-sig")
|
|
lines = [line for line in text.splitlines() if line]
|
|
self.assertGreater(len(lines), 1)
|
|
for line in lines[1:]:
|
|
self.assertEqual(str(self.a_batch_id), line.split(",", 1)[0])
|
|
|
|
def test_cross_company_upload_is_recorded_under_own_company(self) -> None:
|
|
# A uploads B's file bytes while claiming company B in the form; the
|
|
# server must bind the new (duplicate) batch to A from the session.
|
|
status, _, data = self.cashier_a.post_multipart(
|
|
"/api/parse",
|
|
{"company_id": str(self.company_b)},
|
|
CCB_SAMPLE.name,
|
|
CCB_SAMPLE.read_bytes(),
|
|
)
|
|
self.assertEqual(200, status, data)
|
|
payload = as_json(data)
|
|
self.assertEqual("duplicate", payload["status"])
|
|
# Cross-company duplicate: no original batch id, bank, template,
|
|
# period, transaction count or diagnostics may be exposed.
|
|
self.assertNotEqual(self.b_batch_id, payload.get("batch_id"))
|
|
for leaked_key in ("bank", "template", "header_row", "period_start", "period_end", "transactions", "warnings"):
|
|
self.assertNotIn(leaked_key, payload, leaked_key)
|
|
connection = connect(self.db_path)
|
|
try:
|
|
duplicate = connection.execute(
|
|
"SELECT company_id FROM import_batches WHERE status = 'duplicate'"
|
|
).fetchone()
|
|
finally:
|
|
connection.close()
|
|
self.assertIsNotNone(duplicate)
|
|
self.assertEqual(self.company_a, duplicate["company_id"])
|
|
|
|
def test_same_company_duplicate_keeps_idempotent_summary(self) -> None:
|
|
# A re-uploads its own file; the idempotent duplicate response keeps
|
|
# the original batch id and the parsed summary.
|
|
status, _, data = self.cashier_a.post_multipart(
|
|
"/api/parse", {}, CITIC_SAMPLE.name, CITIC_SAMPLE.read_bytes()
|
|
)
|
|
self.assertEqual(200, status, data)
|
|
payload = as_json(data)
|
|
self.assertEqual("duplicate", payload["status"])
|
|
self.assertEqual(self.a_batch_id, payload["batch_id"])
|
|
self.assertTrue(payload.get("sheets"))
|
|
self.assertEqual("中信银行", payload["sheets"][0].get("bank"))
|
|
self.assertTrue(payload["sheets"][0].get("transactions", 0) > 0)
|
|
self.assertIn("period_start", payload["sheets"][0])
|
|
self.assertIn("warnings", payload["sheets"][0])
|
|
|
|
# ------------------------------------------------------------------
|
|
# Disable / reset flows
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_disabled_user_session_and_login_rejected(self) -> None:
|
|
client, password, user_id = self.create_company_user("cashier-disable")
|
|
status, _, _ = client.get("/api/me")
|
|
self.assertEqual(200, status)
|
|
|
|
status, _, data = self.admin.request("POST", f"/api/admin/users/{user_id}/disable")
|
|
self.assertEqual(200, status, data)
|
|
|
|
status, _, _ = client.get("/api/me")
|
|
self.assertEqual(401, status)
|
|
|
|
fresh = self.fresh_client()
|
|
status, _, data = fresh.post_json(
|
|
"/api/login",
|
|
{"username": "cashier-disable", "password": password, "portal": "company"},
|
|
)
|
|
self.assertEqual(403, status, data)
|
|
|
|
def test_reset_password_returns_once_and_revokes_sessions(self) -> None:
|
|
client, old_password, user_id = self.create_company_user("cashier-reset")
|
|
status, _, data = self.admin.request(
|
|
"POST", f"/api/admin/users/{user_id}/reset-password"
|
|
)
|
|
self.assertEqual(200, status, data)
|
|
new_password = as_json(data)["initial_password"]
|
|
self.known_passwords.add(new_password)
|
|
|
|
status, _, _ = client.get("/api/me")
|
|
self.assertEqual(401, status)
|
|
|
|
fresh = self.fresh_client()
|
|
status, _, data = fresh.post_json(
|
|
"/api/login",
|
|
{"username": "cashier-reset", "password": new_password, "portal": "company"},
|
|
)
|
|
self.assertEqual(200, status, data)
|
|
self.assertTrue(as_json(data)["must_change_password"])
|
|
|
|
# ------------------------------------------------------------------
|
|
# Secrets hygiene
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_no_response_contains_password_hash(self) -> None:
|
|
bodies = []
|
|
status, _, data = self.admin.get("/api/admin/users")
|
|
self.assertEqual(200, status)
|
|
bodies.append(data)
|
|
status, _, data = self.admin.get("/api/admin/audit-log?limit=100")
|
|
self.assertEqual(200, status)
|
|
bodies.append(data)
|
|
status, _, data = self.cashier_a.get("/api/me")
|
|
bodies.append(data)
|
|
status, _, data = self.cashier_a.get("/api/batches")
|
|
bodies.append(data)
|
|
for body in bodies:
|
|
self.assertNotIn("password_hash", body.decode("utf-8"))
|
|
|
|
def test_audit_log_contains_no_plaintext_passwords(self) -> None:
|
|
connection = connect(self.db_path)
|
|
try:
|
|
rows = connection.execute(
|
|
"SELECT detail, target FROM audit_log"
|
|
).fetchall()
|
|
finally:
|
|
connection.close()
|
|
for password in self.known_passwords:
|
|
for row in rows:
|
|
self.assertNotIn(password, row["detail"] or "")
|
|
self.assertNotIn(password, row["target"] or "")
|
|
|
|
def test_invalid_reminder_settings_return_400_and_do_not_persist(self) -> None:
|
|
cases = (
|
|
{"monthly_start_day": "0"},
|
|
{"gap_days": "0"},
|
|
{"scan_time": "25:99"},
|
|
)
|
|
for payload in cases:
|
|
with self.subTest(payload=payload):
|
|
status, _, data = self.admin.post_json(
|
|
"/api/admin/reminder-settings", {"settings": payload}
|
|
)
|
|
self.assertEqual(400, status, data)
|
|
self.assertEqual("error", as_json(data)["status"])
|
|
status, _, data = self.admin.get("/api/admin/reminder-settings")
|
|
self.assertEqual(200, status)
|
|
settings = as_json(data)["settings"]
|
|
self.assertEqual("5", settings["monthly_start_day"])
|
|
self.assertEqual("5", settings["gap_days"])
|
|
self.assertEqual("08:00", settings["scan_time"])
|
|
status, _, data = self.admin.get("/api/admin/reminders/pending")
|
|
self.assertEqual(200, status, data)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|