7fb97119ba
移除柱状图/折线图死代码；主从改为 480px+明细宽栏；待审核首页卡、侧栏角标与后端口径同源。 Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>
338 lines
13 KiB
Python
338 lines
13 KiB
Python
"""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_pending_manual_counts_as_medium(self) -> None:
|
|
from bank_importer import auth, manual_records
|
|
|
|
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
|
|
self.connection.commit()
|
|
auth.create_user(
|
|
self.connection, "cashier-a", "CashierPass123", "company", company_id=int(a)
|
|
)
|
|
actor = self.connection.execute(
|
|
"SELECT * FROM users WHERE username = 'cashier-a'"
|
|
).fetchone()
|
|
manual_records.submit(
|
|
self.connection,
|
|
company_id=int(a),
|
|
counterparty_company_id=int(b),
|
|
occurred_at="2026-02-01T09:00:00",
|
|
direction="incoming",
|
|
amount="100.00",
|
|
currency="CNY",
|
|
funding_source="other",
|
|
requested_subject="receivable",
|
|
request_key="hel157-manual-1",
|
|
actor=actor,
|
|
)
|
|
counts = dashboard.audit_counts(self.connection)
|
|
self.assertEqual(1, counts["medium"])
|
|
self.assertEqual(0, counts["high"])
|
|
self.assertEqual(1, counts["total"])
|
|
|
|
def test_audit_total_equals_queue_sum(self) -> None:
|
|
"""Homepage / badge / audit heading must share one backend total."""
|
|
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
|
|
self.connection.commit()
|
|
master_data.submit_bank_account(
|
|
self.connection,
|
|
company_id=int(a),
|
|
bank_name="工行",
|
|
account_type="一般户",
|
|
account_number="6222020000000002",
|
|
start_date="2026-01-01",
|
|
actor=None,
|
|
)
|
|
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', 'unresolved',
|
|
'single', 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.commit()
|
|
|
|
counts = dashboard.audit_counts(self.connection)
|
|
payload = dashboard.build_dashboard(
|
|
self.connection, from_date="2026-01-01", cutoff="2026-08-20"
|
|
)
|
|
self.assertEqual(counts, payload["audit"])
|
|
self.assertEqual(
|
|
counts["total"],
|
|
counts["high"] + counts["medium"] + counts["low"],
|
|
)
|
|
self.assertEqual(1, counts["high"])
|
|
self.assertEqual(1, counts["medium"])
|
|
self.assertEqual(2, 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_audit_counts_sync_after_account_review(self) -> None:
|
|
"""Pending account raises dashboard.audit; approve brings total back down."""
|
|
from bank_importer.db import connect as db_connect
|
|
from bank_importer import master_data as md
|
|
|
|
status, _, raw = self.admin.get("/api/admin/companies")
|
|
company_id = as_json(raw)["companies"][0]["id"]
|
|
|
|
connection = db_connect(self.db_path)
|
|
try:
|
|
account = md.submit_bank_account(
|
|
connection,
|
|
company_id=company_id,
|
|
bank_name="工行",
|
|
account_type="一般户",
|
|
account_number="6222020000000099",
|
|
start_date="2026-01-01",
|
|
actor=None,
|
|
)
|
|
account_id = account["id"]
|
|
finally:
|
|
connection.close()
|
|
|
|
status, _, raw = self.admin.get("/api/admin/dashboard")
|
|
before = as_json(raw)["audit"]
|
|
self.assertEqual(200, status)
|
|
self.assertGreaterEqual(before["medium"], 1)
|
|
self.assertGreaterEqual(before["total"], 1)
|
|
|
|
status, _, raw = self.admin.post_json(
|
|
f"/api/admin/accounts/{account_id}/review",
|
|
{"decision": "approve", "reason": "HEL-157 sync test"},
|
|
)
|
|
self.assertEqual(200, status, raw)
|
|
|
|
status, _, raw = self.admin.get("/api/admin/dashboard")
|
|
after = as_json(raw)["audit"]
|
|
self.assertEqual(after["medium"], before["medium"] - 1)
|
|
self.assertEqual(after["total"], before["total"] - 1)
|
|
self.assertEqual(
|
|
after["total"], after["high"] + after["medium"] + after["low"]
|
|
)
|
|
|
|
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()
|