Merge pull request #23 from lzwjava/refactor-login-cleanup

Refactor and clean up login process

Author: lzwjava

.coverage

@@ -32,5 +32,4 @@ public/config.json
 node_modules/
 package-lock.json
 **/package-lock.json
-node_modules/
 .coverage

.gitignore

@@ -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)

integration_tests/test_exec_integration.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:])))

mini_copilot/exec_tool.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
                 )

mini_copilot/main.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()