"""Drive the real MCP server over stdio, with the real sandbox, as ChatGPT would."""

from __future__ import annotations

import asyncio
import json
import os
import shutil
import sys
from pathlib import Path

import pytest

pytest.importorskip("mcp")
pytestmark = pytest.mark.skipif(shutil.which("bwrap") is None, reason="bubblewrap is required to sandbox commands")

from mcp import ClientSession, StdioServerParameters  # noqa: E402
from mcp.client.stdio import stdio_client  # noqa: E402

ROOT = Path(__file__).resolve().parent.parent


def _payload(result) -> object:
    return json.loads(result.content[0].text)


async def _session(workspace: Path):
    params = StdioServerParameters(
        command=sys.executable,
        args=[str(ROOT / "mcp_server.py"), "--workspace", str(workspace)],
        env={"PATH": os.environ["PATH"], "HOME": os.environ["HOME"]},
    )
    return stdio_client(params)


def _run(workspace: Path, body):
    async def main():
        async with await _session(workspace) as (read, write):
            async with ClientSession(read, write) as session:
                await session.initialize()
                return await body(session)

    return asyncio.run(main())


@pytest.fixture
def workspace(tmp_path: Path) -> Path:
    project = tmp_path / "project"
    (project / "src").mkdir(parents=True)
    return project


def test_the_advertised_tools_are_the_terminal_surface(workspace: Path) -> None:
    names = _run(workspace, lambda s: s.list_tools())
    assert sorted(t.name for t in names.tools) == [
        "list_processes",
        "read_output",
        "run_command",
        "send_input",
        "start_process",
        "stop_process",
        "workspace_info",
    ]


def test_a_command_runs_and_writes_inside_the_workspace(workspace: Path) -> None:
    result = _run(
        workspace,
        lambda s: s.call_tool("run_command", {"command": "echo built > artifact.txt && cat artifact.txt"}),
    )
    assert _payload(result)["exit_code"] == 0
    assert (workspace / "artifact.txt").read_text() == "built\n"


def test_a_write_outside_the_workspace_is_refused(workspace: Path, tmp_path: Path) -> None:
    """The command may believe it succeeded — outside the workspace it is writing
    into the jail's private root, which is discarded when the command exits. What
    matters is that nothing reaches the host."""
    target = tmp_path.parent / "gptbridge-breach"
    _run(workspace, lambda s: s.call_tool("run_command", {"command": f"touch {target}"}))
    assert not target.exists()


def test_a_file_outside_the_workspace_cannot_be_read(workspace: Path, tmp_path: Path) -> None:
    secret = tmp_path / "id_ed25519"
    secret.write_text("PRIVATE KEY MATERIAL", encoding="utf-8")
    result = _run(workspace, lambda s: s.call_tool("run_command", {"command": f"cat {secret}"}))
    assert "PRIVATE KEY MATERIAL" not in _payload(result)["output"]


def test_the_network_is_unreachable_by_default(workspace: Path) -> None:
    result = _run(
        workspace, lambda s: s.call_tool("run_command", {"command": "getent hosts api.openai.com"})
    )
    assert _payload(result)["exit_code"] != 0


def test_a_cwd_outside_the_workspace_is_refused(workspace: Path) -> None:
    result = _run(workspace, lambda s: s.call_tool("run_command", {"command": "pwd", "cwd": "/etc"}))
    assert result.is_error
    assert "escapes the workspace" in result.content[0].text


def test_the_parent_environment_does_not_reach_the_command(workspace: Path) -> None:
    os.environ["GPTBRIDGE_LEAK_CANARY"] = "must-not-appear"
    try:
        result = _run(workspace, lambda s: s.call_tool("run_command", {"command": "env"}))
    finally:
        del os.environ["GPTBRIDGE_LEAK_CANARY"]
    assert "must-not-appear" not in _payload(result)["output"]


def test_a_command_that_overruns_its_timeout_is_killed(workspace: Path) -> None:
    result = _run(
        workspace, lambda s: s.call_tool("run_command", {"command": "sleep 60", "timeout_seconds": 2})
    )
    assert _payload(result)["killed"] == "timed out after 2s"


def test_a_background_process_can_be_read_fed_and_stopped(workspace: Path) -> None:
    async def body(session):
        started = _payload(
            await session.call_tool(
                "start_process", {"command": "while read line; do echo got:$line; done"}
            )
        )
        process_id = started["process_id"]
        await session.call_tool("send_input", {"process_id": process_id, "text": "one\n"})
        first = await _until(session, process_id, 0)
        second_offset = first["next_offset"]
        await session.call_tool("send_input", {"process_id": process_id, "text": "two\n"})
        second = await _until(session, process_id, second_offset)
        listed = _payload(await session.call_tool("list_processes", {}))
        stopped = _payload(await session.call_tool("stop_process", {"process_id": process_id}))
        return first, second, listed, stopped

    first, second, listed, stopped = _run(workspace, body)
    assert first["output"] == "got:one\n"
    assert second["output"] == "got:two\n"
    assert [p["running"] for p in listed["processes"]] == [True]
    assert stopped["status"] == "process proc-1 stopped"


async def _until(session, process_id: str, since: int, tries: int = 100) -> dict:
    for _ in range(tries):
        payload = _payload(await session.call_tool("read_output", {"process_id": process_id, "since": since}))
        if payload["output"]:
            return payload
        await asyncio.sleep(0.1)
    raise AssertionError(f"process {process_id} printed nothing after offset {since}")


def test_workspace_info_reports_what_the_model_is_working_in(workspace: Path) -> None:
    info = _payload(_run(workspace, lambda s: s.call_tool("workspace_info", {})))
    assert info == {
        "workspace": str(workspace),
        "sandbox_mode": "workspace-write",
        "network_access": False,
    }