"""HTTP integration tests for the B-44 intercompany position APIs. Covers the admin balances/pair/events/evidence/subject-review/manual-record endpoints and the company-side scoped reads with masked evidence, plus cross-tenant 404s and company-forbidden admin writes. Every test method spins up its own server with a fresh database so subject confirmations and matched fixtures never leak across tests. """ 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" 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 IntercompanyApiTests(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.initial_passwords: dict[str, str] = {} 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.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.company_a) self.cashier_b = self._login_company("cashier-b", self.company_b) self.cashier_c = self._login_company("cashier-c", self.company_c) self._approve_account(self.company_a, ACCOUNT_A) self._approve_account(self.company_b, ACCOUNT_B) 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.setdefault(username, as_json(data)["initial_password"]) return as_json(data)["company_id"] def _login_company(self, username: str, company_id: int) -> 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) -> None: client = {self.company_a: self.cashier_a, self.company_b: self.cashier_b}[company_id] 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 def _upload_and_confirm(self, 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 _fresh_pair(self, amount: str = "100.00", at: str = "2026-01-05 10:00:00") -> dict: """Upload+confirm a new A<->B pair; returns the pending review item.""" self._upload_and_confirm( self.cashier_a, self.company_a, [outgoing(ACCOUNT_A, ACCOUNT_B, amount, at)], ) self._upload_and_confirm( self.cashier_b, self.company_b, [incoming(ACCOUNT_B, ACCOUNT_A, amount, at.replace("10:", "11:"))], ) status, _, data = self.admin.get( "/api/admin/subject-reviews?from=2026-01-01&cutoff=2026-12-31" ) self.assertEqual(200, status, data) items = as_json(data)["items"] pending = [item for item in items if item["amount"] == amount] self.assertEqual(1, len(pending), data) return pending[0] def _confirm(self, ledger_event_id: int) -> dict: status, _, data = self.admin.get( f"/api/admin/intercompany/events/{ledger_event_id}" ) self.assertEqual(200, status, data) detail = as_json(data) status, _, data = self.admin.post_json( f"/api/admin/intercompany/events/{ledger_event_id}/subject-decisions", { "perspective_company_id": detail["event"]["payer_company_id"], "subject_code": "other_receivable", "reason": "借款确认其他应收", "expected_revision": detail["event"]["ledger_revision_id"], "request_key": f"subj-{ledger_event_id}", }, ) self.assertEqual(200, status, data) return as_json(data)["revision"] def balances(self, client=None, path="/api/admin/intercompany/balances") -> dict: client = client or self.admin status, _, data = client.get(path + "?from=2026-01-01&cutoff=2026-12-31") self.assertEqual(200, status, data) return as_json(data) def _by_company(self, payload: dict, company_id: int) -> dict: return next(item for item in payload["items"] if item["company_id"] == company_id) # ------------------------------------------------------------------ # Admin balances / pair / events # ------------------------------------------------------------------ def test_admin_balances_directory(self) -> None: self._fresh_pair("100.00") payload = self.balances() self.assertEqual("2026-12-31", payload["window"]["cutoff"]) companies = {item["company_id"] for item in payload["items"]} self.assertIn(self.company_a, companies) self.assertIn(self.company_b, companies) for item in payload["items"]: self.assertEqual("unavailable", item["opening"]["status"]) self.assertIsNone(item["opening"]["amount"]) self.assertEqual("period_net_change", item["result"]["kind"]) self.assertIn("gross_amount", item["unresolved"]) self.assertIn("count", item["unresolved"]) self.assertIn("by_reason", item["unresolved"]) self.assertEqual("RMB", item["currency"]) def test_admin_balances_pending_subject_shows_unresolved(self) -> None: self._fresh_pair("100.00") payload = self.balances() item = self._by_company(payload, self.company_a) self.assertEqual(1, item["unresolved"]["count"]) self.assertEqual( "100.00", item["unresolved"]["by_reason"]["subject_review"]["gross_amount"] ) self.assertEqual("0", item["period"]["debit"]) self.assertEqual("0", item["period"]["credit"]) def test_admin_pair_detail_conserves(self) -> None: self._fresh_pair("100.00") status, _, data = self.admin.get( f"/api/admin/intercompany/pairs/{self.company_a}/{self.company_b}" "?from=2026-01-01&cutoff=2026-12-31" ) self.assertEqual(200, status, data) payload = as_json(data) item = payload["items"][0] self.assertTrue(item["conservation"]["opposite"]) self.assertTrue(item["conservation"]["abs_equal"]) self.assertEqual("unavailable", item["opening"]["status"]) self.assertIn("subjects", item) self.assertEqual(1, item["unresolved"]["count"]) def test_admin_subject_decision_flows_to_balances(self) -> None: pending = self._fresh_pair("100.00") revision = self._confirm(pending["ledger_event_id"]) self.assertEqual("confirmed", revision["state"]) self.assertEqual("other_receivable", revision["subject_code"]) payload = self.balances() item = self._by_company(payload, self.company_a) self.assertEqual(0, item["unresolved"]["count"]) self.assertEqual("100.00", item["period"]["debit"]) self.assertEqual("100.00", item["result"]["signed_amount"]) def test_admin_subject_decision_stale_revision_conflicts(self) -> None: pending = self._fresh_pair("100.00") status, _, data = self.admin.post_json( f"/api/admin/intercompany/events/{pending['ledger_event_id']}/subject-decisions", { "perspective_company_id": pending["payer_company_id"], "subject_code": "other_receivable", "reason": "确认", "expected_revision": 999, "request_key": "stale-key", }, ) self.assertEqual(409, status, data) def test_admin_events_list_and_evidence(self) -> None: pending = self._fresh_pair("100.00") revision = self._confirm(pending["ledger_event_id"]) status, _, data = self.admin.get( "/api/admin/intercompany/events?from=2026-01-01&cutoff=2026-12-31" ) self.assertEqual(200, status, data) items = as_json(data)["items"] self.assertTrue(items) event = next( item for item in items if item["ledger_event_id"] == revision["ledger_event_id"] ) self.assertEqual("confirmed", event["state"]) self.assertEqual("other_receivable", event["subject_code"]) self.assertIn("payer_company_name", event) self.assertIn("amount", event) status, _, data = self.admin.get( f"/api/admin/intercompany/events/{event['ledger_event_id']}" ) self.assertEqual(200, status, data) self.assertEqual("confirmed", as_json(data)["state"]) status, _, data = self.admin.get( f"/api/admin/intercompany/events/{event['ledger_event_id']}/evidence" ) self.assertEqual(200, status, data) blocks = as_json(data)["blocks"] self.assertTrue(blocks) for block in blocks: self.assertEqual("visible", block["visibility"]) def test_admin_events_list_includes_pending_subject(self) -> None: pending = self._fresh_pair("100.00") status, _, data = self.admin.get( "/api/admin/intercompany/events?from=2026-01-01&cutoff=2026-12-31" ) self.assertEqual(200, status, data) items = as_json(data)["items"] item = next( item for item in items if item["ledger_event_id"] == pending["ledger_event_id"] ) self.assertEqual("pending_subject", item["state"]) self.assertIsNone(item["subject_code"]) def test_admin_manual_record_decision_via_api(self) -> None: status, _, data = self.cashier_a.post_json( "/api/company/manual-records", { "counterparty_company_id": self.company_b, "occurred_at": "2026-02-01T09:00:00", "direction": "incoming", "amount": "20.00", "currency": "CNY", "funding_source": "other", "requested_subject": "other_receivable", "request_key": "mr-api-1", "summary": "还款", }, ) self.assertEqual(200, status, data) record = as_json(data)["record"] self.assertEqual("pending", record["state"]) status, _, data = self.admin.get("/api/admin/manual-records") self.assertEqual(200, status, data) records = as_json(data)["records"] self.assertTrue(any(item["id"] == record["id"] for item in records)) status, _, data = self.admin.post_json( f"/api/admin/manual-records/{record['id']}/decisions", {"action": "approve_new", "reason": "银行流水中无此事实", "expected_decision_id": record["decision_id"], "request_key": "dec-api-1"}, ) self.assertEqual(200, status, data) self.assertEqual("approved", as_json(data)["decision"]["state"]) status, _, data = self.cashier_a.get("/api/company/manual-records") self.assertEqual(200, status, data) own = [item for item in as_json(data)["records"] if item["id"] == record["id"]] self.assertEqual(1, len(own)) self.assertEqual("approved", own[0]["state"]) status, _, data = self.cashier_b.get("/api/company/manual-records") self.assertEqual(200, status, data) self.assertEqual([], [ item for item in as_json(data)["records"] if item["id"] == record["id"] ]) def test_admin_adjustment_reverse_via_api(self) -> None: pending = self._fresh_pair("100.00") revision = self._confirm(pending["ledger_event_id"]) status, _, data = self.admin.post_json( f"/api/admin/intercompany/events/{revision['ledger_event_id']}/adjustments", {"action": "reverse", "reason": "科目误判，冲销"}, ) self.assertEqual(200, status, data) self.assertEqual("reverse", as_json(data)["action"]) # ------------------------------------------------------------------ # Company scope and masking # ------------------------------------------------------------------ def test_company_balances_scoped_to_own_company(self) -> None: self._fresh_pair("100.00") status, _, data = self.cashier_a.get( "/api/company/intercompany/balances?from=2026-01-01&cutoff=2026-12-31" ) self.assertEqual(200, status, data) payload = as_json(data) for item in payload["items"]: self.assertEqual(self.company_a, item["company_id"]) self.assertIn("counterparties", payload) self.assertTrue(payload["counterparties"]) def test_company_pair_and_events_are_own_scope(self) -> None: self._fresh_pair("100.00") status, _, data = self.cashier_a.get( f"/api/company/intercompany/pairs/{self.company_b}" "?from=2026-01-01&cutoff=2026-12-31" ) self.assertEqual(200, status, data) self.assertEqual(self.company_a, as_json(data)["companies"]["a"]["company_id"]) status, _, data = self.cashier_a.get( "/api/company/intercompany/events?from=2026-01-01&cutoff=2026-12-31" ) self.assertEqual(200, status, data) items = as_json(data)["items"] self.assertTrue(items) for item in items: self.assertIn("direction", item) self.assertIn("own_subject_label", item) self.assertIn(item["direction"], ("incoming", "outgoing")) def test_company_evidence_masks_counterparty_side(self) -> None: pending = self._fresh_pair("100.00") self._confirm(pending["ledger_event_id"]) status, _, data = self.cashier_a.get( f"/api/company/intercompany/events/{pending['ledger_event_id']}/evidence" ) self.assertEqual(200, status, data) blocks = as_json(data)["blocks"] visibilities = {block["visibility"] for block in blocks} self.assertTrue(visibilities & {"visible", "masked"}) for block in blocks: if block["visibility"] == "masked": self.assertEqual("按对方授权不可见", block["fields"].get("note")) self.assertNotIn(ACCOUNT_B, json.dumps(block["fields"], ensure_ascii=False)) def test_company_cannot_read_or_write_admin_intercompany(self) -> None: self._fresh_pair("100.00") status, _, data = self.cashier_a.get("/api/admin/intercompany/balances") self.assertEqual(403, status, data) status, _, data = self.cashier_a.get("/api/admin/manual-records") self.assertEqual(403, status, data) status, _, data = self.cashier_a.post_json( "/api/admin/manual-records/1/decisions", {"action": "approve_new", "reason": "x"} ) self.assertEqual(403, status, data) status, _, data = self.cashier_a.post_json( "/api/admin/intercompany/events/1/adjustments", {"action": "reverse", "reason": "越权"}, ) self.assertEqual(403, status, data) def test_cross_company_reads_are_404(self) -> None: pending = self._fresh_pair("100.00") status, _, data = self.cashier_c.get( f"/api/company/intercompany/events/{pending['ledger_event_id']}" ) self.assertEqual(404, status, data) status, _, data = self.cashier_c.get( f"/api/company/intercompany/events/{pending['ledger_event_id']}/evidence" ) self.assertEqual(404, status, data) if __name__ == "__main__": unittest.main()