"""S2.4 load-bearing seam — the ``http`` transport is TIME-BOUNDED (ingest spec §11's detach-RED
regime).

The hole this closes: ``read_http``'s default transport is the library's ``urllib_get``, which
calls ``urlopen(request)`` with no ``timeout=``. With no timeout argument urllib falls back to the
process-wide default socket timeout, which is ``None`` out of the box — so a source that accepts a
connection and then never answers blocks a run forever. That contradicts the repo invariant that
nothing runs unbounded (stop criteria + budget caps are required at startup, never an open loop).

The fix is a wrapper, NOT a fork: ``ingest.timeout_get`` scopes ``socket.setdefaulttimeout`` around
a delegate call to the library's own ``urllib_get``. This is the documented urllib fallback path,
so the bound applies WITHOUT introducing a second socket path and WITHOUT duplicating the
credential-header logic — S2.4's own verification criterion.

KNOWN CAVEAT (also carried in the implementation's comment): the default socket timeout is
PROCESS-global. Under ``concurrency=k`` the runner is asyncio on a single thread, so the scoping
holds. If ``read_http`` is ever driven from a thread-pool executor, this scoping is NOT
thread-safe and the bound must move to a per-call ``urlopen(timeout=...)`` — which would mean
owning a socket path here.

What each test makes load-bearing:

- SCOPED, NOT LEAKED (both directions): the delegate observes the bound WHILE it runs, and the
  previous process default is restored afterwards. RED when the scoping is removed, RED when it is
  applied AFTER the delegate call, and RED when it is set without being restored — a one-way
  assertion would false-green on an unconditional mutation.
- RESTORED ON FAILURE: a raising delegate still restores the previous default (the ``finally``).
- DEFAULT TRANSPORT IS BOUNDED: a real ``materialize`` run with NO injected ``http_get`` reaches
  the socket path through the wrapper. RED the moment ``materialize`` reverts to handing the
  library a bare ``None`` (i.e. the unwrapped ``urllib_get``) — which is exactly the pre-S2.4 state.
- INJECTED TRANSPORT IS CALLER-OWNED: an explicitly injected transport is NOT wrapped. This is the
  conditional control: it goes RED if the wrapping is made unconditional, and it pins the contract
  that a caller-supplied transport (``mcp_get``, a canned test stub) keeps its own timeout policy
  and never has a process-global side effect imposed on it.
- A REAL HANG IS BOUNDED (the one socket test): the assertions above prove we set the global; only
  this one proves the global actually bites. It opens a LOOPBACK listener that completes the TCP
  handshake and then never answers — no egress, no external host, no credential — and asserts the
  GET fails typed and fast. Run on a daemon thread with a join deadline so that detaching the
  scoping fails the assertion instead of hanging the suite forever.
"""

from __future__ import annotations

import json
import socket
import threading
from pathlib import Path
from typing import Any

import pytest

from portfolio_optimiser.ingest import (
    HTTP_TIMEOUT_SECONDS,
    SourceError,
    _urllib_get,
    materialize,
    timeout_get,
)

_INGESTED_AT = "2026-07-04T12:00:00Z"


def _http_manifest(tmp_path: Path) -> Path:
    manifest = {
        "manifest_version": 1,
        "source": {
            "type": "http",
            "id": "api",
            "base_url": "https://api.example.test/v1",
            "credential_ref": None,
        },
        "bundle_summary": "HTTP extracts from a local mock endpoint (http source type).",
        "extractions": [
            {
                "id": "status",
                "title": "Service status",
                "query": "status",
                "okf_type": "dataset",
                "max_rows": 10,
            }
        ],
    }
    path = tmp_path / "manifest.json"
    path.write_text(json.dumps(manifest), encoding="utf-8")
    return path


# --- LOAD-BEARING: the bound is scoped around the delegate, and restored ------------------------


def test_timeout_get_scopes_socket_default_around_delegate_and_restores() -> None:
    observed: list[float | None] = []

    def delegate(url: str, credential: str | None) -> str:
        observed.append(socket.getdefaulttimeout())
        return "body"

    before = socket.getdefaulttimeout()
    assert timeout_get(delegate, timeout=7.5)("https://h.test/x", None) == "body"

    # DURING the call the bound is in force. RED if the scoping is removed, and RED if it is
    # applied after the delegate returns (both leave the delegate observing `before`).
    assert observed == [7.5]
    # AFTER the call the process default is exactly what it was. RED if the wrapper sets the
    # global without restoring it — the mutation a one-way assertion would false-green on.
    assert socket.getdefaulttimeout() == before


