write_tests #14
@ -178,9 +178,12 @@ def trim_messages(messages: list) -> list:
|
||||
tail = messages[2:]
|
||||
original_task = messages[1]["content"] if len(messages) > 1 else ""
|
||||
|
||||
# Drop the oldest messages first (index 2 onwards) until we are under the limit.
|
||||
# The system prompt (0) and original task (1) are never dropped.
|
||||
while tail and sum(len(m["content"]) for m in head + tail) > MAX_HISTORY_CHARS:
|
||||
tail.pop(0)
|
||||
|
||||
# Inject a reminder so the agent doesn't lose track of its goal after trimming.
|
||||
reminder = {
|
||||
"role": "user",
|
||||
"content": (
|
||||
@ -337,7 +340,7 @@ class CodingAgent:
|
||||
payload = {
|
||||
"model": self.model,
|
||||
"messages": messages,
|
||||
"temperature": 0.2,
|
||||
"temperature": 0.2, # low temperature → deterministic, more reliable tool calls
|
||||
"max_tokens": 4096,
|
||||
"stream": False,
|
||||
}
|
||||
@ -439,7 +442,9 @@ class CodingAgent:
|
||||
result = await dispatch_tool(tool_name, arguments)
|
||||
result = truncate_result(result)
|
||||
|
||||
# Build feedback – nudge agent to replan on errors
|
||||
# Wrap the tool output in an XML tag so the LLM can easily find it.
|
||||
# Append a <replan> tag on errors to force the agent to reconsider
|
||||
# its plan rather than blindly retrying the same failing action.
|
||||
feedback = f'<tool_result tool="{tool_name}">\n{result}\n</tool_result>'
|
||||
if result.startswith("ERROR") or result.startswith("SYNTAX ERROR"):
|
||||
feedback += (
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
"""Chat Manager - Handles chat history and AI communication"""
|
||||
"""Manages the chat history and communication with the AI model API."""
|
||||
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
@ -9,6 +9,13 @@ load_dotenv()
|
||||
|
||||
|
||||
class ChatManager:
|
||||
"""Handles sending messages and maintaining conversation history.
|
||||
|
||||
Connects to an OpenAI-compatible REST endpoint configured via environment
|
||||
variables. All messages (user, assistant, system) are kept in memory so
|
||||
the full conversation is sent with every request.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.api_host = os.getenv("HOST")
|
||||
self.api_port = os.getenv("PORT")
|
||||
@ -22,15 +29,23 @@ class ChatManager:
|
||||
self.chat_history = []
|
||||
|
||||
def add_message(self, role: str, content: str) -> None:
|
||||
"""Append a single message to the conversation history."""
|
||||
self.chat_history.append({"role": role, "content": content})
|
||||
|
||||
def get_history(self) -> list:
|
||||
return self.chat_history
|
||||
"""Return a copy of the conversation history."""
|
||||
return list(self.chat_history)
|
||||
|
||||
def clear_history(self) -> None:
|
||||
"""Wipe the conversation history (starts a fresh chat)."""
|
||||
self.chat_history = []
|
||||
|
||||
def send_message(self, user_message: str) -> str:
|
||||
"""Send a user message to the AI and return its reply.
|
||||
|
||||
Adds the user message to history, calls the API with the full history
|
||||
as context, and appends the AI reply to history before returning it.
|
||||
"""
|
||||
# Add user message to history
|
||||
self.add_message("user", user_message)
|
||||
|
||||
@ -44,6 +59,7 @@ class ChatManager:
|
||||
if self.api_key and self.api_key != "EMPTY":
|
||||
headers["Authorization"] = f"Bearer {self.api_key}"
|
||||
|
||||
# Full history is sent so the model has multi-turn conversation context
|
||||
payload = {
|
||||
"model": self.model,
|
||||
"messages": self.chat_history,
|
||||
@ -91,6 +107,7 @@ class ChatManager:
|
||||
raise Exception(error_msg)
|
||||
|
||||
def get_chat_display(self) -> list:
|
||||
"""Return a copy of the history suitable for display in the UI."""
|
||||
return [
|
||||
{"role": msg["role"], "content": msg["content"]}
|
||||
for msg in self.chat_history
|
||||
|
||||
@ -1,9 +1,66 @@
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
class DebugLogger:
|
||||
"""In-memory logger for code execution events.
|
||||
|
||||
Collects timestamped INFO and ERROR entries during a single run.
|
||||
Call clear() before each new execution to start fresh.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.logs: list[dict] = []
|
||||
|
||||
def log(self, message: str) -> None:
|
||||
"""Append a general info message."""
|
||||
self.logs.append({
|
||||
"level": "INFO",
|
||||
"message": message,
|
||||
"timestamp": datetime.now().strftime("%H:%M:%S"),
|
||||
})
|
||||
|
||||
def log_error(self, error_message: str) -> None:
|
||||
"""Append an error message."""
|
||||
self.logs.append({
|
||||
"level": "ERROR",
|
||||
"message": error_message,
|
||||
"timestamp": datetime.now().strftime("%H:%M:%S"),
|
||||
})
|
||||
|
||||
def get_logs(self) -> list[dict]:
|
||||
"""Return a copy of all collected log entries."""
|
||||
return list(self.logs)
|
||||
|
||||
def clear(self) -> None:
|
||||
"""Reset the log — call before each new execution."""
|
||||
self.logs = []
|
||||
|
||||
def log(self, message):
|
||||
self.logs.append(message)
|
||||
|
||||
def get_logs(self):
|
||||
return self.logs
|
||||
|
||||
def format_debug_output(self, output: dict) -> str:
|
||||
"""Format an ExecutionEngine result dict into a human-readable string.
|
||||
|
||||
Args:
|
||||
output: dict with keys 'stdout', 'stderr', and 'rc'.
|
||||
|
||||
Returns:
|
||||
A formatted string ready for display in the UI.
|
||||
"""
|
||||
lines = []
|
||||
|
||||
status = "SUCCESS" if output.get("rc") == 0 else "FAILED"
|
||||
lines.append(f"[{status}] Exit code: {output.get('rc')}")
|
||||
|
||||
if output.get("stdout"):
|
||||
lines.append("\n--- stdout ---")
|
||||
lines.append(output["stdout"].rstrip())
|
||||
|
||||
if output.get("stderr"):
|
||||
lines.append("\n--- stderr ---")
|
||||
lines.append(output["stderr"].rstrip())
|
||||
|
||||
if not output.get("stdout") and not output.get("stderr"):
|
||||
lines.append("No output produced.")
|
||||
|
||||
for entry in self.logs:
|
||||
lines.append(f"[{entry['timestamp']}] [{entry['level']}] {entry['message']}")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
@ -1,19 +1,38 @@
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
RUN_TIMEOUT = 30 # seconds
|
||||
# Maximum time (seconds) a subprocess is allowed to run before being killed.
|
||||
RUN_TIMEOUT = 30
|
||||
|
||||
|
||||
class ExecutionEngine:
|
||||
"""Runs files from the editor in a subprocess and returns the output.
|
||||
|
||||
Currently supports Python (.py) and LaTeX (.tex) files.
|
||||
Returns a dict with keys: stdout, stderr, rc (return code).
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
|
||||
def run_code(self, active_file: Path) -> dict:
|
||||
"""Execute the given file and return its output.
|
||||
|
||||
Args:
|
||||
active_file: Absolute path to the file that should be run.
|
||||
|
||||
Returns:
|
||||
{"stdout": str, "stderr": str, "rc": int}
|
||||
rc == 0 means success, anything else is an error.
|
||||
"""
|
||||
suffix = active_file.suffix
|
||||
current_dir = active_file.parent.resolve()
|
||||
|
||||
# Build the shell command depending on file type
|
||||
if suffix == ".py":
|
||||
cmd = ["py", active_file.name]
|
||||
elif suffix == ".tex":
|
||||
# pdflatex in non-interactive mode so it never waits for input
|
||||
cmd = [
|
||||
"pdflatex",
|
||||
"-interaction=nonstopmode",
|
||||
@ -22,20 +41,21 @@ class ExecutionEngine:
|
||||
]
|
||||
else:
|
||||
return {"stdout": "", "stderr": f"Unsupported file type: {suffix}", "rc": 1}
|
||||
|
||||
|
||||
try:
|
||||
proc = subprocess.run(
|
||||
cmd,
|
||||
cwd=current_dir,
|
||||
cwd=current_dir, # run inside the file's own directory
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=RUN_TIMEOUT,
|
||||
)
|
||||
return {"stdout": proc.stdout, "stderr": proc.stderr, "rc": proc.returncode}
|
||||
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
return {"stdout": "", "stderr": f"Timed out after {RUN_TIMEOUT}s", "rc": -1}
|
||||
except FileNotFoundError as e:
|
||||
# Raised when the interpreter/compiler binary is not found on PATH
|
||||
return {"stdout": "", "stderr": str(e), "rc": -1}
|
||||
except Exception as e:
|
||||
return {"stdout": "", "stderr": str(e), "rc": -1}
|
||||
return {"stdout": "", "stderr": str(e), "rc": -1}
|
||||
|
||||
@ -1,6 +1,13 @@
|
||||
"""Manages all file and folder operations inside the workspace directory.
|
||||
|
||||
Every method validates that the target path stays inside the workspace before
|
||||
touching the filesystem, preventing path-traversal attacks.
|
||||
"""
|
||||
|
||||
import streamlit as st
|
||||
from pathlib import Path
|
||||
|
||||
# The workspace folder is created at module load so it always exists.
|
||||
WORKSPACE = Path("workspace")
|
||||
WORKSPACE.mkdir(exist_ok=True)
|
||||
|
||||
@ -24,11 +31,12 @@ class FileManager:
|
||||
if not name:
|
||||
st.error(f"Invalid folder name: {name}")
|
||||
return False
|
||||
|
||||
|
||||
# Slashes in the name would silently create nested paths — reject them.
|
||||
if "/" in name or "\\" in name:
|
||||
st.error(f"Invalid folder name (no slashes allowed): {name}")
|
||||
return False
|
||||
|
||||
|
||||
name = Path(name)
|
||||
if relative_path:
|
||||
relative_path = Path(relative_path)
|
||||
@ -37,10 +45,11 @@ class FileManager:
|
||||
|
||||
folder_path = (self.base_path / relative_path / name).resolve()
|
||||
|
||||
# Ensure the resolved path is still inside the workspace (prevents path traversal).
|
||||
if not str(folder_path).startswith(str(self.base_path.resolve())):
|
||||
st.error(f"Access denied: {relative_path}")
|
||||
return False
|
||||
|
||||
|
||||
try:
|
||||
folder_path.mkdir(exist_ok=False)
|
||||
return True
|
||||
@ -66,11 +75,11 @@ class FileManager:
|
||||
if not name or name.strip() == "" :
|
||||
st.error(f"Invalid file name: {name}")
|
||||
return False
|
||||
|
||||
|
||||
name = Path(name)
|
||||
if not name.suffix:
|
||||
name = name.with_suffix(".txt") # Default to .txt if no extension provided
|
||||
|
||||
|
||||
if relative_path:
|
||||
relative_path = Path(relative_path)
|
||||
else:
|
||||
@ -78,10 +87,11 @@ class FileManager:
|
||||
|
||||
file_path = (self.base_path / relative_path / name).resolve()
|
||||
|
||||
# Ensure the resolved path is still inside the workspace (prevents path traversal).
|
||||
if not str(file_path).startswith(str(self.base_path.resolve())):
|
||||
st.error(f"Access denied: {relative_path}")
|
||||
return False
|
||||
|
||||
|
||||
try:
|
||||
file_path.touch(exist_ok=False)
|
||||
return True
|
||||
@ -95,10 +105,11 @@ class FileManager:
|
||||
def read_file(self, relative_path: Path) -> str:
|
||||
"""
|
||||
Reads the content of a file.
|
||||
The relative_path should be the path to the file relative to the base path.
|
||||
Accepts an absolute Path object (as stored in st.session_state.open_files).
|
||||
The path is validated to ensure it stays inside the workspace.
|
||||
|
||||
Args:
|
||||
relative_path (str): The relative path (without base path) to the file to read, including the file name
|
||||
relative_path (Path): Absolute path to the file to read.
|
||||
Returns:
|
||||
str: The content of the file, or an empty string if there was an error.
|
||||
"""
|
||||
@ -110,10 +121,11 @@ class FileManager:
|
||||
if not file_path.is_file():
|
||||
st.error(f"Path is not a file: {relative_path}")
|
||||
return ""
|
||||
# Ensure the resolved path is still inside the workspace (prevents path traversal).
|
||||
if not str(file_path).startswith(str(self.base_path.resolve())):
|
||||
st.error(f"Access denied: {relative_path}")
|
||||
return ""
|
||||
|
||||
|
||||
try:
|
||||
with open(file_path, "r") as f:
|
||||
return f.read()
|
||||
@ -126,21 +138,23 @@ class FileManager:
|
||||
|
||||
def save_file(self, relative_path: str, content: str) -> bool:
|
||||
"""
|
||||
Saves content to a file.
|
||||
The relative_path should be the path to the file relative to the base path.
|
||||
|
||||
Saves content to a file.
|
||||
Accepts an absolute path string (as stored in st.session_state.open_files).
|
||||
The path is validated to ensure it stays inside the workspace.
|
||||
|
||||
Args:
|
||||
relative_path (str): The relative path(without base path) to the file to save, including the file name
|
||||
content (str): The content to write to the file
|
||||
relative_path (str): Absolute path to the file to save, including the file name.
|
||||
content (str): The content to write to the file.
|
||||
Returns:
|
||||
bool: True if save was successful, False otherwise.
|
||||
"""
|
||||
file_path = (Path(relative_path)).resolve()
|
||||
|
||||
# Ensure the resolved path is still inside the workspace (prevents path traversal).
|
||||
if not str(file_path).startswith(str(self.base_path.resolve())):
|
||||
st.error(f"Access denied: {relative_path}")
|
||||
return False
|
||||
|
||||
|
||||
try:
|
||||
with open(file_path, "w") as f:
|
||||
f.write(content)
|
||||
@ -148,7 +162,7 @@ class FileManager:
|
||||
except Exception as e:
|
||||
st.error(f"Error saving file {relative_path}: {str(e)}")
|
||||
return False
|
||||
|
||||
|
||||
def rename_file(self, old_relative_path: str, new_name: str) -> bool:
|
||||
"""
|
||||
Renames a file while keeping the same extension.
|
||||
@ -163,20 +177,22 @@ class FileManager:
|
||||
if not new_name or new_name.strip() == "":
|
||||
st.error(f"Invalid file name: {new_name}")
|
||||
return False
|
||||
|
||||
|
||||
file_type = Path(old_relative_path).suffix
|
||||
new_name = Path(new_name)
|
||||
|
||||
# Force the original extension so the file type cannot be changed by renaming.
|
||||
if not Path(new_name).suffix == file_type:
|
||||
new_name = Path(new_name).with_suffix(file_type) # Ensure the file extension remains the same
|
||||
|
||||
old_file_path = (Path(self.base_path / old_relative_path)).resolve()
|
||||
new_file_path = old_file_path.parent / new_name
|
||||
|
||||
# Both old and new paths must stay inside the workspace.
|
||||
if not str(old_file_path).startswith(str(self.base_path.resolve())) or not str(new_file_path).startswith(str(self.base_path.resolve())):
|
||||
st.error(f"Access denied: {old_relative_path}")
|
||||
return False
|
||||
|
||||
|
||||
try:
|
||||
old_file_path.rename(new_file_path)
|
||||
return True
|
||||
@ -196,6 +212,7 @@ class FileManager:
|
||||
"""
|
||||
folder_path = (self.base_path / relative_path).resolve()
|
||||
|
||||
# Ensure the resolved path is still inside the workspace (prevents path traversal).
|
||||
if not str(folder_path).startswith(str(self.base_path.resolve())):
|
||||
st.error(f"Access denied: {relative_path}")
|
||||
return False
|
||||
@ -223,7 +240,6 @@ class FileManager:
|
||||
"""
|
||||
file_path = Path(relative_path)
|
||||
abs_file_path = (Path(self.base_path) / file_path).resolve()
|
||||
print(f"Absolute file path resolved to: {abs_file_path}") # Debugging info
|
||||
|
||||
if not str(abs_file_path).startswith(str(self.base_path.resolve())):
|
||||
st.error(f"Access denied: {relative_path}")
|
||||
@ -254,11 +270,11 @@ class FileManager:
|
||||
|
||||
for item in sorted(path.iterdir()):
|
||||
if item.is_dir():
|
||||
tree[item.name] = build_tree(item)
|
||||
tree[item.name] = build_tree(item) # recurse into sub-folders
|
||||
else:
|
||||
tree[item.name] = None
|
||||
tree[item.name] = None # leaf node for files
|
||||
return tree
|
||||
return build_tree(self.base_path)
|
||||
|
||||
if __name__ == "__main__":
|
||||
FileManager()
|
||||
FileManager()
|
||||
|
||||
@ -1,18 +1,26 @@
|
||||
"""System Prompter - Builds system prompts with optional file context"""
|
||||
"""Builds the system prompt that is sent to the AI at the start of each chat session."""
|
||||
|
||||
MAX_FILE_CHARS = 4000 # Limit file context to avoid token overflow
|
||||
# Prevents very large files from flooding the context window with tokens.
|
||||
MAX_FILE_CHARS = 4000
|
||||
|
||||
|
||||
class SystemPrompter:
|
||||
"""Generates system prompts for the chat assistant.
|
||||
|
||||
When a file is open in the editor it can be embedded in the prompt so the
|
||||
AI has direct context of the code the user is currently working on.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def generate_prompt(file_context: dict | None = None) -> str:
|
||||
"""Build a system prompt, optionally embedding a file's content.
|
||||
|
||||
Args:
|
||||
file_context: dict with keys 'name' and 'content', or None.
|
||||
file_context: dict with keys 'name' (filename) and 'content' (raw text),
|
||||
or None if no file should be included.
|
||||
|
||||
Returns:
|
||||
A system prompt string.
|
||||
A ready-to-use system prompt string.
|
||||
"""
|
||||
base = (
|
||||
"You are an expert code assistant integrated into a lightweight code editor. "
|
||||
@ -23,9 +31,11 @@ class SystemPrompter:
|
||||
if file_context:
|
||||
name = file_context.get("name", "unknown")
|
||||
content = file_context.get("content", "")
|
||||
# Truncate large files to avoid exceeding token limits
|
||||
|
||||
# Truncate large files to avoid exceeding the model's token limit
|
||||
if len(content) > MAX_FILE_CHARS:
|
||||
content = content[:MAX_FILE_CHARS] + "\n... [truncated]"
|
||||
|
||||
file_section = (
|
||||
f"\n\nThe user currently has the following file open in the editor:\n"
|
||||
f"<file name=\"{name}\">\n"
|
||||
|
||||
@ -1,8 +1,20 @@
|
||||
"""Entry point for the Streamlit app.
|
||||
|
||||
Runs with: streamlit run frontend/app.py
|
||||
|
||||
Responsibilities:
|
||||
- Configure the page layout
|
||||
- Inject global CSS tweaks
|
||||
- Render the sidebar (navigation + file explorer)
|
||||
- Delegate to the correct view (Chat or Code Editor) based on the radio selection
|
||||
"""
|
||||
|
||||
import streamlit as st
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# Add project root to Python path for imports
|
||||
# Add the project root to sys.path so backend imports work regardless of
|
||||
# where streamlit is launched from.
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
|
||||
from frontend.sidebar import render_sidebar
|
||||
@ -10,12 +22,16 @@ from frontend.editor import render_editor
|
||||
from frontend.chat import render_chat
|
||||
from frontend.state import init_state
|
||||
|
||||
# Initialise all session-state keys before any widget is rendered
|
||||
init_state()
|
||||
|
||||
|
||||
def main():
|
||||
st.set_page_config(page_title="Lightweight code editor", layout="wide")
|
||||
|
||||
# Small spacing corrections applied globally:
|
||||
# - Reduce the default top padding of the main content area
|
||||
# - Pull the sidebar content up so the logo sits at the very top
|
||||
st.markdown(
|
||||
"""
|
||||
<style>
|
||||
@ -28,14 +44,17 @@ def main():
|
||||
|
||||
st.title("Lightweight code editor")
|
||||
|
||||
# Re-run init_state to cover any keys that might have been missed on cold start
|
||||
init_state()
|
||||
|
||||
render_sidebar()
|
||||
|
||||
# Switch between the two main views based on the sidebar radio button
|
||||
if st.session_state.get("radio_interface_options") == "Code Editor":
|
||||
render_editor()
|
||||
elif st.session_state.get("radio_interface_options") == "Chat with AI Assistant":
|
||||
render_chat()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
@ -1,3 +1,5 @@
|
||||
"""Chat view — renders both the normal chat interface and the Coding Agent mode."""
|
||||
|
||||
import streamlit as st
|
||||
from backend.managers.chat_manager import ChatManager
|
||||
from backend.managers.system_prompter import SystemPrompter
|
||||
@ -16,8 +18,10 @@ def _run_async(coro):
|
||||
return loop.run_until_complete(coro)
|
||||
|
||||
def _start_agent(task: str):
|
||||
"""Initialise a fresh CodingAgent, start the task,
|
||||
and propose the first action."""
|
||||
"""Create a new CodingAgent, feed it the task, and propose the first action.
|
||||
Stores the agent and its state in session_state so Streamlit can reference
|
||||
them across reruns without losing progress.
|
||||
"""
|
||||
from backend.agent.coding_agent import CodingAgent
|
||||
agent = CodingAgent()
|
||||
agent.start_task(task)
|
||||
@ -29,13 +33,13 @@ def _start_agent(task: str):
|
||||
|
||||
|
||||
def _approve_action():
|
||||
"""Execute the pending action,
|
||||
append it to the log, then propose the next step."""
|
||||
"""Execute the pending action, log it, then immediately propose the next step."""
|
||||
agent = st.session_state.coding_agent
|
||||
pending = st.session_state.agent_pending_action
|
||||
|
||||
result = _run_async(agent.approve())
|
||||
|
||||
# Append a record to the log so the user can review every completed step.
|
||||
st.session_state.agent_log.append({
|
||||
"thought": pending.get("thought", ""),
|
||||
"tool": result["tool"],
|
||||
@ -44,6 +48,7 @@ def _approve_action():
|
||||
})
|
||||
|
||||
if result["is_done"]:
|
||||
# Agent called the "done" tool — task is fully complete.
|
||||
st.session_state.agent_status = "done"
|
||||
st.session_state.agent_pending_action = None
|
||||
else:
|
||||
@ -53,7 +58,10 @@ def _approve_action():
|
||||
|
||||
|
||||
def _reject_action(feedback: str):
|
||||
"""Reject the pending action with optional feedback, then replan."""
|
||||
"""Reject the pending action with feedback so the agent replans.
|
||||
The pending action is discarded; the agent receives the user's feedback and
|
||||
proposes a different approach on the next call to propose_next_action().
|
||||
"""
|
||||
agent = st.session_state.coding_agent
|
||||
agent.reject(feedback or "Please try a different approach.")
|
||||
next_action = _run_async(agent.propose_next_action())
|
||||
@ -62,7 +70,7 @@ def _reject_action(feedback: str):
|
||||
|
||||
|
||||
def _followup_agent(question: str):
|
||||
"""Inject a follow-up question into the finished agent and resume the loop."""
|
||||
"""Continue a finished task by injecting a follow-up question and resuming the loop."""
|
||||
agent = st.session_state.coding_agent
|
||||
agent.follow_up(question)
|
||||
action = _run_async(agent.propose_next_action())
|
||||
@ -71,7 +79,7 @@ def _followup_agent(question: str):
|
||||
|
||||
|
||||
def _reset_agent():
|
||||
"""Reset all agent state back to idle."""
|
||||
"""Clear all agent state and return to the idle (task input) screen."""
|
||||
st.session_state.coding_agent = None
|
||||
st.session_state.agent_status = "idle"
|
||||
st.session_state.agent_log = []
|
||||
@ -81,13 +89,21 @@ def _reset_agent():
|
||||
# ── Agent Mode UI ─────────────────────────────────────────────────────────────
|
||||
|
||||
def render_agent_mode():
|
||||
# Toggle must always be rendered so Streamlit keeps agent_mode=True in session_state
|
||||
"""Render the step-by-step agent UI.
|
||||
|
||||
Three distinct screens based on agent_status:
|
||||
- "idle" → task description input + Start button
|
||||
- "waiting_approval" → show proposed action, Approve / Reject / Abort
|
||||
- "done" → success message, follow-up input, New Task button
|
||||
"""
|
||||
# The toggle must always render so Streamlit keeps agent_mode=True in session_state.
|
||||
st.toggle("Agent Mode", key="agent_mode")
|
||||
|
||||
agent_status = st.session_state.get("agent_status", "idle")
|
||||
agent_log = st.session_state.get("agent_log", [])
|
||||
|
||||
# ── Agent Log ────────────────────────────────────────────────────────────
|
||||
# Collapsed by default so it doesn't clutter the UI during active tasks.
|
||||
if agent_log:
|
||||
with st.expander(f"Agent Log — {len(agent_log)} step(s) completed", expanded=False):
|
||||
for i, step in enumerate(agent_log):
|
||||
@ -97,6 +113,7 @@ def render_agent_mode():
|
||||
if step.get("arguments"):
|
||||
st.json(step["arguments"])
|
||||
result_text = step.get("result", "")
|
||||
# Colour the result based on whether the tool succeeded or failed.
|
||||
if result_text.startswith("ERROR") or result_text.startswith("SYNTAX ERROR"):
|
||||
st.error(result_text)
|
||||
elif result_text.startswith("OK") or result_text.startswith("DONE"):
|
||||
@ -132,6 +149,8 @@ def render_agent_mode():
|
||||
|
||||
args = pending.get("arguments", {})
|
||||
if args:
|
||||
# Show file content separately as a code block for readability;
|
||||
# other arguments are displayed as JSON.
|
||||
if "content" in args:
|
||||
display_args = {k: v for k, v in args.items() if k != "content"}
|
||||
if display_args:
|
||||
@ -194,7 +213,13 @@ def render_agent_mode():
|
||||
# ── Normal Chat ───────────────────────────────────────────────────────────────
|
||||
|
||||
def render_normal_chat():
|
||||
# Chat history as bubbles
|
||||
"""Render the standard multi-turn chat interface.
|
||||
|
||||
On the first message the system prompt is injected into the history.
|
||||
Each subsequent message appends to the same conversation so the AI retains
|
||||
full context throughout the session.
|
||||
"""
|
||||
# Replay the conversation history as chat bubbles (skip system messages).
|
||||
for message in st.session_state.chat_history:
|
||||
role = message["role"]
|
||||
if role == "system":
|
||||
@ -207,16 +232,17 @@ def render_normal_chat():
|
||||
if user_input:
|
||||
chat_manager = st.session_state.chat_manager
|
||||
|
||||
# Inject system prompt on the first message
|
||||
# On the very first user message, prepend the system prompt so the AI
|
||||
# knows it is a code assistant embedded in an editor.
|
||||
if not chat_manager.get_history():
|
||||
system_prompt = SystemPrompter.generate_prompt()
|
||||
chat_manager.add_message("system", system_prompt)
|
||||
|
||||
# Show user message immediately without waiting for response
|
||||
# Show user message immediately without waiting for response.
|
||||
with st.chat_message("user"):
|
||||
st.markdown(user_input)
|
||||
|
||||
# Show response with spinner while API is called
|
||||
# Call the AI and show its response with a spinner while waiting.
|
||||
with st.chat_message("assistant"):
|
||||
with st.spinner("Thinking..."):
|
||||
try:
|
||||
@ -239,6 +265,7 @@ def render_normal_chat():
|
||||
# ── Entry point ───────────────────────────────────────────────────────────────
|
||||
|
||||
def render_chat():
|
||||
"""Top-level chat view — switches between Agent Mode and normal chat."""
|
||||
if st.session_state.get("agent_mode", False):
|
||||
st.subheader("Coding Agent")
|
||||
render_agent_mode()
|
||||
|
||||
@ -1,3 +1,5 @@
|
||||
"""Code Editor view — renders the Ace editor, file tabs, and execution output."""
|
||||
|
||||
import streamlit as st
|
||||
import streamlit_ace as st_ace
|
||||
from pathlib import Path
|
||||
@ -6,6 +8,7 @@ from backend.managers.file_manager import FileManager
|
||||
from backend.managers.execution_engine import ExecutionEngine
|
||||
from backend.managers.debug_logger import DebugLogger
|
||||
|
||||
# Maps file extensions to Ace editor language modes for syntax highlighting.
|
||||
LANG_MAP = {
|
||||
".py": "python", ".tex": "latex", ".js": "javascript",
|
||||
".html": "html", ".css": "css", ".sh": "bash",
|
||||
@ -18,6 +21,10 @@ LANG_MAP = {
|
||||
|
||||
@st.dialog("Rename File")
|
||||
def _rename_dialog(file_path: str):
|
||||
"""Dialog for renaming the given file.
|
||||
Updates open_files, files_content and active_file in session_state so
|
||||
all tabs and the editor reference the new path immediately.
|
||||
"""
|
||||
fm = FileManager()
|
||||
st.write(f"Current name: **{Path(file_path).name}**")
|
||||
new_name = st.text_input("New name:", value=Path(file_path).stem)
|
||||
@ -34,10 +41,13 @@ def _rename_dialog(file_path: str):
|
||||
if fm.rename_file(rel_path, new_name.strip()):
|
||||
ext = Path(file_path).suffix
|
||||
new_file_path = str(Path(file_path).parent / (Path(new_name.strip()).stem + ext))
|
||||
# Update the open-files list in place so the tab order is preserved.
|
||||
i = st.session_state.open_files.index(file_path)
|
||||
st.session_state.open_files[i] = new_file_path
|
||||
# Transfer cached editor content to the new path key.
|
||||
st.session_state.files_content[new_file_path] = \
|
||||
st.session_state.files_content.pop(file_path)
|
||||
# Update active_file if the renamed file was the active one.
|
||||
if st.session_state.active_file == file_path:
|
||||
st.session_state.active_file = new_file_path
|
||||
st.rerun()
|
||||
@ -75,22 +85,30 @@ def _delete_dialog(abs_file_path: str):
|
||||
|
||||
|
||||
def run_active_file():
|
||||
"""Execute the currently active file and store the result in session_state.
|
||||
Returns the execution result dict {stdout, stderr, return_code}, or None
|
||||
if no active file is set.
|
||||
"""
|
||||
active_file = st.session_state.active_file
|
||||
|
||||
if not active_file:
|
||||
st.warning("No active file to run.")
|
||||
return
|
||||
|
||||
|
||||
execution_engine = ExecutionEngine()
|
||||
debug_logger = DebugLogger()
|
||||
|
||||
debug_logger.clear()
|
||||
debug_logger.log(f"Executing code from {active_file}...")
|
||||
|
||||
with st.spinner(f"Running {Path(active_file).name}..."):
|
||||
output = execution_engine.run_code(Path(active_file))
|
||||
|
||||
debug_logger.log("Execution completed.")
|
||||
|
||||
if output["rc"] == 0:
|
||||
debug_logger.log("Execution completed successfully.")
|
||||
else:
|
||||
debug_logger.log_error(f"Execution failed with exit code {output['rc']}.")
|
||||
|
||||
st.session_state.code_execution_output = {
|
||||
"stdout": output["stdout"],
|
||||
"stderr": output["stderr"],
|
||||
@ -100,6 +118,7 @@ def run_active_file():
|
||||
return result
|
||||
|
||||
def render_editor():
|
||||
"""Render the full Code Editor view with tabs, Ace editor, and run output."""
|
||||
st.subheader("Code Editor")
|
||||
|
||||
if not st.session_state.open_files:
|
||||
@ -109,16 +128,19 @@ def render_editor():
|
||||
fm = FileManager()
|
||||
|
||||
# ── Tab bar via st.tabs() ─────────────────────────────────────────────────
|
||||
# Build one tab per open file, named by the file's basename.
|
||||
tab_names = [Path(f).name for f in st.session_state.open_files]
|
||||
tabs = st.tabs(tab_names)
|
||||
|
||||
for idx, file_path in enumerate(st.session_state.open_files):
|
||||
with tabs[idx]:
|
||||
# Load file content from disk on first open; afterwards use the cached version.
|
||||
if file_path not in st.session_state.files_content:
|
||||
st.session_state.files_content[file_path] = fm.read_file(Path(file_path))
|
||||
|
||||
file_language = LANG_MAP.get(Path(file_path).suffix, "text")
|
||||
|
||||
# Ace editor widget — auto_update sends content to Python on each keystroke.
|
||||
code = st_ace.st_ace(
|
||||
value=st.session_state.files_content[file_path],
|
||||
language=file_language,
|
||||
@ -128,6 +150,7 @@ def render_editor():
|
||||
height=400,
|
||||
)
|
||||
|
||||
# Keep the in-memory cache in sync with what the editor currently shows.
|
||||
if code != st.session_state.files_content[file_path]:
|
||||
st.session_state.files_content[file_path] = code
|
||||
|
||||
@ -141,6 +164,7 @@ def render_editor():
|
||||
if st.button("Close File", key=f"close_{file_path}"):
|
||||
st.session_state.open_files.remove(file_path)
|
||||
st.session_state.files_content.pop(file_path, None)
|
||||
# Switch active_file to the next available tab.
|
||||
st.session_state.active_file = (
|
||||
st.session_state.open_files[0]
|
||||
if st.session_state.open_files else None
|
||||
@ -159,33 +183,34 @@ def render_editor():
|
||||
result = run_active_file()
|
||||
if not result:
|
||||
st.stop()
|
||||
|
||||
|
||||
st.subheader("Execution Output")
|
||||
|
||||
|
||||
# Green on exit code 0 (success), red on anything else (error/crash).
|
||||
if result["return_code"] == 0:
|
||||
st.success(f"Exit code: {result['return_code']}")
|
||||
else:
|
||||
st.error(f"Exit code: {result['return_code']}")
|
||||
|
||||
|
||||
if result["stdout"]:
|
||||
st.text_area(
|
||||
"Standard Output",
|
||||
value=result["stdout"],
|
||||
"Standard Output",
|
||||
value=result["stdout"],
|
||||
height=200,
|
||||
disabled=True,
|
||||
disabled=True,
|
||||
key="run_stdout")
|
||||
|
||||
|
||||
if result["stderr"]:
|
||||
st.text_area(
|
||||
"Standard Error",
|
||||
value=result["stderr"],
|
||||
"Standard Error",
|
||||
value=result["stderr"],
|
||||
height=200,
|
||||
disabled=True,
|
||||
disabled=True,
|
||||
key="run_stderr")
|
||||
if not result["stdout"] and not result["stderr"]:
|
||||
st.info("No output produced by the code execution.")
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
render_editor()
|
||||
render_editor()
|
||||
|
||||
@ -1,10 +1,15 @@
|
||||
"""Sidebar — navigation radio, logo, and the workspace file explorer."""
|
||||
|
||||
import streamlit as st
|
||||
from streamlit_arborist import tree_view
|
||||
from pathlib import Path
|
||||
from backend.managers.file_manager import FileManager
|
||||
|
||||
# Shared FileManager instance for all sidebar operations.
|
||||
fm = FileManager()
|
||||
|
||||
# Maps file extensions (and special keys "folder"/"default") to display emojis
|
||||
# shown next to each entry in the file tree.
|
||||
SUFFIX_MAP = {
|
||||
".py": "🐍", # Python
|
||||
".js": "🟨", # JavaScript (Gelbes Quadrat/Logo)
|
||||
@ -29,11 +34,13 @@ SUFFIX_MAP = {
|
||||
|
||||
@st.dialog("Delete Folder")
|
||||
def _delete_folder_dialog(folder_rel: str, folder_name: str):
|
||||
"""Confirmation dialog before permanently deleting a folder and its contents."""
|
||||
st.warning(f"Delete **{folder_name}** and all its contents? This cannot be undone.")
|
||||
col1, col2 = st.columns(2)
|
||||
with col1:
|
||||
if st.button("Delete", type="primary", use_container_width=True):
|
||||
if fm.delete_folder(folder_rel):
|
||||
# Clear the selected-folder state so the action bar disappears.
|
||||
st.session_state.selected_folder = None
|
||||
st.session_state.selected_folder_rel = None
|
||||
st.rerun()
|
||||
@ -187,19 +194,33 @@ def _delete_file_dialog(relative_file_path: str, file_name: str):
|
||||
# ── File tree ─────────────────────────────────────────────────────────────────
|
||||
|
||||
def build_arborist_tree(tree, parent_path=Path()):
|
||||
"""Convert the FileManager dict tree into the node format expected by streamlit-arborist.
|
||||
|
||||
Folders become nodes with a "children" list; files become leaf nodes
|
||||
with an emoji prefix derived from their extension.
|
||||
|
||||
Args:
|
||||
tree: Nested dict from FileManager.get_file_tree()
|
||||
parent_path: Accumulates the relative path while recursing.
|
||||
|
||||
Returns:
|
||||
List of node dicts accepted by tree_view().
|
||||
"""
|
||||
nodes = []
|
||||
|
||||
for name, content in sorted(tree.items()):
|
||||
full_path = parent_path / name
|
||||
node_id = str(full_path.as_posix())
|
||||
node_id = str(full_path.as_posix()) # forward-slash IDs work cross-platform
|
||||
|
||||
if isinstance(content, dict):
|
||||
# Directory — recurse to build child nodes.
|
||||
nodes.append({
|
||||
"id": node_id,
|
||||
"name": f"{name}",
|
||||
"children": build_arborist_tree(content, full_path)
|
||||
})
|
||||
else:
|
||||
# File — pick an emoji based on extension, fall back to default.
|
||||
suffix = Path(name).suffix
|
||||
icon = SUFFIX_MAP.get(suffix, SUFFIX_MAP["default"])
|
||||
|
||||
@ -213,6 +234,7 @@ def build_arborist_tree(tree, parent_path=Path()):
|
||||
|
||||
|
||||
def render_filetree_arborist(tree):
|
||||
"""Render the interactive file tree and return the currently selected node dict."""
|
||||
data = build_arborist_tree(tree)
|
||||
|
||||
selected = tree_view(
|
||||
@ -220,7 +242,7 @@ def render_filetree_arborist(tree):
|
||||
icons={"open": "📂", "closed": "📁"},
|
||||
height=200,
|
||||
selection=None,
|
||||
select_internal_nodes=True,
|
||||
select_internal_nodes=True, # allow clicking folder names, not just files
|
||||
open_by_default=False
|
||||
)
|
||||
|
||||
@ -230,6 +252,12 @@ def render_filetree_arborist(tree):
|
||||
# ── Sidebar ───────────────────────────────────────────────────────────────────
|
||||
|
||||
def render_sidebar():
|
||||
"""Render the full sidebar: navigation radio and workspace file explorer.
|
||||
|
||||
Tree click handling:
|
||||
- Clicking a file → appended to open_files, set as active_file
|
||||
- Clicking a folder → stored in selected_folder so the action bar appears
|
||||
"""
|
||||
st.sidebar.title("Navigation")
|
||||
|
||||
navigation_section = st.sidebar.container()
|
||||
@ -259,11 +287,14 @@ def render_sidebar():
|
||||
if selected:
|
||||
selected_path = selected.get("id")
|
||||
|
||||
# Only react when the user clicks a *different* node to
|
||||
# avoid re-running on every Streamlit rerender.
|
||||
if st.session_state.last_selected != selected_path:
|
||||
st.session_state.last_selected = selected_path
|
||||
abs_path = fm.base_path / selected_path
|
||||
|
||||
if abs_path.is_file():
|
||||
# Open the file in the editor.
|
||||
st.session_state.selected_folder = None
|
||||
st.session_state.selected_folder_rel = None
|
||||
file_str = str(abs_path)
|
||||
@ -273,11 +304,13 @@ def render_sidebar():
|
||||
st.rerun()
|
||||
|
||||
elif abs_path.is_dir():
|
||||
# Select the folder so its action bar appears below.
|
||||
st.session_state.selected_folder = str(abs_path)
|
||||
st.session_state.selected_folder_rel = selected_path
|
||||
st.rerun()
|
||||
|
||||
# Folder actions — rendered outside the selection block so they persist across reruns
|
||||
# Folder action bar — rendered unconditionally outside the selection
|
||||
# block so it persists across reruns even when no new click happens.
|
||||
if st.session_state.get("selected_folder"):
|
||||
folder_name = Path(st.session_state.selected_folder).name
|
||||
folder_rel = st.session_state.selected_folder_rel
|
||||
@ -303,6 +336,7 @@ def render_sidebar():
|
||||
_delete_file_dialog(file_rel, active_file_name)
|
||||
|
||||
with add_more:
|
||||
# Popover for workspace-root actions (not tied to any selected folder).
|
||||
with st.popover("⚙️ Explorer Options", key="popover_options", use_container_width=True):
|
||||
if st.button("Add File", key="btn_add_file", use_container_width=True):
|
||||
_add_file_dialog("")
|
||||
|
||||
@ -1,12 +1,21 @@
|
||||
"""Centralised session-state initialisation for the Streamlit app.
|
||||
|
||||
All keys used throughout the app are declared here with their default values.
|
||||
Calling init_state() at the top of app.py ensures every key exists before any
|
||||
page tries to read it, preventing KeyError on the first load.
|
||||
"""
|
||||
|
||||
import streamlit as st
|
||||
from backend.managers.chat_manager import ChatManager
|
||||
|
||||
|
||||
def init_state():
|
||||
# Sidebar state initialization
|
||||
# last_selected tracks the previously clicked tree node to detect new clicks
|
||||
if "last_selected" not in st.session_state:
|
||||
st.session_state.last_selected = None
|
||||
|
||||
# Absolute path and workspace-relative path of the currently highlighted folder
|
||||
if "selected_folder" not in st.session_state:
|
||||
st.session_state.selected_folder = None
|
||||
|
||||
@ -14,27 +23,32 @@ def init_state():
|
||||
st.session_state.selected_folder_rel = None
|
||||
|
||||
# Chat manager (persists across reruns)
|
||||
# ChatManager keeps the full conversation history in memory across reruns
|
||||
if "chat_manager" not in st.session_state:
|
||||
st.session_state.chat_manager = ChatManager()
|
||||
|
||||
# Editor state initialization
|
||||
# List of absolute file paths that are currently open as tabs
|
||||
if "open_files" not in st.session_state:
|
||||
"""A list of currently open file paths - absolute paths only. The order determines the tab order in the UI.
|
||||
Format: [ "path/to/file1.py", "path/to/file2.js", ... ]
|
||||
"""
|
||||
st.session_state.open_files = []
|
||||
|
||||
# Dict mapping file path → current editor content (may be unsaved)
|
||||
if "files_content" not in st.session_state:
|
||||
"""A dictionary mapping file paths to their current content in the editor.
|
||||
Format: { "path/to/file.py": "file content as string", ... }
|
||||
"""
|
||||
st.session_state.files_content = {}
|
||||
|
||||
# Absolute path of the file whose tab is currently active
|
||||
if "active_file" not in st.session_state:
|
||||
"""The currently active file in the editor (absolute path in string e.g. "/workspace/path/to/file.py").
|
||||
Should be one of the paths in open_files or None if no file is open."""
|
||||
st.session_state.active_file = None
|
||||
|
||||
|
||||
# Index of the active tab (used by st.tabs)
|
||||
if "active_tab" not in st.session_state:
|
||||
st.session_state.active_tab = 0
|
||||
|
||||
@ -44,26 +58,33 @@ def init_state():
|
||||
if "code_suggestions" not in st.session_state:
|
||||
st.session_state.code_suggestions = []
|
||||
|
||||
# Output dict from the last code run: {stdout, stderr, return_code}
|
||||
if "code_execution_output" not in st.session_state:
|
||||
st.session_state.code_execution_output = ""
|
||||
|
||||
# Chat state initialization
|
||||
# Flat list of {"role": ..., "content": ...} dicts shown as chat bubbles
|
||||
if "chat_history" not in st.session_state:
|
||||
st.session_state.chat_history = []
|
||||
|
||||
# Agent Mode state
|
||||
# Whether the UI is currently in Agent Mode (vs normal chat)
|
||||
if "agent_mode" not in st.session_state:
|
||||
st.session_state.agent_mode = False
|
||||
|
||||
# The live CodingAgent instance while a task is running
|
||||
if "coding_agent" not in st.session_state:
|
||||
st.session_state.coding_agent = None
|
||||
|
||||
# Current status of the agent: "idle" | "waiting_approval" | "done"
|
||||
if "agent_status" not in st.session_state:
|
||||
st.session_state.agent_status = "idle"
|
||||
|
||||
# List of completed steps shown in the collapsible Agent Log
|
||||
if "agent_log" not in st.session_state:
|
||||
st.session_state.agent_log = []
|
||||
|
||||
# The action the agent proposed but has not yet been approved or rejected
|
||||
if "agent_pending_action" not in st.session_state:
|
||||
st.session_state.agent_pending_action = None
|
||||
|
||||
|
||||
@ -17,6 +17,7 @@ pandas>=2.0.0
|
||||
# Testing
|
||||
pytest>=7.0.0
|
||||
pytest-cov>=4.0.0
|
||||
pytest-asyncio>=0.23.0
|
||||
|
||||
# Development & Utilities
|
||||
python-dotenv>=1.0.0
|
||||
|
||||
24
tests/conftest.py
Normal file
24
tests/conftest.py
Normal file
@ -0,0 +1,24 @@
|
||||
"""Shared pytest configuration — runs before any test module is imported.
|
||||
|
||||
Patches MCPToolAdapter at the sys.modules level so that importing
|
||||
backend.agent.coding_agent never tries to start real MCP subprocess servers.
|
||||
"""
|
||||
|
||||
import sys
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
# Build a fake adapter instance whose async methods return immediately.
|
||||
_mock_adapter = MagicMock()
|
||||
_mock_adapter.initialize_all_servers = AsyncMock(return_value=None)
|
||||
_mock_adapter.get_all_tools = MagicMock(return_value=[])
|
||||
_mock_adapter.call_tool = AsyncMock(return_value=MagicMock(isError=False, content=[]))
|
||||
|
||||
# Wrap as a class mock: calling MCPToolAdapter() returns _mock_adapter.
|
||||
_mock_adapter_cls = MagicMock(return_value=_mock_adapter)
|
||||
|
||||
# Inject before any test imports coding_agent so the module-level
|
||||
# asyncio.run(adapter.initialize_all_servers()) uses the mock.
|
||||
sys.modules.setdefault(
|
||||
"backend.agent.mcp_server_adapter",
|
||||
MagicMock(MCPToolAdapter=_mock_adapter_cls),
|
||||
)
|
||||
@ -1,242 +1,157 @@
|
||||
"""Test script for ChatManager - Pytest compatible tests"""
|
||||
"""Tests for ChatManager (backend/managers/chat_manager.py)."""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
import pytest
|
||||
import requests
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
# Add project root to Python path
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
|
||||
from backend.managers.chat_manager import ChatManager
|
||||
|
||||
|
||||
class TestChatManager:
|
||||
"""Test suite for ChatManager functionality."""
|
||||
# ── Helpers ──────────────────────────────────────────────────────────────────
|
||||
|
||||
def _mock_ok(content="AI reply"):
|
||||
"""Return a mocked 200 response with a single assistant choice."""
|
||||
mock = MagicMock()
|
||||
mock.status_code = 200
|
||||
mock.json.return_value = {
|
||||
"choices": [{"message": {"role": "assistant", "content": content}}]
|
||||
}
|
||||
mock.text = ""
|
||||
return mock
|
||||
|
||||
|
||||
# ── History management ────────────────────────────────────────────────────────
|
||||
|
||||
class TestHistory:
|
||||
"""Tests for add_message and clear_history."""
|
||||
|
||||
@pytest.fixture
|
||||
def chat_manager(self):
|
||||
def cm(self):
|
||||
return ChatManager()
|
||||
|
||||
def test_initialization(self, chat_manager):
|
||||
"""Test that ChatManager initializes correctly."""
|
||||
assert chat_manager.api_url is not None
|
||||
assert chat_manager.model is not None
|
||||
assert chat_manager.chat_history == []
|
||||
def test_add_message_appends_correct_entry(self, cm):
|
||||
cm.add_message("user", "Hello")
|
||||
assert cm.chat_history == [{"role": "user", "content": "Hello"}]
|
||||
|
||||
def test_add_message(self, chat_manager):
|
||||
"""Test adding messages to chat history."""
|
||||
chat_manager.add_message("user", "Hello")
|
||||
assert len(chat_manager.chat_history) == 1
|
||||
assert chat_manager.chat_history[0]["role"] == "user"
|
||||
assert chat_manager.chat_history[0]["content"] == "Hello"
|
||||
def test_add_multiple_messages_preserves_order(self, cm):
|
||||
cm.add_message("user", "Hi")
|
||||
cm.add_message("assistant", "Hello!")
|
||||
assert cm.chat_history[0]["role"] == "user"
|
||||
assert cm.chat_history[1]["role"] == "assistant"
|
||||
|
||||
def test_get_history(self, chat_manager):
|
||||
"""Test retrieving chat history."""
|
||||
chat_manager.add_message("user", "Hello")
|
||||
chat_manager.add_message("assistant", "Hi there!")
|
||||
|
||||
history = chat_manager.get_history()
|
||||
assert len(history) == 2
|
||||
assert history[0]["role"] == "user"
|
||||
assert history[1]["role"] == "assistant"
|
||||
|
||||
def test_clear_history(self, chat_manager):
|
||||
"""Test clearing chat history."""
|
||||
chat_manager.add_message("user", "Hello")
|
||||
assert len(chat_manager.chat_history) == 1
|
||||
|
||||
chat_manager.clear_history()
|
||||
assert len(chat_manager.chat_history) == 0
|
||||
|
||||
def test_send_message_integration(self, chat_manager):
|
||||
"""
|
||||
Integration test for sending message to AI.
|
||||
This test actually communicates with the API.
|
||||
"""
|
||||
try:
|
||||
# Send a simple test message
|
||||
response = chat_manager.send_message("Hello, what is 2+2?")
|
||||
|
||||
# Verify response is not empty
|
||||
assert isinstance(response, str)
|
||||
assert len(response) > 0
|
||||
|
||||
# Verify message was added to history
|
||||
assert len(chat_manager.chat_history) == 2 # user + assistant
|
||||
assert chat_manager.chat_history[0]["role"] == "user"
|
||||
assert chat_manager.chat_history[1]["role"] == "assistant"
|
||||
|
||||
print(f"API Test Passed")
|
||||
print(f"Response: {response}")
|
||||
|
||||
except Exception as e:
|
||||
# If API is not reachable, mark as skipped
|
||||
pytest.skip(f"API not reachable: {str(e)}")
|
||||
|
||||
def test_multiple_messages(self, chat_manager):
|
||||
"""Test sending multiple messages in a conversation."""
|
||||
try:
|
||||
# Send first message
|
||||
response1 = chat_manager.send_message("What is your name?")
|
||||
assert len(response1) > 0
|
||||
|
||||
# Send follow-up message
|
||||
response2 = chat_manager.send_message("Tell me more")
|
||||
assert len(response2) > 0
|
||||
|
||||
# Verify full conversation is in history
|
||||
assert len(chat_manager.chat_history) == 4 # 2 user + 2 assistant
|
||||
|
||||
print(f"Conversation Test Passed")
|
||||
print(f"Messages: {len(chat_manager.chat_history)}")
|
||||
|
||||
except Exception as e:
|
||||
pytest.skip(f"API not reachable: {str(e)}")
|
||||
def test_clear_history_empties_list(self, cm):
|
||||
cm.add_message("user", "Hi")
|
||||
cm.clear_history()
|
||||
assert cm.chat_history == []
|
||||
|
||||
|
||||
class TestChatManagerSendMessage:
|
||||
"""Unit tests for send_message using mocked HTTP requests."""
|
||||
# ── send_message (mocked HTTP) ────────────────────────────────────────────────
|
||||
|
||||
class TestSendMessage:
|
||||
"""Tests for send_message: history updates, HTTP payload, error handling, and auth headers."""
|
||||
|
||||
@pytest.fixture
|
||||
def chat_manager(self):
|
||||
def cm(self):
|
||||
return ChatManager()
|
||||
|
||||
def _mock_response(self, content="AI reply", status_code=200):
|
||||
mock = MagicMock()
|
||||
mock.status_code = status_code
|
||||
mock.json.return_value = {
|
||||
"choices": [{"message": {"role": "assistant", "content": content}}]
|
||||
}
|
||||
mock.text = "error text"
|
||||
return mock
|
||||
def test_user_message_added_to_history(self, cm):
|
||||
with patch("requests.post", return_value=_mock_ok()):
|
||||
cm.send_message("Hello")
|
||||
assert cm.chat_history[0] == {"role": "user", "content": "Hello"}
|
||||
|
||||
def test_send_message_adds_user_message_to_history(self, chat_manager):
|
||||
with patch("requests.post", return_value=self._mock_response()):
|
||||
chat_manager.send_message("Hello")
|
||||
assert chat_manager.chat_history[0] == {"role": "user", "content": "Hello"}
|
||||
def test_assistant_reply_added_to_history(self, cm):
|
||||
with patch("requests.post", return_value=_mock_ok("Hi there")):
|
||||
cm.send_message("Hello")
|
||||
assert cm.chat_history[1] == {"role": "assistant", "content": "Hi there"}
|
||||
|
||||
def test_send_message_adds_assistant_response_to_history(self, chat_manager):
|
||||
with patch("requests.post", return_value=self._mock_response("Hi there")):
|
||||
chat_manager.send_message("Hello")
|
||||
assert chat_manager.chat_history[1] == {"role": "assistant", "content": "Hi there"}
|
||||
def test_returns_assistant_content_string(self, cm):
|
||||
with patch("requests.post", return_value=_mock_ok("Answer")):
|
||||
result = cm.send_message("Question")
|
||||
assert result == "Answer"
|
||||
|
||||
def test_send_message_returns_ai_content(self, chat_manager):
|
||||
with patch("requests.post", return_value=self._mock_response("Answer")):
|
||||
response = chat_manager.send_message("Question")
|
||||
assert response == "Answer"
|
||||
def test_full_history_sent_in_request_payload(self, cm):
|
||||
"""All prior messages must be forwarded so the model has conversation context."""
|
||||
cm.add_message("system", "You are helpful.")
|
||||
with patch("requests.post", return_value=_mock_ok()) as mock_post:
|
||||
cm.send_message("Hello")
|
||||
payload = mock_post.call_args.kwargs["json"]
|
||||
assert payload["messages"][0]["role"] == "system"
|
||||
assert payload["messages"][1]["role"] == "user"
|
||||
|
||||
def test_send_message_history_grows_with_each_call(self, chat_manager):
|
||||
with patch("requests.post", return_value=self._mock_response()):
|
||||
chat_manager.send_message("First")
|
||||
chat_manager.send_message("Second")
|
||||
assert len(chat_manager.chat_history) == 4 # 2 user + 2 assistant
|
||||
|
||||
def test_send_message_connection_error_raises(self, chat_manager):
|
||||
import requests
|
||||
def test_connection_error_raises_and_adds_error_to_history(self, cm):
|
||||
with patch("requests.post", side_effect=requests.exceptions.ConnectionError("refused")):
|
||||
with pytest.raises(Exception, match="Connection Error"):
|
||||
chat_manager.send_message("Hello")
|
||||
cm.send_message("Hello")
|
||||
assert any("Error" in msg["content"] for msg in cm.chat_history)
|
||||
|
||||
def test_send_message_api_error_status_raises(self, chat_manager):
|
||||
mock = self._mock_response(status_code=500)
|
||||
def test_api_error_status_raises(self, cm):
|
||||
mock = MagicMock()
|
||||
mock.status_code = 500
|
||||
mock.text = "Internal Server Error"
|
||||
with patch("requests.post", return_value=mock):
|
||||
with pytest.raises(Exception, match="API Error 500"):
|
||||
chat_manager.send_message("Hello")
|
||||
cm.send_message("Hello")
|
||||
|
||||
def test_send_message_empty_choices_raises(self, chat_manager):
|
||||
def test_timeout_raises(self, cm):
|
||||
with patch("requests.post", side_effect=requests.exceptions.Timeout()):
|
||||
with pytest.raises(Exception):
|
||||
cm.send_message("Hello")
|
||||
|
||||
def test_empty_choices_raises(self, cm):
|
||||
mock = MagicMock()
|
||||
mock.status_code = 200
|
||||
mock.json.return_value = {"choices": []}
|
||||
with patch("requests.post", return_value=mock):
|
||||
with pytest.raises(Exception, match="Invalid API response format"):
|
||||
chat_manager.send_message("Hello")
|
||||
cm.send_message("Hello")
|
||||
|
||||
def test_send_message_missing_choices_key_raises(self, chat_manager):
|
||||
def test_api_key_included_in_header_when_set(self, cm):
|
||||
cm.api_key = "test-key-123"
|
||||
with patch("requests.post", return_value=_mock_ok()) as mock_post:
|
||||
cm.send_message("Hello")
|
||||
headers = mock_post.call_args.kwargs["headers"]
|
||||
assert headers.get("Authorization") == "Bearer test-key-123"
|
||||
|
||||
def test_api_key_excluded_from_header_when_empty_sentinel(self, cm):
|
||||
# "EMPTY" is the sentinel string the UI writes when the user leaves the key field blank.
|
||||
cm.api_key = "EMPTY"
|
||||
with patch("requests.post", return_value=_mock_ok()) as mock_post:
|
||||
cm.send_message("Hello")
|
||||
headers = mock_post.call_args.kwargs["headers"]
|
||||
assert "Authorization" not in headers
|
||||
|
||||
def test_json_decode_error_raises(self, cm):
|
||||
import json
|
||||
mock = MagicMock()
|
||||
mock.status_code = 200
|
||||
mock.json.return_value = {}
|
||||
mock.json.side_effect = json.JSONDecodeError("bad json", "", 0)
|
||||
with patch("requests.post", return_value=mock):
|
||||
with pytest.raises(Exception):
|
||||
chat_manager.send_message("Hello")
|
||||
|
||||
def test_send_message_timeout_raises(self, chat_manager):
|
||||
import requests
|
||||
with patch("requests.post", side_effect=requests.exceptions.Timeout()):
|
||||
with pytest.raises(Exception):
|
||||
chat_manager.send_message("Hello")
|
||||
with pytest.raises(Exception, match="JSON Decode Error"):
|
||||
cm.send_message("Hello")
|
||||
|
||||
|
||||
class TestChatManagerGetChatDisplay:
|
||||
"""Tests for get_chat_display()."""
|
||||
# ── get_chat_display ──────────────────────────────────────────────────────────
|
||||
|
||||
class TestGetChatDisplay:
|
||||
"""Tests for get_chat_display: correct shape and ordering."""
|
||||
|
||||
@pytest.fixture
|
||||
def chat_manager(self):
|
||||
def cm(self):
|
||||
return ChatManager()
|
||||
|
||||
def test_empty_history_returns_empty_list(self, chat_manager):
|
||||
assert chat_manager.get_chat_display() == []
|
||||
def test_display_has_role_and_content_keys(self, cm):
|
||||
cm.add_message("user", "Hello")
|
||||
entry = cm.get_chat_display()[0]
|
||||
assert "role" in entry
|
||||
assert "content" in entry
|
||||
|
||||
def test_display_contains_role_and_content_keys(self, chat_manager):
|
||||
chat_manager.add_message("user", "Hello")
|
||||
display = chat_manager.get_chat_display()
|
||||
assert "role" in display[0]
|
||||
assert "content" in display[0]
|
||||
|
||||
def test_display_preserves_message_order(self, chat_manager):
|
||||
chat_manager.add_message("user", "First")
|
||||
chat_manager.add_message("assistant", "Second")
|
||||
display = chat_manager.get_chat_display()
|
||||
def test_display_preserves_message_order(self, cm):
|
||||
cm.add_message("user", "First")
|
||||
cm.add_message("assistant", "Second")
|
||||
display = cm.get_chat_display()
|
||||
assert display[0]["role"] == "user"
|
||||
assert display[1]["role"] == "assistant"
|
||||
|
||||
def test_display_matches_history(self, chat_manager):
|
||||
chat_manager.add_message("user", "Hi")
|
||||
chat_manager.add_message("assistant", "Hello!")
|
||||
assert chat_manager.get_chat_display() == chat_manager.get_history()
|
||||
|
||||
def test_system_message_included_in_display(self, chat_manager):
|
||||
chat_manager.add_message("system", "You are a helper.")
|
||||
display = chat_manager.get_chat_display()
|
||||
assert display[0]["role"] == "system"
|
||||
|
||||
|
||||
def test_chat_manager_demo():
|
||||
"""Demo test - Shows interactive chat (can be run manually)."""
|
||||
print("\n" + "=" * 60)
|
||||
print("ChatManager Demo - Interactive Test")
|
||||
print("=" * 60 + "\n")
|
||||
|
||||
chat_manager = ChatManager()
|
||||
|
||||
print(f"Connected to API: {chat_manager.api_url}")
|
||||
print(f"Model: {chat_manager.model}\n")
|
||||
|
||||
# Demo conversation
|
||||
test_messages = ["Hello! What can you do?", "Tell me a joke", "What is Python?"]
|
||||
|
||||
print("Starting conversation...\n")
|
||||
|
||||
for message in test_messages:
|
||||
print(f"User: {message}")
|
||||
|
||||
try:
|
||||
response = chat_manager.send_message(message)
|
||||
print(f"Assistant: {response}\n")
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error: {str(e)}\n")
|
||||
pytest.skip(f"API not reachable: {str(e)}")
|
||||
|
||||
# Display full chat history
|
||||
print("=" * 60)
|
||||
print("Chat History:")
|
||||
print("=" * 60)
|
||||
|
||||
for msg in chat_manager.get_history():
|
||||
print(f"{msg['role'].upper()}: {msg['content']}\n")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Run with: pytest tests/test_chat_manager.py -v -s
|
||||
pytest.main([__file__, "-v", "-s"])
|
||||
|
||||
@ -3,13 +3,16 @@ Tests for CodingAgent (backend/agent/coding_agent.py)
|
||||
|
||||
Structure:
|
||||
TestHelpers – truncate_result, trim_messages, _strip_code_fences
|
||||
TestDispatcher – dispatch_tool routing
|
||||
TestTools – tool functions (read_file, write_file, …) using tmp workspace
|
||||
TestCodingAgentInit – __init__ and start_task
|
||||
TestProposeNextAction – propose_next_action with mocked API
|
||||
TestApprove – approve with mocked API + real tool execution
|
||||
TestReject – reject injects feedback correctly
|
||||
TestFullLoop – integration: real API, skipped if unreachable
|
||||
|
||||
Note: TestTools (write_file, read_file, etc.) and TestDispatcher were removed
|
||||
because those tool functions are now MCP server tools, not standalone functions
|
||||
in coding_agent.py. They will be tested via test_mcp_server_*.py once the MCP
|
||||
servers are finalised.
|
||||
"""
|
||||
|
||||
import json
|
||||
@ -18,24 +21,18 @@ from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
|
||||
from backend.agent.coding_agent import (
|
||||
MAX_HISTORY_CHARS,
|
||||
MAX_ITERATIONS,
|
||||
MAX_RESULT_LENGTH,
|
||||
CodingAgent,
|
||||
_strip_code_fences,
|
||||
dispatch_tool,
|
||||
done,
|
||||
grep_search,
|
||||
list_files,
|
||||
read_file,
|
||||
run_python,
|
||||
truncate_result,
|
||||
trim_messages,
|
||||
validate_python,
|
||||
write_file,
|
||||
)
|
||||
|
||||
|
||||
@ -55,6 +52,8 @@ def _make_api_response(content: str, status_code: int = 200):
|
||||
|
||||
|
||||
def _agent_action_json(tool: str, thought: str = "thinking...", **arguments) -> str:
|
||||
"""Return a JSON string in the exact format the agent expects from the LLM:
|
||||
{"thought": "...", "tool": "<name>", "arguments": {...}}."""
|
||||
return json.dumps({"thought": thought, "tool": tool, "arguments": arguments})
|
||||
|
||||
|
||||
@ -63,6 +62,7 @@ def _agent_action_json(tool: str, thought: str = "thinking...", **arguments) ->
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
class TestTruncateResult:
|
||||
"""Tests for truncate_result(): ensures long tool outputs are capped before entering the message history."""
|
||||
|
||||
def test_short_result_unchanged(self):
|
||||
assert truncate_result("hello") == "hello"
|
||||
@ -73,10 +73,6 @@ class TestTruncateResult:
|
||||
assert len(result) < len(long)
|
||||
assert "TRUNCATED" in result
|
||||
|
||||
def test_exact_limit_not_truncated(self):
|
||||
text = "a" * MAX_RESULT_LENGTH
|
||||
assert truncate_result(text) == text
|
||||
|
||||
def test_truncated_keeps_start_and_end(self):
|
||||
text = "START" + "x" * MAX_RESULT_LENGTH + "END"
|
||||
result = truncate_result(text)
|
||||
@ -85,8 +81,11 @@ class TestTruncateResult:
|
||||
|
||||
|
||||
class TestTrimMessages:
|
||||
"""Tests for trim_messages(): keeps system + original task, drops old turns when history grows too large."""
|
||||
|
||||
def _make_messages(self, n_extra: int, chars_each: int = 100) -> list:
|
||||
"""Build a message list with a fixed system + user header followed by
|
||||
n_extra assistant/user pairs, each pair consuming 2*chars_each characters."""
|
||||
msgs = [
|
||||
{"role": "system", "content": "sys"},
|
||||
{"role": "user", "content": "original task"},
|
||||
@ -105,8 +104,6 @@ class TestTrimMessages:
|
||||
original_total = sum(len(m["content"]) for m in msgs)
|
||||
trimmed = trim_messages(msgs)
|
||||
trimmed_total = sum(len(m["content"]) for m in trimmed)
|
||||
# Must be significantly shorter than the original
|
||||
# (slightly above MAX_HISTORY_CHARS is acceptable due to the injected reminder message)
|
||||
assert trimmed_total < original_total
|
||||
assert len(trimmed) < len(msgs)
|
||||
|
||||
@ -121,6 +118,8 @@ class TestTrimMessages:
|
||||
assert trimmed[1]["content"] == "original task"
|
||||
|
||||
def test_reminder_injected_when_trimmed(self):
|
||||
# trim_messages inserts a "system_note" message so the agent knows that
|
||||
# earlier turns were dropped and it should not reference missing context.
|
||||
msgs = self._make_messages(n_extra=500, chars_each=200)
|
||||
trimmed = trim_messages(msgs)
|
||||
contents = [m["content"] for m in trimmed]
|
||||
@ -128,6 +127,7 @@ class TestTrimMessages:
|
||||
|
||||
|
||||
class TestStripCodeFences:
|
||||
"""Tests for _strip_code_fences(): the LLM sometimes wraps its JSON in markdown fences — this strips them."""
|
||||
|
||||
def test_plain_text_unchanged(self):
|
||||
assert _strip_code_fences("hello") == "hello"
|
||||
@ -140,171 +140,13 @@ class TestStripCodeFences:
|
||||
text = "```\nhello\n```"
|
||||
assert _strip_code_fences(text) == "hello"
|
||||
|
||||
def test_strips_whitespace(self):
|
||||
assert _strip_code_fences(" hello ") == "hello"
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
# TestDispatcher
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
class TestDispatcher:
|
||||
|
||||
def test_unknown_tool_returns_error(self):
|
||||
result = dispatch_tool("nonexistent_tool", {})
|
||||
assert "ERROR" in result
|
||||
assert "nonexistent_tool" in result
|
||||
|
||||
def test_done_tool_dispatched(self):
|
||||
result = dispatch_tool("done", {"summary": "finished"})
|
||||
assert "finished" in result
|
||||
|
||||
def test_wrong_arguments_returns_error(self):
|
||||
result = dispatch_tool("read_file", {"wrong_param": "x"})
|
||||
assert "ERROR" in result
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
# TestTools (patched WORKSPACE → tmp_path)
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
class TestWriteFile:
|
||||
|
||||
def test_write_creates_file(self, tmp_path):
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
result = write_file("hello.py", "print('hi')")
|
||||
assert result.startswith("OK:")
|
||||
assert (tmp_path / "hello.py").read_text() == "print('hi')"
|
||||
|
||||
def test_write_outside_workspace_blocked(self, tmp_path):
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
result = write_file("../evil.py", "bad")
|
||||
assert "ERROR" in result
|
||||
|
||||
def test_write_unsupported_extension_blocked(self, tmp_path):
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
result = write_file("script.sh", "echo hi")
|
||||
assert "ERROR" in result
|
||||
|
||||
|
||||
class TestReadFile:
|
||||
|
||||
def test_read_existing_file(self, tmp_path):
|
||||
(tmp_path / "data.txt").write_text("hello world")
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
result = read_file("data.txt")
|
||||
assert result == "hello world"
|
||||
|
||||
def test_read_nonexistent_file(self, tmp_path):
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
result = read_file("ghost.py")
|
||||
assert "ERROR" in result
|
||||
|
||||
def test_read_outside_workspace_blocked(self, tmp_path):
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
result = read_file("../secret.py")
|
||||
assert "ERROR" in result
|
||||
|
||||
def test_read_unsupported_extension(self, tmp_path):
|
||||
(tmp_path / "data.csv").write_text("a,b")
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
result = read_file("data.csv")
|
||||
assert "ERROR" in result
|
||||
|
||||
|
||||
class TestListFiles:
|
||||
|
||||
def test_empty_workspace(self, tmp_path):
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
result = list_files()
|
||||
assert "No files" in result
|
||||
|
||||
def test_lists_existing_files(self, tmp_path):
|
||||
(tmp_path / "a.py").touch()
|
||||
(tmp_path / "b.txt").touch()
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
result = list_files()
|
||||
assert "a.py" in result
|
||||
assert "b.txt" in result
|
||||
|
||||
def test_glob_filter(self, tmp_path):
|
||||
(tmp_path / "a.py").touch()
|
||||
(tmp_path / "b.txt").touch()
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
result = list_files("*.py")
|
||||
assert "a.py" in result
|
||||
assert "b.txt" not in result
|
||||
|
||||
|
||||
class TestGrepSearch:
|
||||
|
||||
def test_finds_pattern(self, tmp_path):
|
||||
(tmp_path / "code.py").write_text("def hello():\n pass\n")
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
result = grep_search("def hello")
|
||||
assert "code.py" in result
|
||||
assert "def hello" in result
|
||||
|
||||
def test_no_match_returns_message(self, tmp_path):
|
||||
(tmp_path / "code.py").write_text("x = 1\n")
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
result = grep_search("nonexistent_pattern")
|
||||
assert "No matches" in result
|
||||
|
||||
def test_returns_line_number(self, tmp_path):
|
||||
(tmp_path / "code.py").write_text("x = 1\ndef foo():\n pass\n")
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
result = grep_search("def foo")
|
||||
assert ":2:" in result
|
||||
|
||||
|
||||
class TestValidatePython:
|
||||
|
||||
def test_valid_syntax(self, tmp_path):
|
||||
(tmp_path / "good.py").write_text("def f(x):\n return x * 2\n")
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
result = validate_python("good.py")
|
||||
assert result == "OK: syntax is valid."
|
||||
|
||||
def test_invalid_syntax(self, tmp_path):
|
||||
(tmp_path / "bad.py").write_text("def f(x)\n return x\n")
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
result = validate_python("bad.py")
|
||||
assert "SYNTAX ERROR" in result
|
||||
|
||||
def test_file_not_found(self, tmp_path):
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
result = validate_python("ghost.py")
|
||||
assert "ERROR" in result
|
||||
|
||||
|
||||
class TestRunPython:
|
||||
|
||||
def test_successful_execution(self, tmp_path):
|
||||
(tmp_path / "hello.py").write_text("print('hello world')\n")
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
result = run_python("hello.py")
|
||||
assert "hello world" in result
|
||||
assert "Exit code: 0" in result
|
||||
|
||||
def test_runtime_error_captured(self, tmp_path):
|
||||
(tmp_path / "bad.py").write_text("raise ValueError('oops')\n")
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
result = run_python("bad.py")
|
||||
assert "ValueError" in result
|
||||
assert "Exit code: 1" in result
|
||||
|
||||
def test_file_not_found(self, tmp_path):
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
result = run_python("ghost.py")
|
||||
assert "ERROR" in result
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
# TestCodingAgentInit
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
class TestCodingAgentInit:
|
||||
"""Tests for CodingAgent.__init__ and start_task(): state is clean before and after task setup."""
|
||||
|
||||
def test_initial_state_is_clean(self):
|
||||
agent = CodingAgent()
|
||||
@ -313,11 +155,6 @@ class TestCodingAgentInit:
|
||||
assert agent.is_done is False
|
||||
assert agent.iteration == 0
|
||||
|
||||
def test_api_url_is_set(self):
|
||||
agent = CodingAgent()
|
||||
assert agent.api_url.startswith("http://")
|
||||
assert "/v1/chat/completions" in agent.api_url
|
||||
|
||||
def test_start_task_sets_messages(self):
|
||||
agent = CodingAgent()
|
||||
agent.start_task("Write fibonacci.py")
|
||||
@ -348,6 +185,7 @@ class TestCodingAgentInit:
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
class TestProposeNextAction:
|
||||
"""Tests for propose_next_action(): API is mocked so no real HTTP calls are made."""
|
||||
|
||||
@pytest.fixture
|
||||
def agent(self):
|
||||
@ -359,55 +197,72 @@ class TestProposeNextAction:
|
||||
payload = _agent_action_json(tool, thought, **args)
|
||||
agent._call_api = MagicMock(return_value=payload)
|
||||
|
||||
def test_returns_dict_with_required_keys(self, agent):
|
||||
@pytest.mark.asyncio
|
||||
async def test_returns_dict_with_required_keys(self, agent):
|
||||
self._mock_api(agent)
|
||||
action = agent.propose_next_action()
|
||||
action = await agent.propose_next_action()
|
||||
assert "thought" in action
|
||||
assert "tool" in action
|
||||
assert "arguments" in action
|
||||
|
||||
def test_increments_iteration(self, agent):
|
||||
@pytest.mark.asyncio
|
||||
async def test_increments_iteration(self, agent):
|
||||
self._mock_api(agent)
|
||||
agent.propose_next_action()
|
||||
await agent.propose_next_action()
|
||||
assert agent.iteration == 1
|
||||
|
||||
def test_stores_pending_action(self, agent):
|
||||
@pytest.mark.asyncio
|
||||
async def test_stores_pending_action(self, agent):
|
||||
self._mock_api(agent)
|
||||
agent.propose_next_action()
|
||||
await agent.propose_next_action()
|
||||
assert agent.pending_action is not None
|
||||
|
||||
def test_returns_correct_tool(self, agent):
|
||||
@pytest.mark.asyncio
|
||||
async def test_returns_correct_tool(self, agent):
|
||||
self._mock_api(agent, tool="list_files")
|
||||
action = agent.propose_next_action()
|
||||
action = await agent.propose_next_action()
|
||||
assert action["tool"] == "list_files"
|
||||
|
||||
def test_handles_json_parse_error_gracefully(self, agent):
|
||||
@pytest.mark.asyncio
|
||||
async def test_handles_json_parse_error_gracefully(self, agent):
|
||||
agent._call_api = MagicMock(return_value="this is not json {{")
|
||||
action = agent.propose_next_action()
|
||||
action = await agent.propose_next_action()
|
||||
assert action["tool"] == "done"
|
||||
|
||||
def test_handles_api_exception_gracefully(self, agent):
|
||||
@pytest.mark.asyncio
|
||||
async def test_handles_api_exception_gracefully(self, agent):
|
||||
agent._call_api = MagicMock(side_effect=Exception("connection refused"))
|
||||
action = agent.propose_next_action()
|
||||
action = await agent.propose_next_action()
|
||||
assert action["tool"] == "done"
|
||||
|
||||
def test_strips_code_fences_from_response(self, agent):
|
||||
@pytest.mark.asyncio
|
||||
async def test_strips_code_fences_from_response(self, agent):
|
||||
payload = "```json\n" + _agent_action_json("list_files", "thinking") + "\n```"
|
||||
agent._call_api = MagicMock(return_value=payload)
|
||||
action = agent.propose_next_action()
|
||||
action = await agent.propose_next_action()
|
||||
assert action["tool"] == "list_files"
|
||||
|
||||
def test_already_done_returns_done_action(self, agent):
|
||||
@pytest.mark.asyncio
|
||||
async def test_already_done_returns_done_action(self, agent):
|
||||
agent.is_done = True
|
||||
action = agent.propose_next_action()
|
||||
action = await agent.propose_next_action()
|
||||
assert action["tool"] == "done"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_max_iterations_returns_done_without_api_call(self, agent):
|
||||
agent.iteration = MAX_ITERATIONS
|
||||
agent._call_api = MagicMock(side_effect=AssertionError("API must not be called"))
|
||||
action = await agent.propose_next_action()
|
||||
assert action["tool"] == "done"
|
||||
agent._call_api.assert_not_called()
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
# TestApprove (mocked API + real tool execution via tmp_path)
|
||||
# TestApprove (mocked API + mocked dispatch_tool)
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
class TestApprove:
|
||||
"""Tests for approve(): dispatch_tool is mocked so no filesystem or subprocess side-effects occur."""
|
||||
|
||||
@pytest.fixture
|
||||
def agent(self):
|
||||
@ -416,56 +271,68 @@ class TestApprove:
|
||||
return a
|
||||
|
||||
def _set_pending(self, agent, tool: str, **arguments):
|
||||
"""Inject a pending_action into the agent as if propose_next_action() had just run.
|
||||
'raw' holds the original JSON string; 'action' holds the parsed dict."""
|
||||
raw = _agent_action_json(tool, "thought", **arguments)
|
||||
agent.pending_action = {
|
||||
"raw": raw,
|
||||
"action": {"thought": "thought", "tool": tool, "arguments": arguments},
|
||||
}
|
||||
|
||||
def test_approve_without_pending_raises(self, agent):
|
||||
@pytest.mark.asyncio
|
||||
async def test_approve_without_pending_raises(self, agent):
|
||||
with pytest.raises(Exception):
|
||||
agent.approve()
|
||||
await agent.approve()
|
||||
|
||||
def test_approve_done_sets_is_done(self, agent):
|
||||
@pytest.mark.asyncio
|
||||
async def test_approve_done_sets_is_done(self, agent):
|
||||
self._set_pending(agent, "done", summary="all done")
|
||||
result = agent.approve()
|
||||
result = await agent.approve()
|
||||
assert result["is_done"] is True
|
||||
assert agent.is_done is True
|
||||
|
||||
def test_approve_done_returns_summary(self, agent):
|
||||
@pytest.mark.asyncio
|
||||
async def test_approve_done_returns_summary(self, agent):
|
||||
self._set_pending(agent, "done", summary="finished successfully")
|
||||
result = agent.approve()
|
||||
result = await agent.approve()
|
||||
assert "finished successfully" in result["result"]
|
||||
|
||||
def test_approve_clears_pending_action(self, agent):
|
||||
@pytest.mark.asyncio
|
||||
async def test_approve_clears_pending_action(self, agent):
|
||||
self._set_pending(agent, "done", summary="x")
|
||||
agent.approve()
|
||||
await agent.approve()
|
||||
assert agent.pending_action is None
|
||||
|
||||
def test_approve_appends_assistant_message(self, agent):
|
||||
@pytest.mark.asyncio
|
||||
async def test_approve_appends_assistant_message(self, agent):
|
||||
self._set_pending(agent, "done", summary="x")
|
||||
before = len(agent.messages)
|
||||
agent.approve()
|
||||
await agent.approve()
|
||||
assert len(agent.messages) > before
|
||||
|
||||
def test_approve_tool_result_appended_to_messages(self, agent, tmp_path):
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
@pytest.mark.asyncio
|
||||
async def test_approve_tool_result_appended_to_messages(self, agent):
|
||||
with patch("backend.agent.coding_agent.dispatch_tool", return_value="file list"):
|
||||
self._set_pending(agent, "list_files")
|
||||
agent.approve()
|
||||
await agent.approve()
|
||||
tool_results = [m for m in agent.messages if "tool_result" in m["content"]]
|
||||
assert len(tool_results) == 1
|
||||
|
||||
def test_approve_error_result_adds_replan_tag(self, agent, tmp_path):
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
@pytest.mark.asyncio
|
||||
async def test_approve_error_result_adds_replan_tag(self, agent):
|
||||
# When a tool returns an error, approve() adds a "replan" tag to the message
|
||||
# so the LLM knows the last action failed and must choose a different approach.
|
||||
with patch("backend.agent.coding_agent.dispatch_tool", return_value="ERROR: file not found"):
|
||||
self._set_pending(agent, "read_file", path="nonexistent.py")
|
||||
agent.approve()
|
||||
await agent.approve()
|
||||
last_msg = agent.messages[-1]["content"]
|
||||
assert "replan" in last_msg
|
||||
|
||||
def test_approve_returns_tool_name_in_result(self, agent, tmp_path):
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
@pytest.mark.asyncio
|
||||
async def test_approve_returns_tool_name_in_result(self, agent):
|
||||
with patch("backend.agent.coding_agent.dispatch_tool", return_value="(empty)"):
|
||||
self._set_pending(agent, "list_files")
|
||||
result = agent.approve()
|
||||
result = await agent.approve()
|
||||
assert result["tool"] == "list_files"
|
||||
assert result["is_done"] is False
|
||||
|
||||
@ -475,6 +342,7 @@ class TestApprove:
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
class TestReject:
|
||||
"""Tests for reject(): user feedback is injected into the history and the pending action is discarded."""
|
||||
|
||||
@pytest.fixture
|
||||
def agent(self):
|
||||
@ -508,66 +376,12 @@ class TestReject:
|
||||
|
||||
def test_reject_without_pending_does_not_crash(self, agent):
|
||||
agent.pending_action = None
|
||||
agent.reject("no pending action") # should not raise
|
||||
agent.reject("no pending action")
|
||||
|
||||
def test_reject_does_not_execute_tool(self, agent, tmp_path):
|
||||
self._set_pending(agent, "write_file")
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
agent.reject("Do not write anything")
|
||||
assert not list(tmp_path.glob("*")) # no files created
|
||||
assert not list(tmp_path.glob("*"))
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
# TestFullLoop (integration – real API, skipped if unreachable)
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
class TestFullLoop:
|
||||
"""End-to-end test: agent runs a real task against the live API.
|
||||
Skipped automatically if the API is not reachable.
|
||||
"""
|
||||
|
||||
MAX_STEPS = 15 # safety limit for the test loop
|
||||
|
||||
def _run_until_done(self, agent) -> list:
|
||||
"""Drive the agent loop until done or MAX_STEPS reached."""
|
||||
steps = []
|
||||
for _ in range(self.MAX_STEPS):
|
||||
action = agent.propose_next_action()
|
||||
result = agent.approve()
|
||||
steps.append(result)
|
||||
if result["is_done"]:
|
||||
break
|
||||
return steps
|
||||
|
||||
def test_agent_completes_hello_world_task(self, tmp_path):
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
agent = CodingAgent()
|
||||
try:
|
||||
agent.start_task(
|
||||
"Write a Python file called hello.py that prints 'Hello World'. "
|
||||
"Validate it and run it."
|
||||
)
|
||||
steps = self._run_until_done(agent)
|
||||
except Exception as e:
|
||||
pytest.skip(f"API not reachable: {e}")
|
||||
|
||||
assert agent.is_done, "Agent did not reach done state"
|
||||
tools_used = [s["tool"] for s in steps]
|
||||
assert "write_file" in tools_used
|
||||
assert "done" in tools_used
|
||||
|
||||
def test_agent_creates_file_on_disk(self, tmp_path):
|
||||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||||
agent = CodingAgent()
|
||||
try:
|
||||
agent.start_task("Write a file called output.txt containing the text 'test passed'.")
|
||||
self._run_until_done(agent)
|
||||
except Exception as e:
|
||||
pytest.skip(f"API not reachable: {e}")
|
||||
|
||||
py_files = list(tmp_path.glob("*.txt")) + list(tmp_path.glob("*.py"))
|
||||
assert len(py_files) > 0, "Agent did not create any file"
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v", "-s"])
|
||||
|
||||
110
tests/test_debug_logger.py
Normal file
110
tests/test_debug_logger.py
Normal file
@ -0,0 +1,110 @@
|
||||
"""Tests for DebugLogger (backend/managers/debug_logger.py)."""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
|
||||
from backend.managers.debug_logger import DebugLogger
|
||||
|
||||
|
||||
# ── Fixtures ──────────────────────────────────────────────────────────────────
|
||||
|
||||
@pytest.fixture
|
||||
def logger():
|
||||
"""Return a fresh DebugLogger for each test."""
|
||||
return DebugLogger()
|
||||
|
||||
|
||||
# ── log() ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
class TestLog:
|
||||
"""Tests for log(): each call appends an INFO-level entry with message."""
|
||||
|
||||
def test_log_appends_entry(self, logger):
|
||||
logger.log("started")
|
||||
assert len(logger.logs) == 1
|
||||
|
||||
def test_log_sets_level_info(self, logger):
|
||||
logger.log("started")
|
||||
assert logger.logs[0]["level"] == "INFO"
|
||||
|
||||
def test_log_stores_message(self, logger):
|
||||
logger.log("executing file.py")
|
||||
assert logger.logs[0]["message"] == "executing file.py"
|
||||
|
||||
|
||||
# ── log_error() ───────────────────────────────────────────────────────────────
|
||||
|
||||
class TestLogError:
|
||||
"""Tests for log_error(): level is ERROR, not INFO."""
|
||||
|
||||
def test_log_error_sets_level_error(self, logger):
|
||||
logger.log_error("something broke")
|
||||
assert logger.logs[0]["level"] == "ERROR"
|
||||
|
||||
def test_log_and_log_error_are_distinct_levels(self, logger):
|
||||
logger.log("info message")
|
||||
logger.log_error("error message")
|
||||
assert logger.logs[0]["level"] == "INFO"
|
||||
assert logger.logs[1]["level"] == "ERROR"
|
||||
|
||||
|
||||
# ── get_logs() ────────────────────────────────────────────────────────────────
|
||||
|
||||
class TestGetLogs:
|
||||
"""Tests for get_logs()."""
|
||||
|
||||
def test_get_logs_returns_all_entries(self, logger):
|
||||
logger.log("first")
|
||||
logger.log_error("second")
|
||||
assert len(logger.get_logs()) == 2
|
||||
|
||||
|
||||
# ── clear() ───────────────────────────────────────────────────────────────────
|
||||
|
||||
class TestClear:
|
||||
"""Tests for clear()."""
|
||||
|
||||
def test_clear_removes_all_entries(self, logger):
|
||||
logger.log("first")
|
||||
logger.log_error("second")
|
||||
logger.clear()
|
||||
assert logger.logs == []
|
||||
|
||||
|
||||
# ── format_debug_output() ─────────────────────────────────────────────────────
|
||||
|
||||
class TestFormatDebugOutput:
|
||||
"""Tests for format_debug_output(): renders rc, stdout, stderr, and log entries."""
|
||||
|
||||
def test_success_exit_code_shows_success(self, logger):
|
||||
result = logger.format_debug_output({"rc": 0, "stdout": "", "stderr": ""})
|
||||
assert "[SUCCESS]" in result
|
||||
|
||||
def test_nonzero_exit_code_shows_failed(self, logger):
|
||||
result = logger.format_debug_output({"rc": 1, "stdout": "", "stderr": ""})
|
||||
assert "[FAILED]" in result
|
||||
|
||||
def test_stdout_included_when_present(self, logger):
|
||||
result = logger.format_debug_output({"rc": 0, "stdout": "Hello", "stderr": ""})
|
||||
assert "Hello" in result
|
||||
assert "stdout" in result
|
||||
|
||||
def test_stderr_included_when_present(self, logger):
|
||||
result = logger.format_debug_output({"rc": 1, "stdout": "", "stderr": "NameError"})
|
||||
assert "NameError" in result
|
||||
assert "stderr" in result
|
||||
|
||||
def test_log_entries_appended_to_output(self, logger):
|
||||
logger.log("Executing file.py...")
|
||||
logger.log_error("exit code 1")
|
||||
result = logger.format_debug_output({"rc": 1, "stdout": "", "stderr": ""})
|
||||
assert "Executing file.py..." in result
|
||||
assert "exit code 1" in result
|
||||
|
||||
def test_missing_keys_do_not_raise(self, logger):
|
||||
# Defensive: format_debug_output uses .get() so missing keys are safe
|
||||
result = logger.format_debug_output({})
|
||||
assert isinstance(result, str)
|
||||
@ -0,0 +1,220 @@
|
||||
"""Tests for FileManager (backend/managers/file_manager.py)."""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
import pytest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
|
||||
from backend.managers.file_manager import FileManager
|
||||
|
||||
|
||||
# ── Fixtures ──────────────────────────────────────────────────────────────────
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def mock_streamlit():
|
||||
"""Suppress all st.error / st.warning calls — they require a running Streamlit app."""
|
||||
with patch("backend.managers.file_manager.st"):
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fm(tmp_path):
|
||||
"""Return a FileManager whose workspace is an isolated pytest temp directory."""
|
||||
return FileManager(base_path=tmp_path)
|
||||
|
||||
|
||||
# ── create_folder ─────────────────────────────────────────────────────────────
|
||||
|
||||
class TestCreateFolder:
|
||||
"""Tests for create_folder(): name validation, path-traversal protection, and nested creation."""
|
||||
|
||||
def test_creates_folder_successfully(self, fm, tmp_path):
|
||||
result = fm.create_folder("", "myfolder")
|
||||
assert result is True
|
||||
assert (tmp_path / "myfolder").is_dir()
|
||||
|
||||
def test_empty_name_returns_false(self, fm):
|
||||
assert fm.create_folder("", "") is False
|
||||
|
||||
def test_slash_in_name_returns_false(self, fm):
|
||||
assert fm.create_folder("", "a/b") is False
|
||||
|
||||
def test_backslash_in_name_returns_false(self, fm):
|
||||
assert fm.create_folder("", "a\\b") is False
|
||||
|
||||
def test_duplicate_folder_returns_false(self, fm, tmp_path):
|
||||
(tmp_path / "existing").mkdir()
|
||||
assert fm.create_folder("", "existing") is False
|
||||
|
||||
def test_path_traversal_returns_false(self, fm):
|
||||
assert fm.create_folder("../../", "evil") is False
|
||||
|
||||
def test_nested_folder_created_inside_base(self, fm, tmp_path):
|
||||
(tmp_path / "sub").mkdir()
|
||||
result = fm.create_folder("sub", "child")
|
||||
assert result is True
|
||||
assert (tmp_path / "sub" / "child").is_dir()
|
||||
|
||||
|
||||
# ── create_file ───────────────────────────────────────────────────────────────
|
||||
|
||||
class TestCreateFile:
|
||||
"""Tests for create_file(): name validation, auto .txt extension, and path-traversal protection."""
|
||||
|
||||
def test_creates_file_successfully(self, fm, tmp_path):
|
||||
result = fm.create_file("", "test.py")
|
||||
assert result is True
|
||||
assert (tmp_path / "test.py").is_file()
|
||||
|
||||
def test_empty_name_returns_false(self, fm):
|
||||
assert fm.create_file("", "") is False
|
||||
|
||||
def test_whitespace_only_name_returns_false(self, fm):
|
||||
assert fm.create_file("", " ") is False
|
||||
|
||||
def test_no_extension_defaults_to_txt(self, fm, tmp_path):
|
||||
fm.create_file("", "notes")
|
||||
assert (tmp_path / "notes.txt").is_file()
|
||||
|
||||
def test_duplicate_file_returns_false(self, fm, tmp_path):
|
||||
(tmp_path / "existing.py").touch()
|
||||
assert fm.create_file("", "existing.py") is False
|
||||
|
||||
def test_path_traversal_returns_false(self, fm):
|
||||
assert fm.create_file("../../", "evil.py") is False
|
||||
|
||||
|
||||
# ── read_file ─────────────────────────────────────────────────────────────────
|
||||
|
||||
class TestReadFile:
|
||||
"""Tests for read_file(): accepts an absolute Path, validates workspace boundary, returns content or ""."""
|
||||
|
||||
def test_reads_file_content(self, fm, tmp_path):
|
||||
f = tmp_path / "hello.py"
|
||||
f.write_text("print('hello')")
|
||||
assert fm.read_file(f) == "print('hello')"
|
||||
|
||||
def test_nonexistent_file_returns_empty_string(self, fm, tmp_path):
|
||||
assert fm.read_file(tmp_path / "ghost.py") == ""
|
||||
|
||||
def test_file_outside_workspace_returns_empty_string(self, fm, tmp_path):
|
||||
outside = tmp_path.parent / "outside.py"
|
||||
outside.write_text("secret")
|
||||
assert fm.read_file(outside) == ""
|
||||
|
||||
|
||||
# ── save_file ─────────────────────────────────────────────────────────────────
|
||||
|
||||
class TestSaveFile:
|
||||
"""Tests for save_file(): accepts an absolute path string, overwrites content, and blocks path traversal."""
|
||||
|
||||
def test_saves_content_to_file(self, fm, tmp_path):
|
||||
f = tmp_path / "output.py"
|
||||
f.touch()
|
||||
result = fm.save_file(str(f), "x = 1")
|
||||
assert result is True
|
||||
assert f.read_text() == "x = 1"
|
||||
|
||||
def test_overwrites_existing_content(self, fm, tmp_path):
|
||||
f = tmp_path / "script.py"
|
||||
f.write_text("old content")
|
||||
fm.save_file(str(f), "new content")
|
||||
assert f.read_text() == "new content"
|
||||
|
||||
def test_path_traversal_returns_false(self, fm, tmp_path):
|
||||
outside = str(tmp_path.parent / "evil.py")
|
||||
assert fm.save_file(outside, "bad") is False
|
||||
|
||||
|
||||
# ── rename_file ───────────────────────────────────────────────────────────────
|
||||
|
||||
class TestRenameFile:
|
||||
"""Tests for rename_file(): renames by stem only — the original extension is always preserved."""
|
||||
|
||||
def test_renames_file_successfully(self, fm, tmp_path):
|
||||
(tmp_path / "old.py").touch()
|
||||
result = fm.rename_file("old.py", "new")
|
||||
assert result is True
|
||||
assert (tmp_path / "new.py").exists()
|
||||
assert not (tmp_path / "old.py").exists()
|
||||
|
||||
def test_preserves_original_extension(self, fm, tmp_path):
|
||||
# Even if the caller passes a different extension (.txt), rename_file
|
||||
# silently replaces it with the original (.py) to prevent accidental type changes.
|
||||
(tmp_path / "script.py").touch()
|
||||
fm.rename_file("script.py", "renamed.txt")
|
||||
assert (tmp_path / "renamed.py").exists()
|
||||
|
||||
def test_empty_new_name_returns_false(self, fm, tmp_path):
|
||||
(tmp_path / "file.py").touch()
|
||||
assert fm.rename_file("file.py", "") is False
|
||||
|
||||
def test_nonexistent_file_returns_false(self, fm):
|
||||
assert fm.rename_file("ghost.py", "new_name") is False
|
||||
|
||||
def test_path_traversal_returns_false(self, fm):
|
||||
assert fm.rename_file("../../evil.py", "new_name") is False
|
||||
|
||||
|
||||
# ── delete_file ───────────────────────────────────────────────────────────────
|
||||
|
||||
class TestDeleteFile:
|
||||
"""Tests for delete_file(): accepts a relative path, validates workspace boundary, removes the file."""
|
||||
|
||||
def test_deletes_file_successfully(self, fm, tmp_path):
|
||||
f = tmp_path / "todelete.py"
|
||||
f.touch()
|
||||
result = fm.delete_file("todelete.py")
|
||||
assert result is True
|
||||
assert not f.exists()
|
||||
|
||||
def test_nonexistent_file_returns_false(self, fm):
|
||||
assert fm.delete_file("ghost.py") is False
|
||||
|
||||
def test_path_traversal_returns_false(self, fm):
|
||||
assert fm.delete_file("../../evil.py") is False
|
||||
|
||||
|
||||
# ── delete_folder ─────────────────────────────────────────────────────────────
|
||||
|
||||
class TestDeleteFolder:
|
||||
"""Tests for delete_folder(): recursively removes a folder and all its contents."""
|
||||
|
||||
def test_deletes_folder_and_contents(self, fm, tmp_path):
|
||||
sub = tmp_path / "todelete"
|
||||
sub.mkdir()
|
||||
(sub / "file.py").touch()
|
||||
result = fm.delete_folder("todelete")
|
||||
assert result is True
|
||||
assert not sub.exists()
|
||||
|
||||
def test_path_traversal_returns_false(self, fm):
|
||||
assert fm.delete_folder("../../") is False
|
||||
|
||||
|
||||
# ── get_file_tree ─────────────────────────────────────────────────────────────
|
||||
|
||||
class TestGetFileTree:
|
||||
"""Tests for get_file_tree(): returns a nested dict where files map to None and dirs map to dicts."""
|
||||
|
||||
def test_empty_workspace_returns_empty_dict(self, fm):
|
||||
assert fm.get_file_tree() == {}
|
||||
|
||||
def test_file_is_represented_as_none(self, fm, tmp_path):
|
||||
(tmp_path / "main.py").touch()
|
||||
tree = fm.get_file_tree()
|
||||
assert tree["main.py"] is None
|
||||
|
||||
def test_directory_is_represented_as_dict(self, fm, tmp_path):
|
||||
(tmp_path / "src").mkdir()
|
||||
tree = fm.get_file_tree()
|
||||
assert isinstance(tree["src"], dict)
|
||||
|
||||
def test_nested_structure_is_correct(self, fm, tmp_path):
|
||||
(tmp_path / "src").mkdir()
|
||||
(tmp_path / "src" / "app.py").touch()
|
||||
tree = fm.get_file_tree()
|
||||
assert tree["src"]["app.py"] is None
|
||||
|
||||
@ -1,6 +0,0 @@
|
||||
"""Test cases for the main module."""
|
||||
|
||||
|
||||
def test_placeholder():
|
||||
"""Placeholder test."""
|
||||
assert True
|
||||
0
tests/test_mcp_server_code_execution.py
Normal file
0
tests/test_mcp_server_code_execution.py
Normal file
0
tests/test_mcp_server_file_search.py
Normal file
0
tests/test_mcp_server_file_search.py
Normal file
0
tests/test_mcp_server_web_search.py
Normal file
0
tests/test_mcp_server_web_search.py
Normal file
@ -12,23 +12,11 @@ from backend.managers.system_prompter import SystemPrompter, MAX_FILE_CHARS
|
||||
class TestSystemPrompterBasePrompt:
|
||||
"""Tests for generate_prompt() without file context."""
|
||||
|
||||
def test_returns_non_empty_string(self):
|
||||
prompt = SystemPrompter.generate_prompt()
|
||||
assert isinstance(prompt, str)
|
||||
assert len(prompt) > 0
|
||||
|
||||
def test_describes_code_assistant(self):
|
||||
prompt = SystemPrompter.generate_prompt()
|
||||
assert "code assistant" in prompt.lower()
|
||||
|
||||
def test_contains_no_file_xml_tag(self):
|
||||
prompt = SystemPrompter.generate_prompt()
|
||||
assert "<file" not in prompt
|
||||
assert "<code>" not in prompt
|
||||
|
||||
def test_none_equals_no_argument(self):
|
||||
assert SystemPrompter.generate_prompt(file_context=None) == SystemPrompter.generate_prompt()
|
||||
|
||||
|
||||
class TestSystemPrompterWithFileContext:
|
||||
"""Tests for generate_prompt() with file_context provided."""
|
||||
@ -49,15 +37,6 @@ class TestSystemPrompterWithFileContext:
|
||||
prompt = SystemPrompter.generate_prompt(file_context={"name": "f.py", "content": "pass"})
|
||||
assert "<code>" in prompt
|
||||
|
||||
def test_with_context_is_longer_than_base(self):
|
||||
base = SystemPrompter.generate_prompt()
|
||||
with_ctx = SystemPrompter.generate_prompt(file_context={"name": "f.py", "content": "x=1"})
|
||||
assert len(with_ctx) > len(base)
|
||||
|
||||
def test_missing_name_key_uses_unknown(self):
|
||||
prompt = SystemPrompter.generate_prompt(file_context={"content": "some code"})
|
||||
assert "unknown" in prompt
|
||||
|
||||
def test_missing_content_key_does_not_raise(self):
|
||||
prompt = SystemPrompter.generate_prompt(file_context={"name": "empty.py"})
|
||||
assert "empty.py" in prompt
|
||||
@ -77,12 +56,19 @@ class TestSystemPrompterTruncation:
|
||||
assert "[truncated]" not in prompt
|
||||
assert content in prompt
|
||||
|
||||
def test_file_exactly_at_limit_is_not_truncated(self):
|
||||
content = "x" * MAX_FILE_CHARS
|
||||
prompt = SystemPrompter.generate_prompt(file_context={"name": "f.py", "content": content})
|
||||
assert "[truncated]" not in prompt
|
||||
|
||||
def test_file_one_over_limit_is_truncated(self):
|
||||
content = "x" * (MAX_FILE_CHARS + 1)
|
||||
prompt = SystemPrompter.generate_prompt(file_context={"name": "f.py", "content": content})
|
||||
assert "[truncated]" in prompt
|
||||
|
||||
|
||||
class TestSystemPrompterSpecialCharacters:
|
||||
"""Tests that XML special characters in file content are handled without breaking the prompt."""
|
||||
|
||||
def test_xml_tags_in_content_are_preserved_literally(self):
|
||||
# User code often contains HTML or XML. The prompt builder must embed it
|
||||
# verbatim — escaping or stripping tags would corrupt the file content.
|
||||
prompt = SystemPrompter.generate_prompt(
|
||||
file_context={"name": "template.html", "content": "<div>hello</div>"}
|
||||
)
|
||||
assert "<div>hello</div>" in prompt
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user