diff --git a/.coverage b/.coverage deleted file mode 100644 index b82b2b2..0000000 Binary files a/.coverage and /dev/null differ diff --git a/.gitignore b/.gitignore index 4162ab1..6e22d58 100644 --- a/.gitignore +++ b/.gitignore @@ -32,5 +32,4 @@ public/config.json node_modules/ package-lock.json **/package-lock.json -node_modules/ .coverage diff --git a/integration_tests/test_exec_integration.py b/integration_tests/test_exec_integration.py new file mode 100644 index 0000000..c94c869 --- /dev/null +++ b/integration_tests/test_exec_integration.py @@ -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) diff --git a/mini_copilot/exec_tool.py b/mini_copilot/exec_tool.py new file mode 100644 index 0000000..a497a60 --- /dev/null +++ b/mini_copilot/exec_tool.py @@ -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:]))) diff --git a/mini_copilot/main.py b/mini_copilot/main.py index f9f5d96..18331da 100644 --- a/mini_copilot/main.py +++ b/mini_copilot/main.py @@ -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 @@ -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(): @@ -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 ) diff --git a/tests/test_exec_tool.py b/tests/test_exec_tool.py new file mode 100644 index 0000000..f185a53 --- /dev/null +++ b/tests/test_exec_tool.py @@ -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()