"""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="/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/ -- 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 (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 ("/", "/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_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_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()