"""Tests for the SSH remote execution environment backend."""

import json
import os
import shlex
import subprocess
from unittest.mock import MagicMock

import pytest

from tools.environments.ssh import SSHEnvironment
from tools.environments import ssh as ssh_env

# A real sshd to run against: scripts/ci/start_test_sshd.sh starts a throwaway one
# and prints these (the remote-backends CI lane exports them). HERMES_TEST_* rather
# than the product's TERMINAL_SSH_* so the hermetic conftest, which blanks
# TERMINAL_SSH_KEY as a credential, cannot strip them before the tests read them.
_SSH_HOST = os.getenv("HERMES_TEST_SSH_HOST", "")
_SSH_USER = os.getenv("HERMES_TEST_SSH_USER", "")
_SSH_PORT = int(os.getenv("HERMES_TEST_SSH_PORT", "22"))
_SSH_KEY = os.getenv("HERMES_TEST_SSH_KEY", "")

_has_ssh = bool(_SSH_HOST and _SSH_USER)

requires_ssh = pytest.mark.skipif(
    not _has_ssh,
    reason="HERMES_TEST_SSH_HOST / HERMES_TEST_SSH_USER not set (scripts/ci/start_test_sshd.sh)",
)


def _run(command, task_id="ssh_test", **kwargs):
    from tools.terminal_tool import terminal_tool
    return json.loads(terminal_tool(command, task_id=task_id, **kwargs))


def _cleanup(task_id="ssh_test"):
    # The env is cached under the resolved key ("default" without a session key), not
    # the raw task id: cleaning the raw id was a no-op that let one test's env leak
    # into the next.
    from tools.terminal_tool import _resolve_container_task_id
    from tools.terminal_tool_lifecycle import cleanup_vm
    cleanup_vm(_resolve_container_task_id(task_id))


class TestBuildSSHCommand:

    @pytest.fixture(autouse=True)
    def _mock_connection(self, monkeypatch):
        monkeypatch.setattr("tools.environments.ssh.subprocess.run",
                            lambda *a, **k: subprocess.CompletedProcess([], 0))
        monkeypatch.setattr("tools.environments.ssh.subprocess.Popen",
                            lambda *a, **k: MagicMock(stdout=iter([]),
                                                      stderr=iter([]),
                                                      stdin=MagicMock()))
        monkeypatch.setattr("tools.environments.base.time.sleep", lambda _: None)

    def test_base_flags(self, monkeypatch):
        # ControlMaster flags are POSIX-only (#73927): assert them only
        # where multiplexing is enabled so the test passes on Windows too.
        monkeypatch.setattr(ssh_env, "_SSH_MULTIPLEX", True)
        env = SSHEnvironment(host="h", user="u")
        cmd = " ".join(env._build_ssh_command())
        for flag in ("ControlMaster=auto", "ControlPersist=300",
                      "BatchMode=yes", "StrictHostKeyChecking=accept-new"):
            assert flag in cmd

    def test_controlmaster_gated_off_on_windows(self, monkeypatch):
        """#73927: Windows OpenSSH has no Unix-domain ControlMaster, so the
        ControlPath/ControlMaster/ControlPersist options must be omitted —
        passing them fails the connection with 'getsockname failed'."""
        monkeypatch.setattr(ssh_env, "_SSH_MULTIPLEX", False)
        env = SSHEnvironment(host="h", user="u")
        cmd = " ".join(env._build_ssh_command())
        assert "ControlMaster" not in cmd
        assert "ControlPath" not in cmd
        assert "ControlPersist" not in cmd
        # Non-multiplex flags must still be present — the backend works,
        # just without connection pooling.
        assert "BatchMode=yes" in cmd
        assert "StrictHostKeyChecking=accept-new" in cmd
        assert env._build_ssh_command()[-1] == "u@h"


    def test_user_host_suffix(self):
        env = SSHEnvironment(host="h", user="u")
        assert env._build_ssh_command()[-1] == "u@h"

    def _capture_run_bash(self, monkeypatch, env, cmd="echo ok"):
        captured = {}

        def _fake_popen(cmd, stdin_data=None, **kwargs):
            captured["cmd"], captured["env"] = cmd, kwargs.get("env")
            return MagicMock()

        monkeypatch.setattr(ssh_env, "_popen_bash", _fake_popen)
        env._run_bash(cmd)
        return captured

    def test_run_bash_forwards_passthrough_by_sendenv_never_in_remote_argv(self, monkeypatch):
        """#14091: allowlisted names travel as ``-o SendEnv=NAME`` with values only in the ssh client env;
        provider credentials on the allowlist stay behind; a .env value fills an unset shell var."""
        from tools import env_passthrough

        env = SSHEnvironment(host="h", user="u")
        monkeypatch.setenv("NEXTCLOUD_URL", "https://next.example")
        monkeypatch.delenv("NEXTCLOUD_PASS", raising=False)
        monkeypatch.setenv("OPENAI_API_KEY", "sk-must-not-forward")
        monkeypatch.setattr(env_passthrough, "get_all_passthrough",
                            lambda: frozenset({"NEXTCLOUD_URL", "NEXTCLOUD_PASS", "OPENAI_API_KEY"}))
        monkeypatch.setattr(ssh_env, "_load_hermes_env_vars", lambda: {"NEXTCLOUD_PASS": "from-dotenv"})

        captured = self._capture_run_bash(monkeypatch, env)

        sent = {a.split("=", 1)[1] for a in captured["cmd"] if a.startswith("SendEnv=")}
        assert sent == {"NEXTCLOUD_URL", "NEXTCLOUD_PASS"}
        assert captured["env"]["NEXTCLOUD_URL"] == "https://next.example"
        assert captured["env"]["NEXTCLOUD_PASS"] == "from-dotenv"
        remote_text = " ".join(captured["cmd"])
        assert "https://next.example" not in remote_text and "from-dotenv" not in remote_text
        assert "sk-must-not-forward" not in remote_text

    def test_run_bash_without_passthrough_inherits_env_unchanged(self, monkeypatch):
        from tools import env_passthrough

        monkeypatch.setattr(env_passthrough, "get_all_passthrough", lambda: frozenset())
        captured = self._capture_run_bash(monkeypatch, SSHEnvironment(host="h", user="u"))
        assert not any(a.startswith("SendEnv=") for a in captured["cmd"])
        assert captured["env"] is None


class TestControlSocketPath:
    """Regression tests for issue #11840.

    macOS caps Unix domain socket paths at 104 bytes (sun_path). SSH
    appends a 16-byte random suffix to the control socket path when
    operating in ControlMaster mode. An IPv6 host embedded in the
    filename plus the deeply-nested macOS $TMPDIR easily blows past
    the limit, causing every tool call to fail immediately.
    """

    @pytest.fixture(autouse=True)
    def _mock_connection(self, monkeypatch):
        monkeypatch.setattr("tools.environments.ssh.subprocess.run",
                            lambda *a, **k: subprocess.CompletedProcess([], 0))
        monkeypatch.setattr("tools.environments.ssh.subprocess.Popen",
                            lambda *a, **k: MagicMock(stdout=iter([]),
                                                      stderr=iter([]),
                                                      stdin=MagicMock()))
        monkeypatch.setattr("tools.environments.base.time.sleep", lambda _: None)

    # SSH appends ``.XXXXXXXXXXXXXXXX`` (17 bytes) to the ControlPath in
    # ControlMaster mode; the macOS sun_path field is 104 bytes including
    # the NUL terminator, so the usable path length is 103 bytes.
    _SSH_CONTROLMASTER_SUFFIX = 17
    _MAX_SUN_PATH = 103

    def test_fits_under_macos_socket_limit_with_ipv6_host(self, monkeypatch):
        """A realistic macOS $TMPDIR + IPv6 host must still produce a
        control socket path that fits once SSH appends its ControlMaster
        suffix (see issue #11840)."""
        # Simulate the macOS $TMPDIR shape from the issue traceback —
        # 48 bytes, the typical length of ``/var/folders/XX/YYYYYYYYY/T``.
        fake_tmp = "/var/folders/2t/wbkw5yb158jc3zhswgl7tz9c0000gn/T"
        monkeypatch.setattr("tools.environments.ssh.tempfile.gettempdir",
                            lambda: fake_tmp)
        # The simulated path doesn't exist on the test host — skip the
        # real mkdir so __init__ can proceed.
        from pathlib import Path as _Path
        monkeypatch.setattr(_Path, "mkdir", lambda *a, **k: None)

        env = SSHEnvironment(
            host="9373:9b91:4480:558d:708e:e601:24e8:d8d0",
            user="hermes",
            port=22,
        )

        total_len = len(str(env.control_socket)) + self._SSH_CONTROLMASTER_SUFFIX
        assert total_len <= self._MAX_SUN_PATH, (
            f"control socket path would exceed the {self._MAX_SUN_PATH}-byte "
            f"Unix domain socket limit once SSH appends its 16-byte suffix: "
            f"{env.control_socket} (+{self._SSH_CONTROLMASTER_SUFFIX} = {total_len})"
        )

    def test_path_is_deterministic_across_instances(self):
        """Same (user, host, port) must yield the same control socket so
        ControlMaster reuse works across reconnects."""
        first = SSHEnvironment(host="example.com", user="alice", port=2222)
        second = SSHEnvironment(host="example.com", user="alice", port=2222)
        assert first.control_socket == second.control_socket

    def test_path_differs_for_different_targets(self):
        """Different (user, host, port) triples must produce different paths."""
        base = SSHEnvironment(host="h", user="u", port=22).control_socket
        assert SSHEnvironment(host="h", user="u", port=23).control_socket != base
        assert SSHEnvironment(host="h", user="v", port=22).control_socket != base
        assert SSHEnvironment(host="g", user="u", port=22).control_socket != base


class TestTerminalToolConfig:


    def test_ssh_persistent_respects_config(self, monkeypatch):
        """TERMINAL_PERSISTENT_SHELL=false disables SSH persistent by default."""
        monkeypatch.delenv("TERMINAL_SSH_PERSISTENT", raising=False)
        monkeypatch.setenv("TERMINAL_PERSISTENT_SHELL", "false")
        from tools.terminal_tool import _get_env_config
        assert _get_env_config()["ssh_persistent"] is False

    @requires_ssh
    @pytest.mark.parametrize("setting, expected", [(None, True), ("false", False)])
    def test_ssh_persistent_reaches_the_environment(self, monkeypatch, setting, expected):
        """terminal.persistent_shell (bridged to TERMINAL_PERSISTENT_SHELL) reaches SSHEnvironment."""
        from tools.terminal_tool import _get_env_config
        from tools.terminal_tool_lifecycle import _create_configured_env
        _setup_ssh_env(monkeypatch, persistent=True)
        monkeypatch.delenv("TERMINAL_SSH_PERSISTENT")
        if setting is None:
            monkeypatch.delenv("TERMINAL_PERSISTENT_SHELL", raising=False)
        else:
            monkeypatch.setenv("TERMINAL_PERSISTENT_SHELL", setting)
        config = _get_env_config()
        env = _create_configured_env(config, "ssh", image="", cwd="~", timeout=30,
                                     task_id="ssh-persist-config", host_cwd=None)
        try:
            assert env._persist_env is expected
        finally:
            env.cleanup()


class TestSSHPreflight:
    def test_ensure_ssh_available_raises_clear_error_when_missing(self, monkeypatch):
        monkeypatch.setattr(ssh_env.shutil, "which", lambda _name: None)

        with pytest.raises(RuntimeError, match="SSH is not installed or not in PATH"):
            ssh_env._ensure_ssh_available()


    def test_ssh_environment_connects_when_ssh_exists(self, monkeypatch):
        called = {"count": 0}

        monkeypatch.setattr(ssh_env.shutil, "which", lambda _name: "/usr/bin/ssh")

        def _fake_establish(self):
            called["count"] += 1

        monkeypatch.setattr(ssh_env.SSHEnvironment, "_establish_connection", _fake_establish)
        monkeypatch.setattr(ssh_env.SSHEnvironment, "_detect_remote_home", lambda self: "/home/alice")
        monkeypatch.setattr(ssh_env.SSHEnvironment, "_ensure_remote_dirs", lambda self: None)
        monkeypatch.setattr(ssh_env.SSHEnvironment, "init_session", lambda self: None)
        monkeypatch.setattr(ssh_env, "FileSyncManager", lambda **kw: type("M", (), {"sync": lambda self, **k: None})())

        env = ssh_env.SSHEnvironment(host="example.com", user="alice")

        assert called["count"] == 1
        assert env.host == "example.com"
        assert env.user == "alice"

    def test_ssh_environment_can_skip_agent_file_sync(self, monkeypatch):
        monkeypatch.setattr(ssh_env.shutil, "which", lambda _name: "/usr/bin/ssh")
        monkeypatch.setattr(ssh_env.SSHEnvironment, "_establish_connection", lambda self: None)
        monkeypatch.setattr(ssh_env.SSHEnvironment, "_detect_remote_home", lambda self: "/home/alice")
        monkeypatch.setattr(ssh_env.SSHEnvironment, "init_session", lambda self: None)
        monkeypatch.setattr(
            ssh_env.SSHEnvironment,
            "_ensure_remote_dirs",
            lambda self: pytest.fail("workspace browsing must not mutate the SSH target"),
        )
        monkeypatch.setattr(
            ssh_env,
            "FileSyncManager",
            lambda **_kw: pytest.fail("workspace browsing must not start agent file sync"),
        )

        env = ssh_env.SSHEnvironment(
            host="example.com",
            user="alice",
            sync_files=False,
        )

        assert env._sync_manager is None
        env._before_execute()


