464 lines
20 KiB
Python
464 lines
20 KiB
Python
"""HTTP integration tests for the canonical transfer event APIs (B-43).
|
|
|
|
Covers the admin event list/detail, manual decisions and reconcile endpoints,
|
|
personal transit mapping workflow, upload-account persistence, and the
|
|
company-side read scoping with masked counterparty evidence. Uses a real
|
|
``ThreadingHTTPServer`` like ``test_server_auth``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
|
|
from openpyxl import Workbook
|
|
|
|
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"
|
|
|
|
CCB_HEADER = [
|
|
"客户账号", "账户名称", "交易时间", "借方发生额（支取）", "贷方发生额（收入）",
|
|
"余额", "币种", "对方户名", "对方账号", "对方开户机构", "摘要", "备注",
|
|
]
|
|
|
|
ACCOUNT_A = "6222000000000001"
|
|
ACCOUNT_B = "6222000000000002"
|
|
ACCOUNT_C = "6222000000000003"
|
|
|
|
|
|
def workbook_bytes(rows) -> bytes:
|
|
workbook = Workbook()
|
|
sheet = workbook.active
|
|
sheet.title = "正常流水"
|
|
sheet.append(CCB_HEADER)
|
|
for row in rows:
|
|
sheet.append(row)
|
|
buffer = io.BytesIO()
|
|
workbook.save(buffer)
|
|
return buffer.getvalue()
|
|
|
|
|
|
def outgoing(own: str, cp: str, amount: str, at: str = "2026-01-05 10:00:00"):
|
|
return [own, "测试公司", at, amount, "", "50000.00", "RMB", "对方", cp, "某银行", "货款", ""]
|
|
|
|
|
|
def incoming(own: str, cp: str, amount: str, at: str = "2026-01-05 11:00:00"):
|
|
return [own, "测试公司", at, "", amount, "50000.00", "RMB", "对方", cp, "某银行", "收款", ""]
|
|
|
|
|
|
class MatchingApiTests(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)
|
|
assert server.ensure_bootstrap_admin(connection) is None
|
|
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
|
|
|
|
cls.company_a = cls._create_company("甲公司", "cashier-a")
|
|
cls.company_b = cls._create_company("乙公司", "cashier-b")
|
|
cls.company_c = cls._create_company("丙公司", "cashier-c")
|
|
cls.cashier_a = cls._login_company("cashier-a", cls.company_a)
|
|
cls.cashier_b = cls._login_company("cashier-b", cls.company_b)
|
|
cls.cashier_c = cls._login_company("cashier-c", cls.company_c)
|
|
|
|
cls.account_a = cls._approve_account(cls.company_a, ACCOUNT_A)
|
|
cls.account_b = cls._approve_account(cls.company_b, ACCOUNT_B)
|
|
cls.account_c = cls._approve_account(cls.company_c, ACCOUNT_C)
|
|
|
|
# One A<->B matched event and one B<->C matched event.
|
|
cls._upload_and_confirm(cls.cashier_a, cls.company_a, [outgoing(ACCOUNT_A, ACCOUNT_B, "100.00")])
|
|
cls._upload_and_confirm(cls.cashier_b, cls.company_b, [incoming(ACCOUNT_B, ACCOUNT_A, "100.00")])
|
|
cls._upload_and_confirm(cls.cashier_b, cls.company_b, [outgoing(ACCOUNT_B, ACCOUNT_C, "200.00")])
|
|
cls._upload_and_confirm(cls.cashier_c, cls.company_c, [incoming(ACCOUNT_C, ACCOUNT_B, "200.00")])
|
|
|
|
@classmethod
|
|
def _create_company(cls, name: str, username: str) -> int:
|
|
status, _, data = cls.admin.post_json(
|
|
"/api/admin/companies",
|
|
{"name": name, "username": username},
|
|
)
|
|
assert status == 200, data
|
|
cls.initial_passwords.setdefault(username, as_json(data)["initial_password"])
|
|
return as_json(data)["company_id"]
|
|
|
|
initial_passwords: dict[str, str] = {}
|
|
|
|
@classmethod
|
|
def _login_company(cls, username: str, company_id: int) -> Client:
|
|
client = Client("127.0.0.1", cls.port)
|
|
initial = cls.initial_passwords[username]
|
|
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 _approve_account(cls, company_id: int, number: str) -> int:
|
|
# Build the account through company submission + admin review.
|
|
company_client = {
|
|
cls.company_a: cls.cashier_a,
|
|
cls.company_b: cls.cashier_b,
|
|
cls.company_c: cls.cashier_c,
|
|
}[company_id]
|
|
status, _, data = company_client.post_json(
|
|
"/api/company/accounts",
|
|
{"bank_name": "中信银行", "account_type": "基本户",
|
|
"account_number": number, "start_date": "2026-01-01"},
|
|
)
|
|
assert status == 200, data
|
|
account_id = as_json(data)["account"]["id"]
|
|
status, _, data = cls.admin.post_json(
|
|
f"/api/admin/accounts/{account_id}/review",
|
|
{"decision": "approve", "reason": "测试启用",
|
|
"effective_from": "2026-01-01"},
|
|
)
|
|
assert status == 200, data
|
|
return account_id
|
|
|
|
@classmethod
|
|
def _upload_and_confirm(cls, client, company_id: int, rows) -> int:
|
|
content = workbook_bytes(rows)
|
|
status, _, data = cls.admin.post_multipart(
|
|
"/api/parse", {"company_id": str(company_id)}, "账单.xlsx", content
|
|
)
|
|
assert status == 200, data
|
|
batch_id = as_json(data)["batch_id"]
|
|
status, _, data = client.get(f"/api/batches/{batch_id}/sheets")
|
|
assert status == 200, data
|
|
names = [s["sheet_name"] for s in as_json(data)["sheets"] if s["outcome"] == "parsed"]
|
|
status, _, data = client.post_json(
|
|
f"/api/batches/{batch_id}/confirm", {"sheets": names}
|
|
)
|
|
assert status == 200, data
|
|
return batch_id
|
|
|
|
@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 events(self, client) -> list[dict]:
|
|
status, _, data = client.get("/api/admin/transfer-events")
|
|
self.assertEqual(200, status, data)
|
|
return as_json(data)["events"]
|
|
|
|
def company_events(self, client) -> list[dict]:
|
|
status, _, data = client.get("/api/company/transfer-events")
|
|
self.assertEqual(200, status, data)
|
|
return as_json(data)["events"]
|
|
|
|
# ------------------------------------------------------------------
|
|
# Admin event views
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_admin_lists_matched_events(self) -> None:
|
|
events = self.events(self.admin)
|
|
matched = [e for e in events if e["status"] == "matched"]
|
|
self.assertGreaterEqual(len(matched), 2)
|
|
by_amount = {e["amount"]: e for e in matched}
|
|
self.assertEqual(self.company_a, by_amount["100.00"]["payer_company_id"])
|
|
self.assertEqual(self.company_b, by_amount["100.00"]["payee_company_id"])
|
|
self.assertEqual("paired", by_amount["100.00"]["pairing"])
|
|
self.assertEqual(2, by_amount["100.00"]["evidence_count"])
|
|
|
|
def test_admin_event_detail_has_history_and_observations(self) -> None:
|
|
events = self.events(self.admin)
|
|
event = next(e for e in events if e["status"] == "matched")
|
|
status, _, data = self.admin.get(f"/api/admin/transfer-events/{event['event_id']}")
|
|
self.assertEqual(200, status, data)
|
|
detail = as_json(data)["event"]
|
|
self.assertEqual("matched", detail["status"])
|
|
self.assertEqual(2, len(detail["observations"]))
|
|
self.assertTrue(detail["history"])
|
|
self.assertTrue(detail["candidates"])
|
|
|
|
def test_admin_match_exceptions(self) -> None:
|
|
status, _, data = self.admin.get("/api/admin/match-exceptions")
|
|
self.assertEqual(200, status, data)
|
|
self.assertIsInstance(as_json(data)["exceptions"], list)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Admin manual decisions and reconcile via API
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_admin_reconcile_by_batch_is_idempotent(self) -> None:
|
|
# Re-running reconcile on an already settled batch changes nothing.
|
|
status, _, data = self.admin.post_json(
|
|
"/api/admin/transfer-events/reconcile", {"batch_id": 1}
|
|
)
|
|
self.assertEqual(200, status, data)
|
|
result = as_json(data)["matching"]
|
|
self.assertEqual(0, result["created_events"])
|
|
self.assertEqual(0, result["updated_events"])
|
|
|
|
def test_admin_confirm_single_and_reverse_via_api(self) -> None:
|
|
# A uploads its side only -> internal_single.
|
|
status, _, data = self.admin.post_multipart(
|
|
"/api/parse", {"company_id": str(self.company_a)},
|
|
"单边.xlsx", workbook_bytes([outgoing(ACCOUNT_A, ACCOUNT_B, "300.00", "2026-02-01 10:00:00")]),
|
|
)
|
|
self.assertEqual(200, status, data)
|
|
batch_id = as_json(data)["batch_id"]
|
|
status, _, data = self.cashier_a.get(f"/api/batches/{batch_id}/sheets")
|
|
names = [s["sheet_name"] for s in as_json(data)["sheets"] if s["outcome"] == "parsed"]
|
|
status, _, data = self.cashier_a.post_json(
|
|
f"/api/batches/{batch_id}/confirm", {"sheets": names}
|
|
)
|
|
self.assertEqual(200, status, data)
|
|
events = self.events(self.admin)
|
|
single = next(e for e in events if e["status"] == "internal_single" and e["amount"] == "300.00")
|
|
detail = self._admin_detail(single["event_id"])
|
|
revision = detail["revision"]
|
|
|
|
# Company users cannot write manual decisions.
|
|
status, _, data = self.cashier_a.post_json(
|
|
f"/api/admin/transfer-events/{single['event_id']}/decisions",
|
|
{"action": "reverse", "reason": "不应允许", "expected_revision": revision},
|
|
)
|
|
self.assertEqual(403, status, data)
|
|
|
|
# Admin confirms the single event based on evidence.
|
|
status, _, data = self.admin.post_json(
|
|
f"/api/admin/transfer-events/{single['event_id']}/decisions",
|
|
{"action": "assign_participant", "reason": "函证确认",
|
|
"expected_revision": revision, "request_key": "confirm-300",
|
|
"participant": {"role": "payee", "company_id": self.company_b}},
|
|
)
|
|
self.assertEqual(200, status, data)
|
|
decision = as_json(data)["decision"]
|
|
self.assertEqual("intercompany", decision["classification"])
|
|
self.assertTrue(decision["locked"])
|
|
|
|
# Replaying the same request key is idempotent.
|
|
status, _, data = self.admin.post_json(
|
|
f"/api/admin/transfer-events/{single['event_id']}/decisions",
|
|
{"action": "assign_participant", "reason": "函证确认",
|
|
"expected_revision": revision, "request_key": "confirm-300",
|
|
"participant": {"role": "payee", "company_id": self.company_b}},
|
|
)
|
|
self.assertEqual(200, status, data)
|
|
|
|
# Stale revision conflicts.
|
|
status, _, data = self.admin.post_json(
|
|
f"/api/admin/transfer-events/{single['event_id']}/decisions",
|
|
{"action": "reverse", "reason": "撤销", "expected_revision": revision},
|
|
)
|
|
self.assertEqual(409, status, data)
|
|
|
|
def _admin_detail(self, event_id: int) -> dict:
|
|
status, _, data = self.admin.get(f"/api/admin/transfer-events/{event_id}")
|
|
self.assertEqual(200, status, data)
|
|
return as_json(data)["event"]
|
|
|
|
# ------------------------------------------------------------------
|
|
# Company scope and masking
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_company_sees_only_events_it_participates_in(self) -> None:
|
|
events = self.company_events(self.cashier_a)
|
|
self.assertTrue(events)
|
|
for event in events:
|
|
self.assertEqual(self.company_a, event["own_company_id"])
|
|
matched = [e for e in events if e["status"] == "matched"]
|
|
self.assertTrue(matched)
|
|
self.assertEqual(
|
|
{"乙公司"},
|
|
{e["counterparty_company_name"] for e in matched},
|
|
)
|
|
# B<->C event is invisible to A.
|
|
status, _, data = self.cashier_a.get("/api/company/transfer-events")
|
|
bc_events = [
|
|
e for e in as_json(data)["events"]
|
|
if e["counterparty_company_name"] == "丙公司"
|
|
]
|
|
self.assertEqual([], bc_events)
|
|
|
|
def test_company_detail_masks_counterparty_and_own_rows_only(self) -> None:
|
|
events = self.company_events(self.cashier_a)
|
|
event = next(e for e in events if e["status"] == "matched")
|
|
status, _, data = self.cashier_a.get(
|
|
f"/api/company/transfer-events/{event['event_id']}"
|
|
)
|
|
self.assertEqual(200, status, data)
|
|
detail = as_json(data)["event"]
|
|
self.assertEqual("乙公司", detail["counterparty"]["company_name"])
|
|
self.assertTrue(detail["counterparty"].get("account_number_masked", "").startswith("****"))
|
|
# The counterparty's full account number never leaves the server.
|
|
self.assertNotIn(ACCOUNT_B, json.dumps(detail, ensure_ascii=False))
|
|
# Only A's own observation rows are exposed.
|
|
self.assertEqual(1, len(detail["observations"]))
|
|
observation = detail["observations"][0]
|
|
self.assertTrue(observation["own_account_masked"].startswith("****"))
|
|
self.assertEqual(self.company_a, observation.get("batch_company_id"))
|
|
|
|
def test_company_cannot_read_event_it_does_not_participate_in(self) -> None:
|
|
# The B<->C event id comes from B's own list; A probing it returns 404.
|
|
status, _, data = self.cashier_b.get("/api/company/transfer-events")
|
|
bc_event_ids = [
|
|
e["event_id"] for e in as_json(data)["events"]
|
|
if e["counterparty_company_name"] == "丙公司"
|
|
]
|
|
self.assertTrue(bc_event_ids)
|
|
for event_id in bc_event_ids:
|
|
status, _, data = self.cashier_a.get(f"/api/company/transfer-events/{event_id}")
|
|
self.assertEqual(404, status, data)
|
|
status, _, data = self.cashier_a.get(f"/api/admin/transfer-events/{event_id}")
|
|
self.assertEqual(403, status, data)
|
|
|
|
def test_company_match_exceptions_scoped_to_own_company(self) -> None:
|
|
status, _, data = self.admin.post_multipart(
|
|
"/api/parse", {"company_id": str(self.company_a)},
|
|
"未决.xlsx",
|
|
workbook_bytes([outgoing(ACCOUNT_A, "9999999999999999", "77.00", "2026-02-05 10:00:00")]),
|
|
)
|
|
self.assertEqual(200, status, data)
|
|
batch_id = as_json(data)["batch_id"]
|
|
status, _, data = self.cashier_a.get(f"/api/batches/{batch_id}/sheets")
|
|
names = [s["sheet_name"] for s in as_json(data)["sheets"] if s["outcome"] == "parsed"]
|
|
status, _, data = self.cashier_a.post_json(
|
|
f"/api/batches/{batch_id}/confirm", {"sheets": names}
|
|
)
|
|
self.assertEqual(200, status, data)
|
|
|
|
status, _, data = self.cashier_a.get("/api/company/match-exceptions")
|
|
self.assertEqual(200, status, data)
|
|
mine = [e for e in as_json(data)["exceptions"] if e["amount"] == "77.00"]
|
|
self.assertEqual(1, len(mine))
|
|
self.assertEqual("unresolved", mine[0]["status"])
|
|
|
|
status, _, data = self.cashier_b.get("/api/company/match-exceptions")
|
|
self.assertEqual(200, status, data)
|
|
self.assertEqual(
|
|
[],
|
|
[e for e in as_json(data)["exceptions"] if e["amount"] == "77.00"],
|
|
)
|
|
|
|
def test_company_forbidden_on_admin_event_endpoints(self) -> None:
|
|
for call in (
|
|
lambda: self.cashier_a.get("/api/admin/transfer-events"),
|
|
lambda: self.cashier_a.get("/api/admin/match-exceptions"),
|
|
lambda: self.cashier_a.post_json(
|
|
"/api/admin/transfer-events/reconcile", {"batch_id": 1}
|
|
),
|
|
lambda: self.cashier_a.post_json(
|
|
"/api/admin/personal-transit-mappings", {}
|
|
),
|
|
lambda: self.cashier_a.get("/api/admin/personal-transit-mappings"),
|
|
):
|
|
status, _, data = call()
|
|
self.assertEqual(403, status, data)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Personal transit mappings
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_personal_transit_mapping_workflow(self) -> None:
|
|
status, _, data = self.admin.post_json(
|
|
"/api/admin/personal-transit-mappings",
|
|
{"account_number": "880088008800", "account_name": "张个人",
|
|
"represented_company_id": self.company_b,
|
|
"allowed_direction": "incoming", "effective_from": "2026-01-01"},
|
|
)
|
|
self.assertEqual(200, status, data)
|
|
mapping_id = as_json(data)["mapping"]["id"]
|
|
self.assertEqual("pending", as_json(data)["mapping"]["status"])
|
|
|
|
status, _, data = self.admin.post_json(
|
|
f"/api/admin/personal-transit-mappings/{mapping_id}/review",
|
|
{"decision": "approve", "reason": "资料齐全", "effective_from": "2026-01-01"},
|
|
)
|
|
self.assertEqual(200, status, data)
|
|
self.assertEqual("active", as_json(data)["mapping"]["status"])
|
|
|
|
status, _, data = self.admin.get("/api/admin/personal-transit-mappings")
|
|
self.assertEqual(200, status, data)
|
|
mappings = as_json(data)["mappings"]
|
|
self.assertTrue(any(m["id"] == mapping_id for m in mappings))
|
|
|
|
# Duplicate account number conflicts.
|
|
status, _, data = self.admin.post_json(
|
|
"/api/admin/personal-transit-mappings",
|
|
{"account_number": "880088008800", "account_name": "张个人",
|
|
"represented_company_id": self.company_b, "allowed_direction": "both"},
|
|
)
|
|
self.assertEqual(409, status, data)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Upload account persistence
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_upload_persists_approved_bank_account(self) -> None:
|
|
content = workbook_bytes([outgoing(ACCOUNT_A, ACCOUNT_B, "55.00", "2026-03-01 10:00:00")])
|
|
status, _, data = self.cashier_a.post_multipart(
|
|
"/api/parse", {"bank_account_id": str(self.account_a)},
|
|
"带账户.xlsx", content,
|
|
)
|
|
self.assertEqual(200, status, data)
|
|
batch_id = as_json(data)["batch_id"]
|
|
connection = connect(self.db_path)
|
|
try:
|
|
row = connection.execute(
|
|
"SELECT upload_bank_account_id FROM import_batches WHERE id = ?",
|
|
(batch_id,),
|
|
).fetchone()
|
|
finally:
|
|
connection.close()
|
|
self.assertEqual(self.account_a, row["upload_bank_account_id"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|