"""Per-op timeout guard for SFTP operations (SP27 Task 8).

paramiko is synchronous; without a timeout, a hung ``listdir_attr``
freezes the worker thread indefinitely. The 06/25 silent hang was
exactly this — the MFT server TCP-acked but stopped responding, the
scheduler's ``asyncio.to_thread`` waited forever, and the operator
had no signal that polling had stalled.

This test pins the fix: the SFTP client's only async-wrapped call
site (``async_list_inbound``) applies an ``asyncio.wait_for(...
timeout=N)`` so the event loop can give up after the configured
bound. The bound is read from ``CYCLONE_SFTP_OP_TIMEOUT_SECONDS``
at call time (default 30s).

The other SFTP methods (``list_inbound_names``, ``download_inbound``,
``read_file``, ``write_file``) intentionally remain sync — they're
called from operator-triggered paths (admin endpoints, CLI) where
the operator can Ctrl-C the request. Wrapping them would require
making the FastAPI handlers themselves async, which is out of
scope for Task 8. Tracked as a follow-up.
"""
from __future__ import annotations

import asyncio
import time

import pytest

from cyclone.clearhouse import SftpClient
from cyclone.providers import SftpBlock


def _block() -> SftpBlock:
    return SftpBlock(
        host="mft.example.com",
        port=22,
        username="user",
        auth={"password_keychain_account": "x"},
        paths={"inbound": "/inbound", "outbound": "/outbound"},
        stub=True,
    )


class _HangingClient(SftpClient):
    """SftpClient that hangs on list_inbound — exercises the
    asyncio.wait_for timeout. The hang runs in a thread, so
    ``asyncio.to_thread`` alone would never give up; only the
    wait_for wrapper can break it.

    The hang is short (3s) so the per-test thread pool doesn't
    accumulate stragglers across the timeout tests. Each test sets
    ``CYCLONE_SFTP_OP_TIMEOUT_SECONDS=1`` so wait_for fires well
    before the sleep completes.
    """

    def list_inbound(self):
        time.sleep(3)
        return []


# ---- env-var parsing --------------------------------------------------------


def test_op_timeout_default_is_30_seconds(monkeypatch):
    """Default is 30s when CYCLONE_SFTP_OP_TIMEOUT_SECONDS is unset.

    Pins the documented default so a refactor that changes it has to
    update both the constant and the spec.
    """
    monkeypatch.delenv("CYCLONE_SFTP_OP_TIMEOUT_SECONDS", raising=False)
    from cyclone.clearhouse import _op_timeout_seconds
    assert _op_timeout_seconds() == 30.0


def test_op_timeout_reads_env_var(monkeypatch):
    """Operator can override via env var without a restart-rebuild."""
    monkeypatch.setenv("CYCLONE_SFTP_OP_TIMEOUT_SECONDS", "7.5")
    from cyclone.clearhouse import _op_timeout_seconds
    assert _op_timeout_seconds() == 7.5


def test_op_timeout_rejects_unparseable_env_var(monkeypatch):
    """A non-float value (e.g. "30s") falls back to the default.

    Pins the ValueError branch — a refactor that drops the try/except
    would crash the SFTP wrapper on every call when the env var is
    set to anything non-numeric.
    """
    monkeypatch.setenv("CYCLONE_SFTP_OP_TIMEOUT_SECONDS", "30s")
    from cyclone.clearhouse import _op_timeout_seconds
    assert _op_timeout_seconds() == 30.0


def test_op_timeout_rejects_zero_and_negative(monkeypatch):
    """Zero / negative timeouts would silently turn every SFTP call
    into an instant failure (asyncio.wait_for(timeout=0) raises
    immediately; timeout<0 is undefined). They fall back to default.
    """
    from cyclone.clearhouse import _op_timeout_seconds
    for bad in ("0", "0.0", "-1", "-0.5"):
        monkeypatch.setenv("CYCLONE_SFTP_OP_TIMEOUT_SECONDS", bad)
        assert _op_timeout_seconds() == 30.0, f"value {bad!r} should fall back to default"


# ---- async wrapper behavior -------------------------------------------------


@pytest.mark.asyncio
async def test_async_list_inbound_times_out_when_paramiko_hangs(monkeypatch):
    """A hanging list_inbound must be cancelled by wait_for, not the thread.

    The 06/25 silent hang was exactly this scenario — paramiko's
    listdir_attr TCP-acked then went silent, freezing the worker
    thread. The wait_for wrapper is the only thing that can break
    out of that.
    """
    monkeypatch.setenv("CYCLONE_SFTP_OP_TIMEOUT_SECONDS", "1")
    client = _HangingClient(_block())
    start = time.monotonic()
    with pytest.raises(asyncio.TimeoutError):
        await client.async_list_inbound()
    elapsed = time.monotonic() - start
    # Loose upper bound: timeout fires at ~1s, allow 3s for asyncio
    # scheduling jitter on slow CI runners. If this ever fires at the
    # hang duration (3s) the wait_for wrapper is missing.
    assert elapsed < 3, f"timeout fired too late: {elapsed:.2f}s"


@pytest.mark.asyncio
async def test_async_list_inbound_returns_normally_when_fast(monkeypatch, tmp_path):
    """Sanity check: a non-hanging call returns the stub's empty list
    without raising — pins that the wrapper doesn't break the happy path."""
    monkeypatch.setenv("CYCLONE_SFTP_OP_TIMEOUT_SECONDS", "5")
    block = _block()
    block.staging_dir = str(tmp_path)
    client = SftpClient(block)
    # stub mode + no files staged = empty list, no hang, no timeout
    result = await client.async_list_inbound()
    assert result == []