leefer/caiwuzongzhang
1
0
Fork 0
Code Issues Pull Requests 4 Actions Packages Projects Releases Wiki Activity
Files
b27016d71dc1d4df4a8cc18bf9785191c84efe99
caiwuzongzhang/tests/test_company_transfer_events.py
T
总工Cursormultica-agent
 40f7a91a0f HEL-176: 公司端转账往来明细筛选与导出 
新增 events 组合筛选 + keyset 分页，以及仅导出已确认明细的 CSV（含审计）；
与 B-44 ledger /events 按查询参数分发，不改库表与确认语义。

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: multica-agent <github@multica.ai>
2026-08-27 13:14:02 +00:00

501 lines
20 KiB
Python
Raw Blame History

"""HTTP tests for company intercompany events list + CSV export (HEL-176).
Covers combined filters, keyset pagination, empty state, lateral access,
forged company_id, ID guessing, export isolation and audit logging.
"""
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 CompanyIntercompanyEventsTests(unittest.TestCase):
def setUp(self) -> None:
self.temp_dir = tempfile.TemporaryDirectory()
self.addCleanup(self.temp_dir.cleanup)
root = Path(self.temp_dir.name)
self.db_path = root / "app.db"
self.storage = root / "files"
self._old_db_path = server.DB_PATH
self._old_storage = server.STORAGE_DIR
server.DB_PATH = self.db_path
server.STORAGE_DIR = self.storage
os.environ["APP_BOOTSTRAP_ADMIN_PASSWORD"] = BOOTSTRAP_PASSWORD
connection = connect(self.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
self.httpd = server.ThreadingHTTPServer(("127.0.0.1", 0), QuietHandler)
self.port = self.httpd.server_address[1]
self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True)
self.thread.start()
self.admin = Client("127.0.0.1", self.port)
status, _, data = self.admin.post_json(
"/api/login",
{"username": "group-admin", "password": BOOTSTRAP_PASSWORD, "portal": "admin"},
)
assert status == 200, data
status, _, data = self.admin.post_json(
"/api/password/change",
{"old_password": BOOTSTRAP_PASSWORD, "new_password": ADMIN_PASSWORD},
)
assert status == 200, data
self.initial_passwords: dict[str, str] = {}
self.company_a = self._create_company("甲公司", "cashier-a")
self.company_b = self._create_company("乙公司", "cashier-b")
self.company_c = self._create_company("丙公司", "cashier-c")
self.cashier_a = self._login_company("cashier-a")
self.cashier_b = self._login_company("cashier-b")
self.cashier_c = self._login_company("cashier-c")
self._approve_account(self.company_a, ACCOUNT_A, self.cashier_a)
self._approve_account(self.company_b, ACCOUNT_B, self.cashier_b)
self._approve_account(self.company_c, ACCOUNT_C, self.cashier_c)
def tearDown(self) -> None:
self.httpd.shutdown()
self.httpd.server_close()
server.DB_PATH = self._old_db_path
server.STORAGE_DIR = self._old_storage
os.environ.pop("APP_BOOTSTRAP_ADMIN_PASSWORD", None)
def _create_company(self, name: str, username: str) -> int:
status, _, data = self.admin.post_json(
"/api/admin/companies", {"name": name, "username": username}
)
assert status == 200, data
self.initial_passwords[username] = as_json(data)["initial_password"]
return as_json(data)["company_id"]
def _login_company(self, username: str) -> Client:
client = Client("127.0.0.1", self.port)
initial = self.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
def _approve_account(self, company_id: int, number: str, client: Client) -> int:
status, _, data = 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 = self.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
def _upload_and_confirm(self, client: Client, company_id: int, rows) -> int:
content = workbook_bytes(rows)
status, _, data = self.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")
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
def _lock_single(self, amount: str, at: str = "2026-03-01 10:00:00") -> dict:
self._upload_and_confirm(
self.cashier_a,
self.company_a,
[outgoing(ACCOUNT_A, ACCOUNT_B, amount, at)],
)
status, _, data = self.admin.get("/api/admin/transfer-events")
self.assertEqual(200, status, data)
single = next(
e
for e in as_json(data)["events"]
if e["status"] == "internal_single" and e["amount"] == amount
)
status, _, data = self.admin.get(f"/api/admin/transfer-events/{single['event_id']}")
self.assertEqual(200, status, data)
revision = as_json(data)["event"]["revision"]
status, _, data = self.admin.post_json(
f"/api/admin/transfer-events/{single['event_id']}/decisions",
{
"action": "assign_participant",
"reason": "函证确认",
"expected_revision": revision,
"request_key": f"lock-{amount}-{at}",
"participant": {"role": "payee", "company_id": self.company_b},
},
)
self.assertEqual(200, status, data)
return as_json(data)["decision"]
def _seed_mixed(self) -> None:
"""Paired A→B 100, B→A 40, locked A→B 25, pending A→B 7, B→C 200."""
self._upload_and_confirm(
self.cashier_a, self.company_a,
[outgoing(ACCOUNT_A, ACCOUNT_B, "100.00", "2026-01-05 10:00:00")],
)
self._upload_and_confirm(
self.cashier_b, self.company_b,
[incoming(ACCOUNT_B, ACCOUNT_A, "100.00", "2026-01-05 11:00:00")],
)
self._upload_and_confirm(
self.cashier_b, self.company_b,
[outgoing(ACCOUNT_B, ACCOUNT_A, "40.00", "2026-01-10 10:00:00")],
)
self._upload_and_confirm(
self.cashier_a, self.company_a,
[incoming(ACCOUNT_A, ACCOUNT_B, "40.00", "2026-01-10 11:00:00")],
)
self._lock_single("25.00", "2026-02-01 10:00:00")
self._upload_and_confirm(
self.cashier_a, self.company_a,
[outgoing(ACCOUNT_A, ACCOUNT_B, "7.00", "2026-02-15 10:00:00")],
)
self._upload_and_confirm(
self.cashier_b, self.company_b,
[outgoing(ACCOUNT_B, ACCOUNT_C, "200.00", "2026-01-20 10:00:00")],
)
self._upload_and_confirm(
self.cashier_c, self.company_c,
[incoming(ACCOUNT_C, ACCOUNT_B, "200.00", "2026-01-20 11:00:00")],
)
def _events(self, client: Client, query: str = "from=2026-01-01&to=2026-12-31"):
status, _, data = client.get(f"/api/company/intercompany/events?{query}")
return status, as_json(data) if data else {}
def _export(self, client: Client, query: str = "from=2026-01-01&to=2026-12-31"):
return client.get(f"/api/company/intercompany/export.csv?{query}")
# ------------------------------------------------------------------
# Auth / parameter guards
# ------------------------------------------------------------------
def test_requires_company_role(self) -> None:
status, payload = self._events(self.admin)
self.assertEqual(403, status, payload)
def test_rejects_forged_company_id_on_events_and_export(self) -> None:
status, payload = self._events(
self.cashier_a,
f"from=2026-01-01&to=2026-12-31&company_id={self.company_b}",
)
self.assertEqual(400, status, payload)
status, _, data = self._export(
self.cashier_a,
f"from=2026-01-01&to=2026-12-31&company_id={self.company_a}",
)
self.assertEqual(400, status, data)
def test_rejects_bad_filters(self) -> None:
status, payload = self._events(
self.cashier_a, "from=2026-01-01&to=2026-12-31&direction=sideways"
)
self.assertEqual(400, status, payload)
status, payload = self._events(
self.cashier_a, "from=2026-01-01&to=2026-12-31&state=maybe"
)
self.assertEqual(400, status, payload)
status, payload = self._events(
self.cashier_a, "from=2026-13-40&to=2026-12-31"
)
self.assertEqual(400, status, payload)
def test_empty_window(self) -> None:
status, payload = self._events(self.cashier_a)
self.assertEqual(200, status, payload)
self.assertEqual([], payload["events"])
self.assertFalse(payload["has_more"])
self.assertIsNone(payload["next_cursor"])
# ------------------------------------------------------------------
# Combined filters + confirmed/pending separation
# ------------------------------------------------------------------
def test_combined_filters_and_state_split(self) -> None:
self._seed_mixed()
status, payload = self._events(self.cashier_a)
self.assertEqual(200, status, payload)
events = payload["events"]
# A sees: out 100, in 40, locked out 25, pending out 7 — not B↔C 200
self.assertEqual(4, len(events))
amounts = {e["amount"] for e in events}
self.assertEqual({"100.00", "40.00", "25.00", "7.00"}, amounts)
for event in events:
self.assertNotEqual(self.company_c, event["counterparty_company_id"])
self.assertIn(event["state"], ("confirmed", "pending"))
self.assertIn(event["direction"], ("out", "in"))
status, confirmed = self._events(
self.cashier_a, "from=2026-01-01&to=2026-12-31&state=confirmed"
)
self.assertEqual(200, status, confirmed)
self.assertEqual(3, len(confirmed["events"]))
self.assertTrue(all(e["state"] == "confirmed" for e in confirmed["events"]))
self.assertNotIn("7.00", {e["amount"] for e in confirmed["events"]})
status, pending = self._events(
self.cashier_a, "from=2026-01-01&to=2026-12-31&state=pending"
)
self.assertEqual(200, status, pending)
self.assertEqual(1, len(pending["events"]))
self.assertEqual("7.00", pending["events"][0]["amount"])
self.assertEqual("pending", pending["events"][0]["state"])
status, outs = self._events(
self.cashier_a, "from=2026-01-01&to=2026-12-31&direction=out&state=confirmed"
)
self.assertEqual(200, status, outs)
self.assertEqual({"100.00", "25.00"}, {e["amount"] for e in outs["events"]})
self.assertTrue(all(e["direction"] == "out" for e in outs["events"]))
status, by_cp = self._events(
self.cashier_a,
f"from=2026-01-01&to=2026-12-31&counterparty_id={self.company_b}&state=confirmed",
)
self.assertEqual(200, status, by_cp)
self.assertEqual(3, len(by_cp["events"]))
# Date window excludes Feb locked/pending
status, jan = self._events(
self.cashier_a, "from=2026-01-01&to=2026-01-31&state=confirmed"
)
self.assertEqual(200, status, jan)
self.assertEqual({"100.00", "40.00"}, {e["amount"] for e in jan["events"]})
def test_keyset_pagination_no_dup_no_gap(self) -> None:
# Three confirmed A→B outs on distinct days
for i, amount in enumerate(("11.00", "12.00", "13.00", "14.00", "15.00")):
day = 5 + i
self._upload_and_confirm(
self.cashier_a, self.company_a,
[outgoing(ACCOUNT_A, ACCOUNT_B, amount, f"2026-01-{day:02d} 10:00:00")],
)
self._upload_and_confirm(
self.cashier_b, self.company_b,
[incoming(ACCOUNT_B, ACCOUNT_A, amount, f"2026-01-{day:02d} 11:00:00")],
)
status, page1 = self._events(
self.cashier_a,
"from=2026-01-01&to=2026-12-31&state=confirmed&direction=out&limit=2",
)
self.assertEqual(200, status, page1)
self.assertEqual(2, len(page1["events"]))
self.assertTrue(page1["has_more"])
self.assertIsNotNone(page1["next_cursor"])
status, page2 = self._events(
self.cashier_a,
"from=2026-01-01&to=2026-12-31&state=confirmed&direction=out"
f"&limit=2&cursor={page1['next_cursor']}",
)
self.assertEqual(200, status, page2)
self.assertEqual(2, len(page2["events"]))
self.assertTrue(page2["has_more"])
status, page3 = self._events(
self.cashier_a,
"from=2026-01-01&to=2026-12-31&state=confirmed&direction=out"
f"&limit=2&cursor={page2['next_cursor']}",
)
self.assertEqual(200, status, page3)
self.assertEqual(1, len(page3["events"]))
self.assertFalse(page3["has_more"])
self.assertIsNone(page3["next_cursor"])
ids = [e["event_id"] for e in page1["events"] + page2["events"] + page3["events"]]
self.assertEqual(5, len(ids))
self.assertEqual(len(ids), len(set(ids)))
# Descending by effective_at then event_id
amounts = [e["amount"] for e in page1["events"] + page2["events"] + page3["events"]]
self.assertEqual(["15.00", "14.00", "13.00", "12.00", "11.00"], amounts)
# ------------------------------------------------------------------
# Isolation / ID guess / detail reuse
# ------------------------------------------------------------------
def test_lateral_isolation_and_id_guess_404(self) -> None:
self._seed_mixed()
status, payload_a = self._events(self.cashier_a)
self.assertEqual(200, status, payload_a)
a_ids = {e["event_id"] for e in payload_a["events"]}
status, payload_c = self._events(self.cashier_c)
self.assertEqual(200, status, payload_c)
# C only participates in B↔C 200
self.assertTrue(payload_c["events"])
for event in payload_c["events"]:
self.assertEqual("200.00", event["amount"])
self.assertNotIn(event["event_id"], a_ids)
# C guessing A's event id via transfer-events detail → 404
a_event_id = next(iter(a_ids))
status, _, data = self.cashier_c.get(
f"/api/company/transfer-events/{a_event_id}"
)
self.assertEqual(404, status, data)
# A can open own event; counterparty account masked; only own observations
status, _, data = self.cashier_a.get(
f"/api/company/transfer-events/{a_event_id}"
)
self.assertEqual(200, status, data)
detail = as_json(data)["event"]
if detail.get("counterparty") and "account_number_masked" in detail["counterparty"]:
masked = detail["counterparty"]["account_number_masked"]
self.assertTrue(str(masked).startswith("****"))
self.assertNotIn(ACCOUNT_B, masked)
for obs in detail["observations"]:
self.assertEqual(self.company_a, obs["batch_company_id"])
self.assertNotIn(ACCOUNT_B, json.dumps(obs, ensure_ascii=False))
# Filtering by counterparty C still cannot leak B↔C into A's list
status, filtered = self._events(
self.cashier_a,
f"from=2026-01-01&to=2026-12-31&counterparty_id={self.company_c}",
)
self.assertEqual(200, status, filtered)
self.assertEqual([], filtered["events"])
def test_b44_events_path_still_works_with_cutoff(self) -> None:
# Without HEL-176 discriminators, /events stays on B-44 ledger list.
status, _, data = self.cashier_a.get(
"/api/company/intercompany/events?from=2026-01-01&cutoff=2026-12-31"
)
self.assertEqual(200, status, data)
payload = as_json(data)
self.assertIn("items", payload)
self.assertNotIn("events", payload)
# ------------------------------------------------------------------
# Export: confirmed only + audit + isolation
# ------------------------------------------------------------------
def test_export_confirmed_only_isolated_and_audited(self) -> None:
self._seed_mixed()
status, headers, data = self._export(self.cashier_a)
self.assertEqual(200, status, data)
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.assertGreaterEqual(len(lines), 2)
body = "\n".join(lines[1:])
self.assertIn("100.00", body)
self.assertIn("40.00", body)
self.assertIn("25.00", body)
self.assertNotIn("7.00", body) # pending excluded
self.assertNotIn("200.00", body) # B↔C excluded
# Explicit pending state rejected
status, _, data = self._export(
self.cashier_a, "from=2026-01-01&to=2026-12-31&state=pending"
)
self.assertEqual(400, status, data)
# C export must not contain A's amounts
status, _, data = self._export(self.cashier_c)
self.assertEqual(200, status, data)
text_c = data.decode("utf-8-sig")
self.assertNotIn("100.00", text_c)
self.assertNotIn("25.00", text_c)
self.assertIn("200.00", text_c)
status, _, data = self.admin.get("/api/admin/audit-log?limit=50")
self.assertEqual(200, status, data)
actions = [row["action"] for row in as_json(data)["entries"]]
self.assertIn("export_intercompany_csv", actions)
export_rows = [
row
for row in as_json(data)["entries"]
if row["action"] == "export_intercompany_csv"
]
self.assertTrue(export_rows)
self.assertTrue(
any(f"company:{self.company_a}" == row.get("target") for row in export_rows)
)
self.assertTrue(any("state:confirmed" in (row.get("detail") or "") for row in export_rows))
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink