"""Ревью-тесты стейт-машины логина без реального Telegram.

Фейковый TelegramClient + фейковый telethon.errors/functions: каждый сценарий
(SMS, 2FA, отмена, QR, sweeper, осиротевшие строки) проверяется против
реального login_flow и реальной SQLite-схемы. Тесты self-contained: БД и
клиент создаются в тесте, соединение закрывается до выхода (тот же loop).
"""

import asyncio
import sys
import types
from datetime import datetime, timedelta, timezone
from pathlib import Path
from types import SimpleNamespace

import pytest

BACKEND = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(BACKEND))


# --- фейковый telethon.errors/functions -----------------------------------

class _TGError(Exception):  # noqa: N818
    def __init__(self, *args, **kwargs):
        super().__init__(args or kwargs)
        if args and isinstance(args[0], int):
            self.seconds = args[0]


PHONE_INVALID = type("PhoneNumberInvalidError", (_TGError,), {})
FLOOD_WAIT = type("FloodWaitError", (_TGError,), {})
API_ID_INVALID = type("ApiIdInvalidError", (_TGError,), {})
CODE_INVALID = type("PhoneCodeInvalidError", (_TGError,), {})
CODE_EXPIRED = type("PhoneCodeExpiredError", (_TGError,), {})
PASSWORD_NEEDED = type("SessionPasswordNeededError", (_TGError,), {})
PASSWORD_HASH_INVALID = type("PasswordHashInvalidError", (_TGError,), {})

fake_errors = types.SimpleNamespace(
    PhoneNumberInvalidError=PHONE_INVALID, FloodWaitError=FLOOD_WAIT,
    ApiIdInvalidError=API_ID_INVALID, PhoneCodeInvalidError=CODE_INVALID,
    PhoneCodeExpiredError=CODE_EXPIRED, SessionPasswordNeededError=PASSWORD_NEEDED,
    PasswordHashInvalidError=PASSWORD_HASH_INVALID,
)


class FakeInner:
    """Поведение задаётся тестом: sign_in_effect / qr-эффекты / hang_cancel."""

    def __init__(self):
        self.alive = False
        self.phone = ""
        self.phone_code_hash = "h" * 32
        self.hang_cancel = False
        self.cancel_calls = 0
        self.sign_in_effect = None   # None = успех; класс-исключение
        self.user_me = SimpleNamespace(
            id=42, phone="+79991234567", username="tester",
            first_name="Тест", last_name="Юзер",
        )

    async def connect(self):
        self.alive = True

    def is_connected(self):
        return self.alive

    async def send_code_request(self, phone):
        assert self.alive
        self.phone = phone
        return SimpleNamespace(phone_code_hash=self.phone_code_hash)

    async def sign_in(self, *args, **kwargs):
        if self.sign_in_effect is not None:
            raise self.sign_in_effect()
        return SimpleNamespace()

    async def get_me(self):
        return self.user_me

    async def disconnect(self):
        self.alive = False


class FakeQR:
    def __init__(self):
        self.urls = ["tg://login?token=AAA", "tg://login?token=BBB"]
        self.n = 0
        self.wait_effect = None      # None = успех (скан принят)
        self.recreate_effect = None  # исключение при первом recreate
        self.user_me = None
        self.url_prop = ""  # пустой url имитирует потерю токена

    @property
    def url(self):
        return self.url_prop or self.urls[min(self.n, len(self.urls) - 1)]

    async def wait(self, *a, **kw):
        if self.wait_effect is not None:
            raise self.wait_effect()
        return self.user_me

    async def recreate(self):
        if self.recreate_effect is not None and self.n == 0:
            raise self.recreate_effect()
        self.n += 1


class FakeClient:
    def __init__(self, api_id, api_hash):
        self.inner = FakeInner()
        self.api_id = api_id
        self.qr = None

    @property
    def session(self):
        return SimpleNamespace(save=lambda: "session-string")

    async def connect(self):
        await self.inner.connect()
        return self

    def is_connected(self):
        return self.inner.is_connected()

    async def send_code_request(self, phone):
        return await self.inner.send_code_request(phone)

    async def qr_login(self):
        self.qr = FakeQR()
        return self.qr

    async def sign_in(self, *args, **kwargs):
        return await self.inner.sign_in(*args, **kwargs)

    async def get_me(self):
        return await self.inner.get_me()

    async def disconnect(self):
        await self.inner.disconnect()

    async def __call__(self, request):
        # fake_functions возвращает ("cancel", kw) — так тест видит сам запрос
        if isinstance(request, tuple) and request[0] == "cancel":
            self.inner.cancel_calls += 1
            if self.inner.hang_cancel:
                await asyncio.sleep(3600)
        return SimpleNamespace()


