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,40 @@
+import os
+import unittest
+from mini_copilot.main import MiniCopilot
+from tools.patch_tool import PatchTool
+
+
+class TestEditToolIntegration(unittest.TestCase):
+    def setUp(self):
+        self.copilot = MiniCopilot()
+        self.test_file = "integration_test_file.py"
+        with open(self.test_file, "w") as f:
+            f.write("def hello():\n    print('hello world')\n")
+
+    def tearDown(self):
+        if os.path.exists(self.test_file):
+            os.remove(self.test_file)
+
+    def test_apply_patch_integration(self):
+        # We simulate what the agent would do: generate a patch and call PatchTool
+        patch = f"""--- {self.test_file}
++++ {self.test_file}
+@@ -1,2 +1,2 @@
+ def hello():
+-    print('hello world')
++    print('hello integration')
+"""
+        new_content = PatchTool.apply_patch(self.test_file, patch)
+
+        with open(self.test_file, "w") as f:
+            f.write(new_content)
+
+        with open(self.test_file, "r") as f:
+            updated_content = f.read()
+
+        self.assertIn("print('hello integration')", updated_content)
+        self.assertNotIn("print('hello world')", updated_content)
+
+
+if __name__ == "__main__":
+    unittest.main()

integration_tests/test_edit_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)

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

tests/test_exec_tool.py

@@ -0,0 +1,62 @@
+import unittest
+import os
+from tools.patch_tool import PatchTool
+
+
+class TestPatchTool(unittest.TestCase):
+    def setUp(self):
+        self.test_file = "test_file.txt"
+        with open(self.test_file, "w") as f:
+            f.write("Line 1\nLine 2\nLine 3\nLine 4\nLine 5\n")
+
+    def tearDown(self):
+        if os.path.exists(self.test_file):
+            os.remove(self.test_file)
+
+    def test_basic_replace(self):
+        patch = """--- test_file.txt
++++ test_file.txt
+@@ -2,1 +2,1 @@
+-Line 2
++Line Two Modified
+"""
+        new_content = PatchTool.apply_patch(self.test_file, patch)
+        self.assertIn("Line Two Modified\n", new_content)
+        self.assertNotIn("Line 2\n", new_content)
+        self.assertEqual(new_content.splitlines()[1], "Line Two Modified")
+
+    def test_multi_line_hunk(self):
+        patch = """--- test_file.txt
++++ test_file.txt
+@@ -3,2 +3,3 @@
+-Line 3
+-Line 4
++Line Three
++Line Three.Five
++Line Four
+"""
+        new_content = PatchTool.apply_patch(self.test_file, patch)
+        lines = new_content.splitlines()
+        self.assertEqual(lines[2], "Line Three")
+        self.assertEqual(lines[3], "Line Three.Five")
+        self.assertEqual(lines[4], "Line Four")
+        self.assertEqual(len(lines), 6)
+
+    def test_multiple_hunks(self):
+        patch = """--- test_file.txt
++++ test_file.txt
+@@ -1,1 +1,1 @@
+-Line 1
++First Line
+@@ -5,1 +5,1 @@
+-Line 5
++Last Line
+"""
+        new_content = PatchTool.apply_patch(self.test_file, patch)
+        lines = new_content.splitlines()
+        self.assertEqual(lines[0], "First Line")
+        self.assertEqual(lines[-1], "Last Line")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_patch_tool.py

@@ -0,0 +1,81 @@
+import os
+import re
+
+
+class PatchTool:
+    @staticmethod
+    def apply_patch(file_path: str, patch_content: str) -> str:
+        """
+        Applies a unified diff patch to a file.
+        Returns the new content of the file.
+        """
+        if not os.path.exists(file_path):
+            content = []
+        else:
+            with open(file_path, "r", encoding="utf-8") as f:
+                content = f.readlines()
+
+        patch_lines = patch_content.splitlines(keepends=True)
+        hunk_re = re.compile(r"^@@ -(\d+),?(\d*) \+(\d+),?(\d*) @@")
+
+        hunks = []
+        current_hunk = None
+
+        for line in patch_lines:
+            if line.startswith("---") or line.startswith("+++"):
+                continue
+
+            match = hunk_re.match(line)
+            if match:
+                if current_hunk:
+                    hunks.append(current_hunk)
+                current_hunk = {
+                    "start_old": int(match.group(1)),
+                    "len_old": int(match.group(2) or 1),
+                    "start_new": int(match.group(3)),
+                    "len_new": int(match.group(4) or 1),
+                    "lines": [],
+                }
+            elif current_hunk:
+                current_hunk["lines"].append(line)
+
+        if current_hunk:
+            hunks.append(current_hunk)
+
+        result_lines = list(content)
+        offset = 0
+
+        for hunk in hunks:
+            start_in_file = hunk["start_old"] - 1 + offset
+            old_len = hunk["len_old"]
+
+            new_hunk_lines = []
+            for h_line in hunk["lines"]:
+                if h_line.startswith(" "):
+                    new_hunk_lines.append(h_line[1:])
+                elif h_line.startswith("+"):
+                    new_hunk_lines.append(h_line[1:])
+                elif h_line.startswith("-"):
+                    pass
+
+            # Pad result_lines if the patch references lines beyond current EOF
+            while len(result_lines) < start_in_file:
+                result_lines.append("\n")
+
+            result_lines[start_in_file : start_in_file + old_len] = new_hunk_lines
+            offset += len(new_hunk_lines) - old_len
+
+        return "".join(result_lines)
+
+
+if __name__ == "__main__":
+    import sys
+
+    if len(sys.argv) == 3:
+        path, patch_file = sys.argv[1], sys.argv[2]
+        with open(patch_file, "r") as pf:
+            patch = pf.read()
+        new_text = PatchTool.apply_patch(path, patch)
+        with open(path, "w") as f:
+            f.write(new_text)
+        print(f"Applied patch to {path}")