"""HEL-168: company workspace pending unilateral counts sync after confirm.""" from __future__ import annotations import io import os from pathlib import Path import tempfile import threading import unittest from openpyxl import Workbook import server from test_server_auth import Client, as_json BOOTSTRAP_PASSWORD = "BootAdmin123" ADMIN_PASSWORD = "AdminPass123" CASHIER_PASSWORD = "Cashier123" CCB_HEADER = [ "客户账号", "账户名称", "交易时间", "借方发生额（支取）", "贷方发生额（收入）", "余额", "币种", "对方户名", "对方账号", "对方开户机构", "摘要", "备注", ] ACCOUNT_A = "6222000000001001" ACCOUNT_B = "6222000000001002" 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) -> list: return [own, "测试公司", at, amount, "", "50000.00", "RMB", "对方", cp, "某银行", "货款", ""] class CompanyWorkspaceTodoTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.temp_dir = tempfile.TemporaryDirectory() root = Path(cls.temp_dir.name) cls.db_path = root / "app.db" cls.storage = root / "files" cls._old_db_path = server.DB_PATH cls._old_storage = server.STORAGE_DIR server.DB_PATH = cls.db_path server.STORAGE_DIR = cls.storage os.environ["APP_BOOTSTRAP_ADMIN_PASSWORD"] = BOOTSTRAP_PASSWORD from bank_importer.db import connect, migrate connection = connect(cls.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 cls.httpd = server.ThreadingHTTPServer(("127.0.0.1", 0), QuietHandler) cls.port = cls.httpd.server_address[1] cls.thread = threading.Thread(target=cls.httpd.serve_forever, daemon=True) cls.thread.start() cls.admin = Client("127.0.0.1", cls.port) status, _, data = cls.admin.post_json( "/api/login", {"username": "group-admin", "password": BOOTSTRAP_PASSWORD, "portal": "admin"}, ) assert status == 200, data status, _, data = cls.admin.post_json( "/api/password/change", {"old_password": BOOTSTRAP_PASSWORD, "new_password": ADMIN_PASSWORD}, ) assert status == 200, data cls.initial_passwords: dict[str, str] = {} cls.company_a = cls._create_company("甲公司", "ws-cashier-a") cls.company_b = cls._create_company("乙公司", "ws-cashier-b") cls.cashier_a = cls._login_company("ws-cashier-a") cls.cashier_b = cls._login_company("ws-cashier-b") cls._approve_account(cls.cashier_a, ACCOUNT_A) cls._approve_account(cls.cashier_b, ACCOUNT_B) @classmethod def tearDownClass(cls) -> None: cls.httpd.shutdown() cls.httpd.server_close() server.DB_PATH = cls._old_db_path server.STORAGE_DIR = cls._old_storage os.environ.pop("APP_BOOTSTRAP_ADMIN_PASSWORD", None) cls.temp_dir.cleanup() @classmethod def _create_company(cls, name: str, username: str) -> int: status, _, data = cls.admin.post_json( "/api/admin/companies", {"name": name, "username": username}, ) assert status == 200, data payload = as_json(data) cls.initial_passwords[username] = payload["initial_password"] return payload["company_id"] @classmethod def _login_company(cls, username: str) -> Client: client = Client("127.0.0.1", cls.port) initial = cls.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 @classmethod def _approve_account(cls, company_client: Client, number: str) -> int: status, _, data = company_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 = cls.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 _seed_unilateral(self, amount: str, at: str, cp_account: str = "9999999999999999") -> dict: status, _, data = self.admin.post_multipart( "/api/parse", {"company_id": str(self.company_a)}, f"单边-{amount}.xlsx", workbook_bytes([outgoing(ACCOUNT_A, cp_account, amount, at)]), ) self.assertEqual(200, status, data) batch_id = as_json(data)["batch_id"] status, _, data = self.cashier_a.get(f"/api/batches/{batch_id}/sheets") self.assertEqual(200, status, data) names = [s["sheet_name"] for s in as_json(data)["sheets"] if s["outcome"] == "parsed"] status, _, data = self.cashier_a.post_json( f"/api/batches/{batch_id}/confirm", {"sheets": names} ) self.assertEqual(200, status, data) status, _, data = self.cashier_a.get("/api/company/workspace") self.assertEqual(200, status, data) events = [ e for e in as_json(data)["unilateral_events"] if e["amount"] == amount ] self.assertEqual(1, len(events), data) return events[0] def _workspace(self, client: Client) -> dict: status, _, data = client.get("/api/company/workspace") self.assertEqual(200, status, data) payload = as_json(data) self.assertEqual("ok", payload["status"]) return payload def test_seed_three_confirm_sync_and_isolation(self) -> None: # Seed 3 pending unilaterals for company A. e1 = self._seed_unilateral("11.00", "2026-03-01 10:00:00") e2 = self._seed_unilateral("22.00", "2026-03-02 10:00:00") e3 = self._seed_unilateral("33.00", "2026-03-03 10:00:00") seeded_ids = {e1["event_id"], e2["event_id"], e3["event_id"]} ws = self._workspace(self.cashier_a) own = [e for e in ws["unilateral_events"] if e["event_id"] in seeded_ids] self.assertEqual(3, len(own)) self.assertEqual(ws["pending_unilateral"], ws["pending_total"]) self.assertGreaterEqual(ws["pending_unilateral"], 3) # 两处数字与列表同一权威口径 self.assertEqual(ws["pending_unilateral"], len(ws["unilateral_events"])) # 另一公司看不到甲公司这 3 笔 ws_b = self._workspace(self.cashier_b) self.assertEqual( [], [e for e in ws_b["unilateral_events"] if e["event_id"] in seeded_ids], ) # match-exceptions 与 workspace 同集合 status, _, data = self.cashier_a.get("/api/company/match-exceptions") self.assertEqual(200, status, data) exceptions = as_json(data)["exceptions"] exception_ids = {e["event_id"] for e in exceptions} self.assertTrue(seeded_ids.issubset(exception_ids)) remaining = [e1, e2, e3] for expected in (2, 1, 0): event = remaining.pop(0) status, _, data = self.cashier_a.post_json( f"/api/company/transfer-events/{event['event_id']}/confirm", { "expected_revision": event["revision"], "request_key": f"hel168-{event['event_id']}-ok", "counterparty_company_id": self.company_b, "reason": "公司端确认单边流水", }, ) self.assertEqual(200, status, data) body = as_json(data) self.assertIn("workspace", body) left = [ e for e in body["workspace"]["unilateral_events"] if e["event_id"] in seeded_ids ] self.assertEqual(expected, len(left)) self.assertEqual(body["workspace"]["pending_unilateral"], len(body["workspace"]["unilateral_events"])) # 刷新等价：重新 GET 仍为同步后的数字 ws = self._workspace(self.cashier_a) left = [e for e in ws["unilateral_events"] if e["event_id"] in seeded_ids] self.assertEqual(expected, len(left)) # 重复确认：事件已离开待确认集合且 request_key 不同 → 404，数字不得误减 status, _, data = self.cashier_a.post_json( f"/api/company/transfer-events/{e1['event_id']}/confirm", { "expected_revision": e1["revision"], "request_key": f"hel168-{e1['event_id']}-again", "counterparty_company_id": self.company_b, "reason": "公司端确认单边流水", }, ) self.assertEqual(404, status, data) ws = self._workspace(self.cashier_a) left = [e for e in ws["unilateral_events"] if e["event_id"] in seeded_ids] self.assertEqual(0, len(left)) # 同一 request_key 重放：幂等成功，数字不误减 e4 = self._seed_unilateral("44.00", "2026-03-04 10:00:00") key = f"hel168-{e4['event_id']}-replay" status, _, data = self.cashier_a.post_json( f"/api/company/transfer-events/{e4['event_id']}/confirm", { "expected_revision": e4["revision"], "request_key": key, "counterparty_company_id": self.company_b, "reason": "公司端确认单边流水", }, ) self.assertEqual(200, status, data) after_first = self._workspace(self.cashier_a)["pending_unilateral"] status, _, data = self.cashier_a.post_json( f"/api/company/transfer-events/{e4['event_id']}/confirm", { "expected_revision": e4["revision"], "request_key": key, "counterparty_company_id": self.company_b, "reason": "公司端确认单边流水", }, ) self.assertEqual(200, status, data) self.assertEqual(after_first, as_json(data)["workspace"]["pending_unilateral"]) self.assertEqual(after_first, self._workspace(self.cashier_a)["pending_unilateral"]) # 失败请求不得误减：用不存在的事件 before = self._workspace(self.cashier_a)["pending_unilateral"] status, _, data = self.cashier_a.post_json( "/api/company/transfer-events/999999/confirm", { "expected_revision": 1, "request_key": "hel168-fail", "counterparty_company_id": self.company_b, }, ) self.assertEqual(404, status, data) self.assertEqual(before, self._workspace(self.cashier_a)["pending_unilateral"]) # 乙公司不能确认甲公司事件（即使猜到 id） status, _, data = self.cashier_b.post_json( f"/api/company/transfer-events/{e2['event_id']}/confirm", { "expected_revision": e2["revision"], "request_key": "hel168-cross", "counterparty_company_id": self.company_a, }, ) self.assertEqual(404, status, data) def test_admin_forbidden_on_company_workspace(self) -> None: status, _, data = self.admin.get("/api/company/workspace") self.assertEqual(403, status, data) if __name__ == "__main__": unittest.main()