class FakeManager:
    def __init__(self):
        self.pending = {}
        self.drops = []

    def put_pending(self, login_id, client):
        self.pending[login_id] = client

    async def drop_pending(self, login_id, disconnect=True):
        client = self.pending.pop(login_id, None)
        if client is not None and disconnect:
            await client.disconnect()

    def pending_client(self, login_id):
        return self.pending.get(login_id)

    async def drop(self, account_id):
        self.drops.append(account_id)


fake_functions = types.SimpleNamespace(
    auth=types.SimpleNamespace(
        CancelCodeRequest=lambda **kw: ("cancel", kw),
    ),
)


# --- каркас ----------------------------------------------------------------

async def setup(tmp_path, monkeypatch):
    monkeypatch.setenv("TGCLIENT_DB_PATH", str(tmp_path / "t.db"))
    monkeypatch.setenv("TGCLIENT_API_ID", "1")
    monkeypatch.setenv("TGCLIENT_API_HASH", "hash")
    monkeypatch.setenv("TGCLIENT_AUTH_CLIENT_ID", "")

    import app.config as config
    import app.db as appdb
    import app.main as appmain
    import app.tg.login_flow as flow

    config.get_settings.cache_clear()
    await appdb.init_db()
    conn = appdb.get_db()
    now = datetime.now(timezone.utc).isoformat()
    await conn.execute(
        "INSERT OR IGNORE INTO users (user_id, email, created_at, updated_at)"
        " VALUES ('u1','owner@example.com',?,?)", (now, now))
    await conn.commit()

    manager = FakeManager()
    monkeypatch.setattr(appmain, "get_account_manager", lambda: manager, raising=False)

    flow.errors = fake_errors
    flow.functions = fake_functions
    flow.TelegramClientWithSession = FakeClient

    reports = []

    def fake_report(event, value):  # synapse_report.report — sync fire-and-forget
        reports.append((event, value))

    monkeypatch.setattr(flow, "report", fake_report)
    return flow, manager, reports


def iso(minutes: float, *, past: bool = False) -> str:
    now = datetime.now(timezone.utc)
    delta = now + timedelta(minutes=minutes)
    if past:
        delta = now - timedelta(minutes=minutes)
    return delta.isoformat()


async def db_row(login_id) -> dict:
    import app.db as appdb
    c = await appdb.get_db().execute(
        "SELECT * FROM login_sessions WHERE id = ?", (login_id,))
    r = await c.fetchone()
    return dict(r) if r is not None else None


pytestmark = pytest.mark.asyncio


# --- SMS-путь --------------------------------------------------------------

@pytest.mark.asyncio
async def test_sms_start_reuse_wrong_code_3x_410(monkeypatch, tmp_path):
    flow, manager, reports = await setup(tmp_path, monkeypatch)
    out = await flow.login_start("+79991234567", "u1", "work")
    assert out["step"] == "awaiting_code"
    assert out["phone_masked"] == "799***567"  # маска first3+***+last3
    assert out["login_id"] in manager.pending

    # reuse: второй start того же phone возвращает ту же сессию
    out2 = await flow.login_start("+79991234567", "u1", "")
    assert out2["reused"] is True
    assert out2["login_id"] == out["login_id"]

    # три неправильных кода: первые два — attempts_left, третий — 410 + чистка
    fake = flow.TelegramClientWithSession(1, "h")
    fake.inner.sign_in_effect = CODE_INVALID
    manager.put_pending(out["login_id"], fake)  # подменяем живой клиент
    r1 = await flow.login_code(out["login_id"], "11111", "u1")
    assert r1["attempts_left"] == 2
    r2 = await flow.login_code(out["login_id"], "11111", "u1")
    assert r2["attempts_left"] == 1
    with pytest.raises(flow.DomainError) as e3:
        await flow.login_code(out["login_id"], "11111", "u1")
    assert getattr(e3.value, "code", None) == 410
    assert manager.pending == {}
    assert await db_row(out["login_id"]) is None  # строка тоже чистится


@pytest.mark.asyncio
async def test_sms_password_2fa_and_attempts(monkeypatch, tmp_path):
    flow, manager, reports = await setup(tmp_path, monkeypatch)
    out = await flow.login_start("+79991234567", "u1", "")
    fake = manager.pending[out["login_id"]]
    fake.inner.sign_in_effect = PASSWORD_NEEDED
    r = await flow.login_code(out["login_id"], "12345", "u1")
    assert r == {"login_id": out["login_id"], "step": "awaiting_password"}

    fake.inner.sign_in_effect = PASSWORD_HASH_INVALID
    r2 = await flow.login_password(out["login_id"], "bad", "u1")
    assert r2["attempts_left"] == 1
    with pytest.raises(flow.DomainError):
        await flow.login_password(out["login_id"], "bad", "u1")
    assert manager.pending == {}


