Add self-hosted usage analytics: pageview tracking + admin stats dashboard
Records pageviews (path, referrer, browser/OS/device, IP, GeoIP country) via a POST /api/track beacon into a new PageView table, and exposes aggregated daily/breakdown stats behind a token-gated GET /api/admin/stats endpoint with brute-force lockout. Frontend gets a /admin dashboard (charts + breakdown tables) built on recharts, with a switchable per-day pageviews chart. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 5
@@ -0,0 +1,180 @@
|
||||
"""Testy self-hosted usage analytics (api/stats.py).
|
||||
|
||||
Bezia na docasnom SQLite subore, rovnaky pattern ako tests/test_history.py.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
import uuid
|
||||
from types import SimpleNamespace
|
||||
|
||||
# Nastav DB/ENCRYPTION_KEY PRED importom db/api modulov -- engine sa vytvara pri importe.
|
||||
_DB_FILE = os.path.join(tempfile.gettempdir(), f"bridzik_test_{uuid.uuid4().hex}.db")
|
||||
os.environ["DATABASE_URL"] = "sqlite+aiosqlite:///" + _DB_FILE.replace("\\", "/")
|
||||
|
||||
from cryptography.fernet import Fernet # noqa: E402
|
||||
|
||||
os.environ.setdefault("ENCRYPTION_KEY", Fernet.generate_key().decode())
|
||||
|
||||
import pyotp # noqa: E402
|
||||
|
||||
import api as api_module # noqa: E402
|
||||
from api import auth, history, stats # noqa: E402
|
||||
from db.db import init_db # noqa: E402
|
||||
|
||||
CHROME_UA = (
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||||
"(KHTML, like Gecko) Chrome/120.0 Safari/537.36"
|
||||
)
|
||||
|
||||
|
||||
def run(coro):
|
||||
return asyncio.run(coro)
|
||||
|
||||
|
||||
def make_core(completed=True):
|
||||
"""Stub jednej hry s jednym dohratym kolom (seria 0, kolo 0)."""
|
||||
rnd = SimpleNamespace(
|
||||
round_number=0,
|
||||
guesses={0: 2, 1: 1, 2: 0, 3: 1},
|
||||
is_completed=lambda: True,
|
||||
get_points_summary=lambda: [12, 0, 10, 11],
|
||||
)
|
||||
series = SimpleNamespace(
|
||||
series_number=0, rounds=[rnd], get_last_round=lambda: rnd
|
||||
)
|
||||
return SimpleNamespace(series=[series], is_completed=lambda: completed)
|
||||
|
||||
|
||||
class StatsCase(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
run(init_db())
|
||||
|
||||
def _make_players(self, n=4):
|
||||
ids = []
|
||||
for _ in range(n):
|
||||
username = "u_" + uuid.uuid4().hex[:8]
|
||||
data = run(auth.register_account(username))
|
||||
ident = run(auth.login(username, pyotp.TOTP(data["secret"]).now()))
|
||||
ids.append(ident["player_id"])
|
||||
return ids
|
||||
|
||||
def test_record_pageview_parses_user_agent(self):
|
||||
run(stats.record_pageview(
|
||||
path="/", referrer="https://example.com", user_agent=CHROME_UA, ip="203.0.113.5",
|
||||
))
|
||||
|
||||
data = run(stats.get_daily_stats())
|
||||
self.assertGreaterEqual(sum(data["pageviews_per_day"].values()), 1)
|
||||
self.assertIn("Chrome", data["browsers"])
|
||||
self.assertIn("Windows", data["operating_systems"])
|
||||
self.assertGreaterEqual(data["device_types"].get("pc", 0), 1)
|
||||
self.assertGreaterEqual(data["top_referrers"].get("https://example.com", 0), 1)
|
||||
self.assertGreaterEqual(data["top_paths"].get("/", 0), 1)
|
||||
|
||||
# Bez GEOIP_DB_PATH (v testoch nenastaveny) sa krajina jednoducho nerozlusi,
|
||||
# ale IP sa uz do page_views ulozi.
|
||||
async def _last_ip():
|
||||
from sqlalchemy import select as sa_select
|
||||
|
||||
from db.db import async_session
|
||||
from db.models import PageView
|
||||
|
||||
async with async_session() as session:
|
||||
row = (
|
||||
await session.execute(
|
||||
sa_select(PageView).order_by(PageView.id.desc()).limit(1)
|
||||
)
|
||||
).scalar_one()
|
||||
return row.ip, row.country
|
||||
|
||||
ip, country = run(_last_ip())
|
||||
self.assertEqual(ip, "203.0.113.5")
|
||||
self.assertEqual(country, "")
|
||||
|
||||
def test_country_for_ip_without_geoip_db_path_returns_empty(self):
|
||||
os.environ.pop("GEOIP_DB_PATH", None)
|
||||
stats._geoip_reader = None
|
||||
stats._geoip_load_attempted = False
|
||||
self.assertEqual(stats._country_for_ip("203.0.113.5"), "")
|
||||
self.assertEqual(stats._country_for_ip(""), "")
|
||||
|
||||
def test_pageviews_per_day_breakdown_by_dimension(self):
|
||||
run(stats.record_pageview(path="/", referrer="", user_agent=CHROME_UA))
|
||||
data = run(stats.get_daily_stats())
|
||||
today = next(iter(data["pageviews_per_day_by_device"]))
|
||||
self.assertGreaterEqual(data["pageviews_per_day_by_device"][today].get("pc", 0), 1)
|
||||
self.assertIn("Chrome", data["pageviews_per_day_by_browser"][today])
|
||||
self.assertIn("Windows", data["pageviews_per_day_by_os"][today])
|
||||
|
||||
def test_daily_stats_reflect_games_and_players(self):
|
||||
before = run(stats.get_daily_stats())
|
||||
base_total_players = before["total_players"]
|
||||
|
||||
ids = self._make_players()
|
||||
gid = str(uuid.uuid4())
|
||||
run(history.record_game_started(gid, "Test", ids))
|
||||
run(history.record_completed_rounds(gid, make_core()))
|
||||
|
||||
data = run(stats.get_daily_stats())
|
||||
self.assertEqual(data["total_players"], base_total_players + 4)
|
||||
self.assertEqual(data["completion_rate"], 1.0)
|
||||
self.assertGreaterEqual(sum(data["games_per_day"].values()), 1)
|
||||
self.assertGreaterEqual(sum(data["rounds_per_day"].values()), 4)
|
||||
|
||||
def test_empty_referrer_excluded_from_top_referrers(self):
|
||||
run(stats.record_pageview(path="/lobby", referrer="", user_agent=CHROME_UA))
|
||||
data = run(stats.get_daily_stats())
|
||||
self.assertNotIn("", data["top_referrers"])
|
||||
|
||||
|
||||
def _scope(ip="1.2.3.4", token=None):
|
||||
headers = [(b"x-forwarded-for", ip.encode())]
|
||||
if token is not None:
|
||||
headers.append((b"authorization", f"Bearer {token}".encode()))
|
||||
return {"headers": headers, "client": ("9.9.9.9", 0)}
|
||||
|
||||
|
||||
class AdminAuthCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
os.environ["ADMIN_TOKEN"] = "secret-token"
|
||||
api_module._admin_failed_attempts.clear()
|
||||
|
||||
def tearDown(self):
|
||||
os.environ.pop("ADMIN_TOKEN", None)
|
||||
|
||||
def test_client_ip_prefers_x_forwarded_for(self):
|
||||
self.assertEqual(api_module._client_ip(_scope(ip="5.6.7.8")), "5.6.7.8")
|
||||
self.assertEqual(api_module._client_ip({"headers": [], "client": ("9.9.9.9", 0)}), "9.9.9.9")
|
||||
|
||||
def test_wrong_token_rejected_and_locked_out_after_repeated_failures(self):
|
||||
ip = "1.1.1.1"
|
||||
self.assertFalse(api_module._admin_authorized(_scope(ip=ip, token="wrong")))
|
||||
|
||||
for _ in range(api_module._ADMIN_ATTEMPT_LIMIT):
|
||||
self.assertFalse(api_module._admin_locked_out(ip))
|
||||
api_module._register_admin_failure(ip)
|
||||
|
||||
self.assertTrue(api_module._admin_locked_out(ip))
|
||||
# Iny IP nie je zamknuty
|
||||
self.assertFalse(api_module._admin_locked_out("2.2.2.2"))
|
||||
|
||||
def test_correct_token_authorized(self):
|
||||
self.assertTrue(api_module._admin_authorized(_scope(token="secret-token")))
|
||||
|
||||
|
||||
def tearDownModule():
|
||||
from db.db import engine
|
||||
|
||||
run(engine.dispose())
|
||||
try:
|
||||
os.remove(_DB_FILE)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user