tests/test_fullstack_parity.py

"""E2E-ish tests covering the new fullstack-parity wiring.

These tests don't require Docker / Postgres — they:

* exercise i18n keys for completeness,
* exercise the initData HMAC verifier (positive + negative cases),
* drive the gatekeeper handler with a stubbed bot + in-memory DB,
* spin up the real aiohttp app with stub bot+db and hit /api/me, /api/stats.

To run:

    BOT_TOKEN=test:abc WEBHOOK_BASE_URL=https://example.com \\
    WEBHOOK_SECRET=secret DATABASE_URL=postgres://x \\
    ADMIN_USER_IDS=111 \\
    .venv/Scripts/python.exe -m pytest tests/test_fullstack_parity.py -v
"""
from __future__ import annotations

import asyncio
import hashlib
import hmac
import json
import os
import time
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional

import pytest
import pytest_asyncio

# --- env so config.load_settings() doesn't blow up at import time -----------
os.environ.setdefault("BOT_TOKEN", "test:abc")
os.environ.setdefault("WEBHOOK_BASE_URL", "https://example.com")
os.environ.setdefault("WEBHOOK_SECRET", "secret")
os.environ.setdefault("DATABASE_URL", "postgres://x")
os.environ.setdefault("ADMIN_USER_IDS", "111")

from app.config import load_settings  # noqa: E402
from app.i18n import LOCALES, t, infer_lang_from_telegram  # noqa: E402
from app.webapp import _verify_init_data, setup_webapp  # noqa: E402


# ----------------------------------------------------------------------------
# i18n: every key in EN must exist in RU and AZ
# ----------------------------------------------------------------------------
def test_i18n_locale_completeness():
    en_keys = set(LOCALES["en"].keys())
    ru_keys = set(LOCALES["ru"].keys())
    az_keys = set(LOCALES["az"].keys())
    missing_ru = en_keys - ru_keys
    missing_az = en_keys - az_keys
    assert not missing_ru, f"RU missing keys: {missing_ru}"
    assert not missing_az, f"AZ missing keys: {missing_az}"


def test_i18n_inference():
    assert infer_lang_from_telegram("ru") == "ru"
    assert infer_lang_from_telegram("ru-RU") == "ru"
    assert infer_lang_from_telegram("az-Latn") == "az"
    assert infer_lang_from_telegram("en-US") == "en"
    assert infer_lang_from_telegram("fr") == "en"
    assert infer_lang_from_telegram(None) == "en"


def test_i18n_format_safety():
    """Missing variables in templates fall back to the raw template, not crash."""
    out = t("ru", "gk_owner_dm_body", title="x")
    assert "{added_by}" in out or "x" in out  # didn't raise


# ----------------------------------------------------------------------------
# initData verifier
# ----------------------------------------------------------------------------
def _signed_init_data(token: str, *, age: int = 0, override_user: dict | None = None) -> str:
    auth_date = int(time.time()) - age
    user_obj = override_user or {"id": 42, "first_name": "A", "username": "a"}
    pairs = {
        "auth_date": str(auth_date),
        "query_id": "AAH",
        "user": json.dumps(user_obj),
    }
    data_check = "\n".join(f"{k}={pairs[k]}" for k in sorted(pairs))
    secret = hmac.new(b"WebAppData", token.encode(), hashlib.sha256).digest()
    h = hmac.new(secret, data_check.encode(), hashlib.sha256).hexdigest()
    return "&".join(f"{k}={v}" for k, v in pairs.items()) + f"&hash={h}"


def test_init_data_valid():
    raw = _signed_init_data("test:abc")
    parsed = _verify_init_data(raw, "test:abc")
    assert parsed is not None
    assert parsed["user"]["id"] == 42


def test_init_data_wrong_token():
    raw = _signed_init_data("test:abc")
    assert _verify_init_data(raw, "different:token") is None


def test_init_data_tampered_hash():
    raw = _signed_init_data("test:abc")
    # Replace the hash with all zeros
    parts = raw.split("&hash=")
    tampered = parts[0] + "&hash=" + ("0" * 64)
    assert _verify_init_data(tampered, "test:abc") is None


def test_init_data_expired():
    raw = _signed_init_data("test:abc", age=2 * 24 * 3600)
    assert _verify_init_data(raw, "test:abc") is None


def test_init_data_empty():
    assert _verify_init_data("", "test:abc") is None
    assert _verify_init_data(None, "test:abc") is None  # type: ignore[arg-type]


# ----------------------------------------------------------------------------
# Stub DB + Bot for gatekeeper / API integration
# ----------------------------------------------------------------------------
@dataclass
class StubChat:
    chat_id: int
    title: str = "Test Chat"
    added_by: Optional[int] = None
    status: str = "pending"
    status_at: Any = None
    status_by: Optional[int] = None
    member_count: Optional[int] = None
    added_at: Any = field(default_factory=lambda: __import__("datetime").datetime.utcnow())