@pytest.fixture
def _mock_ssh_runtime(monkeypatch, tmp_path):
    hooks = {
        "_establish_connection": MagicMock(),
        "_detect_remote_home": MagicMock(return_value="/home/alice"),
        "_ensure_remote_dirs": MagicMock(),
        "init_session": MagicMock(),
    }
    monkeypatch.setattr(ssh_env.tempfile, "gettempdir", lambda: str(tmp_path))
    monkeypatch.setattr(ssh_env.shutil, "which", lambda _name: "/usr/bin/ssh")
    for name, hook in hooks.items():
        monkeypatch.setattr(ssh_env.SSHEnvironment, name, hook)
    hooks["sync_factory"] = MagicMock(return_value=MagicMock())
    monkeypatch.setattr(ssh_env, "FileSyncManager", hooks["sync_factory"])
    return hooks


class TestSSHProbeOnly:
    def test_probe_only_skips_state_sync_and_session_setup(self, _mock_ssh_runtime):
        env = ssh_env.SSHEnvironment(host="example.com", user="alice", probe_only=True)
        env._before_execute()
        env.cleanup()

        _mock_ssh_runtime["_establish_connection"].assert_called_once_with()
        _mock_ssh_runtime["_detect_remote_home"].assert_not_called()
        _mock_ssh_runtime["_ensure_remote_dirs"].assert_not_called()
        _mock_ssh_runtime["sync_factory"].assert_not_called()
        _mock_ssh_runtime["init_session"].assert_not_called()

    def test_probe_only_control_socket_is_isolated(self, monkeypatch, _mock_ssh_runtime):
        control_exit_calls = []

        def _fake_run(*args, **kwargs):
            control_exit_calls.append(args[0])
            return subprocess.CompletedProcess([], 0)

        monkeypatch.setattr(ssh_env.subprocess, "run", _fake_run)

        normal = ssh_env.SSHEnvironment(host="example.com", user="alice")
        first_probe = ssh_env.SSHEnvironment(host="example.com", user="alice", probe_only=True)
        second_probe = ssh_env.SSHEnvironment(host="example.com", user="alice", probe_only=True)

        assert first_probe.control_socket != normal.control_socket
        assert second_probe.control_socket != first_probe.control_socket
        assert len(first_probe.control_socket.name) == len(normal.control_socket.name)

        normal.control_socket.touch()
        first_probe.control_socket.touch()
        first_probe.cleanup()

        assert normal.control_socket.exists()
        assert not first_probe.control_socket.exists()
        assert len(control_exit_calls) == 1


def _setup_ssh_env(monkeypatch, persistent: bool):
    monkeypatch.setenv("TERMINAL_ENV", "ssh")
    monkeypatch.setenv("TERMINAL_SSH_HOST", _SSH_HOST)
    monkeypatch.setenv("TERMINAL_SSH_USER", _SSH_USER)
    monkeypatch.setenv("TERMINAL_SSH_PERSISTENT", "true" if persistent else "false")
    if _SSH_PORT != 22:
        monkeypatch.setenv("TERMINAL_SSH_PORT", str(_SSH_PORT))
    if _SSH_KEY:
        monkeypatch.setenv("TERMINAL_SSH_KEY", _SSH_KEY)


