diff --git a/server.py b/server.py index c334678..1afc22f 100644 --- a/server.py +++ b/server.py @@ -11,8 +11,8 @@ from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import parse_qs, urlparse from bank_importer import ( - auth, dashboard, importing, ledger_events, manual_records, master_data, matching, - multipart, personal_transit, positions, settings, subjects, + auth, company_transfers, dashboard, importing, ledger_events, manual_records, + master_data, matching, multipart, personal_transit, positions, settings, subjects, ) from bank_importer.db import connect, migrate, utc_now @@ -155,6 +155,11 @@ class AppHandler(SimpleHTTPRequestHandler): self._handle_admin_intercompany_evidence(int(admin_evidence.group(1))) return + # Company transfer-summary (HEL-169/HEL-175, eligible_intercompany_events) + if path == "/api/company/intercompany/summary": + self._handle_company_intercompany_summary(query) + return + # B-44 intercompany positions (company, own-company scope) if path == "/api/company/intercompany/balances": self._handle_company_intercompany_balances(query) @@ -2858,6 +2863,32 @@ class AppHandler(SimpleHTTPRequestHandler): return None, None return user, user["company_id"] + def _handle_company_intercompany_summary(self, query: dict[str, list[str]]) -> None: + """Own-company transfer summary; company_id is session-only (HEL-175).""" + connection = connect(DB_PATH) + try: + user, company_id = self._company_intercompany_scope(connection) + if company_id is None: + return + # Front-end must never supply company_id; reject even matching values. + if (query.get("company_id") or [None])[0] is not None: + self._send_json( + 400, + {"status": "error", "message": "不允许传入 company_id 参数。"}, + ) + return + as_of = (query.get("as_of") or [None])[0] + try: + payload = company_transfers.company_intercompany_summary( + connection, company_id=int(company_id), as_of=as_of + ) + 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_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 new file mode 100644 index 0000000..035b01a --- /dev/null +++ b/src/bank_importer/company_transfers.py @@ -0,0 +1,298 @@ +"""Company-portal intercompany transfer summary (HEL-175 / HEL-169). + +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. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from decimal import Decimal, InvalidOperation +import re +import sqlite3 + +from . import settings + +_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") +_ZERO = Decimal("0.00") + + +class TransferSummaryInputError(ValueError): + """Invalid query parameters (mapped to HTTP 400).""" + + +def today_shanghai() -> str: + return datetime.now(timezone(timedelta(hours=8))).date().isoformat() + + +def _validate_date(value: object, label: str) -> str: + text = str(value or "").strip() + if not text or not _DATE_RE.fullmatch(text): + raise TransferSummaryInputError(f"{label}必须是 YYYY-MM-DD 格式。") + try: + datetime.strptime(text, "%Y-%m-%d") + except ValueError as exc: + raise TransferSummaryInputError(f"{label}不是有效日期。") from exc + return text + + +def _money(value: Decimal) -> str: + return format(value.quantize(Decimal("0.01")), "f") + + +def _as_decimal(raw: object) -> Decimal: + try: + return Decimal(str(raw)) + except (InvalidOperation, TypeError) as exc: + raise TransferSummaryInputError("金额数据无效，无法汇总。") from exc + + +def _company_row(connection: sqlite3.Connection, company_id: int) -> dict[str, object]: + row = connection.execute( + "SELECT id, name FROM companies WHERE id = ?", (company_id,) + ).fetchone() + if row is None: + raise TransferSummaryInputError("本公司不存在。") + return {"id": int(row["id"]), "name": row["name"]} + + +def _window_bounds( + connection: sqlite3.Connection, as_of: str | None +) -> tuple[str, str]: + end = _validate_date(as_of, "as_of") if as_of else today_shanghai() + start = settings.get_settings(connection).get("start_date") or "2026-01-01" + start = _validate_date(start, "start_date") + if start > end: + # Opening / start-date plumbing may lag; clamp rather than 500. + start = end + return start, end + + +def _load_confirmed( + connection: sqlite3.Connection, company_id: int, start: str, end: str +) -> list[sqlite3.Row]: + return connection.execute( + """ + SELECT e.event_id, e.decision_id, e.effective_at, e.amount, e.currency, + e.payer_company_id, e.payee_company_id, e.pairing, + cpayer.name AS payer_company_name, + cpayee.name AS payee_company_name + FROM eligible_intercompany_events e + JOIN companies cpayer ON cpayer.id = e.payer_company_id + JOIN companies cpayee ON cpayee.id = e.payee_company_id + WHERE (e.payer_company_id = ? OR e.payee_company_id = ?) + AND e.effective_at >= ? + AND e.effective_at <= ? + ORDER BY e.event_id + """, + (company_id, company_id, start, end + "T23:59:59"), + ).fetchall() + + +def _load_pending( + connection: sqlite3.Connection, company_id: int, start: str, end: str +) -> list[sqlite3.Row]: + """Pending = not eligible confirmed, still company-visible for tip only. + + Includes unresolved / needs_review / internal_single, plus any + intercompany decision that is neither paired nor locked. + """ + return connection.execute( + """ + SELECT c.event_id, d.id AS decision_id, d.effective_at, d.amount, + d.currency, d.classification, d.pairing, d.locked, + payer.company_id AS payer_company_id, + payee.company_id AS payee_company_id, + cpayer.name AS payer_company_name, + cpayee.name AS payee_company_name + 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 + WHERE e.lifecycle = 'active' + AND (payer.company_id = ? OR payee.company_id = ?) + AND d.effective_at >= ? + AND d.effective_at <= ? + AND ( + d.classification IN ('unresolved', 'needs_review', 'internal_single') + OR ( + d.classification = 'intercompany' + AND d.pairing != 'paired' + AND d.locked = 0 + ) + ) + ORDER BY c.event_id + """, + (company_id, company_id, start, end + "T23:59:59"), + ).fetchall() + + +def _counterparty_of(row: sqlite3.Row, company_id: int) -> tuple[int | None, str | None]: + payer = row["payer_company_id"] + payee = row["payee_company_id"] + if payer is not None and int(payer) == int(company_id): + if payee is None: + return None, None + return int(payee), row["payee_company_name"] + if payee is not None and int(payee) == int(company_id): + if payer is None: + return None, None + return int(payer), row["payer_company_name"] + return None, None + + +def _is_outflow(row: sqlite3.Row, company_id: int) -> bool: + return row["payer_company_id"] is not None and int(row["payer_company_id"]) == int( + company_id + ) + + +def company_intercompany_summary( + connection: sqlite3.Connection, + *, + company_id: int, + as_of: str | None = None, +) -> dict[str, object]: + """Build the HEL-169 summary payload for one company session.""" + own = _company_row(connection, company_id) + start, end = _window_bounds(connection, as_of) + + confirmed_rows = _load_confirmed(connection, company_id, start, end) + pending_rows = _load_pending(connection, company_id, start, end) + + seen_confirmed: set[int] = set() + outflow = _ZERO + inflow = _ZERO + outflow_count = 0 + inflow_count = 0 + + # counterparty_id -> bucket + buckets: dict[int, dict[str, object]] = {} + + def bucket_for(cp_id: int, cp_name: str | None) -> dict[str, object]: + bucket = buckets.get(cp_id) + if bucket is None: + bucket = { + "company_id": cp_id, + "company_name": cp_name or "", + "confirmed_outflow": _ZERO, + "confirmed_inflow": _ZERO, + "pending_count": 0, + "last_effective_at": None, + } + buckets[cp_id] = bucket + elif cp_name and not bucket["company_name"]: + bucket["company_name"] = cp_name + return bucket + + def touch_last(bucket: dict[str, object], effective_at: str | None) -> None: + if not effective_at: + return + previous = bucket["last_effective_at"] + if previous is None or str(effective_at) > str(previous): + bucket["last_effective_at"] = effective_at + + for row in confirmed_rows: + event_id = int(row["event_id"]) + if event_id in seen_confirmed: + continue + seen_confirmed.add(event_id) + amount = _as_decimal(row["amount"]) + cp_id, cp_name = _counterparty_of(row, company_id) + if _is_outflow(row, company_id): + outflow += amount + outflow_count += 1 + if cp_id is not None: + bucket = bucket_for(cp_id, cp_name) + bucket["confirmed_outflow"] += amount + touch_last(bucket, row["effective_at"]) + else: + inflow += amount + inflow_count += 1 + if cp_id is not None: + bucket = bucket_for(cp_id, cp_name) + bucket["confirmed_inflow"] += amount + touch_last(bucket, row["effective_at"]) + + pending_total = _ZERO + seen_pending: set[int] = set() + for row in pending_rows: + event_id = int(row["event_id"]) + if event_id in seen_pending or event_id in seen_confirmed: + continue + seen_pending.add(event_id) + amount = _as_decimal(row["amount"]) + pending_total += amount + cp_id, cp_name = _counterparty_of(row, company_id) + if cp_id is not None: + bucket = bucket_for(cp_id, cp_name) + bucket["pending_count"] = int(bucket["pending_count"]) + 1 + touch_last(bucket, row["effective_at"]) + + net = outflow - inflow + if net > 0: + net_direction: str | None = "receivable" + elif net < 0: + net_direction = "payable" + else: + net_direction = None + + counterparties = [] + for cp_id in sorted( + buckets.keys(), + key=lambda i: ( + -( + buckets[i]["confirmed_outflow"] # type: ignore[operator] + + buckets[i]["confirmed_inflow"] + ), + buckets[i]["company_name"] or "", + i, + ), + ): + bucket = buckets[cp_id] + conf_out = bucket["confirmed_outflow"] + conf_in = bucket["confirmed_inflow"] + assert isinstance(conf_out, Decimal) and isinstance(conf_in, Decimal) + counterparties.append( + { + "company_id": cp_id, + "company_name": bucket["company_name"], + "confirmed_outflow": _money(conf_out), + "confirmed_inflow": _money(conf_in), + "net": _money(conf_out - conf_in), + "pending_count": int(bucket["pending_count"]), + "last_effective_at": bucket["last_effective_at"], + } + ) + + return { + "own_company": own, + "window": { + "start": start, + "end": end, + "has_opening": False, + # Reserved for opening-balance rollout; callers must not invent balances. + "opening": None, + "ending": None, + }, + "confirmed": { + "outflow_total": _money(outflow), + "outflow_count": outflow_count, + "inflow_total": _money(inflow), + "inflow_count": inflow_count, + "net_change": _money(net), + "net_direction": net_direction, + }, + "pending": { + "count": len(seen_pending), + "amount_total": _money(pending_total), + }, + "counterparties": counterparties, + } diff --git a/tests/test_company_transfer_summary.py b/tests/test_company_transfer_summary.py new file mode 100644 index 0000000..d4b3b80 --- /dev/null +++ b/tests/test_company_transfer_summary.py @@ -0,0 +1,362 @@ +"""HTTP tests for GET /api/company/intercompany/summary (HEL-175). + +Covers session-scoped company_id, forged company_id rejection, confirmed vs +pending separation, Decimal net math, dual-company isolation and empty data. +""" + +from __future__ import annotations + +import io +import json +import os +from decimal import Decimal +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 CompanyIntercompanySummaryTests(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: + """A-side only upload → admin locks as intercompany single.""" + 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 _summary(self, client: Client, query: str = "as_of=2026-12-31"): + status, _, data = client.get(f"/api/company/intercompany/summary?{query}") + return status, as_json(data) if data else {} + + # ------------------------------------------------------------------ + # Auth / parameter guards + # ------------------------------------------------------------------ + + def test_requires_company_role(self) -> None: + status, payload = self._summary(self.admin) + self.assertEqual(403, status, payload) + + def test_rejects_forged_company_id(self) -> None: + status, payload = self._summary( + self.cashier_a, f"as_of=2026-12-31&company_id={self.company_b}" + ) + self.assertEqual(400, status, payload) + self.assertIn("company_id", payload.get("message", "")) + + def test_rejects_own_company_id_param(self) -> None: + # Even matching the session company is forbidden. + status, payload = self._summary( + self.cashier_a, f"as_of=2026-12-31&company_id={self.company_a}" + ) + self.assertEqual(400, status, payload) + + def test_rejects_bad_as_of(self) -> None: + status, payload = self._summary(self.cashier_a, "as_of=2026-13-40") + self.assertEqual(400, status, payload) + + def test_anonymous_is_unauthorized(self) -> None: + anon = Client("127.0.0.1", self.port) + status, _, data = anon.get("/api/company/intercompany/summary?as_of=2026-12-31") + self.assertIn(status, (401, 403)) + + # ------------------------------------------------------------------ + # Empty / confirmed math / pending isolation + # ------------------------------------------------------------------ + + def test_empty_window_returns_zeros(self) -> None: + status, payload = self._summary(self.cashier_a) + self.assertEqual(200, status, payload) + self.assertEqual(self.company_a, payload["own_company"]["id"]) + self.assertFalse(payload["window"]["has_opening"]) + self.assertIsNone(payload["window"]["opening"]) + self.assertEqual("0.00", payload["confirmed"]["outflow_total"]) + self.assertEqual("0.00", payload["confirmed"]["inflow_total"]) + self.assertEqual("0.00", payload["confirmed"]["net_change"]) + self.assertEqual(0, payload["pending"]["count"]) + self.assertEqual([], payload["counterparties"]) + # Amounts must be strings, never floats. + self.assertIsInstance(payload["confirmed"]["net_change"], str) + self.assertNotIsInstance(payload["confirmed"]["net_change"], float) + + def test_paired_locked_pending_math_and_isolation(self) -> None: + # Paired A→B 100 + B→A 40 → A net outflow 60 + 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")], + ) + + # Locked single A→B 25 (confirmed) + self._lock_single("25.00", "2026-02-01 10:00:00") + + # Pending unilateral A→B 7 (internal_single, not confirmed) + self._upload_and_confirm( + self.cashier_a, self.company_a, + [outgoing(ACCOUNT_A, ACCOUNT_B, "7.00", "2026-02-15 10:00:00")], + ) + + # B↔C paired 200 — must not appear in A's summary + 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")], + ) + + status, payload = self._summary(self.cashier_a) + self.assertEqual(200, status, payload) + + confirmed = payload["confirmed"] + self.assertEqual("125.00", confirmed["outflow_total"]) # 100 + 25 + self.assertEqual(2, confirmed["outflow_count"]) + self.assertEqual("40.00", confirmed["inflow_total"]) + self.assertEqual(1, confirmed["inflow_count"]) + self.assertEqual("85.00", confirmed["net_change"]) # 125 - 40 + self.assertEqual("receivable", confirmed["net_direction"]) + + # Pending tip only — never folded into confirmed totals + self.assertEqual(1, payload["pending"]["count"]) + self.assertEqual("7.00", payload["pending"]["amount_total"]) + + # Decimal identity: net = outflow - inflow, no float drift + net = Decimal(confirmed["net_change"]) + self.assertEqual( + Decimal(confirmed["outflow_total"]) - Decimal(confirmed["inflow_total"]), + net, + ) + + counterparties = {row["company_id"]: row for row in payload["counterparties"]} + self.assertIn(self.company_b, counterparties) + self.assertNotIn(self.company_c, counterparties) + row_b = counterparties[self.company_b] + self.assertEqual("125.00", row_b["confirmed_outflow"]) + self.assertEqual("40.00", row_b["confirmed_inflow"]) + self.assertEqual("85.00", row_b["net"]) + self.assertEqual(1, row_b["pending_count"]) + + # B must not see C-only? B sees C; A must not see B's C totals via forgery + status_b, payload_b = self._summary(self.cashier_b) + self.assertEqual(200, status_b, payload_b) + cps_b = {row["company_id"] for row in payload_b["counterparties"]} + self.assertIn(self.company_c, cps_b) + # A's view still excludes C + self.assertNotIn(self.company_c, counterparties) + + # Same event counted once: eligible event count equals outflow+inflow counts + self.assertEqual( + confirmed["outflow_count"] + confirmed["inflow_count"], + 3, + ) + + def test_company_b_cannot_see_a_only_pending(self) -> None: + self._upload_and_confirm( + self.cashier_a, self.company_a, + [outgoing(ACCOUNT_A, ACCOUNT_B, "9.00", "2026-04-01 10:00:00")], + ) + status_a, payload_a = self._summary(self.cashier_a) + self.assertEqual(200, status_a, payload_a) + self.assertEqual(1, payload_a["pending"]["count"]) + + status_c, payload_c = self._summary(self.cashier_c) + self.assertEqual(200, status_c, payload_c) + self.assertEqual(0, payload_c["pending"]["count"]) + self.assertEqual("0.00", payload_c["confirmed"]["outflow_total"]) + self.assertEqual([], payload_c["counterparties"]) + + +if __name__ == "__main__": + unittest.main()