@pytest.mark.asyncio
async def test_sms_success_finalizes_and_marks_done_row(monkeypatch, tmp_path):
    flow, manager, reports = await setup(tmp_path, monkeypatch)
    out = await flow.login_start("+79991234567", "u1", "т")
    await flow.login_code(out["login_id"], "12345", "u1")

    r = await db_row(out["login_id"])
    assert r["step"] == "done"
    assert r["phone_code_hash"] == ""            # хеш кода стёрт
    assert manager.pending == {}                 # RAM-клиент отдан пулу

    import app.db as appdb
    c = await appdb.get_db().execute(
        "SELECT id FROM accounts WHERE phone = '79991234567'")
    assert (await c.fetchone()) is not None
    assert any(er == "tg_account_logged_in" for er, _ in reports)

    # полл читает настоящий успех
    st = await flow.login_status(out["login_id"], "u1")
    assert st["step"] == "done"

    # повторный start того же phone: строка done не лочит новый логин
    out3 = await flow.login_start("+79991234567", "u1", "")
    assert out3.get("reused") is None and out3["login_id"] != out["login_id"]


@pytest.mark.asyncio
async def test_login_code_on_done_row_conflict(monkeypatch, tmp_path):
    flow, manager, reports = await setup(tmp_path, monkeypatch)
    # done-строка недоступна для второго кода: 409, а не тихий успех
    out = await flow.login_start("+79991234567", "u1", "")
    fake = manager.pending[out["login_id"]]
    await flow.login_code(out["login_id"], "12345", "u1")  # → done
    with pytest.raises(flow.DomainError) as e:
        await flow.login_code(out["login_id"], "12345", "u1")
    assert getattr(e.value, "code", None) == 409


@pytest.mark.asyncio
async def test_cancel_with_hanging_mtproto_connection(monkeypatch, tmp_path):
    """Кнопка отмены при полумёртвом коннекте: cancelCode под таймаутом,
    отмена не зависает, строка чистится, вызов кода был сделан."""
    flow, manager, reports = await setup(tmp_path, monkeypatch)
    out = await flow.login_start("+79991234567", "u1", "")
    fake = manager.pending[out["login_id"]]
    fake.inner.hang_cancel = True

    await asyncio.wait_for(flow.login_cancel(out["login_id"], "u1"), timeout=10)
    assert fake.inner.cancel_calls == 1
    assert manager.pending == {}
    assert await db_row(out["login_id"]) is None


# --- QR-путь ---------------------------------------------------------------

@pytest.mark.asyncio
async def test_qr_scan_success_finalizes(monkeypatch, tmp_path):
    flow, manager, reports = await setup(tmp_path, monkeypatch)
    out = await flow.login_qr_start("u1", "qr")
    assert out["step"] == "awaiting_scan" and out["qr_url"].startswith("tg://login")
    fake = manager.pending[out["login_id"]]
    fake.qr.user_me = fake.inner.user_me  # скан принят: wait() возвращает юзера

    await asyncio.sleep(0.3)  # waiter финализирует
    r = await db_row(out["login_id"])
    assert r["step"] == "done"
    assert r["phone"] == "+79991234567"
    st = await flow.login_status(out["login_id"], "u1")
    assert st["step"] == "done"
    assert manager.pending == {}
    assert any(er == "tg_account_logged_in" for er, _ in reports)


@pytest.mark.asyncio
async def test_qr_password_step_via_status(monkeypatch, tmp_path):
    flow, manager, reports = await setup(tmp_path, monkeypatch)
    out = await flow.login_qr_start("u1", "")
    fake = manager.pending[out["login_id"]]
    fake.qr.wait_effect = PASSWORD_NEEDED  # юзер с 2FA

    await asyncio.sleep(0.3)
    st = await flow.login_status(out["login_id"], "u1")
    assert st["step"] == "awaiting_password"
    # 2FA добивается обычным пароль-маршрутом
    fake.inner.sign_in_effect = PASSWORD_HASH_INVALID
    r2 = await flow.login_password(out["login_id"], "bad", "u1")
    assert r2["attempts_left"] == 1
    with pytest.raises(flow.DomainError):
        await flow.login_password(out["login_id"], "bad", "u1")
    r = await db_row(out["login_id"])
    assert r is None  # 2 промаха: строка чистится


@pytest.mark.asyncio
async def test_qr_recreate_failure_marks_error(monkeypatch, tmp_path):
    flow, manager, reports = await setup(tmp_path, monkeypatch)
    out = await flow.login_qr_start("u1", "")
    fake = manager.pending[out["login_id"]]
    fake.qr.wait_effect = asyncio.TimeoutError       # токен истёк
    fake.qr.recreate_effect = OSError("no network")  # и обновить не вышло

    await asyncio.sleep(0.5)
    st = await flow.login_status(out["login_id"], "u1")
    assert st["step"] == "error"
    assert st["last_error"]
    r = await db_row(out["login_id"])
    assert r["phone_code_hash"] == ""


