from __future__ import annotations

import asyncio
import os
import shlex
import time
from pathlib import Path

import pytest

from EvoScientist.native_sandbox import (
    NativeSandboxExecutor,
    NativeWorkspaceBackend,
    ensure_native_sandbox_ready,
)

pytestmark = pytest.mark.skipif(
    os.getenv("EVOSCIENTIST_RUN_NATIVE_SANDBOX_TESTS") != "1",
    reason="set EVOSCIENTIST_RUN_NATIVE_SANDBOX_TESTS=1 on a supported host",
)


@pytest.fixture
def executor(tmp_path: Path) -> NativeSandboxExecutor:
    ensure_native_sandbox_ready()
    files = tmp_path / "files"
    runtime = tmp_path / "runtime"
    files.mkdir()
    runtime.mkdir()
    return NativeSandboxExecutor(files, runtime, timeout=5)


def test_real_sandbox_scrubs_secrets_and_keeps_command_tmp_private(
    executor: NativeSandboxExecutor, monkeypatch: pytest.MonkeyPatch
):
    monkeypatch.setenv("OPENAI_API_KEY", "must-not-leak")
    result = executor.execute(
        'env | sort; printf result > result.txt; printf temp > "$TMPDIR/value"'
    )

    assert result.exit_code == 0
    assert "OPENAI_API_KEY" not in result.output
    assert "PROXY_PASSWORD" not in result.output
    assert "HTTP_PROXY" not in result.output
    assert (executor.files_dir / "result.txt").read_text(encoding="utf-8") == "result"
    assert not list((executor.runtime_dir / "tmp").glob("*/tmp/value"))


def test_real_sandbox_drains_but_caps_output(
    executor: NativeSandboxExecutor, monkeypatch: pytest.MonkeyPatch
):
    monkeypatch.setenv("EVOSCIENTIST_SANDBOX_MAX_OUTPUT_BYTES", "64")
    code = "import sys; sys.stdout.write('x' * 10000)"

    result = executor.execute(f"python3 -c {shlex.quote(code)}")

    assert result.exit_code == 0
    assert result.truncated is True
    assert len(result.output.encode("utf-8")) <= 64


def test_real_sandbox_enforces_timeout_and_file_limit(
    executor: NativeSandboxExecutor, monkeypatch: pytest.MonkeyPatch
):
    started = time.monotonic()
    timed_out = executor.execute("sleep 5", timeout=1)
    assert timed_out.exit_code == 124
    assert time.monotonic() - started < 3

    monkeypatch.setenv("EVOSCIENTIST_SANDBOX_FILE_SIZE_BYTES", "1024")
    code = "f=open('large.bin','wb'); f.write(b'x' * 4096); f.flush()"
    limited = executor.execute(f"python3 -c {shlex.quote(code)}")
    assert limited.exit_code != 0
    assert (executor.files_dir / "large.bin").stat().st_size <= 1024


def test_real_sandbox_kills_background_descendants(executor: NativeSandboxExecutor):
    result = executor.execute("sleep 30 >/dev/null 2>&1 & echo $! > child.pid")
    assert result.exit_code == 0
    child_pid = int((executor.files_dir / "child.pid").read_text(encoding="ascii"))
    time.sleep(0.1)
    with pytest.raises(ProcessLookupError):
        os.kill(child_pid, 0)


def test_file_tools_and_pandoc_share_one_workspace(tmp_path: Path):
    ensure_native_sandbox_ready()
    files = tmp_path / "files"
    runtime = tmp_path / "runtime"
    files.mkdir()
    runtime.mkdir()
    backend = NativeWorkspaceBackend(files, runtime, timeout=20)
    assert (
        backend.write("/workspace/source.md", "# 火电分析\n\n同一会话文件。\n").error
        is None
    )

    result = backend.execute(
        "mkdir -p uploads results && "
        'pandoc source.md -o "uploads/湖南 火电.docx" && '
        'pandoc "uploads/湖南 火电.docx" -o results/report.html'
    )

    assert result.exit_code == 0, result.output
    assert (
        backend.read("/workspace/uploads/湖南 火电.docx").file_data["encoding"]
        == "base64"
    )
    report = backend.read("/workspace/results/report.html")
    assert report.error is None
    assert "火电分析" in report.file_data["content"]
    assert backend.download_files(["/workspace/results/report.html"])[0].content


@pytest.mark.asyncio
async def test_async_cancellation_terminates_process_group(tmp_path: Path):
    ensure_native_sandbox_ready()
    files = tmp_path / "files"
    runtime = tmp_path / "runtime"
    files.mkdir()
    runtime.mkdir()
    backend = NativeWorkspaceBackend(files, runtime, timeout=30)

    task = asyncio.create_task(
        backend.aexecute("echo $$ > shell.pid; sleep 30", timeout=30)
    )
    for _ in range(50):
        if (files / "shell.pid").exists():
            break
        await asyncio.sleep(0.02)
    assert (files / "shell.pid").exists()
    shell_pid = int((files / "shell.pid").read_text(encoding="ascii"))

    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task
    await asyncio.sleep(0.1)
    with pytest.raises(ProcessLookupError):
        os.kill(shell_pid, 0)