Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Binary file removed .coverage
Binary file not shown.
1 change: 0 additions & 1 deletion .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -32,5 +32,4 @@ public/config.json
node_modules/
package-lock.json
**/package-lock.json
node_modules/
.coverage
86 changes: 86 additions & 0 deletions integration_tests/test_exec_integration.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,86 @@
import subprocess
import time
import sys
import os


def run_integration_test():
print("🚀 Starting Integration Test for 'exec' tool...")

# Ensure current directory is in PYTHONPATH for the subprocess
env = os.environ.copy()
env["PYTHONPATH"] = os.getcwd() + ":" + env.get("PYTHONPATH", "")

# Start the mini-copilot process
process = subprocess.Popen(
[sys.executable, "-m", "mini_copilot.main"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT, # Combine stdout and stderr
text=True,
bufsize=1,
cwd=os.getcwd(),
env=env,
)

try:
# 1. Wait for prompt
print("Waiting for prompt...")
output = ""
start_time = time.time()
while time.time() - start_time < 30:
char = process.stdout.read(1)
if not char:
break
output += char
if "> " in output:
print("✅ Prompt detected.")
break

if "> " not in output:
print(f"❌ Timed out waiting for prompt. Last output: {output}")
return False

# 2. Trigger tool
print("Sending message to trigger 'exec' tool...")
process.stdin.write("run command: echo integration_test_success\n")
process.stdin.flush()

# 3. Look for tool execution
print("Monitoring for tool invocation and response...")
found_tool = False
found_output = False
start_time = time.time()
output = ""
while time.time() - start_time < 60:
char = process.stdout.read(1)
if not char:
break
output += char
sys.stdout.write(char) # Echo for observability
sys.stdout.flush()

if "[exec] Running command: echo integration_test_success" in output:
found_tool = True

if "integration_test_success" in output and found_tool:
found_output = True
print("\n✅ Found expected output in responses!")
break

if found_output:
print("\n🎉 Integration Test PASSED!")
return True
else:
print("\n❌ Integration Test FAILED.")
return False

finally:
process.terminate()


if __name__ == "__main__":
if run_integration_test():
sys.exit(0)
else:
sys.exit(1)
26 changes: 26 additions & 0 deletions mini_copilot/exec_tool.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
import subprocess


def exec_command(command: str) -> str:
"""Execute a shell command and return its output (stdout or stderr)."""
print(f"[exec] Running command: {command}")
try:
# Run command with a 30s timeout, capturing both stdout and stderr
result = subprocess.run(
command, shell=True, capture_output=True, text=True, timeout=30
)
output = result.stdout if result.returncode == 0 else result.stderr
if not output.strip():
output = f"(Process exited with code {result.returncode})"
return output
except subprocess.TimeoutExpired:
return "Error: Command timed out after 30 seconds."
except Exception as e:
return f"Error executing command: {str(e)}"


if __name__ == "__main__":
import sys

if len(sys.argv) > 1:
print(exec_command(" ".join(sys.argv[1:])))
34 changes: 33 additions & 1 deletion mini_copilot/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ def completer(text, state):

from mini_copilot.github_api import chat, get_copilot_token
from mini_copilot.web_search import web_search
from mini_copilot.exec_tool import exec_command as exec
from mini_copilot.commands.auth import handle_login_command
from mini_copilot.commands.model import handle_model_command
from mini_copilot.commands.search_provider import handle_search_provider_command
Expand Down Expand Up @@ -59,7 +60,25 @@ def completer(text, state):
},
},
}
TOOLS = [WEB_SEARCH_TOOL]

EXEC_COMMAND_TOOL = {
"type": "function",
"function": {
"name": "exec",
"description": "Execute a shell command on the local system and return the output.",
"parameters": {
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "The shell command to execute.",
},
},
"required": ["command"],
},
},
}
TOOLS = [WEB_SEARCH_TOOL, EXEC_COMMAND_TOOL]


def load_github_token():
Expand Down Expand Up @@ -167,6 +186,19 @@ def main():
"content": search_context,
}
)

if function_name == "exec":
command = function_args.get("command")
output = exec(command)

messages.append(
{
"tool_call_id": tool_call["id"],
"role": "tool",
"name": function_name,
"content": output,
}
)
response_message = chat(
messages, copilot_token, current_model, tools=TOOLS
)
Expand Down
21 changes: 21 additions & 0 deletions tests/test_exec_tool.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
import unittest
from mini_copilot.exec_tool import exec_command


class TestExecTool(unittest.TestCase):
def test_exec_echo(self):
output = exec_command("echo 'hello world'")
self.assertEqual(output.strip(), "hello world")

def test_exec_ls(self):
output = exec_command("ls pyproject.toml")
self.assertIn("pyproject.toml", output)

def test_exec_error(self):
# Depending on the system, the error message might vary, but it should contain some indication of failure
output = exec_command("non_existent_command_12345")
self.assertIn("not found", output.lower())


if __name__ == "__main__":
unittest.main()
Loading