更新

backend/tests/test_dashboard_account_stats.py

@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+from collections import namedtuple
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+
+BACKEND_DIR = Path(__file__).resolve().parents[1]
+os.environ["KEFU_DB_TYPE"] = "sqlite"
+os.environ["KEFU_DATABASE_URL"] = ""
+os.environ["KEFU_DB_PATH"] = str(BACKEND_DIR / "kefu.db")
+if str(BACKEND_DIR) not in sys.path:
+    sys.path.insert(0, str(BACKEND_DIR))
+
+import main
+
+
+_AggregateRow = namedtuple(
+    "_AggregateRow",
+    (
+        "total_accounts",
+        "online_accounts",
+        "my_accounts",
+        "my_online_accounts",
+    ),
+)
+
+
+class _AggregateResult:
+    def __init__(self, row):
+        self._row = row
+
+    def one(self):
+        return self._row
+
+
+class DashboardAccountStatsTests(unittest.IsolatedAsyncioTestCase):
+    async def test_conditional_aggregate_maps_global_and_personal_counts(self):
+        row = _AggregateRow(
+            total_accounts=12,
+            online_accounts=5,
+            my_accounts=3,
+            my_online_accounts=2,
+        )
+        db = SimpleNamespace(
+            execute=AsyncMock(return_value=_AggregateResult(row))
+        )
+        user = SimpleNamespace(id=42, role="operator")
+
+        response = await main.get_dashboard_account_stats(db=db, user=user)
+
+        self.assertIsInstance(response, main.DashboardAccountStatsResponse)
+        self.assertEqual(response.total_accounts, 12)
+        self.assertEqual(response.online_accounts, 5)
+        self.assertEqual(response.my_accounts, 3)
+        self.assertEqual(response.my_online_accounts, 2)
+        db.execute.assert_awaited_once()
+
+    async def test_owner_scope_is_only_inside_personal_aggregates(self):
+        row = _AggregateRow(
+            total_accounts=8,
+            online_accounts=4,
+            my_accounts=2,
+            my_online_accounts=1,
+        )
+        db = SimpleNamespace(
+            execute=AsyncMock(return_value=_AggregateResult(row))
+        )
+        user = SimpleNamespace(id=73, role="viewer")
+
+        await main.get_dashboard_account_stats(db=db, user=user)
+
+        statement = db.execute.await_args.args[0]
+        sql = " ".join(str(statement).lower().split())
+        compiled_params = list(statement.compile().params.values())
+
+        # All roles receive the same global totals.  The current user id may
+        # appear in CASE expressions for the two personal counters, but must
+        # never filter the entire aggregate query through a global WHERE.
+        self.assertIn("owner_id", sql)
+        self.assertGreaterEqual(sql.count("case when"), 3)
+        self.assertEqual(sql.count("accounts.owner_id"), 2)
+        self.assertIn(73, compiled_params)
+        self.assertNotIn(" where ", f" {sql} ")
+        self.assertEqual(db.execute.await_count, 1)
+
+
+if __name__ == "__main__":
+    unittest.main()