From 40f7a91a0fe8793acc56d5f4da67e2770dd208e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=80=BB=E5=B7=A5?= Date: Thu, 27 Aug 2026 13:14:02 +0000 Subject: [PATCH] =?UTF-8?q?HEL-176:=20=E5=85=AC=E5=8F=B8=E7=AB=AF=E8=BD=AC?= =?UTF-8?q?=E8=B4=A6=E5=BE=80=E6=9D=A5=E6=98=8E=E7=BB=86=E7=AD=9B=E9=80=89?= =?UTF-8?q?=E4=B8=8E=E5=AF=BC=E5=87=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 events 组合筛选 + keyset 分页，以及仅导出已确认明细的 CSV（含审计）； 与 B-44 ledger /events 按查询参数分发，不改库表与确认语义。 Co-authored-by: Cursor Co-authored-by: multica-agent --- server.py | 117 +++++- src/bank_importer/company_transfers.py | 446 +++++++++++++++++++++- tests/test_company_transfer_events.py | 500 +++++++++++++++++++++++++ 3 files changed, 1059 insertions(+), 4 deletions(-) create mode 100644 tests/test_company_transfer_events.py diff --git a/server.py b/server.py index 1afc22f..e1f9bdf 100644 --- a/server.py +++ b/server.py @@ -155,10 +155,13 @@ class AppHandler(SimpleHTTPRequestHandler): self._handle_admin_intercompany_evidence(int(admin_evidence.group(1))) return - # Company transfer-summary (HEL-169/HEL-175, eligible_intercompany_events) + # Company transfer-summary (HEL-169/HEL-175/HEL-176) if path == "/api/company/intercompany/summary": self._handle_company_intercompany_summary(query) return + if path == "/api/company/intercompany/export.csv": + self._handle_company_intercompany_export_csv(query) + return # B-44 intercompany positions (company, own-company scope) if path == "/api/company/intercompany/balances": @@ -169,7 +172,12 @@ class AppHandler(SimpleHTTPRequestHandler): self._handle_company_intercompany_pair(int(company_pair.group(1)), query) return if path == "/api/company/intercompany/events": - self._handle_company_intercompany_events(query) + # HEL-176 transfer-detail list shares this path with B-44 ledger + # events; dispatch by distinctive query params (to/direction/…). + if company_transfers.is_transfer_summary_events_query(query): + self._handle_company_transfer_summary_events(query) + else: + self._handle_company_intercompany_events(query) return if path == "/api/company/manual-records": self._handle_company_manual_records(query) @@ -2889,6 +2897,111 @@ class AppHandler(SimpleHTTPRequestHandler): finally: connection.close() + def _reject_forged_company_id(self, query: dict[str, list[str]]) -> bool: + """Return True when the handler already sent a 400 for company_id.""" + if (query.get("company_id") or [None])[0] is not None: + self._send_json( + 400, + {"status": "error", "message": "不允许传入 company_id 参数。"}, + ) + return True + return False + + def _handle_company_transfer_summary_events( + self, query: dict[str, list[str]] + ) -> None: + """HEL-176 filtered transfer-detail list (keyset pagination).""" + connection = connect(DB_PATH) + try: + user, company_id = self._company_intercompany_scope(connection) + if company_id is None: + return + if self._reject_forged_company_id(query): + return + try: + payload = company_transfers.company_intercompany_events( + connection, + company_id=int(company_id), + from_=(query.get("from") or [None])[0], + to=(query.get("to") or [None])[0], + counterparty_id=(query.get("counterparty_id") or [None])[0], + direction=(query.get("direction") or [None])[0], + state=(query.get("state") or [None])[0], + limit=(query.get("limit") or [None])[0], + cursor=(query.get("cursor") or [None])[0], + ) + except company_transfers.TransferSummaryInputError as exc: + self._send_json(400, {"status": "error", "message": str(exc)}) + return + self._send_json(200, {"status": "ok", **payload}) + finally: + connection.close() + + def _handle_company_intercompany_export_csv( + self, query: dict[str, list[str]] + ) -> None: + """Confirmed-only CSV export; session company scope + audit (HEL-176).""" + connection = connect(DB_PATH) + try: + user, company_id = self._company_intercompany_scope(connection) + if company_id is None: + return + if self._reject_forged_company_id(query): + return + # Pending must never leave via export, even if a client sends state=. + if (query.get("state") or [None])[0] not in (None, "", "confirmed"): + self._send_json( + 400, + { + "status": "error", + "message": "导出仅支持已确认明细，不允许导出待确认。", + }, + ) + return + try: + items, meta = company_transfers.company_intercompany_export_rows( + connection, + company_id=int(company_id), + from_=(query.get("from") or [None])[0], + to=(query.get("to") or [None])[0], + counterparty_id=(query.get("counterparty_id") or [None])[0], + direction=(query.get("direction") or [None])[0], + ) + except company_transfers.TransferSummaryInputError as exc: + self._send_json(400, {"status": "error", "message": str(exc)}) + return + content = company_transfers.render_intercompany_export_csv(items) + detail_parts = [ + f"rows:{meta['row_count']}", + f"from:{meta['start']}", + f"to:{meta['end']}", + "state:confirmed", + ] + if meta["counterparty_id"] is not None: + detail_parts.append(f"counterparty_id:{meta['counterparty_id']}") + if meta["direction"] is not None: + detail_parts.append(f"direction:{meta['direction']}") + auth.audit( + connection, + "export_intercompany_csv", + actor=user, + target=f"company:{company_id}", + detail=";".join(detail_parts), + ip=self._client_ip, + ) + self.send_response(200) + self.send_header("Content-Type", "text/csv; charset=utf-8") + self.send_header( + "Content-Disposition", + 'attachment; filename="intercompany-export.csv"', + ) + self.send_header("Content-Length", str(len(content))) + self.send_header("Cache-Control", "no-store") + self.end_headers() + self.wfile.write(content) + finally: + connection.close() + def _handle_company_intercompany_balances(self, query: dict[str, list[str]]) -> None: connection = connect(DB_PATH) try: diff --git a/src/bank_importer/company_transfers.py b/src/bank_importer/company_transfers.py index 035b01a..2955596 100644 --- a/src/bank_importer/company_transfers.py +++ b/src/bank_importer/company_transfers.py @@ -1,22 +1,32 @@ -"""Company-portal intercompany transfer summary (HEL-175 / HEL-169). +"""Company-portal intercompany transfer summary / detail / export (HEL-175/176). Confirmed totals reuse the authoritative ``eligible_intercompany_events`` view (intercompany + paired or locked). Pending counts/amounts are listed separately and never enter outflow, inflow or net. All money math uses ``Decimal`` on stored TEXT amounts — never float or SQLite SUM. + +Detail listing (HEL-176) applies counterparty / date / direction / state +filters inside SQL before LIMIT, and uses keyset pagination on +(effective_at, event_id) descending. Export only ships confirmed rows. """ from __future__ import annotations +import base64 +import csv +import io from datetime import datetime, timedelta, timezone from decimal import Decimal, InvalidOperation import re import sqlite3 -from . import settings +from . import matching, settings _DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") _ZERO = Decimal("0.00") +_DEFAULT_PAGE = 50 +_MAX_PAGE = 200 +_MAX_EXPORT_ROWS = 20000 class TransferSummaryInputError(ValueError): @@ -296,3 +306,435 @@ def company_intercompany_summary( }, "counterparties": counterparties, } + + +# --------------------------------------------------------------------------- +# Detail list + CSV export (HEL-176) +# --------------------------------------------------------------------------- + + +def encode_cursor(effective_at: str, event_id: int) -> str: + raw = f"{effective_at}|{event_id}" + return base64.urlsafe_b64encode(raw.encode("utf-8")).decode("ascii") + + +def decode_cursor(cursor: str | None) -> tuple[str, int] | None: + if not cursor: + return None + try: + raw = base64.urlsafe_b64decode(cursor.encode("ascii")).decode("utf-8") + effective_at, event_id_text = raw.split("|", 1) + return effective_at, int(event_id_text) + except Exception as exc: + raise TransferSummaryInputError("分页游标无效。") from exc + + +def _parse_limit(raw: object | None, default: int = _DEFAULT_PAGE) -> int: + if raw is None or str(raw).strip() == "": + return default + try: + value = int(str(raw).strip()) + except (TypeError, ValueError) as exc: + raise TransferSummaryInputError("limit 必须是正整数。") from exc + if value < 1: + raise TransferSummaryInputError("limit 必须是正整数。") + return min(value, _MAX_PAGE) + + +def _parse_optional_date(raw: object | None, label: str) -> str | None: + if raw is None or str(raw).strip() == "": + return None + return _validate_date(raw, label) + + +def _parse_direction(raw: object | None) -> str | None: + if raw is None or str(raw).strip() == "": + return None + value = str(raw).strip().lower() + if value not in ("out", "in"): + raise TransferSummaryInputError("direction 只能是 out 或 in。") + return value + + +def _parse_state(raw: object | None) -> str | None: + if raw is None or str(raw).strip() == "": + return None + value = str(raw).strip().lower() + if value not in ("confirmed", "pending"): + raise TransferSummaryInputError("state 只能是 confirmed 或 pending。") + return value + + +def _parse_counterparty_id(raw: object | None) -> int | None: + if raw is None or str(raw).strip() == "": + return None + try: + value = int(str(raw).strip()) + except (TypeError, ValueError) as exc: + raise TransferSummaryInputError("counterparty_id 参数无效。") from exc + if value < 1: + raise TransferSummaryInputError("counterparty_id 参数无效。") + return value + + +def _event_window( + connection: sqlite3.Connection, + *, + from_: str | None, + to: str | None, +) -> tuple[str, str]: + end = _parse_optional_date(to, "to") or today_shanghai() + start_default = settings.get_settings(connection).get("start_date") or "2026-01-01" + start = _parse_optional_date(from_, "from") or _validate_date( + start_default, "start_date" + ) + if start > end: + raise TransferSummaryInputError("from 不能晚于 to。") + return start, end + + +# Confirmed = eligible_intercompany_events. Pending matches summary tip set. +_CONFIRMED_PREDICATE = """ + d.classification = 'intercompany' + AND (d.pairing = 'paired' OR d.locked = 1) +""" + +_PENDING_PREDICATE = """ + ( + d.classification IN ('unresolved', 'needs_review', 'internal_single') + OR ( + d.classification = 'intercompany' + AND d.pairing != 'paired' + AND d.locked = 0 + ) + ) +""" + +_BOTH_PREDICATE = f""" + ( + ({_CONFIRMED_PREDICATE}) + OR ({_PENDING_PREDICATE}) + ) +""" + + +def _state_predicate(state: str | None) -> str: + if state == "confirmed": + return f"({_CONFIRMED_PREDICATE})" + if state == "pending": + return f"({_PENDING_PREDICATE})" + return _BOTH_PREDICATE + + +def _list_select_sql() -> str: + return """ + SELECT c.event_id, d.id AS decision_id, d.revision, d.classification, + d.pairing, d.amount, d.currency, d.effective_at, d.mode, + d.locked, d.rule_version, d.created_at, d.reason, + payer.company_id AS payer_company_id, + payee.company_id AS payee_company_id, + payer.bank_account_id AS payer_account_id, + payee.bank_account_id AS payee_account_id, + cpayer.name AS payer_company_name, + cpayee.name AS payee_company_name, + (SELECT COUNT(*) FROM transfer_decision_observations o + WHERE o.decision_id = d.id) AS evidence_count, + (SELECT r.summary + FROM transfer_decision_observations o + JOIN source_rows r ON r.id = o.source_row_id + JOIN sheet_batches s ON s.id = r.sheet_batch_id + JOIN import_batches b ON b.id = s.import_batch_id + WHERE o.decision_id = d.id AND b.company_id = ? + ORDER BY o.id + LIMIT 1) AS summary + FROM current_transfer_decisions c + JOIN transfer_match_decisions d ON d.id = c.decision_id + JOIN canonical_transfer_events e ON e.id = c.event_id + LEFT JOIN transfer_decision_participants payer + ON payer.decision_id = d.id AND payer.role = 'payer' + LEFT JOIN transfer_decision_participants payee + ON payee.decision_id = d.id AND payee.role = 'payee' + LEFT JOIN companies cpayer ON cpayer.id = payer.company_id + LEFT JOIN companies cpayee ON cpayee.id = payee.company_id + """ + + +def _build_event_filters( + *, + company_id: int, + start: str, + end: str, + counterparty_id: int | None, + direction: str | None, + state: str | None, + cursor: tuple[str, int] | None, +) -> tuple[str, list[object]]: + conditions = [ + "e.lifecycle = 'active'", + "(payer.company_id = ? OR payee.company_id = ?)", + "d.effective_at >= ?", + "d.effective_at <= ?", + _state_predicate(state), + ] + params: list[object] = [ + company_id, + company_id, + start, + end + "T23:59:59", + ] + + if counterparty_id is not None: + # Counterparty is the other participant; own company stays forced above. + conditions.append( + """ + ( + (payer.company_id = ? AND payee.company_id = ?) + OR (payee.company_id = ? AND payer.company_id = ?) + ) + """ + ) + params.extend([company_id, counterparty_id, company_id, counterparty_id]) + + if direction == "out": + conditions.append("payer.company_id = ?") + params.append(company_id) + elif direction == "in": + conditions.append("payee.company_id = ?") + params.append(company_id) + + if cursor is not None: + cursor_at, cursor_id = cursor + conditions.append( + """ + ( + d.effective_at < ? + OR (d.effective_at = ? AND c.event_id < ?) + ) + """ + ) + params.extend([cursor_at, cursor_at, cursor_id]) + + where = " WHERE " + " AND ".join(conditions) + return where, params + + +def _row_state(row: sqlite3.Row) -> str: + classification = row["classification"] + pairing = row["pairing"] + locked = bool(row["locked"]) + if classification == "intercompany" and (pairing == "paired" or locked): + return "confirmed" + return "pending" + + +def _row_direction(row: sqlite3.Row, company_id: int) -> str | None: + if row["payer_company_id"] is not None and int(row["payer_company_id"]) == int( + company_id + ): + return "out" + if row["payee_company_id"] is not None and int(row["payee_company_id"]) == int( + company_id + ): + return "in" + return None + + +def _event_list_item(row: sqlite3.Row, company_id: int) -> dict[str, object]: + direction = _row_direction(row, company_id) + state = _row_state(row) + if direction == "out": + counterparty_company_id = row["payee_company_id"] + counterparty_company_name = row["payee_company_name"] + else: + counterparty_company_id = row["payer_company_id"] + counterparty_company_name = row["payer_company_name"] + summary = row["summary"] or row["reason"] or "" + return { + "event_id": int(row["event_id"]), + "decision_id": int(row["decision_id"]), + "revision": row["revision"], + "classification": row["classification"], + "pairing": row["pairing"], + "status": matching.exposed_status(row), + "state": state, + "direction": direction, + "amount": row["amount"], + "currency": row["currency"], + "effective_at": row["effective_at"], + "mode": row["mode"], + "locked": bool(row["locked"]), + "rule_version": row["rule_version"], + "summary": summary, + "own_company_id": company_id, + "counterparty_company_id": ( + int(counterparty_company_id) if counterparty_company_id is not None else None + ), + "counterparty_company_name": counterparty_company_name, + "evidence_count": row["evidence_count"], + } + + +def company_intercompany_events( + connection: sqlite3.Connection, + *, + company_id: int, + from_: str | None = None, + to: str | None = None, + counterparty_id: int | None | object = None, + direction: str | None | object = None, + state: str | None | object = None, + limit: object | None = None, + cursor: str | None = None, +) -> dict[str, object]: + """Filtered keyset page of company-visible transfer events (HEL-176).""" + start, end = _event_window(connection, from_=from_, to=to) + cp_id = _parse_counterparty_id(counterparty_id) + direction_value = _parse_direction(direction) + state_value = _parse_state(state) + page_size = _parse_limit(limit) + cursor_tuple = decode_cursor(cursor) + + where, params = _build_event_filters( + company_id=company_id, + start=start, + end=end, + counterparty_id=cp_id, + direction=direction_value, + state=state_value, + cursor=cursor_tuple, + ) + # summary subquery binds own company_id first. + sql = ( + _list_select_sql() + + where + + " ORDER BY d.effective_at DESC, c.event_id DESC LIMIT ?" + ) + rows = connection.execute( + sql, (company_id, *params, page_size + 1) + ).fetchall() + + has_more = len(rows) > page_size + page = rows[:page_size] + items = [_event_list_item(row, company_id) for row in page] + next_cursor = None + if has_more and page: + last = page[-1] + next_cursor = encode_cursor(str(last["effective_at"]), int(last["event_id"])) + + return { + "window": {"start": start, "end": end}, + "events": items, + "next_cursor": next_cursor, + "has_more": has_more, + } + + +def company_intercompany_export_rows( + connection: sqlite3.Connection, + *, + company_id: int, + from_: str | None = None, + to: str | None = None, + counterparty_id: int | None | object = None, + direction: str | None | object = None, +) -> tuple[list[dict[str, object]], dict[str, object]]: + """Confirmed-only rows for CSV export; pending never included.""" + start, end = _event_window(connection, from_=from_, to=to) + cp_id = _parse_counterparty_id(counterparty_id) + direction_value = _parse_direction(direction) + + where, params = _build_event_filters( + company_id=company_id, + start=start, + end=end, + counterparty_id=cp_id, + direction=direction_value, + state="confirmed", + cursor=None, + ) + sql = ( + _list_select_sql() + + where + + " ORDER BY d.effective_at DESC, c.event_id DESC LIMIT ?" + ) + rows = connection.execute( + sql, (company_id, *params, _MAX_EXPORT_ROWS + 1) + ).fetchall() + if len(rows) > _MAX_EXPORT_ROWS: + raise TransferSummaryInputError( + f"导出行数超过上限 {_MAX_EXPORT_ROWS}，请缩小筛选范围。" + ) + items = [_event_list_item(row, company_id) for row in rows] + meta = { + "start": start, + "end": end, + "counterparty_id": cp_id, + "direction": direction_value, + "state": "confirmed", + "row_count": len(items), + } + return items, meta + + +def render_intercompany_export_csv(items: list[dict[str, object]]) -> bytes: + buffer = io.StringIO() + writer = csv.writer(buffer) + writer.writerow( + [ + "日期", + "方向", + "对方公司", + "金额", + "币种", + "摘要", + "状态", + "配对", + "事件ID", + "决策ID", + ] + ) + direction_label = {"out": "转出", "in": "转入"} + for item in items: + writer.writerow( + [ + item.get("effective_at") or "", + direction_label.get(str(item.get("direction") or ""), ""), + item.get("counterparty_company_name") or "", + item.get("amount") or "", + item.get("currency") or "", + item.get("summary") or "", + "已确认", + item.get("pairing") or "", + item.get("event_id") or "", + item.get("decision_id") or "", + ] + ) + # UTF-8 BOM so Excel opens the CSV with the right encoding. + return (chr(0xFEFF) + buffer.getvalue()).encode("utf-8") + + +def is_transfer_summary_events_query(query: dict[str, list[str]]) -> bool: + """Discriminate HEL-176 transfer list from B-44 ledger ``/events``. + + B-44 uses ``cutoff`` / subject / posting_kind / source_kind / pending_subject. + HEL-176 uses ``to`` / direction / counterparty_id / state=pending|confirmed + (without ledger-only knobs). + """ + if (query.get("direction") or [None])[0] is not None: + return True + if (query.get("counterparty_id") or [None])[0] is not None: + return True + if (query.get("to") or [None])[0] is not None: + return True + state = (query.get("state") or [None])[0] + if state in ("pending", "confirmed") and (query.get("cutoff") or [None])[0] is None: + # Bare state=confirmed without cutoff is the transfer-summary list; + # B-44 confirmed always pairs with cutoff in existing callers/tests. + if (query.get("subject") or [None])[0] is not None: + return False + if (query.get("posting_kind") or [None])[0] is not None: + return False + if (query.get("source_kind") or [None])[0] is not None: + return False + return True + return False diff --git a/tests/test_company_transfer_events.py b/tests/test_company_transfer_events.py new file mode 100644 index 0000000..98ba4bd --- /dev/null +++ b/tests/test_company_transfer_events.py @@ -0,0 +1,500 @@ +"""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()