tim/bridzik
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
8e18325660128753a0b6cb2e34b4b3aaab099b28
bridzik/tests/test_stats.py
T
timandClaude Fable 5 5b9d6342ad Auth: nedokoncena registracia sa da dokoncit novym QR kodom 
Ucet bez potvrdeneho kodu (auth_token is NULL a totp_last_step == 0)
uz neblokuje meno: opakovany register_account vyda novy secret (stary
QR prestane platit) a login vyhodi RegistrationIncomplete, na ktoru
server odpovie novym QR -- klient sa prepne na registracny tab.
Admin statistiky vykazuju nedokoncene registracie osobitne, Hraci
celkom pocita len potvrdene ucty. Novy event register v analytike.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-04 17:34:03 +02:00

392 lines
17 KiB
Python
Raw Blame History

"""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.confirm_account(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="/history", 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("/history", 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_dynamic_path_segments_normalized(self):
self.assertEqual(stats._normalize_path("/lobby/abc-123"), "/lobby")
self.assertEqual(stats._normalize_path("/game/abc-123"), "/game")
self.assertEqual(stats._normalize_path("/lobby"), "/lobby")
self.assertEqual(stats._normalize_path("/history"), "/history")
# /gameXYZ nie je /game/<id> -- nesmie sa orezat
self.assertEqual(stats._normalize_path("/gamex"), "/gamex")
def test_skipped_paths_not_recorded(self):
# "/", "/lobby" a "/game" (aj s dynamickym ID) sa vobec nezapisuju --
# vysoka frekvencia bez analytickej hodnoty; "/auth" zas preto, ze je
# to vzdy 1:1 duplicita eventu "landing" (api/stats.py _SKIPPED_PATHS).
async def _count():
from sqlalchemy import func, select as sa_select
from db.db import async_session
from db.models import PageView
async with async_session() as session:
return (
await session.execute(sa_select(func.count()).select_from(PageView))
).scalar()
before = run(_count())
for path in ("/", "/auth", "/lobby", "/lobby/abc-123", "/game", "/game/xyz-789"):
run(stats.record_pageview(path=path, referrer="", user_agent=CHROME_UA))
after = run(_count())
self.assertEqual(after, before)
def test_event_path_stores_player_id(self):
# Konvencia: nazov bez "/" na zaciatku = pomenovany event (napr. "login"
# zapisovany zo socket handlera po uspesnom prihlaseni), nie URL cesta.
ids = self._make_players(n=1)
run(stats.record_pageview(
path="login", referrer="", user_agent=CHROME_UA, player_id=ids[0],
))
async def _last():
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.path, row.player_id
path, player_id = run(_last())
self.assertEqual(path, "login")
self.assertEqual(player_id, ids[0])
data = run(stats.get_daily_stats())
self.assertGreaterEqual(data["top_paths"].get("login", 0), 1)
def test_landing_event_preserves_referrer(self):
# "landing" event (main.tsx, jeden na kazdy plny load stranky) nesie
# referrer prveho dotyku -- jediny zaznam z navstevy, ktora zacina na
# "/" (preklik z FB a pod.), kedze "/" aj REPLACE redirecty sa skipuju.
run(stats.record_pageview(
path="landing",
referrer="https://facebook.com/",
user_agent=CHROME_UA,
ip="203.0.113.99",
))
data = run(stats.get_daily_stats())
self.assertGreaterEqual(data["top_paths"].get("landing", 0), 1)
self.assertGreaterEqual(data["top_referrers"].get("https://facebook.com/", 0), 1)
def test_rules_view_event_recorded_without_player_id(self):
run(stats.record_pageview(path="rules_view", referrer="", user_agent=CHROME_UA))
data = run(stats.get_daily_stats())
self.assertGreaterEqual(data["top_paths"].get("rules_view", 0), 1)
def test_logged_in_only_scope_counts_login_events_only(self):
# Bezne beacony uz neposielaju player_id vobec -- scope "logged_in"
# preto filtruje priamo podla path == "login", nie podla pritomnosti
# player_id. Obycajna navsteva (aj s player_id) sa do neho nepocita.
ids = self._make_players(n=1)
ip = "203.0.113.99"
before_all = run(stats.get_daily_stats(logged_in_only=False))
before_logged_in = run(stats.get_daily_stats(logged_in_only=True))
all_before = sum(before_all["pageviews_per_day"].values())
logged_in_before = sum(before_logged_in["pageviews_per_day"].values())
# Bezna navsteva -- ma pribudnut len vo "vsetci", aj keby mala player_id.
run(stats.record_pageview(
path="/history", referrer="", user_agent=CHROME_UA, ip=ip, player_id=ids[0],
))
# Login event -- ma pribudnut v oboch.
run(stats.record_pageview(
path="login", referrer="", user_agent=CHROME_UA, ip=ip, player_id=ids[0],
))
after_all = run(stats.get_daily_stats(logged_in_only=False))
after_logged_in = run(stats.get_daily_stats(logged_in_only=True))
self.assertEqual(sum(after_all["pageviews_per_day"].values()), all_before + 2)
self.assertEqual(sum(after_logged_in["pageviews_per_day"].values()), logged_in_before + 1)
def test_logged_in_only_breakdown_tables_count_each_login_not_deduped(self):
# browsers/os/device_types/top_referrers/countries pouzivaju v scope
# "logged_in" priamy pocet login-eventov, nie unikatny navstevnicky
# den -- 2x prihlasenie tym istym prehliadacom/IP v ten isty den sa
# ma prejavit ako 2, rovnako ako v grafe pageviews_per_day.
ids = self._make_players(n=1)
before = run(stats.get_daily_stats(logged_in_only=True))
before_browsers = sum(before["browsers"].values())
run(stats.record_pageview(path="login", referrer="", user_agent=CHROME_UA, player_id=ids[0]))
run(stats.record_pageview(path="login", referrer="", user_agent=CHROME_UA, player_id=ids[0]))
data = run(stats.get_daily_stats(logged_in_only=True))
self.assertEqual(sum(data["browsers"].values()), before_browsers + 2)
def test_logged_in_visitors_per_day_counts_distinct_players_not_ip(self):
# V scope "logged_in" ma visitors_per_day znamenat unikatnych HRACOV
# (player_id) za den, nie unikatne IP+UA -- 2x prihlasenie toho
# isteho hraca (aj z inej IP/prehliadaca) sa ma pocitat len raz, na
# rozdiel od pageviews_per_day, kde sa kazdy login pocita zvlast.
ids = self._make_players(n=2)
before = run(stats.get_daily_stats(logged_in_only=True))
before_visitors = sum(before["visitors_per_day"].values())
before_pageviews = sum(before["pageviews_per_day"].values())
# Ten isty hrac, 2x prihlasenie z roznych "zariadeni" (rozne IP/UA).
run(stats.record_pageview(
path="login", referrer="", user_agent=CHROME_UA, ip="203.0.113.10", player_id=ids[0],
))
run(stats.record_pageview(
path="login", referrer="", user_agent=CHROME_UA, ip="203.0.113.11", player_id=ids[0],
))
# Iny hrac, 1x prihlasenie.
run(stats.record_pageview(
path="login", referrer="", user_agent=CHROME_UA, ip="203.0.113.12", player_id=ids[1],
))
data = run(stats.get_daily_stats(logged_in_only=True))
self.assertEqual(sum(data["pageviews_per_day"].values()), before_pageviews + 3)
self.assertEqual(sum(data["visitors_per_day"].values()), before_visitors + 2)
def test_visitors_counted_once_per_day_per_ip(self):
# 3 kliky z tej istej IP v ten isty den -> pageviews +3, visitors iba 1.
ip = "198.51.100.77"
for path in ("/history", "/history", "/history"):
run(stats.record_pageview(path=path, referrer="", user_agent=CHROME_UA, ip=ip))
data = run(stats.get_daily_stats())
today = next(iter(data["visitors_per_day"]))
# V testovej DB su vsetky zaznamy z dneska; unikatnych IP je menej nez klikov.
self.assertLess(data["visitors_per_day"][today], data["pageviews_per_day"][today])
# Tabulky zlozenia publika pocitaju den+IP raz -- 3 kliky tej istej IP
# nesmu zdvihnut "Chrome" o 3. Overime, ze pocet je mensi nez pocet klikov.
self.assertLess(data["browsers"].get("Chrome", 0), sum(data["pageviews_per_day"].values()))
def test_same_ip_different_user_agent_counts_as_two_visitors(self):
ip = "192.0.2.44"
firefox_ua = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:126.0) "
"Gecko/20100101 Firefox/126.0"
)
before = run(stats.get_daily_stats())
today_before = sum(before["visitors_per_day"].values())
# Ta ista IP, dva rozne prehliadace -> 2 navstevnici (kazdy klikne 2x).
for _ in range(2):
run(stats.record_pageview(path="/history", referrer="", user_agent=CHROME_UA, ip=ip))
run(stats.record_pageview(path="/history", referrer="", user_agent=firefox_ua, ip=ip))
after = run(stats.get_daily_stats())
self.assertEqual(sum(after["visitors_per_day"].values()), today_before + 2)
def test_pageviews_per_day_breakdown_by_dimension(self):
run(stats.record_pageview(path="/history", 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_all_scope_chart_breakdown_matches_composition_tables(self):
# Bug hlaseny uzivatelom: v scope "vsetci" graf (pageviews_per_day_by_*)
# pocital kazdy klik zvlast, zatial co tabulky nizsie (browsers/os/
# device_types) pocitali unikatny "navstevnicky den" -- cisla si tak
# nesedeli. Oboje ma teraz rovnaku dedup logiku (jeden navstevnik +
# kategoria + den = 1), takze sucty musia byt zhodne.
ip = "203.0.113.50"
for _ in range(3):
run(stats.record_pageview(path="/history", referrer="", user_agent=CHROME_UA, ip=ip))
data = run(stats.get_daily_stats(logged_in_only=False))
chart_pc_total = sum(day.get("pc", 0) for day in data["pageviews_per_day_by_device"].values())
chart_chrome_total = sum(
day.get("Chrome", 0) for day in data["pageviews_per_day_by_browser"].values()
)
self.assertEqual(chart_pc_total, data["device_types"].get("pc", 0))
self.assertEqual(chart_chrome_total, data["browsers"].get("Chrome", 0))
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_unconfirmed_players_counted_separately(self):
before = run(stats.get_daily_stats())
# Registracia bez potvrdenia kodu -> nedokonceny ucet
username = "ghost_" + uuid.uuid4().hex[:8]
data = run(auth.register_account(username))
after = run(stats.get_daily_stats())
self.assertEqual(after["total_players"], before["total_players"])
self.assertEqual(after["unconfirmed_players"], before["unconfirmed_players"] + 1)
# Po potvrdeni sa presunie medzi potvrdenych hracov
run(auth.confirm_account(username, pyotp.TOTP(data["secret"]).now()))
confirmed = run(stats.get_daily_stats())
self.assertEqual(confirmed["total_players"], before["total_players"] + 1)
self.assertEqual(confirmed["unconfirmed_players"], before["unconfirmed_players"])
def test_empty_referrer_excluded_from_top_referrers(self):
run(stats.record_pageview(path="/history", 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 View Git Blame Copy Permalink