def test_timeout_get_restores_socket_default_when_the_delegate_raises() -> None:
    def failing(url: str, credential: str | None) -> str:
        raise SourceError("http GET failed", code="http_transport")

    before = socket.getdefaulttimeout()
    with pytest.raises(SourceError):
        timeout_get(failing, timeout=7.5)("https://h.test/x", None)
    # RED when the restore is not in a `finally`: a transport failure would leak the bound onto
    # every socket the process opens afterwards.
    assert socket.getdefaulttimeout() == before


# --- LOAD-BEARING: the DEFAULT transport is the bounded one --------------------------------------


def test_materialize_default_transport_is_bounded(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """No ``http_get`` injected → the real socket path must still be reached through the wrapper.

    The socket path itself is stubbed at the module global the wrapper delegates to, so this runs
    end-to-end through the unchanged library without opening a socket. RED when ``materialize``
    goes back to passing the library a bare ``None``: the library then resolves its own unwrapped
    ``urllib_get`` and the recorded timeout is the untouched process default.
    """
    observed: list[float | None] = []

    def stub_urllib_get(url: str, credential: str | None) -> str:
        observed.append(socket.getdefaulttimeout())
        return "service ok"

    monkeypatch.setattr("portfolio_optimiser.ingest._urllib_get", stub_urllib_get)

    written = materialize(
        _http_manifest(tmp_path),
        tmp_path / "b",
        ingested_at=_INGESTED_AT,
        allow_network=True,
        # NOTE: no http_get= — this is the default-path assertion.
    )

    assert any(p.name == "ingest-status.md" for p in written)
    assert observed == [HTTP_TIMEOUT_SECONDS]


def test_materialize_does_not_wrap_an_injected_transport(tmp_path: Path) -> None:
    """A caller-supplied transport keeps its own timeout policy (the conditional control).

    ``mcp_get`` fronts a subprocess, not a socket, and already carries its own
    ``timeout_seconds``; canned test stubs touch nothing. Imposing a process-global socket bound on
    caller-owned code would be a side effect we do not own. RED if the wrapping is made
    unconditional — which would also false-green the default-path test above.
    """
    observed: list[float | None] = []
    before = socket.getdefaulttimeout()

    def injected(url: str, credential: str | None) -> str:
        observed.append(socket.getdefaulttimeout())
        return "service ok"

    materialize(
        _http_manifest(tmp_path),
        tmp_path / "b",
        ingested_at=_INGESTED_AT,
        allow_network=True,
        http_get=injected,
    )

    assert observed == [before]


# --- LOAD-BEARING: the bound actually bites on a real blocked connection -------------------------


def test_bounded_transport_fails_fast_against_a_hanging_loopback_server() -> None:
    """The only socket in this suite: a loopback listener that never answers.

    It binds 127.0.0.1 on an ephemeral port and never calls ``accept``; the OS backlog still
    completes the handshake, so the client connects and then blocks reading a response that never
    comes — the exact shape of the hang S2.4 closes. No egress, no external host, no credential.

    Detaching the scoping does not merely change a value here, it removes the only thing that ends
    the read. The call therefore runs on a daemon thread with a join deadline: a detached seam
    leaves the thread alive and fails the assertion, instead of hanging the suite forever.
    """
    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    listener.bind(("127.0.0.1", 0))
    listener.listen(1)
    host, port = listener.getsockname()
    before = socket.getdefaulttimeout()
    outcome: list[Any] = []

    def attempt() -> None:
        try:
            outcome.append(
                timeout_get(_urllib_get, timeout=0.25)(f"http://{host}:{port}/hang", None)
            )
        except BaseException as exc:  # noqa: BLE001 - the outcome IS the assertion subject
            outcome.append(exc)

    worker = threading.Thread(target=attempt, daemon=True)
    worker.start()
    worker.join(10.0)
    try:
        assert not worker.is_alive(), (
            "unbounded: the GET was still blocked after 10s — the socket carried no timeout"
        )
        assert isinstance(outcome[0], SourceError)
        # Typed, never a leaked OSError; the library's own transport categorisation rides through.
        assert outcome[0].code == "http_transport"
    finally:
        listener.close()

    assert socket.getdefaulttimeout() == before