class StubDB:
    def __init__(self):
        self.chats: Dict[int, StubChat] = {}
        self.user_lang: Dict[int, str] = {}
        self.events: List[Dict[str, Any]] = []
        self.settings: Dict[int, Dict[str, Any]] = {}
        self.connected = True

    async def upsert_chat(self, chat_id, title, added_by, member_count=None, initial_status="pending"):
        existing = self.chats.get(chat_id)
        if existing:
            if existing.status not in ("approved", "rejected"):
                existing.status = "pending"
            existing.title = title
            return existing.status
        c = StubChat(chat_id=chat_id, title=title, added_by=added_by,
                     member_count=member_count, status=initial_status)
        self.chats[chat_id] = c
        return c.status

    async def set_chat_status(self, chat_id, status, decided_by=None):
        c = self.chats.get(chat_id)
        if c:
            c.status = status
            c.status_by = decided_by

    async def get_chat(self, chat_id):
        c = self.chats.get(chat_id)
        if not c:
            return None
        return {
            "chat_id": c.chat_id, "title": c.title, "added_by": c.added_by,
            "status": c.status, "status_at": c.status_at, "status_by": c.status_by,
            "member_count": c.member_count, "added_at": c.added_at,
        }

    async def is_chat_approved(self, chat_id):
        c = self.chats.get(chat_id)
        return bool(c and c.status == "approved")

    async def list_chats_for_admin(self, _admin_user_id):
        return [
            {"chat_id": c.chat_id, "title": c.title, "status": c.status,
             "member_count": c.member_count, "added_at": c.added_at, "status_at": c.status_at}
            for c in self.chats.values() if c.status == "approved"
        ]

    async def mark_chat_removed(self, chat_id):
        await self.set_chat_status(chat_id, "removed")

    async def delete_chat(self, chat_id):
        self.chats.pop(chat_id, None)

    async def get_user_lang(self, user_id):
        return self.user_lang.get(user_id, "en")

    async def set_user_lang(self, user_id, lang):
        self.user_lang[user_id] = lang

    async def stats_for_chat(self, chat_id):
        return {
            "d24": 1, "week": 5, "week_delta_pct": 50, "total": 10,
            "by_category": [{"key": "porn", "count": 3}],
            "daily": [{"day": "2026-05-29", "count": 1}],
            "top_violators": [{"user_id": 7, "name": "X", "count": 3}],
        }

    async def events_for_chat(self, chat_id, kinds=None, limit=100):
        return list(self.events)

    async def get_settings(self, chat_id):
        from app.database import ChatSettings
        return ChatSettings(chat_id=chat_id)

    async def update_setting(self, chat_id, key, value):
        self.settings.setdefault(chat_id, {})[key] = value

    async def record_event(self, **kw):
        self.events.append(kw)


class StubBot:
    """Mimics aiogram Bot calls used by the webapp module."""
    def __init__(self):
        self.sent: List[Dict[str, Any]] = []
        self.left: List[int] = []
        self.member_admin_for: Dict[tuple[int, int], bool] = {}

    async def send_message(self, chat_id, text, **kw):
        self.sent.append({"chat_id": chat_id, "text": text, **kw})
        return type("Msg", (), {"message_id": 1})()

    async def leave_chat(self, chat_id):
        self.left.append(chat_id)
        return True

    async def get_chat_member_count(self, chat_id):
        return 7

    async def get_chat_member(self, chat_id, user_id):
        is_admin = self.member_admin_for.get((chat_id, user_id), False)
        from aiogram.enums import ChatMemberStatus
        cls = type("M", (), {"status": (
            ChatMemberStatus.ADMINISTRATOR if is_admin else ChatMemberStatus.MEMBER
        )})
        return cls()


# ----------------------------------------------------------------------------
# Gatekeeper logic with stub DB
# ----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_gatekeeper_owner_self_add_auto_approves():
    db = StubDB()
    settings = load_settings()
    owner = settings.owner_id
    assert owner is not None
    # Random user adds → pending
    status_random = await db.upsert_chat(
        chat_id=-100, title="random", added_by=999,
        member_count=5, initial_status="pending",
    )
    assert status_random == "pending"
    # Owner adds another chat → approved
    status_owner = await db.upsert_chat(
        chat_id=-200, title="owner", added_by=owner,
        member_count=5, initial_status="approved",
    )
    assert status_owner == "approved"


@pytest.mark.asyncio
async def test_gatekeeper_re_add_keeps_decision():
    db = StubDB()
    # Initial: pending
    await db.upsert_chat(chat_id=-100, title="t", added_by=1, initial_status="pending")
    # Owner approves
    await db.set_chat_status(-100, "approved", decided_by=111)
    # Re-add by random — must keep 'approved'
    again = await db.upsert_chat(chat_id=-100, title="t", added_by=2, initial_status="pending")
    assert again == "approved"


