"""Dashboard aggregate helpers and admin HTTP endpoints (no openpyxl dependency).""" from __future__ import annotations import os from pathlib import Path import tempfile import threading import unittest from bank_importer import dashboard, master_data from bank_importer.db import connect, migrate import server from test_server_auth import Client, as_json BOOTSTRAP_PASSWORD = "BootAdmin123" ADMIN_PASSWORD = "AdminPass123" class DashboardUnitTests(unittest.TestCase): def setUp(self) -> None: self.connection = connect(":memory:") migrate(self.connection) def tearDown(self) -> None: self.connection.close() def test_empty_dashboard(self) -> None: payload = dashboard.build_dashboard( self.connection, from_date="2026-01-01", cutoff="2026-08-20" ) self.assertEqual(0, payload["audit"]["total"]) self.assertEqual([], payload["companies"]) self.assertEqual(7, len(payload["weekly_flow"]["labels"])) def test_pending_account_counts_as_medium(self) -> None: now = master_data.utc_now() cursor = self.connection.execute( "INSERT INTO companies (name, credit_code, cashier_name, status, created_at, updated_at) " "VALUES ('甲公司', NULL, NULL, 'active', ?, ?)", (now, now), ) company_id = int(cursor.lastrowid) self.connection.commit() master_data.submit_bank_account( self.connection, company_id=company_id, bank_name="工行", account_type="一般户", account_number="6222020000000001", start_date="2026-01-01", actor=None, ) counts = dashboard.audit_counts(self.connection) self.assertEqual(1, counts["medium"]) self.assertEqual(0, counts["high"]) self.assertEqual(1, counts["total"]) def test_company_summaries_from_eligible_events(self) -> None: now = master_data.utc_now() a = self.connection.execute( "INSERT INTO companies (name, credit_code, cashier_name, status, created_at, updated_at) " "VALUES ('甲公司', NULL, NULL, 'active', ?, ?)", (now, now), ).lastrowid b = self.connection.execute( "INSERT INTO companies (name, credit_code, cashier_name, status, created_at, updated_at) " "VALUES ('乙公司', NULL, NULL, 'active', ?, ?)", (now, now), ).lastrowid event_id = self.connection.execute( "INSERT INTO canonical_transfer_events (lifecycle, created_at) VALUES ('active', ?)", (now,), ).lastrowid decision_id = self.connection.execute( """ INSERT INTO transfer_match_decisions ( event_id, revision, effective_at, amount, currency, classification, pairing, locked, mode, rule_version, created_at ) VALUES (?, 1, '2026-07-05T10:00:00', '100000.00', 'CNY', 'intercompany', 'paired', 0, 'manual', 'test', ?) """, (event_id, now), ).lastrowid self.connection.execute( "INSERT INTO current_transfer_decisions (event_id, decision_id) VALUES (?, ?)", (event_id, decision_id), ) self.connection.execute( """ INSERT INTO transfer_decision_participants ( decision_id, role, company_id, bank_account_id, resolve_method, created_at ) VALUES (?, 'payer', ?, NULL, 'manual', ?), (?, 'payee', ?, NULL, 'manual', ?) """, (decision_id, a, now, decision_id, b, now), ) self.connection.commit() items, totals = dashboard.company_summaries( self.connection, from_date="2026-01-01", cutoff="2026-08-20" ) self.assertEqual(1, totals["detail_count"]) by_name = {row["name"]: row for row in items} self.assertEqual(1, by_name["甲公司"]["detail_count"]) self.assertEqual("10.00", by_name["甲公司"]["debit_wan"]) self.assertEqual("-10.00", by_name["甲公司"]["net_wan"]) self.assertEqual("10.00", by_name["乙公司"]["credit_wan"]) self.assertEqual("10.00", by_name["乙公司"]["net_wan"]) detail = dashboard.company_peer_groups( self.connection, int(a), from_date="2026-01-01", cutoff="2026-08-20" ) self.assertEqual(1, len(detail["groups"])) self.assertEqual("乙公司", detail["groups"][0]["peer_name"]) self.assertIsNone(detail["groups"][0]["opening"]) class DashboardApiTests(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 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 status, _, data = cls.admin.post_json( "/api/admin/companies", {"name": "甲公司", "username": "cashier-a"} ) assert status == 200, data @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 cls.temp_dir.cleanup() def test_dashboard_ok_for_admin(self) -> None: status, _, raw = self.admin.get( "/api/admin/dashboard?from=2026-01-01&cutoff=2026-08-20" ) data = as_json(raw) self.assertEqual(200, status) self.assertEqual("ok", data["status"]) self.assertIn("audit", data) self.assertEqual(1, len(data["companies"])) self.assertEqual("甲公司", data["companies"][0]["name"]) def test_company_detail_missing(self) -> None: status, _, raw = self.admin.get( "/api/admin/dashboard/companies/999999?from=2026-01-01&cutoff=2026-08-20" ) data = as_json(raw) self.assertEqual(404, status) self.assertEqual("error", data["status"]) def test_company_detail_ok(self) -> None: status, _, raw = self.admin.get("/api/admin/dashboard") data = as_json(raw) company_id = data["companies"][0]["id"] status, _, raw = self.admin.get( f"/api/admin/dashboard/companies/{company_id}?from=2026-01-01&cutoff=2026-08-20" ) detail = as_json(raw) self.assertEqual(200, status) self.assertEqual([], detail["groups"]) if __name__ == "__main__": unittest.main()