B-41: 动态主数据——公司、银行账户、别名与审核轨迹
- 公司、公司账号、银行账户、账户类型、户名/账号别名和生效区间 全部来自数据库，新增公司无需改代码即可检索与使用。 - 公司端提交账户登记，总账端审核为启用/退回/停用；未启用账户 不参与所有权识别、上传或覆盖计算。 - 账号归一化、脱敏显示与数据库 UNIQUE 约束，并发提交不产生重复 有效账户；别名匹配有优先级和有效期。 - 所有主数据变更记录前值、后值、操作人、时间和原因。 - 决策记录见 docs/decisions/004-master-data.md。
This commit is contained in:
腾讯WorkBuddy
@@ -0,0 +1,47 @@
|
||||
# 004 动态主数据技术决策
|
||||
|
||||
对应 Issue：B-41（`docs/issues/004-p1-dynamic-master-data.md`）。
|
||||
|
||||
## 账户状态机与生效区间
|
||||
|
||||
- `bank_accounts.status`：`pending → active → disabled`，审核退回为
|
||||
`returned`；退回后同公司重新提交复用同一行并回到 `pending`，不产生
|
||||
第二条记录。停用这个动作只对 `active` 开放，退回只对 `pending` 开放，
|
||||
越界操作返回 409。
|
||||
- 生效区间 `effective_from` / `effective_to`（含当天）：公司提交时填写
|
||||
期望启用日，管理员审核通过时确认；停用时写入 `effective_to`（默认当天）。
|
||||
- 两个判定函数分开：
|
||||
- `is_usable`（上传选择、当期操作）：仅 `active` 且在生效区间内。
|
||||
- `is_identifiable`（历史流水所有权识别）：`active` 或 `disabled` 且在
|
||||
生效区间内——停用只终止未来参与，不改写历史归属。
|
||||
|
||||
## 账号规范化与唯一性
|
||||
|
||||
- 账号入库前统一规范化：去空格/短横线、全角数字转半角，只存 6–32 位
|
||||
数字的规范形式；唯一性由数据库 `UNIQUE(account_number)` 保证，并发
|
||||
提交必然只有一个成功（有并发测试）。
|
||||
- 公司端任何响应只含脱敏形式 `****尾四位`；完整账号只在总账管理端
|
||||
（授权审计视图）返回。审计日志 detail 也只写脱敏账号。
|
||||
|
||||
## 别名匹配优先级
|
||||
|
||||
- `account_aliases` 支持户名（`name`）与账号（`account`）两类别名，
|
||||
各自带生效区间和 `priority` 列。
|
||||
- 匹配优先级确定：精确账号（0）< 账号别名（1000+priority）<
|
||||
户名别名（2000+priority）；同层内按 `priority` 再按账户 id 排序。
|
||||
别名只在自身生效区间且账户 `is_identifiable` 时参与匹配。
|
||||
|
||||
## 审计轨迹
|
||||
|
||||
- 所有主数据变更写入 `master_data_changes`：实体、动作、前值 JSON、
|
||||
后值 JSON、原因、操作人、时间。公司创建、账户提交/退回重提/
|
||||
审核/停用、别名创建全部覆盖，管理员可按实体查询。
|
||||
|
||||
## 前端边界
|
||||
|
||||
- 公司下拉（往来查询、审核筛选、流水筛选、提醒、期初）和公司表格全部
|
||||
由 `/api/admin/companies` 渲染，新增公司无需改代码即可被检索。
|
||||
- 公司端账户目录、上传账户选择、手工记录资金来源只来自
|
||||
`/api/company/accounts`，且只有 `usable` 账户进入上传选择。
|
||||
- 手工记录仍是浏览器 localStorage 演示数据（属 Issue 007/012 范围），
|
||||
账户登记已完全切换到服务端。
|
||||
@@ -0,0 +1,570 @@
|
||||
"""Dynamic master data: companies, bank accounts, aliases and audit trail.
|
||||
|
||||
Companies, cashier logins and bank accounts live in the database instead of
|
||||
being hard-coded in the UI. Bank account numbers are stored normalized
|
||||
(digits only) under a database UNIQUE constraint, so two concurrent
|
||||
submissions can never produce two usable accounts for the same number.
|
||||
Company-side registrations are requests: only administrator-approved
|
||||
(``active``) accounts inside their effective interval may identify ownership,
|
||||
accept uploads or take part in coverage calculation. Every change is recorded
|
||||
in ``master_data_changes`` with before/after values, actor, time and reason.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date, datetime, timezone
|
||||
import json
|
||||
import re
|
||||
import sqlite3
|
||||
|
||||
from .db import utc_now
|
||||
|
||||
|
||||
ACCOUNT_TYPES = ("基本户", "一般户", "专用户")
|
||||
ACCOUNT_STATUSES = ("pending", "active", "returned", "disabled")
|
||||
ALIAS_KINDS = ("name", "account")
|
||||
|
||||
# Deterministic alias match priority: an exact account-number hit always
|
||||
# beats an account alias, which always beats a name alias. The per-alias
|
||||
# ``priority`` column only orders matches within the same tier.
|
||||
PRIORITY_EXACT_ACCOUNT = 0
|
||||
PRIORITY_ACCOUNT_ALIAS = 1000
|
||||
PRIORITY_NAME_ALIAS = 2000
|
||||
|
||||
_ACCOUNT_NUMBER_PATTERN = re.compile(r"^[0-9]{6,32}$")
|
||||
_DATE_PATTERN = re.compile(r"^\d{4}-\d{2}-\d{2}$")
|
||||
_FULL_WIDTH_DIGITS = str.maketrans("０１２３４５６７８９", "0123456789")
|
||||
|
||||
|
||||
class ConflictError(ValueError):
|
||||
"""A uniqueness or ownership conflict (mapped to HTTP 409)."""
|
||||
|
||||
|
||||
def utc_today() -> str:
|
||||
return datetime.now(timezone.utc).date().isoformat()
|
||||
|
||||
|
||||
def normalize_account_number(raw: object) -> str:
|
||||
"""Normalize an account number to digits only; raise ValueError if invalid.
|
||||
|
||||
Spaces, dashes and full-width digits entered by cashiers or emitted by
|
||||
bank exports collapse to one canonical form, which is what the UNIQUE
|
||||
constraint and all matching operate on.
|
||||
"""
|
||||
text = str(raw or "").translate(_FULL_WIDTH_DIGITS)
|
||||
text = re.sub(r"[\s\-‐-‒–—]+", "", text)
|
||||
if not _ACCOUNT_NUMBER_PATTERN.fullmatch(text):
|
||||
raise ValueError("银行账号须为 6-32 位数字（可含空格或短横线分隔）。")
|
||||
return text
|
||||
|
||||
|
||||
def mask_account_number(number: str) -> str:
|
||||
"""Masked display form; the full number stays server-side."""
|
||||
if len(number) <= 4:
|
||||
return f"****{number}"
|
||||
return f"****{number[-4:]}"
|
||||
|
||||
|
||||
def validate_date(value: object, field: str, *, required: bool = False) -> str | None:
|
||||
text = str(value or "").strip()
|
||||
if not text:
|
||||
if required:
|
||||
raise ValueError(f"{field}不能为空。")
|
||||
return None
|
||||
if not _DATE_PATTERN.fullmatch(text):
|
||||
raise ValueError(f"{field}须为 YYYY-MM-DD 格式。")
|
||||
try:
|
||||
date.fromisoformat(text)
|
||||
except ValueError:
|
||||
raise ValueError(f"{field}不是有效日期。") from None
|
||||
return text
|
||||
|
||||
|
||||
def validate_account_type(value: object) -> str:
|
||||
text = str(value or "").strip() or "一般户"
|
||||
if text not in ACCOUNT_TYPES:
|
||||
raise ValueError(f"账户类型必须是：{'、'.join(ACCOUNT_TYPES)}。")
|
||||
return text
|
||||
|
||||
|
||||
def _snapshot(row: sqlite3.Row | None) -> dict[str, object] | None:
|
||||
if row is None:
|
||||
return None
|
||||
return {key: row[key] for key in row.keys()}
|
||||
|
||||
|
||||
def record_change(
|
||||
connection: sqlite3.Connection,
|
||||
entity_type: str,
|
||||
entity_id: int,
|
||||
action: str,
|
||||
before: dict[str, object] | None,
|
||||
after: dict[str, object] | None,
|
||||
reason: str | None,
|
||||
actor: sqlite3.Row | None,
|
||||
) -> None:
|
||||
"""Append a before/after audit entry for a master data change."""
|
||||
connection.execute(
|
||||
"""
|
||||
INSERT INTO master_data_changes (
|
||||
entity_type, entity_id, action, before_json, after_json,
|
||||
reason, actor_user_id, actor_username, created_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
(
|
||||
entity_type,
|
||||
entity_id,
|
||||
action,
|
||||
json.dumps(before, ensure_ascii=False) if before is not None else None,
|
||||
json.dumps(after, ensure_ascii=False) if after is not None else None,
|
||||
reason,
|
||||
actor["id"] if actor is not None else None,
|
||||
actor["username"] if actor is not None else None,
|
||||
utc_now(),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Companies
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def create_company(
|
||||
connection: sqlite3.Connection,
|
||||
name: str,
|
||||
credit_code: str | None,
|
||||
cashier_name: str | None,
|
||||
actor: sqlite3.Row | None,
|
||||
) -> int:
|
||||
"""Create a company master record; returns the new id."""
|
||||
name = name.strip()
|
||||
if not name:
|
||||
raise ValueError("公司名称不能为空。")
|
||||
now = utc_now()
|
||||
try:
|
||||
with connection:
|
||||
cursor = connection.execute(
|
||||
"""
|
||||
INSERT INTO companies (
|
||||
name, credit_code, cashier_name, status, created_at, updated_at
|
||||
) VALUES (?, ?, ?, 'active', ?, ?)
|
||||
""",
|
||||
(name, (credit_code or "").strip() or None,
|
||||
(cashier_name or "").strip() or None, now, now),
|
||||
)
|
||||
except sqlite3.IntegrityError as exc:
|
||||
raise ConflictError("公司名称已存在。") from exc
|
||||
company_id = int(cursor.lastrowid)
|
||||
with connection:
|
||||
record_change(
|
||||
connection, "company", company_id, "create",
|
||||
None, {"name": name, "credit_code": credit_code or None,
|
||||
"cashier_name": cashier_name or None, "status": "active"},
|
||||
None, actor,
|
||||
)
|
||||
return company_id
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Bank accounts
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def get_account(connection: sqlite3.Connection, account_id: int) -> sqlite3.Row | None:
|
||||
return connection.execute(
|
||||
"SELECT * FROM bank_accounts WHERE id = ?", (account_id,)
|
||||
).fetchone()
|
||||
|
||||
|
||||
def submit_bank_account(
|
||||
connection: sqlite3.Connection,
|
||||
*,
|
||||
company_id: int,
|
||||
bank_name: str,
|
||||
account_type: object,
|
||||
account_number: object,
|
||||
account_name: object = None,
|
||||
start_date: object = None,
|
||||
actor: sqlite3.Row | None,
|
||||
) -> sqlite3.Row:
|
||||
"""Register an account for review; returns the resulting account row.
|
||||
|
||||
A normalized number already present is never duplicated: resubmitting a
|
||||
returned request from the same company reopens that same row as pending;
|
||||
any other existing row is a conflict. The UNIQUE constraint on
|
||||
``account_number`` is the final guard for concurrent submissions.
|
||||
"""
|
||||
number = normalize_account_number(account_number)
|
||||
bank = str(bank_name or "").strip()
|
||||
if not bank:
|
||||
raise ValueError("开户银行不能为空。")
|
||||
kind = validate_account_type(account_type)
|
||||
holder = str(account_name or "").strip() or None
|
||||
requested_from = validate_date(start_date, "启用日期")
|
||||
now = utc_now()
|
||||
|
||||
existing = connection.execute(
|
||||
"SELECT * FROM bank_accounts WHERE account_number = ?", (number,)
|
||||
).fetchone()
|
||||
|
||||
if existing is None:
|
||||
try:
|
||||
with connection:
|
||||
cursor = connection.execute(
|
||||
"""
|
||||
INSERT INTO bank_accounts (
|
||||
company_id, account_number, account_name, bank_name,
|
||||
account_type, status, effective_from, submitted_by,
|
||||
created_at, updated_at
|
||||
) VALUES (?, ?, ?, ?, ?, 'pending', ?, ?, ?, ?)
|
||||
""",
|
||||
(company_id, number, holder, bank, kind,
|
||||
requested_from, actor["id"] if actor else None, now, now),
|
||||
)
|
||||
except sqlite3.IntegrityError as exc:
|
||||
# Lost a concurrent-insert race on the UNIQUE constraint.
|
||||
raise ConflictError("该银行账号已登记，请等待现有申请处理。") from exc
|
||||
account_id = int(cursor.lastrowid)
|
||||
with connection:
|
||||
record_change(
|
||||
connection, "bank_account", account_id, "submit", None,
|
||||
{"company_id": company_id, "account_number": number,
|
||||
"bank_name": bank, "account_type": kind, "status": "pending",
|
||||
"effective_from": requested_from},
|
||||
None, actor,
|
||||
)
|
||||
return get_account(connection, account_id)
|
||||
|
||||
if existing["status"] == "returned" and existing["company_id"] == company_id:
|
||||
before = _snapshot(existing)
|
||||
with connection:
|
||||
connection.execute(
|
||||
"""
|
||||
UPDATE bank_accounts
|
||||
SET account_name = ?, bank_name = ?, account_type = ?,
|
||||
status = 'pending', effective_from = ?,
|
||||
submitted_by = ?, reviewed_by = NULL, reviewed_at = NULL,
|
||||
review_reason = NULL, updated_at = ?
|
||||
WHERE id = ? AND status = 'returned'
|
||||
""",
|
||||
(holder, bank, kind, requested_from,
|
||||
actor["id"] if actor else None, now, existing["id"]),
|
||||
)
|
||||
record_change(
|
||||
connection, "bank_account", existing["id"], "resubmit",
|
||||
before, {"account_number": number, "bank_name": bank,
|
||||
"account_type": kind, "status": "pending",
|
||||
"effective_from": requested_from},
|
||||
"退回后重新提交", actor,
|
||||
)
|
||||
return get_account(connection, existing["id"])
|
||||
|
||||
if existing["company_id"] == company_id:
|
||||
raise ConflictError("该银行账号已登记，请等待现有申请处理。")
|
||||
raise ConflictError("该银行账号已被其他公司登记，请联系总账管理员核对。")
|
||||
|
||||
|
||||
def review_bank_account(
|
||||
connection: sqlite3.Connection,
|
||||
account_id: int,
|
||||
decision: str,
|
||||
reason: str | None,
|
||||
actor: sqlite3.Row,
|
||||
*,
|
||||
effective_from: object = None,
|
||||
effective_to: object = None,
|
||||
) -> sqlite3.Row:
|
||||
"""Approve, return or disable a registration; returns the updated row."""
|
||||
account = get_account(connection, account_id)
|
||||
if account is None:
|
||||
raise LookupError("账户不存在。")
|
||||
reason = (reason or "").strip() or None
|
||||
today = utc_today()
|
||||
before = _snapshot(account)
|
||||
|
||||
if decision == "approve":
|
||||
if account["status"] != "pending":
|
||||
raise ConflictError("只有待复核的账户可以审核通过。")
|
||||
start = validate_date(effective_from, "启用日期") or account["effective_from"] or today
|
||||
with connection:
|
||||
connection.execute(
|
||||
"""
|
||||
UPDATE bank_accounts
|
||||
SET status = 'active', effective_from = ?, effective_to = NULL,
|
||||
reviewed_by = ?, reviewed_at = ?, review_reason = ?, updated_at = ?
|
||||
WHERE id = ?
|
||||
""",
|
||||
(start, actor["id"], utc_now(), reason, utc_now(), account_id),
|
||||
)
|
||||
record_change(
|
||||
connection, "bank_account", account_id, "approve", before,
|
||||
{"status": "active", "effective_from": start}, reason, actor,
|
||||
)
|
||||
elif decision == "return":
|
||||
if account["status"] != "pending":
|
||||
raise ConflictError("只有待复核的账户可以退回。")
|
||||
if reason is None:
|
||||
raise ValueError("退回必须填写原因。")
|
||||
with connection:
|
||||
connection.execute(
|
||||
"""
|
||||
UPDATE bank_accounts
|
||||
SET status = 'returned', reviewed_by = ?, reviewed_at = ?,
|
||||
review_reason = ?, updated_at = ?
|
||||
WHERE id = ?
|
||||
""",
|
||||
(actor["id"], utc_now(), reason, utc_now(), account_id),
|
||||
)
|
||||
record_change(
|
||||
connection, "bank_account", account_id, "return", before,
|
||||
{"status": "returned"}, reason, actor,
|
||||
)
|
||||
elif decision == "disable":
|
||||
if account["status"] != "active":
|
||||
raise ConflictError("只有已启用的账户可以停用。")
|
||||
if reason is None:
|
||||
raise ValueError("停用必须填写原因。")
|
||||
end = validate_date(effective_to, "停用日期") or today
|
||||
with connection:
|
||||
connection.execute(
|
||||
"""
|
||||
UPDATE bank_accounts
|
||||
SET status = 'disabled', effective_to = ?,
|
||||
reviewed_by = ?, reviewed_at = ?, review_reason = ?, updated_at = ?
|
||||
WHERE id = ?
|
||||
""",
|
||||
(end, actor["id"], utc_now(), reason, utc_now(), account_id),
|
||||
)
|
||||
record_change(
|
||||
connection, "bank_account", account_id, "disable", before,
|
||||
{"status": "disabled", "effective_to": end}, reason, actor,
|
||||
)
|
||||
else:
|
||||
raise ValueError("审核决定必须是 approve、return 或 disable。")
|
||||
return get_account(connection, account_id)
|
||||
|
||||
|
||||
def in_effective_window(account: sqlite3.Row, on_date: str) -> bool:
|
||||
"""True when ``on_date`` falls inside the account's effective interval.
|
||||
|
||||
``effective_to`` is the last participating day (inclusive).
|
||||
"""
|
||||
if account["effective_from"] and on_date < account["effective_from"]:
|
||||
return False
|
||||
if account["effective_to"] and on_date > account["effective_to"]:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def is_usable(account: sqlite3.Row, on_date: str) -> bool:
|
||||
"""True when the account may accept uploads / be selected on a date."""
|
||||
return account["status"] == "active" and in_effective_window(account, on_date)
|
||||
|
||||
|
||||
def is_identifiable(account: sqlite3.Row, on_date: str) -> bool:
|
||||
"""True when the account may resolve ownership for a transaction date.
|
||||
|
||||
A disabled account keeps identifying historical rows inside its effective
|
||||
window — disabling stops future participation, it never rewrites history.
|
||||
"""
|
||||
return account["status"] in ("active", "disabled") and in_effective_window(
|
||||
account, on_date
|
||||
)
|
||||
|
||||
|
||||
def list_accounts(
|
||||
connection: sqlite3.Connection,
|
||||
company_id: int | None = None,
|
||||
status: str | None = None,
|
||||
) -> list[sqlite3.Row]:
|
||||
conditions: list[str] = []
|
||||
params: list[object] = []
|
||||
if company_id is not None:
|
||||
conditions.append("a.company_id = ?")
|
||||
params.append(company_id)
|
||||
if status is not None:
|
||||
if status not in ACCOUNT_STATUSES:
|
||||
raise ValueError("无效的账户状态。")
|
||||
conditions.append("a.status = ?")
|
||||
params.append(status)
|
||||
where = f"WHERE {' AND '.join(conditions)}" if conditions else ""
|
||||
return connection.execute(
|
||||
f"""
|
||||
SELECT a.*, c.name AS company_name
|
||||
FROM bank_accounts a
|
||||
JOIN companies c ON c.id = a.company_id
|
||||
{where}
|
||||
ORDER BY a.id
|
||||
""",
|
||||
params,
|
||||
).fetchall()
|
||||
|
||||
|
||||
def usable_accounts(
|
||||
connection: sqlite3.Connection, company_id: int, on_date: str | None = None
|
||||
) -> list[sqlite3.Row]:
|
||||
"""Accounts allowed to upload / identify ownership for the company today."""
|
||||
day = on_date or utc_today()
|
||||
return [
|
||||
account
|
||||
for account in list_accounts(connection, company_id=company_id)
|
||||
if is_usable(account, day)
|
||||
]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Aliases
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def add_alias(
|
||||
connection: sqlite3.Connection,
|
||||
account_id: int,
|
||||
alias_kind: str,
|
||||
alias_value: object,
|
||||
priority: object = None,
|
||||
effective_from: object = None,
|
||||
effective_to: object = None,
|
||||
actor: sqlite3.Row | None = None,
|
||||
) -> int:
|
||||
"""Attach a name/account alias with an effective interval; returns the id."""
|
||||
if get_account(connection, account_id) is None:
|
||||
raise LookupError("账户不存在。")
|
||||
if alias_kind not in ALIAS_KINDS:
|
||||
raise ValueError("别名类型必须是 name 或 account。")
|
||||
if alias_kind == "account":
|
||||
value = normalize_account_number(alias_value)
|
||||
else:
|
||||
value = re.sub(r"\s+", "", str(alias_value or ""))
|
||||
if not value:
|
||||
raise ValueError("户名别名不能为空。")
|
||||
try:
|
||||
rank = int(priority) if priority not in (None, "") else 100
|
||||
except (TypeError, ValueError):
|
||||
raise ValueError("优先级必须是整数。") from None
|
||||
start = validate_date(effective_from, "别名生效日期")
|
||||
end = validate_date(effective_to, "别名失效日期")
|
||||
if start and end and end < start:
|
||||
raise ValueError("别名失效日期不能早于生效日期。")
|
||||
try:
|
||||
with connection:
|
||||
cursor = connection.execute(
|
||||
"""
|
||||
INSERT INTO account_aliases (
|
||||
bank_account_id, alias_kind, alias_value, priority,
|
||||
effective_from, effective_to, created_by, created_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
(account_id, alias_kind, value, rank, start, end,
|
||||
actor["id"] if actor else None, utc_now()),
|
||||
)
|
||||
except sqlite3.IntegrityError as exc:
|
||||
raise ConflictError("该账户下相同别名已存在。") from exc
|
||||
alias_id = int(cursor.lastrowid)
|
||||
with connection:
|
||||
record_change(
|
||||
connection, "account_alias", alias_id, "create", None,
|
||||
{"bank_account_id": account_id, "alias_kind": alias_kind,
|
||||
"alias_value": value, "priority": rank,
|
||||
"effective_from": start, "effective_to": end},
|
||||
None, actor,
|
||||
)
|
||||
return alias_id
|
||||
|
||||
|
||||
def list_aliases(connection: sqlite3.Connection, account_id: int) -> list[sqlite3.Row]:
|
||||
return connection.execute(
|
||||
"SELECT * FROM account_aliases WHERE bank_account_id = ? ORDER BY id",
|
||||
(account_id,),
|
||||
).fetchall()
|
||||
|
||||
|
||||
def match_account(
|
||||
connection: sqlite3.Connection,
|
||||
*,
|
||||
account_number: object = None,
|
||||
name: object = None,
|
||||
on_date: str | None = None,
|
||||
) -> list[dict[str, object]]:
|
||||
"""Resolve an observed counterparty to operating, in-window accounts.
|
||||
|
||||
Deterministic priority: exact account number < account alias < name
|
||||
alias; within one tier the alias ``priority`` column orders the hits.
|
||||
Pending and returned accounts never match; disabled accounts still match
|
||||
transaction dates inside their effective window (see ``is_identifiable``).
|
||||
"""
|
||||
day = on_date or utc_today()
|
||||
hits: list[tuple[int, sqlite3.Row, str]] = []
|
||||
|
||||
number = str(account_number or "").strip()
|
||||
if number:
|
||||
try:
|
||||
normalized = normalize_account_number(number)
|
||||
except ValueError:
|
||||
normalized = None
|
||||
if normalized is not None:
|
||||
exact = connection.execute(
|
||||
"SELECT * FROM bank_accounts WHERE account_number = ?",
|
||||
(normalized,),
|
||||
).fetchone()
|
||||
if exact is not None and is_identifiable(exact, day):
|
||||
hits.append((PRIORITY_EXACT_ACCOUNT, exact, "account_exact"))
|
||||
alias_rows = connection.execute(
|
||||
"""
|
||||
SELECT a.*, al.priority AS alias_priority,
|
||||
al.effective_from AS alias_from, al.effective_to AS alias_to
|
||||
FROM account_aliases al
|
||||
JOIN bank_accounts a ON a.id = al.bank_account_id
|
||||
WHERE al.alias_kind = 'account' AND al.alias_value = ?
|
||||
""",
|
||||
(normalized,),
|
||||
).fetchall()
|
||||
for alias in alias_rows:
|
||||
if _alias_in_window(alias, day) and is_identifiable(alias, day):
|
||||
hits.append(
|
||||
(PRIORITY_ACCOUNT_ALIAS + alias["alias_priority"], alias, "account_alias")
|
||||
)
|
||||
|
||||
normalized_name = re.sub(r"\s+", "", str(name or ""))
|
||||
if normalized_name:
|
||||
alias_rows = connection.execute(
|
||||
"""
|
||||
SELECT a.*, al.priority AS alias_priority,
|
||||
al.effective_from AS alias_from, al.effective_to AS alias_to
|
||||
FROM account_aliases al
|
||||
JOIN bank_accounts a ON a.id = al.bank_account_id
|
||||
WHERE al.alias_kind = 'name' AND al.alias_value = ?
|
||||
""",
|
||||
(normalized_name,),
|
||||
).fetchall()
|
||||
for alias in alias_rows:
|
||||
if _alias_in_window(alias, day) and is_identifiable(alias, day):
|
||||
hits.append(
|
||||
(PRIORITY_NAME_ALIAS + alias["alias_priority"], alias, "name_alias")
|
||||
)
|
||||
|
||||
hits.sort(key=lambda item: (item[0], item[1]["id"]))
|
||||
seen: set[int] = set()
|
||||
results: list[dict[str, object]] = []
|
||||
for rank, account, via in hits:
|
||||
if account["id"] in seen:
|
||||
continue
|
||||
seen.add(account["id"])
|
||||
results.append(
|
||||
{
|
||||
"bank_account_id": account["id"],
|
||||
"company_id": account["company_id"],
|
||||
"via": via,
|
||||
"priority": rank,
|
||||
}
|
||||
)
|
||||
return results
|
||||
|
||||
|
||||
def _alias_in_window(alias: sqlite3.Row, on_date: str) -> bool:
|
||||
if alias["alias_from"] and on_date < alias["alias_from"]:
|
||||
return False
|
||||
if alias["alias_to"] and on_date > alias["alias_to"]:
|
||||
return False
|
||||
return True
|
||||
@@ -0,0 +1,593 @@
|
||||
"""Tests for dynamic master data (companies, bank accounts, aliases).
|
||||
|
||||
Unit tests exercise normalization, masking, effective-window boundaries and
|
||||
alias-match priority against an in-memory database; the HTTP integration
|
||||
class drives a live server to cover the submission -> review -> disable
|
||||
workflow, uniqueness under concurrency, masking boundaries and RBAC.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
|
||||
from bank_importer import auth, master_data
|
||||
from bank_importer.db import connect, migrate
|
||||
|
||||
import server
|
||||
from test_server_auth import Client, as_json
|
||||
|
||||
|
||||
BOOTSTRAP_PASSWORD = "BootAdmin123"
|
||||
ADMIN_PASSWORD = "AdminPass123"
|
||||
CASHIER_PASSWORD = "Cashier123"
|
||||
|
||||
|
||||
class MasterDataUnitTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.connection = connect(":memory:")
|
||||
migrate(self.connection)
|
||||
now = master_data.utc_now()
|
||||
cursor = self.connection.execute(
|
||||
"INSERT INTO companies (name, created_at, updated_at) VALUES ('甲公司', ?, ?)",
|
||||
(now, now),
|
||||
)
|
||||
self.company_id = int(cursor.lastrowid)
|
||||
self.connection.commit()
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.connection.close()
|
||||
|
||||
def _active_account(self, number: str = "1234567890123") -> object:
|
||||
return master_data.submit_bank_account(
|
||||
self.connection,
|
||||
company_id=self.company_id,
|
||||
bank_name="中信银行",
|
||||
account_type="基本户",
|
||||
account_number=number,
|
||||
start_date="2026-01-01",
|
||||
actor=None,
|
||||
)
|
||||
|
||||
def test_normalize_account_number(self) -> None:
|
||||
self.assertEqual("1234567890", master_data.normalize_account_number("1234 5678-90"))
|
||||
self.assertEqual("1234567890", master_data.normalize_account_number("１２３４５６７８９０"))
|
||||
for invalid in ("", "123", "abc123456", "12345678901234567890123456789012345"):
|
||||
with self.assertRaises(ValueError):
|
||||
master_data.normalize_account_number(invalid)
|
||||
|
||||
def test_mask_account_number(self) -> None:
|
||||
self.assertEqual("****9012", master_data.mask_account_number("123456789012"))
|
||||
self.assertEqual("****123", master_data.mask_account_number("123"))
|
||||
|
||||
def test_is_usable_window_boundaries(self) -> None:
|
||||
account = self._active_account()
|
||||
account = master_data.review_bank_account(
|
||||
self.connection, account["id"], "approve", None, self._admin(),
|
||||
effective_from="2026-02-01",
|
||||
)
|
||||
self.assertFalse(master_data.is_usable(account, "2026-01-31"))
|
||||
self.assertTrue(master_data.is_usable(account, "2026-02-01"))
|
||||
# A scheduled effective_to bounds the usable window (inclusive).
|
||||
with self.connection:
|
||||
self.connection.execute(
|
||||
"UPDATE bank_accounts SET effective_to = '2026-06-30' WHERE id = ?",
|
||||
(account["id"],),
|
||||
)
|
||||
account = master_data.get_account(self.connection, account["id"])
|
||||
self.assertTrue(master_data.is_usable(account, "2026-06-30"))
|
||||
self.assertFalse(master_data.is_usable(account, "2026-07-01"))
|
||||
# Disabled accounts never accept new uploads, even inside the window.
|
||||
account = master_data.review_bank_account(
|
||||
self.connection, account["id"], "disable", "账户销户", self._admin(),
|
||||
effective_to="2026-06-30",
|
||||
)
|
||||
self.assertFalse(master_data.is_usable(account, "2026-06-30"))
|
||||
# …but they still identify historical rows inside the window.
|
||||
self.assertTrue(master_data.is_identifiable(account, "2026-06-30"))
|
||||
self.assertFalse(master_data.is_identifiable(account, "2026-07-01"))
|
||||
|
||||
def _admin(self):
|
||||
user = self.connection.execute(
|
||||
"SELECT * FROM users WHERE username = 'admin-u'"
|
||||
).fetchone()
|
||||
if user is None:
|
||||
auth.create_user(self.connection, "admin-u", "AdminPass123", "admin")
|
||||
user = self.connection.execute(
|
||||
"SELECT * FROM users WHERE username = 'admin-u'"
|
||||
).fetchone()
|
||||
return user
|
||||
|
||||
def test_pending_and_returned_accounts_never_usable(self) -> None:
|
||||
account = master_data.submit_bank_account(
|
||||
self.connection,
|
||||
company_id=self.company_id,
|
||||
bank_name="中信银行",
|
||||
account_type="一般户",
|
||||
account_number="9988776655",
|
||||
start_date="2026-01-01",
|
||||
actor=None,
|
||||
)
|
||||
self.assertFalse(master_data.is_usable(account, "2026-06-01"))
|
||||
account = master_data.review_bank_account(
|
||||
self.connection, account["id"], "return", "资料不全", self._admin()
|
||||
)
|
||||
self.assertEqual("returned", account["status"])
|
||||
self.assertFalse(master_data.is_usable(account, "2026-06-01"))
|
||||
|
||||
def test_alias_match_priority_and_window(self) -> None:
|
||||
exact_account = self._active_account("1111222233334")
|
||||
exact_account = master_data.review_bank_account(
|
||||
self.connection, exact_account["id"], "approve", None, self._admin(),
|
||||
effective_from="2026-01-01",
|
||||
)
|
||||
alias_account = master_data.submit_bank_account(
|
||||
self.connection,
|
||||
company_id=self.company_id,
|
||||
bank_name="建设银行",
|
||||
account_type="一般户",
|
||||
account_number="5555666677778",
|
||||
start_date="2026-01-01",
|
||||
actor=None,
|
||||
)
|
||||
alias_account = master_data.review_bank_account(
|
||||
self.connection, alias_account["id"], "approve", None, self._admin(),
|
||||
effective_from="2026-01-01",
|
||||
)
|
||||
master_data.add_alias(
|
||||
self.connection, alias_account["id"], "account", "9999000011112",
|
||||
priority=5, actor=None,
|
||||
)
|
||||
master_data.add_alias(
|
||||
self.connection, alias_account["id"], "name", "甲公司郑州分部",
|
||||
effective_from="2026-03-01", effective_to="2026-03-31", actor=None,
|
||||
)
|
||||
|
||||
# Exact account number beats the account alias tier.
|
||||
master_data.add_alias(
|
||||
self.connection, alias_account["id"], "account", "1111222233334",
|
||||
priority=1, actor=None,
|
||||
)
|
||||
hits = master_data.match_account(
|
||||
self.connection, account_number="1111-2222 3333 4", on_date="2026-06-01"
|
||||
)
|
||||
self.assertEqual(exact_account["id"], hits[0]["bank_account_id"])
|
||||
self.assertEqual("account_exact", hits[0]["via"])
|
||||
self.assertEqual(alias_account["id"], hits[1]["bank_account_id"])
|
||||
self.assertEqual("account_alias", hits[1]["via"])
|
||||
|
||||
# Name alias only matches inside its effective window.
|
||||
hits = master_data.match_account(
|
||||
self.connection, name="甲公司郑州分部 ", on_date="2026-03-15"
|
||||
)
|
||||
self.assertEqual(alias_account["id"], hits[0]["bank_account_id"])
|
||||
self.assertEqual("name_alias", hits[0]["via"])
|
||||
self.assertEqual(
|
||||
[],
|
||||
master_data.match_account(
|
||||
self.connection, name="甲公司郑州分部", on_date="2026-04-01"
|
||||
),
|
||||
)
|
||||
|
||||
def test_match_never_returns_pending_or_disabled(self) -> None:
|
||||
pending = master_data.submit_bank_account(
|
||||
self.connection,
|
||||
company_id=self.company_id,
|
||||
bank_name="郑州银行",
|
||||
account_type="一般户",
|
||||
account_number="7777888899990",
|
||||
actor=None,
|
||||
)
|
||||
self.assertEqual(
|
||||
[],
|
||||
master_data.match_account(self.connection, account_number="7777888899990"),
|
||||
)
|
||||
account = master_data.review_bank_account(
|
||||
self.connection, pending["id"], "approve", None, self._admin(),
|
||||
effective_from="2026-01-01",
|
||||
)
|
||||
self.assertTrue(
|
||||
master_data.match_account(self.connection, account_number="7777888899990")
|
||||
)
|
||||
master_data.review_bank_account(
|
||||
self.connection, account["id"], "disable", "销户", self._admin(),
|
||||
effective_to="2026-06-30",
|
||||
)
|
||||
# Disabled accounts still resolve dates inside their effective window…
|
||||
self.assertTrue(
|
||||
master_data.match_account(
|
||||
self.connection, account_number="7777888899990", on_date="2026-06-30"
|
||||
)
|
||||
)
|
||||
# …and never resolve dates after it.
|
||||
self.assertEqual(
|
||||
[],
|
||||
master_data.match_account(
|
||||
self.connection, account_number="7777888899990", on_date="2026-07-01"
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
class MasterDataApiTests(unittest.TestCase):
|
||||
"""Live-server workflow tests for account registration and review."""
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls) -> None:
|
||||
import os
|
||||
|
||||
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)
|
||||
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
|
||||
|
||||
# Company A created together with its cashier login in one call.
|
||||
status, _, data = cls.admin.post_json(
|
||||
"/api/admin/companies",
|
||||
{"name": "甲公司", "credit_code": "91410100TEST", "cashier_name": "牛女士",
|
||||
"username": "cashier-a"},
|
||||
)
|
||||
assert status == 200, data
|
||||
payload = as_json(data)
|
||||
cls.company_a = payload["company_id"]
|
||||
assert payload["initial_password"] == "cashier-a"
|
||||
|
||||
status, _, data = cls.admin.post_json("/api/admin/companies", {"name": "乙公司"})
|
||||
assert status == 200, data
|
||||
cls.company_b = as_json(data)["company_id"]
|
||||
status, _, data = cls.admin.post_json(
|
||||
"/api/admin/users", {"username": "cashier-b", "company_id": cls.company_b}
|
||||
)
|
||||
assert status == 200, data
|
||||
|
||||
cls.cashier_a = cls._login_company_user("cashier-a", "cashier-a")
|
||||
cls.cashier_b = cls._login_company_user("cashier-b", "cashier-b")
|
||||
|
||||
@classmethod
|
||||
def _login_company_user(cls, username: str, initial: str) -> Client:
|
||||
client = Client("127.0.0.1", cls.port)
|
||||
status, _, data = client.post_json(
|
||||
"/api/login", {"username": username, "password": initial, "portal": "company"}
|
||||
)
|
||||
assert status == 200, data
|
||||
status, _, data = client.post_json(
|
||||
"/api/password/change",
|
||||
{"old_password": initial, "new_password": CASHIER_PASSWORD},
|
||||
)
|
||||
assert status == 200, data
|
||||
return client
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls) -> None:
|
||||
import os
|
||||
|
||||
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 submit_account(self, client: Client, number: str, bank: str = "郑州银行"):
|
||||
return client.post_json(
|
||||
"/api/company/accounts",
|
||||
{"bank_name": bank, "account_type": "一般户",
|
||||
"account_number": number, "start_date": "2026-08-01"},
|
||||
)
|
||||
|
||||
def review(self, account_id: int, decision: str, reason: str = ""):
|
||||
return self.admin.post_json(
|
||||
f"/api/admin/accounts/{account_id}/review",
|
||||
{"decision": decision, "reason": reason},
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Company management
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_new_company_appears_in_admin_list_with_user(self) -> None:
|
||||
status, _, data = self.admin.get("/api/admin/companies")
|
||||
self.assertEqual(200, status)
|
||||
companies = {c["name"]: c for c in as_json(data)["companies"]}
|
||||
self.assertIn("甲公司", companies)
|
||||
self.assertEqual("91410100TEST", companies["甲公司"]["credit_code"])
|
||||
self.assertEqual("牛女士", companies["甲公司"]["cashier_name"])
|
||||
self.assertIn("cashier-a", companies["甲公司"]["usernames"])
|
||||
self.assertEqual(0, companies["乙公司"]["account_count"])
|
||||
|
||||
def test_company_user_cannot_change_own_company_binding(self) -> None:
|
||||
# A company_id in the submission body is ignored: the account is
|
||||
# always bound to the session company.
|
||||
status, _, data = self.cashier_a.post_json(
|
||||
"/api/company/accounts",
|
||||
{"bank_name": "民生银行", "account_type": "一般户",
|
||||
"account_number": "6001000100010001", "company_id": self.company_b},
|
||||
)
|
||||
self.assertEqual(200, status, data)
|
||||
self.assertEqual(self.company_a, as_json(data)["account"]["company_id"])
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Account registration -> review workflow
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_full_registration_workflow(self) -> None:
|
||||
status, _, data = self.submit_account(self.cashier_a, "6222 0210-0100 8888")
|
||||
self.assertEqual(200, status, data)
|
||||
account = as_json(data)["account"]
|
||||
account_id = account["id"]
|
||||
self.assertEqual("pending", account["status"])
|
||||
# Company view is masked only.
|
||||
self.assertEqual("****8888", account["account_number_masked"])
|
||||
self.assertNotIn("account_number", account)
|
||||
self.assertFalse(account.get("usable", True))
|
||||
|
||||
# Admin audit view shows the full normalized number.
|
||||
status, _, data = self.admin.get("/api/admin/accounts")
|
||||
self.assertEqual(200, status)
|
||||
full = [a for a in as_json(data)["accounts"] if a["id"] == account_id][0]
|
||||
self.assertEqual("62220210010088 88".replace(" ", ""), full["account_number"])
|
||||
self.assertEqual("甲公司", full["company_name"])
|
||||
|
||||
# Pending account cannot upload.
|
||||
sample = Path("流水模板/中信银行账户流水.xlsx")
|
||||
status, _, data = self.cashier_a.post_multipart(
|
||||
"/api/parse", {"bank_account_id": str(account_id)},
|
||||
sample.name, sample.read_bytes(),
|
||||
)
|
||||
self.assertEqual(409, status, data)
|
||||
|
||||
# Approve -> active and usable.
|
||||
status, _, data = self.review(account_id, "approve")
|
||||
self.assertEqual(200, status, data)
|
||||
self.assertEqual("active", as_json(data)["account"]["status"])
|
||||
self.assertEqual("2026-08-01", as_json(data)["account"]["effective_from"])
|
||||
|
||||
status, _, data = self.cashier_a.get("/api/company/accounts")
|
||||
mine = [a for a in as_json(data)["accounts"] if a["id"] == account_id][0]
|
||||
self.assertTrue(mine["usable"])
|
||||
|
||||
# Approved account accepts uploads.
|
||||
status, _, data = self.cashier_a.post_multipart(
|
||||
"/api/parse", {"bank_account_id": str(account_id)},
|
||||
sample.name, sample.read_bytes(),
|
||||
)
|
||||
self.assertEqual(200, status, data)
|
||||
|
||||
# Disable -> leaves the usable window.
|
||||
status, _, data = self.review(account_id, "disable", "账户销户")
|
||||
self.assertEqual(200, status, data)
|
||||
disabled = as_json(data)["account"]
|
||||
self.assertEqual("disabled", disabled["status"])
|
||||
self.assertIsNotNone(disabled["effective_to"])
|
||||
|
||||
status, _, data = self.cashier_a.get("/api/company/accounts")
|
||||
mine = [a for a in as_json(data)["accounts"] if a["id"] == account_id][0]
|
||||
self.assertFalse(mine["usable"])
|
||||
|
||||
def test_return_and_resubmit_reuses_same_row(self) -> None:
|
||||
status, _, data = self.submit_account(self.cashier_a, "3100998877665")
|
||||
self.assertEqual(200, status, data)
|
||||
account_id = as_json(data)["account"]["id"]
|
||||
|
||||
status, _, data = self.review(account_id, "return", "开户许可证模糊")
|
||||
self.assertEqual(200, status, data)
|
||||
self.assertEqual("returned", as_json(data)["account"]["status"])
|
||||
|
||||
# Return requires a reason.
|
||||
status, _, data = self.submit_account(self.cashier_a, "3100998877666")
|
||||
other_id = as_json(data)["account"]["id"]
|
||||
status, _, _ = self.review(other_id, "return")
|
||||
self.assertEqual(400, status)
|
||||
|
||||
# Resubmission reopens the same row as pending.
|
||||
status, _, data = self.submit_account(self.cashier_a, "3100 9988-7766 5")
|
||||
self.assertEqual(200, status, data)
|
||||
resubmitted = as_json(data)["account"]
|
||||
self.assertEqual(account_id, resubmitted["id"])
|
||||
self.assertEqual("pending", resubmitted["status"])
|
||||
|
||||
# The whole trail is auditable with actor, time and reason.
|
||||
status, _, data = self.admin.get(
|
||||
f"/api/admin/master-changes?entity_type=bank_account&entity_id={account_id}"
|
||||
)
|
||||
self.assertEqual(200, status)
|
||||
changes = as_json(data)["changes"]
|
||||
actions = [change["action"] for change in changes]
|
||||
self.assertEqual(["resubmit", "return", "submit"], actions)
|
||||
returned = [c for c in changes if c["action"] == "return"][0]
|
||||
self.assertEqual("开户许可证模糊", returned["reason"])
|
||||
self.assertEqual("group-admin", returned["actor_username"])
|
||||
self.assertIn('"pending"', returned["before_json"])
|
||||
self.assertIn('"returned"', returned["after_json"])
|
||||
|
||||
def test_duplicate_normalized_number_conflicts(self) -> None:
|
||||
status, _, _ = self.submit_account(self.cashier_a, "4501111222233")
|
||||
self.assertEqual(200, status)
|
||||
# Same company, same number with different separators.
|
||||
status, _, data = self.submit_account(self.cashier_a, "4501-1112 2223-3")
|
||||
self.assertEqual(409, status, data)
|
||||
# Another company registering the same number conflicts too.
|
||||
status, _, data = self.submit_account(self.cashier_b, "4501111222233")
|
||||
self.assertEqual(409, status, data)
|
||||
self.assertIn("其他公司", as_json(data)["message"])
|
||||
|
||||
def test_concurrent_submissions_create_only_one_account(self) -> None:
|
||||
number = "8800123456789"
|
||||
results: list[int] = []
|
||||
|
||||
def submit() -> None:
|
||||
client = Client("127.0.0.1", self.port)
|
||||
client.cookies.update(self.cashier_a.cookies)
|
||||
status, _, _ = client.post_json(
|
||||
"/api/company/accounts",
|
||||
{"bank_name": "中信银行", "account_type": "基本户",
|
||||
"account_number": number},
|
||||
)
|
||||
results.append(status)
|
||||
|
||||
threads = [threading.Thread(target=submit) for _ in range(2)]
|
||||
for thread in threads:
|
||||
thread.start()
|
||||
for thread in threads:
|
||||
thread.join()
|
||||
self.assertEqual(sorted(results), [200, 409])
|
||||
|
||||
connection = connect(self.db_path)
|
||||
try:
|
||||
count = connection.execute(
|
||||
"SELECT COUNT(*) AS n FROM bank_accounts WHERE account_number = ?",
|
||||
(number,),
|
||||
).fetchone()["n"]
|
||||
finally:
|
||||
connection.close()
|
||||
self.assertEqual(1, count)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# RBAC and masking
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_company_user_forbidden_on_admin_account_endpoints(self) -> None:
|
||||
for call in (
|
||||
lambda: self.cashier_a.get("/api/admin/accounts"),
|
||||
lambda: self.cashier_a.post_json(
|
||||
"/api/admin/accounts/1/review", {"decision": "approve"}
|
||||
),
|
||||
lambda: self.cashier_a.get("/api/admin/accounts/1/aliases"),
|
||||
lambda: self.cashier_a.post_json(
|
||||
"/api/admin/accounts/1/aliases",
|
||||
{"alias_kind": "name", "alias_value": "x"},
|
||||
),
|
||||
lambda: self.cashier_a.get("/api/admin/master-changes"),
|
||||
):
|
||||
status, _, data = call()
|
||||
self.assertEqual(403, status, data)
|
||||
|
||||
def test_admin_cannot_use_company_account_endpoint(self) -> None:
|
||||
status, _, _ = self.admin.get("/api/company/accounts")
|
||||
self.assertEqual(403, status)
|
||||
status, _, _ = self.submit_account(self.admin, "1000200030004")
|
||||
self.assertEqual(403, status)
|
||||
|
||||
def test_unauthenticated_master_data_calls_return_401(self) -> None:
|
||||
anon = Client("127.0.0.1", self.port)
|
||||
for call in (
|
||||
lambda: anon.get("/api/company/accounts"),
|
||||
lambda: anon.get("/api/admin/accounts"),
|
||||
lambda: anon.post_json("/api/company/accounts", {}),
|
||||
lambda: anon.get("/api/admin/master-changes"),
|
||||
):
|
||||
status, _, _ = call()
|
||||
self.assertEqual(401, status)
|
||||
|
||||
def test_company_account_list_scoped_and_masked(self) -> None:
|
||||
status, _, data = self.cashier_b.get("/api/company/accounts")
|
||||
self.assertEqual(200, status)
|
||||
for account in as_json(data)["accounts"]:
|
||||
self.assertEqual(self.company_b, account["company_id"])
|
||||
self.assertNotIn("account_number", account)
|
||||
self.assertTrue(account["account_number_masked"].startswith("****"))
|
||||
|
||||
def test_upload_with_other_companys_account_returns_404(self) -> None:
|
||||
status, _, data = self.submit_account(self.cashier_a, "5550001112223")
|
||||
account_id = as_json(data)["account"]["id"]
|
||||
status, _, _ = self.review(account_id, "approve")
|
||||
self.assertEqual(200, status)
|
||||
sample = Path("流水模板/中信银行账户流水.xlsx")
|
||||
status, _, data = self.cashier_b.post_multipart(
|
||||
"/api/parse", {"bank_account_id": str(account_id)},
|
||||
sample.name, sample.read_bytes(),
|
||||
)
|
||||
self.assertEqual(404, status, data)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Aliases via API
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_alias_crud_and_audit(self) -> None:
|
||||
status, _, data = self.submit_account(self.cashier_a, "6601234509876")
|
||||
account_id = as_json(data)["account"]["id"]
|
||||
status, _, _ = self.review(account_id, "approve")
|
||||
self.assertEqual(200, status)
|
||||
|
||||
status, _, data = self.admin.post_json(
|
||||
f"/api/admin/accounts/{account_id}/aliases",
|
||||
{"alias_kind": "name", "alias_value": "甲公司工会",
|
||||
"effective_from": "2026-01-01", "effective_to": "2026-12-31",
|
||||
"priority": 10},
|
||||
)
|
||||
self.assertEqual(200, status, data)
|
||||
alias_id = as_json(data)["alias_id"]
|
||||
|
||||
# Duplicate alias on the same account conflicts.
|
||||
status, _, _ = self.admin.post_json(
|
||||
f"/api/admin/accounts/{account_id}/aliases",
|
||||
{"alias_kind": "name", "alias_value": "甲公司工会"},
|
||||
)
|
||||
self.assertEqual(409, status)
|
||||
|
||||
status, _, data = self.admin.get(f"/api/admin/accounts/{account_id}/aliases")
|
||||
self.assertEqual(200, status)
|
||||
aliases = as_json(data)["aliases"]
|
||||
self.assertEqual(1, len(aliases))
|
||||
self.assertEqual("甲公司工会", aliases[0]["alias_value"])
|
||||
self.assertEqual(10, aliases[0]["priority"])
|
||||
|
||||
status, _, data = self.admin.get(
|
||||
f"/api/admin/master-changes?entity_type=account_alias&entity_id={alias_id}"
|
||||
)
|
||||
changes = as_json(data)["changes"]
|
||||
self.assertEqual("create", changes[0]["action"])
|
||||
self.assertIn("2026-12-31", changes[0]["after_json"])
|
||||
|
||||
# Invalid alias input is rejected, unknown account is 404.
|
||||
status, _, _ = self.admin.post_json(
|
||||
f"/api/admin/accounts/{account_id}/aliases",
|
||||
{"alias_kind": "account", "alias_value": "abc"},
|
||||
)
|
||||
self.assertEqual(400, status)
|
||||
status, _, _ = self.admin.post_json(
|
||||
"/api/admin/accounts/99999/aliases",
|
||||
{"alias_kind": "name", "alias_value": "x"},
|
||||
)
|
||||
self.assertEqual(404, status)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user