@pytest.mark.asyncio
async def test_gatekeeper_blocks_moderation():
    db = StubDB()
    await db.upsert_chat(chat_id=-100, title="t", added_by=1, initial_status="pending")
    assert (await db.is_chat_approved(-100)) is False
    await db.set_chat_status(-100, "approved")
    assert (await db.is_chat_approved(-100)) is True


# ----------------------------------------------------------------------------
# aiohttp app: hit /api/me with valid + invalid initData
# ----------------------------------------------------------------------------
@pytest_asyncio.fixture
async def http_app(aiohttp_client):
    from aiohttp import web
    app = web.Application()
    db = StubDB()
    bot = StubBot()
    settings = load_settings()
    setup_webapp(app, bot=bot, db=db, settings=settings)
    client = await aiohttp_client(app)
    return client, db, bot, settings


@pytest.mark.asyncio
async def test_api_me_valid(http_app):
    client, db, bot, settings = http_app
    raw = _signed_init_data(settings.bot_token, override_user={"id": 5, "first_name": "U", "language_code": "ru"})
    r = await client.get("/api/me", headers={"X-Telegram-Init-Data": raw})
    assert r.status == 200
    body = await r.json()
    assert body["id"] == 5
    assert body["lang"] == "ru"
    assert body["is_owner"] is False


@pytest.mark.asyncio
async def test_api_me_unauthorized(http_app):
    client, *_ = http_app
    r = await client.get("/api/me", headers={"X-Telegram-Init-Data": "garbage"})
    assert r.status == 401


@pytest.mark.asyncio
async def test_api_chat_admin_required(http_app):
    client, db, bot, settings = http_app
    raw = _signed_init_data(settings.bot_token, override_user={"id": 5, "first_name": "U"})
    # Chat exists and is approved, but user 5 is not admin
    await db.upsert_chat(-100, "t", added_by=1, initial_status="approved")
    r = await client.get("/api/stats/-100", headers={"X-Telegram-Init-Data": raw})
    assert r.status == 403
    body = await r.json()
    assert body["error"]["code"] == "not_admin"
    # Now make the user an admin (clear the in-process TTL cache so the new
    # value is honored without waiting 60s).
    from app import webapp as _wm
    _wm._admin_cache.clear()
    bot.member_admin_for[(-100, 5)] = True
    r2 = await client.get("/api/stats/-100", headers={"X-Telegram-Init-Data": raw})
    assert r2.status == 200
    s = await r2.json()
    assert s["d24"] == 1
    assert s["total"] == 10


@pytest.mark.asyncio
async def test_api_static_app_present(http_app):
    client, *_ = http_app
    r = await client.get("/app/index.html")
    # Either the index is served (200) or the static dir exists with the file we need.
    assert r.status in (200, 404), "static route must respond"


@pytest.mark.asyncio
async def test_api_patch_settings_writes_event(http_app):
    client, db, bot, settings = http_app
    raw = _signed_init_data(settings.bot_token, override_user={"id": 7, "first_name": "U"})
    await db.upsert_chat(-100, "t", added_by=1, initial_status="approved")
    bot.member_admin_for[(-100, 7)] = True
    r = await client.patch(
        "/api/settings/-100",
        headers={"X-Telegram-Init-Data": raw, "Content-Type": "application/json"},
        data=json.dumps({"threshold": 85, "block_sexy": True, "garbage_key": "ignored"}),
    )
    assert r.status == 200, await r.text()
    body = await r.json()
    assert body["applied"]["threshold"] == 85
    assert body["applied"]["block_sexy"] is True
    assert "garbage_key" not in body["applied"]
    # Audit event recorded
    assert any(e.get("kind") == "settings_changed" for e in db.events)


@pytest.mark.asyncio
async def test_api_patch_antispam_keys_accepted(http_app):
    """Regression: the frontend used to send local UI keys (antispam_warn,
    violation_window, ban_duration, delete_warning_seconds) that the backend
    silently dropped. After the frontend key-map fix it sends the real column
    names — assert the backend persists them."""
    client, db, bot, settings = http_app
    raw = _signed_init_data(settings.bot_token, override_user={"id": 9, "first_name": "U"})
    await db.upsert_chat(-100, "t", added_by=1, initial_status="approved")
    bot.member_admin_for[(-100, 9)] = True
    payload = {
        "warn_user": False,
        "violation_window_minutes": 360,
        "ban_duration_minutes": 0,        # frontend maps "forever" → 0
        "warn_delete_seconds": 60,
    }
    r = await client.patch(
        "/api/settings/-100",
        headers={"X-Telegram-Init-Data": raw, "Content-Type": "application/json"},
        data=json.dumps(payload),
    )
    assert r.status == 200, await r.text()
    applied = (await r.json())["applied"]
    for k, v in payload.items():
        assert applied.get(k) == v, f"{k} not applied (got {applied.get(k)})"