tests/test_executor_seam.py

"""
Network-free tests for the executor seam (ADR-0007 Phase 0,
`src/managers/execution/base.py`).

Covers executor SELECTION via the EXECUTOR env var (in_process | sandbox +
unknown), that the default factory returns an in-process executor conforming to
the `Executor` protocol, that it round-trips a trivial `execute(...)`, that the
`sandbox` value returns a `SandboxedExecutor` (ADR-0007 Phase 1 — was a
NotImplementedError stub in Phase 0), and that an unknown value raises ValueError.

The factory dispatch / error paths need no heavy deps. The round-trip touches
`PythonExecutor.reset_environment`, which imports pandas/numpy — guarded with
`importorskip` so the dispatch/error assertions still run where those are absent.
"""

from __future__ import annotations

import importlib.util
import sys
import types
from pathlib import Path

import pytest

# Load the execution package's `base` + `python_executor` modules DIRECTLY from
# their files, WITHOUT importing `managers.__init__` (which pulls in the full
# agent stack: langchain_core, rich, …). This keeps the seam test dependency-light
# — only the factory's own deps (none for dispatch/errors; pandas/numpy only for
# the guarded round-trip).
#
# The stand-in packages are registered under a PRIVATE root (`_seam_mgr`), never
# under the real name `managers`. That matters beyond tidiness: pytest imports
# every test module during collection, so a stand-in left at `managers` (with an
# empty __path__) shadows the real package for every module collected afterwards
# — `from managers import ConsoleDisplay` then fails with "unknown location",
# which previously made test_upload_ui_gate.py and test_tool_registry.py
# uncollectable in a full-suite run while passing standalone. A private root
# cannot collide, so no cleanup is needed and collection order stops mattering.
#
# Everything under src/managers/execution imports RELATIVELY (`from .base import
# …`, `from ..namespace_kernel import …`), and a relative import resolves through
# the importing module's own __package__. So inside this alias tree every such
# import lands back in the tree, and the classes the tests hold are the same
# objects the factory returns — the module-identity property an alias preserves
# but deleting the entries after load would destroy.
_EXEC_DIR = Path(__file__).parent.parent / "src" / "managers" / "execution"
_ALIAS = "_seam_mgr"


def _load_execution_package():
    for name, path in ((_ALIAS, []), (f"{_ALIAS}.execution", [str(_EXEC_DIR)])):
        if name not in sys.modules:
            pkg = types.ModuleType(name)
            pkg.__path__ = path
            sys.modules[name] = pkg
    # The alias package's __path__ points at the real directory, so ordinary
    # import machinery finds the submodules (and their relative-import deps,
    # e.g. python_executor -> namespace_kernel) without hand-loading each file.
    return (
        importlib.import_module(f"{_ALIAS}.execution.base"),
        importlib.import_module(f"{_ALIAS}.execution.python_executor"),
    )


_base, _py = _load_execution_package()
Executor = _base.Executor
get_executor = _base.get_executor
PythonExecutor = _py.PythonExecutor


# --------------------------------------------------------------------------- #
# Interface surface + structural conformance
# --------------------------------------------------------------------------- #
def test_executor_protocol_surface():
    """The Executor protocol captures exactly the current PythonExecutor surface."""
    for method in ("reset_environment", "send_functions", "send_variables", "execute", "__call__"):
        assert hasattr(Executor, method), f"Executor missing {method}"


def test_python_executor_conforms_structurally():
    """PythonExecutor conforms to the runtime_checkable Executor protocol."""
    assert issubclass(PythonExecutor, Executor)


def test_seam_is_reexported_from_init_files():
    """Executor / get_executor are wired into the package __init__ exports.

    Asserted by reading the __init__ source (not importing it) so the check
    stays free of the heavy `managers` import chain (langchain_core, rich, …).
    """
    exec_init = (_EXEC_DIR / "__init__.py").read_text()
    assert "get_executor" in exec_init and "Executor" in exec_init

    managers_init = (_EXEC_DIR.parent / "__init__.py").read_text()
    assert "get_executor" in managers_init and "Executor" in managers_init


# --------------------------------------------------------------------------- #
# Factory dispatch (no heavy deps needed for these three)
# --------------------------------------------------------------------------- #
def test_default_is_in_process(monkeypatch):
    """No EXECUTOR set → in_process → a PythonExecutor conforming to Executor."""
    monkeypatch.delenv("EXECUTOR", raising=False)
    ex = get_executor()
    assert isinstance(ex, PythonExecutor)
    assert isinstance(ex, Executor)


def test_explicit_in_process_kind():
    """Explicit kind='in_process' returns a PythonExecutor regardless of env."""
    ex = get_executor("in_process")
    assert isinstance(ex, PythonExecutor)


def test_env_selects_in_process(monkeypatch):
    """EXECUTOR=in_process (case/space-insensitive) selects the in-process executor."""
    monkeypatch.setenv("EXECUTOR", "  In_Process ")
    assert isinstance(get_executor(), PythonExecutor)


def test_sandbox_selects_sandboxed_executor(monkeypatch):
    """EXECUTOR=sandbox now returns a SandboxedExecutor (ADR-0007 Phase 1 landed).

    Was a NotImplementedError stub in Phase 0. With the subprocess launcher (so no
    Docker is needed) the factory constructs the real client — verified by class
    name so this seam test stays free of the sandbox package's own imports. No
    kernel is started here (construction is lazy; start happens on first call).
    """
    monkeypatch.setenv("EXECUTOR", "sandbox")
    monkeypatch.setenv("SANDBOX_LAUNCHER", "subprocess")
    ex = get_executor()
    assert type(ex).__name__ == "SandboxedExecutor"
    assert isinstance(ex, Executor)  # conforms to the same protocol
    # explicit-kind path too
    ex2 = get_executor("sandbox")
    assert type(ex2).__name__ == "SandboxedExecutor"


def test_unknown_value_raises_value_error(monkeypatch):
    """An unknown EXECUTOR value fails loud with ValueError."""
    monkeypatch.setenv("EXECUTOR", "wasm")
    with pytest.raises(ValueError) as exc:
        get_executor()
    assert "wasm" in str(exc.value)


# --------------------------------------------------------------------------- #
# Round-trip through the default executor (needs pandas/numpy)
# --------------------------------------------------------------------------- #
def test_in_process_round_trip():
    """The default executor executes code and returns captured stdout."""
    pytest.importorskip("pandas")
    pytest.importorskip("numpy")
    ex = get_executor("in_process")
    assert ex.execute("print(1 + 1)") == "2"
    # __call__ is an alias for execute
    assert ex("print('hi')") == "hi"
    # empty output path
    assert ex.execute("x = 5") == "Code executed successfully"


def test_in_process_namespace_persists():
    """State persists across calls (Jupyter-kernel semantics)."""
    pytest.importorskip("pandas")
    pytest.importorskip("numpy")
    ex = get_executor("in_process")
    ex.send_variables({"seed": 41})
    assert ex.execute("print(seed + 1)") == "42"