94 lines
2.8 KiB
Python
94 lines
2.8 KiB
Python
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()
|