"""Regression tests for the audit-hook backstop *re-arming* paths.

The egress audit hook reads a per-thread ``EgressContext``. ``threading.local``
is NOT inherited by pool / scheduler worker threads, so several call-sites must
explicitly *re-arm* the backstop on the worker:

* ``BackgroundJobScheduler._arm_egress_backstop`` builds a context from the
  user's saved settings and arms it on the APScheduler worker thread.
* ``parallel_search_engine`` captures the submitter's context once and
  re-applies it per pool task (clearing it in a finally so it does not leak to
  the next task on a reused worker).
* ``news_strategy`` captures the context and re-arms it via
  ``active_egress_context`` inside a ThreadPoolExecutor worker.

These tests pin the *real* end-to-end behaviour of those re-arm paths and the
two extraction edge cases (bytearray host, IPv6 4-tuple) that the existing
suites do not cover. They intentionally do NOT duplicate the unit cases already
in ``test_egress_audit_hook.py`` (ADAPTIVE rejection, bytes-under-PRIVATE_ONLY,
nesting) or ``test_scheduler.py`` (mocked private_only arm, bad-settings swallow).
"""

from __future__ import annotations

import concurrent.futures
import socket

from unittest.mock import MagicMock

import pytest

from local_deep_research.security import install_audit_hook
from local_deep_research.security.egress.audit_hook import (
    _audit_hook,
    _extract_host,
    active_egress_context,
    clear_active_context,
    get_active_context,
    set_active_context,
)
from local_deep_research.security.egress.policy import (
    EgressContext,
    EgressScope,
    PolicyDeniedError,
)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def make_ctx(
    scope: EgressScope = EgressScope.BOTH,
    primary: str = "wikipedia",
    require_local_llm: bool = False,
    require_local_embeddings: bool = False,
) -> EgressContext:
    """Mirror of ``make_ctx`` in test_egress_policy.py."""
    return EgressContext(
        scope=scope,
        primary_engine=primary,
        require_local_llm=require_local_llm,
        require_local_embeddings=require_local_embeddings,
    )


@pytest.fixture(autouse=True)
def _clear_context_between_tests():
    """The hook reads a process-wide thread-local; a context leaking across
    tests (or from an armed scheduler test) would mask the next test's setup.
    """
    clear_active_context()
    yield
    clear_active_context()


def _probe(host: str, port: int = 53, timeout: float = 0.5):
    """Open a real AF_INET socket and attempt connect. The installed audit
    hook fires BEFORE the SYN, so the kernel-level outcome is irrelevant; we
    only care whether ``PolicyDeniedError`` was raised. Returns the raised
    exception, or None if the connect somehow returned.
    """
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(timeout)
    try:
        sock.connect((host, port))
        return None
    except PolicyDeniedError as exc:
        return exc
    except OSError as exc:
        return exc
    finally:
        sock.close()


def _fake_settings_manager(snapshot, tool="auto"):
    sm = MagicMock()
    sm.get_settings_snapshot.return_value = snapshot
    sm.get_setting.side_effect = lambda k, d=None: (
        tool if k == "search.tool" else d
    )
    return sm


# ---------------------------------------------------------------------------
# 1. BackgroundJobScheduler._arm_egress_backstop — REAL wiring (not mocked)
# ---------------------------------------------------------------------------


def test_arm_backstop_really_arms_enforcing_context_and_blocks_public():
    """End-to-end: a private_only snapshot must arm a context that the
    *installed* hook actually enforces — a public connect on the scheduler
    thread is blocked. The existing scheduler test mocks set_active_context, so
    it never proves the real thread-local + installed hook actually engage.

    Also verifies the username is threaded into the built context (a settings
    DB lookup for engine classification depends on it).
    """
    install_audit_hook()
    from local_deep_research.scheduler.background import BackgroundJobScheduler

    sched = BackgroundJobScheduler()
    sm = _fake_settings_manager(
        {"policy.egress_scope": "private_only", "search.tool": "library"},
        tool="library",
    )

    sched._arm_egress_backstop(sm, "alice")

    ctx = get_active_context()
    assert ctx is not None, "backstop did not arm a context"
    assert ctx.scope == EgressScope.PRIVATE_ONLY
    assert ctx.username == "alice", "username not threaded into context"

    # The armed context must be live: a real public connect is denied.
    err = _probe("8.8.8.8", 53)
    assert isinstance(err, PolicyDeniedError), (
        f"armed PRIVATE_ONLY backstop did not block public host: {err!r}"
    )
    assert err.decision.reason == "scope_mismatch_private_only"


def test_arm_backstop_resolves_adaptive_before_activation():
    """An ``adaptive`` scope must be RESOLVED to a concrete scope by
    context_from_snapshot before set_active_context stores it. If resolution
    regressed and a raw ADAPTIVE context reached set_active_context, that call
    raises ValueError (swallowed best-effort) and NOTHING would be armed.

    adaptive + a meta-picker primary ("auto") resolves to BOTH — concrete and
    storable. We assert a context IS armed and it is not ADAPTIVE.
    """
    install_audit_hook()
    from local_deep_research.scheduler.background import BackgroundJobScheduler

    sched = BackgroundJobScheduler()
    sm = _fake_settings_manager(
        {"policy.egress_scope": "adaptive", "search.tool": "auto"},
        tool="auto",
    )

    sched._arm_egress_backstop(sm, "bob")

    ctx = get_active_context()
    assert ctx is not None, (
        "adaptive backstop armed nothing — ADAPTIVE likely reached "
        "set_active_context and was rejected"
    )
    assert ctx.scope == EgressScope.BOTH
    assert ctx.scope != EgressScope.ADAPTIVE