@requires_ssh
class TestOneShotSSH:

    @pytest.fixture(autouse=True)
    def _setup(self, monkeypatch):
        _setup_ssh_env(monkeypatch, persistent=False)
        yield
        _cleanup()

    def test_echo(self):
        r = _run("echo hello")
        assert r["exit_code"] == 0
        assert "hello" in r["output"]


    def test_state_does_not_persist(self):
        """Exports stay in their own command; cwd still carries (file tools rely on it)."""
        home = _run("pwd")["output"].strip()
        _run("mkdir -p oneshot && cd oneshot && export HERMES_ONESHOT_TEST=yes")
        r = _run("echo ${HERMES_ONESHOT_TEST:-unset}; pwd")
        assert r["output"].split() == ["unset", f"{home}/oneshot"]


@requires_ssh
class TestPersistentSSH:

    @pytest.fixture(autouse=True)
    def _setup(self, monkeypatch):
        _setup_ssh_env(monkeypatch, persistent=True)
        yield
        _cleanup()

    def test_echo(self):
        r = _run("echo hello-persistent")
        assert r["exit_code"] == 0
        assert "hello-persistent" in r["output"]

    def test_env_var_persists(self):
        _run("export HERMES_PERSIST_TEST=works")
        r = _run("echo $HERMES_PERSIST_TEST")
        assert r["output"].strip() == "works"


    def test_large_output(self):
        r = _run("seq 1 1000")
        assert r["exit_code"] == 0
        lines = r["output"].strip().splitlines()
        assert len(lines) == 1000
        assert lines[0] == "1"
        assert lines[-1] == "1000"


@requires_ssh
class TestRealSSHBackend:
    """terminal_tool and the file tools over a real sshd, with the default (persistent) config."""

    @pytest.fixture(autouse=True)
    def _setup(self, monkeypatch):
        _setup_ssh_env(monkeypatch, persistent=True)
        yield
        _cleanup()

    def test_exit_code_and_stderr_come_back(self):
        r = _run("echo out; echo err >&2; exit 3")
        assert r["exit_code"] == 3
        assert "out" in r["output"] and "err" in r["output"]

    def test_cwd_persists_and_workdir_does_not(self):
        home = _run("pwd")["output"].strip()
        assert home.startswith("/")
        _run("mkdir -p 'proj/with space' && cd 'proj/with space'")
        assert _run("pwd")["output"].strip() == f"{home}/proj/with space"
        assert _run("pwd", workdir="/usr")["output"].strip() == "/usr"
        assert _run("pwd")["output"].strip() == f"{home}/proj/with space"

    def test_quoting_survives_the_ssh_hop(self):
        # One quoting layer for bash -c, one for the remote login shell sshd runs it under.
        payload = "it's \"quoted\" $HOME `tick` \\back; echo injected"
        r = _run(f"printf '%s\\n' {shlex.quote(payload)}")
        assert r["output"].rstrip("\n") == payload

    def test_passthrough_env_reaches_remote_via_sendenv(self, monkeypatch):
        """#14091 end to end: the name rides SendEnv, the value the ssh client env, and the
        test sshd's ``AcceptEnv HERMES_RB_*`` hands it to the remote shell."""
        from tools import env_passthrough

        monkeypatch.setenv("HERMES_RB_PASS", "through-sendenv")
        monkeypatch.setenv("HERMES_RB_PRIVATE", "stays-local")
        env_passthrough.register_env_passthrough(["HERMES_RB_PASS"])
        try:
            r = _run('echo "p=$HERMES_RB_PASS q=${HERMES_RB_PRIVATE:-unset}"')
        finally:
            env_passthrough.clear_env_passthrough()
        assert r["output"].strip() == "p=through-sendenv q=unset"

    def test_file_tools_write_and_read_on_the_remote(self):
        from tools.file_tools import read_file_tool, write_file_tool

        home = _run("pwd")["output"].strip()
        path = f"{home}/notes/remote.txt"
        content = "alpha\nbeta: ünïcödé ✓\n"
        wrote = json.loads(write_file_tool(path, content, task_id="ssh_test"))
        assert "error" not in wrote, wrote
        assert _run(f"wc -c < {shlex.quote(path)}")["output"].strip() == str(len(content.encode()))
        read = json.loads(read_file_tool(path, task_id="ssh_test"))
        assert "beta: ünïcödé ✓" in read["content"], read