@pytest.mark.asyncio
async def test_qr_cancel_stops_waiter(monkeypatch, tmp_path):
    flow, manager, reports = await setup(tmp_path, monkeypatch)
    out = await flow.login_qr_start("u1", "")
    fake = manager.pending[out["login_id"]]
    fake.qr.wait_effect = asyncio.futures.Future()  # ждёт скана вечно

    await flow.login_cancel(out["login_id"], "u1")
    assert await db_row(out["login_id"]) is None
    assert manager.pending == {}
    await asyncio.sleep(0.2)  # waiter погашен cancel, без побочных


@pytest.mark.asyncio
async def test_qr_service_restart_orphan(monkeypatch, tmp_path):
    """Рестарт: строка есть, waiter-таски нет — статус честный 410 и чистка."""
    import app.db as appdb
    flow, manager, reports = await setup(tmp_path, monkeypatch)
    conn = appdb.get_db()
    await conn.execute(
        "INSERT INTO login_sessions (id, user_id, phone, phone_code_hash, step,"
        " expires_at, created_at) VALUES ('orph','u1','','','awaiting_scan',?,?)",
        (iso(15), iso(0)),
    )
    await conn.commit()
    with pytest.raises(flow.DomainError) as e:
        await flow.login_status("orph", "u1")
    assert getattr(e.value, "code", None) == 410
    c = await conn.execute("SELECT COUNT(*) AS n FROM login_sessions")
    assert (await c.fetchone())["n"] == 0


# --- sweeper ---------------------------------------------------------------

@pytest.mark.asyncio
async def test_sweeper_skips_terminal_and_purges_orphans(monkeypatch, tmp_path):
    import app.db as appdb
    flow, manager, reports = await setup(tmp_path, monkeypatch)
    conn = appdb.get_db()
    await conn.execute(
        "INSERT INTO login_sessions (id, user_id, phone, phone_code_hash, step,"
        " expires_at, created_at) VALUES ('a','u1','+79991234561','h','awaiting_code',?,?)",
        (iso(5), iso(20, past=True)))
    # осиротевший свежий
    await conn.execute(
        "INSERT INTO login_sessions (id, user_id, phone, phone_code_hash, step,"
        " expires_at, created_at) VALUES ('b','u1','+79991234562','h','awaiting_scan',?,?)",
        (iso(10), iso(0)))
    # терминальный done (свежий) — не трогаем
    await conn.execute(
        "INSERT INTO login_sessions (id, user_id, phone, phone_code_hash, step,"
        " expires_at, created_at) VALUES ('c','u1','+79991234563','','done',?,?)",
        (iso(0.5), iso(30, past=True)))
    # терминальный done истёк — удалить
    await conn.execute(
        "INSERT INTO login_sessions (id, user_id, phone, phone_code_hash, step,"
        " expires_at, created_at) VALUES ('d','u1','+79991234564','','done',?,?)",
        (iso(0.5, past=True), iso(30, past=True)))
    await conn.commit()

    cancelled = await flow.purge_stale_logins()
    assert cancelled == 0  # RAM пуст у живых — осиротевшие, не cancel

    c = await conn.execute("SELECT id FROM login_sessions ORDER BY id")
    # свежий осиротевший 'b' живёт до TTL, свежий done 'c' — до своего expires
    assert [r["id"] for r in await c.fetchall()] == ["b", "c"]  # свежий done живёт

    # а его expires_at наступил — чистится в следующем прогоне
    await conn.execute("UPDATE login_sessions SET expires_at = ? WHERE id = 'c'",
                       (iso(0.1, past=True),))
    await conn.commit()
    await flow.purge_stale_logins()
    c = await conn.execute("SELECT id FROM login_sessions ORDER BY id")
    assert [r["id"] for r in await c.fetchall()] == ["b"]  # 'b' живёт до TTL
    assert await db_row("b") is not None
    await appdb.close_db()


@pytest.mark.asyncio
async def test_sweeper_cancels_alive_stale_login(monkeypatch, tmp_path):
    """RAM-клиент жив, логин завис старше stale_min — настоящая отмена."""
    flow, manager, reports = await setup(tmp_path, monkeypatch)
    out = await flow.login_start("+79991234567", "u1", "")
    # искусственно состарим created_at, сохранив expires_at
    import app.db as appdb
    conn = appdb.get_db()
    await conn.execute("UPDATE login_sessions SET created_at = ? WHERE id = ?",
                       (iso(30, past=True), out["login_id"]))
    await conn.commit()
    cancelled = await flow.purge_stale_logins()
    assert cancelled == 1
    assert manager.pending == {}
    assert await db_row(out["login_id"]) is None
    await appdb.close_db()