"""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()