# ---------------------------------------------------------------------------
# 2. Pool / worker capture-then-rearm pattern (parallel_search / news_strategy)
# ---------------------------------------------------------------------------


def test_news_strategy_rearm_pattern_enforces_in_worker_and_preserves_parent():
    """Mirror news_strategy: capture the parent thread's context with
    get_active_context(), then re-arm it in a worker via active_egress_context.
    The worker must enforce the captured PRIVATE_ONLY scope (a fresh worker
    thread does NOT inherit the thread-local), and the parent's context must be
    intact afterwards (the worker had no parent context, so the manager clears
    on exit without disturbing the submitter thread).
    """
    install_audit_hook()
    set_active_context(make_ctx(EgressScope.PRIVATE_ONLY))
    captured = get_active_context()

    result = {}

    def _worker():
        # Worker thread starts with NO inherited context.
        result["before"] = get_active_context()
        with active_egress_context(captured):
            result["err"] = _probe("8.8.8.8", 53)
        result["after"] = get_active_context()

    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ex:
        ex.submit(_worker).result()

    assert result["before"] is None, "thread-local leaked into fresh worker"
    assert isinstance(result["err"], PolicyDeniedError), (
        f"re-armed worker did not enforce PRIVATE_ONLY: {result['err']!r}"
    )
    assert result["err"].decision.reason == "scope_mismatch_private_only"
    assert result["after"] is None, "worker did not clear re-armed context"

    # Submitter thread's context is untouched by the worker's re-arm/clear.
    assert get_active_context() is captured


def test_pool_wrapper_clears_context_so_it_does_not_leak_to_next_task():
    """Mirror parallel_search_engine._run_with_context: a pool worker re-arms
    the captured context, runs, then clears in a finally. The CRITICAL property
    is that a SECOND task landing on the SAME reused worker thread (with no
    captured context) is not silently governed by the first task's PRIVATE_ONLY
    policy. max_workers=1 forces both tasks onto one thread.
    """
    install_audit_hook()
    private_ctx = make_ctx(EgressScope.PRIVATE_ONLY)

    def _pool_task(captured_ctx, host):
        # Re-arm only if a context was captured (matches the production guard).
        if captured_ctx is not None:
            set_active_context(captured_ctx)
        try:
            return _probe(host, 53)
        finally:
            if captured_ctx is not None:
                clear_active_context()

    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ex:
        first = ex.submit(_pool_task, private_ctx, "8.8.8.8").result()
        # Second task: no captured context -> wrapper must not re-arm anything.
        second = ex.submit(_pool_task, None, "8.8.8.8").result()

    assert isinstance(first, PolicyDeniedError), (
        f"first task did not enforce PRIVATE_ONLY: {first!r}"
    )
    assert not isinstance(second, PolicyDeniedError), (
        "first task's PRIVATE_ONLY context leaked to the next task on the "
        f"reused worker: {second!r}"
    )


def test_rearm_preserves_allow_side_private_host():
    """The re-arm path must not over-block: a re-armed PRIVATE_ONLY context in
    a worker still ALLOWS a private host (the local LLM / settings DB). Pairs
    with the deny-side worker test above so a "deny everything" regression in
    the worker path is caught.
    """
    install_audit_hook()
    captured = make_ctx(EgressScope.PRIVATE_ONLY)
    result = {}

    def _worker():
        with active_egress_context(captured):
            # 127.0.0.1 is private -> allowed; only a non-policy OSError may
            # surface (refused/timeout), never PolicyDeniedError.
            result["err"] = _probe("127.0.0.1", 1)

    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ex:
        ex.submit(_worker).result()

    assert not isinstance(result["err"], PolicyDeniedError), (
        f"re-armed PRIVATE_ONLY wrongly blocked a private host: {result['err']!r}"
    )


# ---------------------------------------------------------------------------
# 3. _extract_host edge cases not covered elsewhere
# ---------------------------------------------------------------------------


def test_extract_host_decodes_bytearray():
    """CPython may fire socket.connect with a bytearray host; the existing
    suite only covers ``bytes``. A bytearray must decode identically, else a
    bytearray-host caller would bypass the PRIVATE_ONLY/STRICT backstop.
    """
    assert _extract_host((bytearray(b"8.8.8.8"), 80)) == "8.8.8.8"
    assert _extract_host((bytearray(b"10.0.0.5"), 443)) == "10.0.0.5"


def test_extract_host_ipv6_four_tuple():
    """AF_INET6 connect addresses are ``(host, port, flowinfo, scope_id)``.
    _extract_host must read the host from a 4-tuple, not only a 2-tuple.
    """
    assert _extract_host(("::1", 0, 0, 0)) == "::1"
    assert _extract_host(("2001:4860:4860::8888", 53, 0, 0)) == (
        "2001:4860:4860::8888"
    )


def test_audit_hook_blocks_bytes_public_host_under_strict():
    """The existing bytes-host test only covers PRIVATE_ONLY. STRICT also bans
    public hosts, and the bytes-decode path must apply there too — a bytes
    public host under STRICT must be denied with the STRICT reason.
    """
    set_active_context(make_ctx(EgressScope.STRICT))
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        with pytest.raises(PolicyDeniedError) as excinfo:
            _audit_hook("socket.connect", (sock, (b"8.8.8.8", 80)))
    finally:
        sock.close()
    assert excinfo.value.decision.reason == "strict_public_host"
