import json
import os
import shutil
import subprocess
import sys
import tempfile
import time
import unittest
from unittest import mock

import shellmcp

HERE = os.path.dirname(os.path.abspath(__file__))


class Run(unittest.TestCase):
    def test_exit_code_and_combined_output(self):
        self.assertEqual(shellmcp.run("echo hi; echo oops >&2; exit 3", HERE),
                         "exit code 3\nhi\noops\n")

    def test_long_output_keeps_head_and_tail(self):
        out = shellmcp.run("printf 'a%.0s' $(seq 100); printf 'b%.0s' $(seq 100)", HERE, limit=20)
        self.assertEqual(out, "exit code 0\naaaaaaaaaa\n[... 180 characters cut ...]\nbbbbbbbbbb")

    def test_timeout_kills_the_whole_group(self):
        start = time.monotonic()
        out = shellmcp.run("sleep 30 & sleep 30", HERE, timeout=1)
        self.assertLess(time.monotonic() - start, 5)
        self.assertEqual(out, "timed out after 1 s, killed\n")

    def test_no_stdin_and_no_pager(self):
        self.assertEqual(shellmcp.run("cat; echo $PAGER", HERE), "exit code 0\ncat\n")

@unittest.skipUnless(shutil.which("bwrap"), "needs bwrap")
class Sandbox(unittest.TestCase):
    def test_readonly_cannot_write_and_hides_paths(self):
        with tempfile.TemporaryDirectory(dir=HERE) as d:  # not /tmp: the sandbox replaces /tmp
            os.mkdir(os.path.join(d, "secret"))
            open(os.path.join(d, "secret", "key"), "w").close()
            prefix = shellmcp.sandbox(d, [os.path.join(d, "secret")])
            out = shellmcp.run("touch new; ls secret; echo > /tmp/x && echo tmp-ok", d,
                               prefix=prefix)
            self.assertIn("Read-only file system", out)
            self.assertNotIn("key", out)
            self.assertIn("tmp-ok", out)
            self.assertFalse(os.path.exists(os.path.join(d, "new")))

class Describe(unittest.TestCase):
    def test_names_only_installed_tools_and_the_mode(self):
        with mock.patch("shutil.which", lambda name: name in ("fd", "rg")):
            text = shellmcp.describe("readonly", "/srv", 60, 8000)
        self.assertIn("in /srv.", text)
        self.assertIn("Read-only sandbox", text)
        self.assertIn("prefer these: rg: search file contents (instead of grep); "
                      "fd: find files (instead of find).", text)
        self.assertNotIn("jq", text)
        self.assertIn("full permissions", shellmcp.describe("all", "/srv", 60, 8000))

class Protocol(unittest.TestCase):
    def talk(self, *messages):
        lines = "".join(json.dumps(m) + "\n" for m in messages) + "not json\n"
        done = subprocess.run([sys.executable, os.path.join(HERE, "shellmcp.py"), "--mode", "all",
                               "--cwd", HERE], input=lines, capture_output=True, text=True,
                              timeout=10)
        return [json.loads(line) for line in done.stdout.splitlines()]

    def test_initialize_list_and_call(self):
        replies = self.talk(
            {"jsonrpc": "2.0", "id": 1, "method": "initialize",
             "params": {"protocolVersion": "2025-06-18", "capabilities": {},
                        "clientInfo": {"name": "test", "version": "0"}}},
            {"jsonrpc": "2.0", "method": "notifications/initialized"},
            {"jsonrpc": "2.0", "id": 2, "method": "tools/list"},
            {"jsonrpc": "2.0", "id": 3, "method": "tools/call",
             "params": {"name": "run", "arguments": {"command": "echo hi"}}},
            {"jsonrpc": "2.0", "id": 4, "method": "tools/call",
             "params": {"name": "nope", "arguments": {}}},
            {"jsonrpc": "2.0", "id": 5, "method": "resources/list"})
        self.assertEqual([r.get("id") for r in replies], [1, 2, 3, 4, 5, None])
        self.assertEqual(replies[0]["result"]["protocolVersion"], "2025-06-18")
        self.assertEqual(replies[1]["result"]["tools"][0]["name"], "run")
        self.assertEqual(replies[2]["result"],
                         {"content": [{"type": "text", "text": "exit code 0\nhi\n"}],
                          "isError": False})
        self.assertEqual([r["error"]["code"] for r in replies[3:]], [-32602, -32601, -32700])

    def test_unknown_protocol_gets_the_newest(self):
        reply = self.talk({"jsonrpc": "2.0", "id": 1, "method": "initialize",
                           "params": {"protocolVersion": "1999-01-01"}})[0]
        self.assertEqual(reply["result"]["protocolVersion"], shellmcp.PROTOCOLS[0])