bridzik/tests/test_stats.py

"""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()