diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..f62b0f0 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,30 @@ +[build-system] +requires = ["setuptools>=68.0"] +build-backend = "setuptools.build_meta" + +[project] +name = "crewai-gui-qt" +version = "0.1.0" +description = "A Node-Based Frontend for CrewAI" +readme = "README.md" +license = {text = "MIT"} +requires-python = ">=3.10" +dependencies = [ + "PySide6>=6.6", + "crewai>=0.30", + "crewai-tools>=0.1", + "langchain-community>=0.0.10", + "langchain-openai>=0.1", + "networkx>=3.0", + "pyautogui>=0.9", + "requests>=2.31", +] + +[project.optional-dependencies] +dev = [ + "pytest>=7.0", + "pytest-cov>=4.0", +] + +[tool.setuptools.packages.find] +where = ["src"] diff --git a/src/AdditionalTools.py b/src/AdditionalTools.py index eedc233..f1fd9d3 100644 --- a/src/AdditionalTools.py +++ b/src/AdditionalTools.py @@ -1,22 +1,75 @@ +""" +Potentially dangerous tool wrappers for CrewAI. + +These tools can interact with the local system. Use with care. +""" + import os import requests import subprocess -from crewai_tools import BaseTool +import logging +from crewai.tools import BaseTool # type: ignore[import-untyped] from typing import Optional +logger = logging.getLogger(__name__) + +# Commands that are NEVER allowed regardless of context. +DENIED_COMMANDS = [ + "rm -rf /", + "rm -rf ~", + "dd if=", + "mkfs.", + "fdisk", + "> /dev/", + "chmod 777 /", + "wget ", # remote downloads — handled by restricted network + "curl ", # same +] + +# Allowed command prefixes for ``SystemCommandTool``. +SAFE_PREFIXES = [ + "ls", + "cat", + "echo", + "head", + "tail", + "grep", + "wc", + "find", + "pwd", + "whoami", + "date", + "uname", + "df", + "du", + "ps aux", + "which", + "python3 --version", + "pip list", +] + + class WebRequestTool(BaseTool): + """Make HTTP requests to external URLs.""" + name: str = "WebRequestTool" - description: str = "A tool for making web requests and fetching content from URLs." + description: str = ( + "A tool for making web requests and fetching content from URLs. " + "Only accessible to admin-level agents." + ) def _run(self, url: str, method: str = "GET", params: Optional[dict] = None) -> str: try: - response = requests.request(method, url, params=params) + response = requests.request(method, url, params=params, timeout=30) response.raise_for_status() return response.text except requests.RequestException as e: return f"Error making request: {str(e)}" + class FileOperationTool(BaseTool): + """Read from and write to local files.""" + name: str = "FileOperationTool" description: str = "A tool for reading from and writing to files." @@ -29,21 +82,61 @@ def _run(self, action: str, path: str, content: Optional[str] = None) -> str: return f"Error reading file: {str(e)}" elif action == "write": try: + # Safety: prevent write outside project directory + allowed = os.path.abspath(path).startswith(os.getcwd()) + if not allowed: + return "Error: writing to paths outside the working directory is not allowed." with open(path, 'w') as file: - file.write(content) + if content is not None: + file.write(content) return f"Successfully wrote to {path}" except IOError as e: return f"Error writing to file: {str(e)}" else: return "Invalid action. Supported actions are 'read' and 'write'." + class SystemCommandTool(BaseTool): + """Execute safe system commands. Denied and unknown commands are blocked.""" + name: str = "SystemCommandTool" - description: str = "A tool for executing system commands." + description: str = ( + "A tool for executing safe system commands. " + "Only read-only commands are allowed (ls, cat, echo, pwd, etc.). " + "Destructive or unknown commands are blocked." + ) + + @staticmethod + def _is_safe(command: str) -> bool: + stripped = command.strip() + # Block denied patterns first + for denied in DENIED_COMMANDS: + if denied in stripped: + return False + # Allow only known safe prefixes + for prefix in SAFE_PREFIXES: + if stripped.startswith(prefix): + return True + return False def _run(self, command: str) -> str: + if not self._is_safe(command): + logger.warning("Blocked unsafe command: %s", command) + return ( + f"Error: command '{command}' is not in the allowed list. " + f"Allowed: {', '.join(SAFE_PREFIXES)}" + ) try: - result = subprocess.run(command, shell=True, check=True, capture_output=True, text=True) + result = subprocess.run( + command, + shell=True, + check=True, + capture_output=True, + text=True, + timeout=30, + ) return result.stdout except subprocess.CalledProcessError as e: - return f"Error executing command: {str(e)}" \ No newline at end of file + return f"Error executing command: {str(e)}" + except subprocess.TimeoutExpired: + return "Error: command timed out after 30 seconds" diff --git a/src/KeyboardMouseTool.py b/src/KeyboardMouseTool.py index 53f8021..aa80349 100644 --- a/src/KeyboardMouseTool.py +++ b/src/KeyboardMouseTool.py @@ -1,42 +1,84 @@ +""" +Keyboard and mouse control tool for CrewAI agents. + +WARNING: This tool can physically control the user's input devices. +Use only in controlled environments and with explicit user approval. +""" + import pyautogui import time -from crewai_tools import BaseTool +from crewai.tools import BaseTool # type: ignore[import-untyped] from typing import Optional + class KeyboardMouseTool(BaseTool): + """Control keyboard and mouse, and take screenshots. + + This tool requires **explicit user approval** before execution. + It is disabled by default and must be enabled via the GUI. + """ + name: str = "KeyboardMouseTool" - description: str = "A tool for keyboard and mouse control, as well as taking screenshots." + description: str = ( + "A tool for keyboard and mouse control, as well as taking screenshots. " + "IMPORTANT: This requires explicit user approval. " + "Never use this without asking the user first." + ) + + # Flag to require user confirmation before any action. + # Set to False only if you have explicitly confirmed with the user. + _enabled: bool = False def _run(self, action: str, params: Optional[dict] = None) -> str: - if action == "type": - text = params.get("text", "") - pyautogui.typewrite(text) - return f"Typed: {text}" - elif action == "click": - x = params.get("x") - y = params.get("y") - if x is not None and y is not None: - pyautogui.click(x, y) - return f"Clicked at ({x}, {y})" - else: - pyautogui.click() - return "Clicked at current position" - elif action == "move": - x = params.get("x") - y = params.get("y") - pyautogui.moveTo(x, y) - return f"Moved to ({x}, {y})" - elif action == "screenshot": - filename = params.get("filename", "screenshot.png") - folder = params.get("folder", ".") - path = f"{folder}/{filename}" - screenshot = pyautogui.screenshot() - screenshot.save(path) - return f"Screenshot saved as {path}" - else: - return "Invalid action. Supported actions are 'type', 'click', 'move', and 'screenshot'." + if not self._enabled: + return ( + "KeyboardMouseTool is disabled by default for safety. " + "Enable it in the agent settings if you trust the agent." + ) + + actions = { + "type": self._do_type, + "click": self._do_click, + "move": self._do_move, + "screenshot": self._do_screenshot, + } + + handler = actions.get(action) + if handler is None: + return "Invalid action. Supported: type, click, move, screenshot." + return handler(params or {}) + + def _do_type(self, params: dict) -> str: + text = params.get("text", "") + pyautogui.typewrite(text) + return f"Typed: {text}" + + def _do_click(self, params: dict) -> str: + x = params.get("x") + y = params.get("y") + if x is not None and y is not None: + pyautogui.click(x, y) + return f"Clicked at ({x}, {y})" + pyautogui.click() + return "Clicked at current position" + + def _do_move(self, params: dict) -> str: + x = params.get("x") + y = params.get("y") + if x is None or y is None: + return "Error: x and y coordinates required." + pyautogui.moveTo(x, y) + return f"Moved to ({x}, {y})" + + def _do_screenshot(self, params: dict) -> str: + filename = params.get("filename", "screenshot.png") + folder = params.get("folder", ".") + path = f"{folder}/{filename}" + screenshot = pyautogui.screenshot() + screenshot.save(path) + return f"Screenshot saved as {path}" def _arun(self, action: str, params: Optional[dict] = None) -> str: - # For async operations, you might want to add a small delay + """Async wrapper — adds a small delay, then calls ``_run``.""" time.sleep(0.1) - return self._run(action, params) \ No newline at end of file + return self._run(action, params) diff --git a/src/Node.py b/src/Node.py index c2306ed..5455d25 100644 --- a/src/Node.py +++ b/src/Node.py @@ -93,19 +93,23 @@ def remove_node(self): for edge in self.output_port.edges[:]: edge.remove() + scene = self.scene() + if scene is None: + return + # Update prevs and nexts of connected nodes for prev_id in self.data.prevs: - prev_node = self.scene().get_node_by_id(prev_id) + prev_node = scene.get_node_by_id(prev_id) if prev_node: prev_node.data.nexts.remove(self.data.uniq_id) for next_id in self.data.nexts: - next_node = self.scene().get_node_by_id(next_id) + next_node = scene.get_node_by_id(next_id) if next_node: next_node.data.prevs.remove(self.data.uniq_id) # Finally, remove this node from the scene - self.scene().removeItem(self) + scene.removeItem(self) def hoverEnterEvent(self, event): self.hovered = True diff --git a/src/NodeData.py b/src/NodeData.py index a5d6e1b..4c40545 100644 --- a/src/NodeData.py +++ b/src/NodeData.py @@ -1,56 +1,72 @@ -# NodeData.py +""" +Data models for the CrewAI-GUI node graph. +""" from dataclasses import dataclass, asdict, field from typing import List + @dataclass class Serializable: + """Mixin that adds ``to_dict`` / ``from_dict`` round-tripping.""" + def to_dict(self): return asdict(self) - + @classmethod - def from_dict(cls, data): + def from_dict(cls, data: dict): return cls(**data) + @dataclass class NodeData(Serializable): - - # "None", "Start", "Agent", "Task", "Step", "Team" + """A single node in the visual workflow graph. + + Every node holds position, size, and type-specific configuration. + Connection topology is stored via ``uniq_id`` references in ``prevs`` + (upstream nodes) and ``nexts`` (downstream nodes). + """ + + # Node type: "None", "Start", "Agent", "Task", "Step", "Team" type: str = "" - # New field for tools + # Tools assigned to this node's agent tools: List[str] = field(default_factory=list) + # Unique identifier (string, not int — JSON keys are strings) uniq_id: str = "" + + # Position and size in the editor canvas pos_x: float = 0.0 pos_y: float = 0.0 - width: float = 200.0 # Default width - height: float = 200.0 # Default height + width: float = 200.0 + height: float = 200.0 + # Display name name: str = "" - # Agent + # Agent configuration role: str = "" goal: str = "" backstory: str = "" - # Task + # Task configuration agent: str = "" description: str = "" expected_output: str = "" - # Step + # Step configuration tool: str = "" arg: str = "" output_var: str = "" + # Graph topology — IDs are strings matching ``uniq_id`` + nexts: List[str] = field(default_factory=list) + prevs: List[str] = field(default_factory=list) - nexts: List[int] = field(default_factory=list) - prevs: List[int] = field(default_factory=list) - - def to_dict(self): + def to_dict(self) -> dict: return asdict(self) @classmethod - def from_dict(cls, data): + def from_dict(cls, data: dict): return cls(**data) diff --git a/src/WorkFlow.py b/src/WorkFlow.py index bcef619..1214f80 100644 --- a/src/WorkFlow.py +++ b/src/WorkFlow.py @@ -1,35 +1,54 @@ +""" +CrewAI workflow engine — load, validate, and execute DAG-based CrewAI workflows. + +Provides JSON deserialization of node graphs, topological sorting, +and orchestration of CrewAI agents/tasks from a visual editor export. +""" + import os import json import configparser -from typing import Dict, List +from collections import deque +from typing import Dict, List, Optional + from NodeData import NodeData -from crewai import Agent, Task, Crew, Process -from langchain_community.llms import Ollama -from langchain_community.chat_models import ChatOpenAI -from crewai_tools import FileReadTool, BaseTool +from crewai import Agent, Task, Crew, Process # type: ignore[import-untyped] +from langchain_community.llms import Ollama # type: ignore[import-untyped] +from langchain_openai import ChatOpenAI # type: ignore[import-untyped] +from crewai_tools import FileReadTool # type: ignore[import-untyped] +from crewai.tools import BaseTool # type: ignore[import-untyped] import networkx as nx -from KeyboardMouseTool import KeyboardMouseTool -from AdditionalTools import WebRequestTool, FileOperationTool, SystemCommandTool +# KeyboardMouseTool and AdditionalTools are imported lazily in create_agent() +# to avoid requiring a display (pyautogui dependency) at module load time. + def load_nodes_from_json(filename: str) -> Dict[str, NodeData]: + """Load a node map from a JSON file exported by the GUI editor.""" with open(filename, 'r') as file: data = json.load(file) - node_map = {} + node_map: Dict[str, NodeData] = {} for node_data in data["nodes"]: node = NodeData.from_dict(node_data) node_map[node.uniq_id] = node return node_map + def find_nodes_by_type(node_map: Dict[str, NodeData], node_type: str) -> List[NodeData]: + """Return every node in *node_map* whose ``.type`` equals *node_type*.""" return [node for node in node_map.values() if node.type == node_type] -def find_node_by_type(node_map: Dict[str, NodeData], node_type: str) -> NodeData: + +def find_node_by_type(node_map: Dict[str, NodeData], node_type: str) -> Optional[NodeData]: + """Return the first node matching *node_type*, or ``None``.""" for node in node_map.values(): if node.type == node_type: return node return None + class FileWriterTool(BaseTool): + """Write *content* to *filename* on the local filesystem.""" + name: str = "FileWriter" description: str = "Writes given content to a specified file." @@ -40,6 +59,11 @@ def _run(self, filename: str, content: str) -> str: def create_agent(node: NodeData, llm) -> Agent: + """Build a CrewAI ``Agent`` from a visual-editor *node*.""" + # Lazy imports to avoid requiring a display (pyautogui) at import time. + from KeyboardMouseTool import KeyboardMouseTool + from AdditionalTools import WebRequestTool, FileOperationTool, SystemCommandTool + tools = [] for tool_name in node.tools: if tool_name == "KeyboardMouseTool": @@ -50,7 +74,7 @@ def create_agent(node: NodeData, llm) -> Agent: tools.append(FileOperationTool()) elif tool_name == "SystemCommandTool": tools.append(SystemCommandTool()) - + return Agent( role=node.role, goal=node.goal, @@ -58,10 +82,17 @@ def create_agent(node: NodeData, llm) -> Agent: verbose=True, allow_delegation=False, llm=llm, - tools=tools + tools=tools, ) -def create_task(node: NodeData, agent: Agent, node_map: Dict[str, NodeData], task_map: Dict[str, Task]) -> Task: + +def create_task( + node: NodeData, + agent: Agent, + node_map: Dict[str, NodeData], + task_map: Dict[str, Task], +) -> Task: + """Build a CrewAI ``Task`` from a visual-editor task node.""" steps = [] for step_id in node.nexts: step_node = node_map[step_id] @@ -73,11 +104,10 @@ def create_task(node: NodeData, agent: Agent, node_map: Dict[str, NodeData], tas step = { 'tool': tool_instance, 'args': step_node.arg, - 'output_var': step_node.output_var + 'output_var': step_node.output_var, } steps.append(step) - - # Resolve dependencies with actual Task instances + dependencies = [task_map[dep_id] for dep_id in node.prevs if dep_id in task_map] return Task( @@ -85,32 +115,28 @@ def create_task(node: NodeData, agent: Agent, node_map: Dict[str, NodeData], tas expected_output=node.expected_output, agent=agent, steps=steps, - dependencies=dependencies + dependencies=dependencies, ) + def topological_sort_tasks(task_nodes: List[NodeData]) -> List[NodeData]: + """Topologically sort *task_nodes* so dependencies come before dependents.""" graph = nx.DiGraph() - # Add nodes to the graph for node in task_nodes: graph.add_node(node.uniq_id) - - # Add edges to the graph for node in task_nodes: for prev_id in node.prevs: if prev_id in graph: graph.add_edge(prev_id, node.uniq_id) - - # Perform topological sort + sorted_ids = list(nx.topological_sort(graph)) - - # Return nodes in sorted order id_to_node = {node.uniq_id: node for node in task_nodes} - sorted_tasks = [id_to_node[node_id] for node_id in sorted_ids] - - return sorted_tasks + return [id_to_node[nid] for nid in sorted_ids] -def RunWorkFlow(node: NodeData, node_map: Dict[str, NodeData], llm): + +def RunWorkFlow(node: NodeData, node_map: Dict[str, NodeData], llm) -> None: + """Execute a CrewAI workflow starting from the Start *node*.""" print(f"Start root ID: {node.uniq_id}") # from root find team @@ -123,18 +149,17 @@ def RunWorkFlow(node: NodeData, node_map: Dict[str, NodeData], llm): print(f"Processing Team {team_node.name} ID: {team_node.uniq_id}") # from team find agents - agent_map = {next_id: node_map[next_id] for next_id in team_node.nexts} agent_nodes = find_nodes_by_type(node_map, "Agent") agents = {agent_node.name: create_agent(agent_node, llm) for agent_node in agent_nodes} for agent_node in agent_nodes: print(f"Agent {agent_node.name} ID: {agent_node.uniq_id}") - # Use BFS to collect all task nodes - task_nodes = [] - queue = find_nodes_by_type(sub_node_map, "Task") - + # BFS to collect all task nodes + task_nodes: List[NodeData] = [] + queue: deque = deque(find_nodes_by_type(sub_node_map, "Task")) + while queue: - current_node = queue.pop(0) + current_node = queue.popleft() if current_node not in task_nodes: print(f"Processing task_node ID: {current_node.uniq_id}") task_nodes.append(current_node) @@ -144,10 +169,9 @@ def RunWorkFlow(node: NodeData, node_map: Dict[str, NodeData], llm): # Sort tasks topologically to respect dependencies sorted_task_nodes = topological_sort_tasks(task_nodes) - tasks = [] - task_map = {} - - # Create tasks with dependencies resolved + tasks: List[Task] = [] + task_map: Dict[str, Task] = {} + for task_node in sorted_task_nodes: if task_node: print(f"Processing task_node ID: {task_node.description}") @@ -162,33 +186,17 @@ def RunWorkFlow(node: NodeData, node_map: Dict[str, NodeData], llm): crew = Crew( agents=list(agents.values()), tasks=tasks, - verbose=2 + verbose=2, ) - + result = crew.kickoff() print("######################") print(result) -def run_workflow_from_file(filename: str, llm): + +def run_workflow_from_file(filename: str, llm) -> None: + """Load a JSON workflow from *filename* and execute every Start node.""" node_map = load_nodes_from_json(filename) start_nodes = find_nodes_by_type(node_map, "Start") for start_node in start_nodes: RunWorkFlow(start_node, node_map, llm) - - - - - - - - - - - - - - - - - -# ... (rest of the file remains the same) \ No newline at end of file diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..521670b --- /dev/null +++ b/src/__init__.py @@ -0,0 +1 @@ +# src package diff --git a/src/backend.py b/src/backend.py index 11810c2..2aaa80b 100644 --- a/src/backend.py +++ b/src/backend.py @@ -3,7 +3,7 @@ import configparser from WorkFlow import run_workflow_from_file from langchain_community.llms import Ollama -from langchain_community.chat_models import ChatOpenAI +from langchain_openai import ChatOpenAI from Tee import Tee def get_openai_api_key(config_path=None): diff --git a/src/frontend.py b/src/frontend.py index acf8c3a..55e10a0 100644 --- a/src/frontend.py +++ b/src/frontend.py @@ -7,7 +7,7 @@ def initialize_main_window(): window = MainWindow() - window.setWindowTitle("Json Node Editor") + window.setWindowTitle("CrewAI-GUI") window.setGeometry(100, 100, 800, 600) # Set initial size to 800x600 # Set up a timer to refresh the map view periodically diff --git a/tests/test_backend.py b/tests/test_backend.py new file mode 100644 index 0000000..9d2da20 --- /dev/null +++ b/tests/test_backend.py @@ -0,0 +1,235 @@ +"""Tests for backend modules that don't require a Qt display.""" + +import json +import os +import tempfile +import pytest +from NodeData import NodeData +from AdditionalTools import WebRequestTool, FileOperationTool, SystemCommandTool + +# ── NodeData ───────────────────────────────────────────────────────────────── + +class TestNodeData: + def test_defaults(self): + n = NodeData() + assert n.type == "" + assert n.tools == [] + assert n.width == 200.0 + assert n.height == 200.0 + + def test_to_dict_roundtrip(self): + n = NodeData( + type="Agent", + uniq_id="abc123", + name="test-agent", + role="Helper", + goal="Help", + backstory="A helpful agent", + tools=["WebRequestTool"], + ) + d = n.to_dict() + assert d["type"] == "Agent" + assert d["name"] == "test-agent" + assert d["tools"] == ["WebRequestTool"] + + restored = NodeData.from_dict(d) + assert restored.type == "Agent" + assert restored.name == "test-agent" + assert restored.uniq_id == "abc123" + assert restored.tools == ["WebRequestTool"] + assert restored.role == "Helper" + + def test_position(self): + n = NodeData(pos_x=42.5, pos_y=13.7) + assert n.pos_x == 42.5 + assert n.pos_y == 13.7 + + def test_nexts_prevs(self): + n = NodeData(nexts=["b", "c"], prevs=["a"]) + assert n.nexts == ["b", "c"] + assert n.prevs == ["a"] + + def test_from_dict_full(self): + raw = { + "type": "Task", + "uniq_id": "task1", + "name": "My Task", + "description": "Do the thing", + "expected_output": "Done", + "agent": "helper-agent", + "tools": [], + "nexts": [], + "prevs": [], + "pos_x": 100.0, + "pos_y": 200.0, + "width": 250.0, + "height": 180.0, + } + n = NodeData.from_dict(raw) + assert n.type == "Task" + assert n.description == "Do the thing" + assert n.expected_output == "Done" + assert n.width == 250.0 + assert n.height == 180.0 + + +# ── WorkFlow helpers ───────────────────────────────────────────────────────── + +from WorkFlow import load_nodes_from_json, find_nodes_by_type, find_node_by_type + +class TestWorkFlow: + def test_load_nodes_from_json(self): + data = { + "nodes": [ + { + "type": "Start", + "uniq_id": "start1", + "name": "Start", + "tools": [], + "nexts": ["team1"], + "prevs": [], + "pos_x": 0.0, + "pos_y": 0.0, + "width": 200.0, + "height": 200.0, + }, + { + "type": "Team", + "uniq_id": "team1", + "name": "MyTeam", + "tools": [], + "nexts": [], + "prevs": ["start1"], + "pos_x": 300.0, + "pos_y": 0.0, + "width": 200.0, + "height": 200.0, + }, + ] + } + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: + json.dump(data, f) + path = f.name + try: + node_map = load_nodes_from_json(path) + assert len(node_map) == 2 + assert "start1" in node_map + assert "team1" in node_map + assert node_map["start1"].type == "Start" + assert node_map["team1"].name == "MyTeam" + finally: + os.unlink(path) + + def test_find_nodes_by_type(self): + n1 = NodeData(type="Agent", uniq_id="a1", name="Alpha") + n2 = NodeData(type="Task", uniq_id="t1", name="Task1") + n3 = NodeData(type="Agent", uniq_id="a2", name="Beta") + node_map = {"a1": n1, "t1": n2, "a2": n3} + agents = find_nodes_by_type(node_map, "Agent") + assert len(agents) == 2 + assert agents[0].name == "Alpha" + assert agents[1].name == "Beta" + + def test_find_node_by_type(self): + n1 = NodeData(type="Start", uniq_id="s1") + n2 = NodeData(type="Team", uniq_id="tm1") + node_map = {"s1": n1, "tm1": n2} + team = find_node_by_type(node_map, "Team") + assert team is not None + assert team.uniq_id == "tm1" + not_found = find_node_by_type(node_map, "Agent") + assert not_found is None + + +# ── Topological sort ───────────────────────────────────────────────────────── + +from WorkFlow import topological_sort_tasks + +class TestTopologicalSort: + def test_simple_chain(self): + t1 = NodeData(uniq_id="t1", prevs=[], nexts=["t2"], type="Task", name="1") + t2 = NodeData(uniq_id="t2", prevs=["t1"], nexts=["t3"], type="Task", name="2") + t3 = NodeData(uniq_id="t3", prevs=["t2"], nexts=[], type="Task", name="3") + ids = [n.uniq_id for n in topological_sort_tasks([t3, t1, t2])] + assert ids == ["t1", "t2", "t3"], f"got {ids}" + + def test_diamond(self): + t1 = NodeData(uniq_id="s", prevs=[], nexts=["a", "b"], type="Task") + ta = NodeData(uniq_id="a", prevs=["s"], nexts=["e"], type="Task") + tb = NodeData(uniq_id="b", prevs=["s"], nexts=["e"], type="Task") + te = NodeData(uniq_id="e", prevs=["a", "b"], nexts=[], type="Task") + sorted_ids = [n.uniq_id for n in topological_sort_tasks([te, tb, ta, t1])] + assert sorted_ids[0] == "s" + assert sorted_ids[-1] == "e" + # a and b order doesn't matter, but both must come before e + assert sorted_ids.index("a") < sorted_ids.index("e") + assert sorted_ids.index("b") < sorted_ids.index("e") + + def test_single_node(self): + n = NodeData(uniq_id="only", prevs=[], nexts=[], type="Task") + assert [n.uniq_id for n in topological_sort_tasks([n])] == ["only"] + + +# ── SystemCommandTool safety ───────────────────────────────────────────────── + +class TestSystemCommandTool: + def setup_method(self): + self.tool = SystemCommandTool() + + def test_safe_ls(self): + result = self.tool._run("ls") + # should succeed or return "Error: ..." (varies by env) + assert not result.startswith("Error: command 'ls' is not in the allowed list") + + def test_safe_echo(self): + result = self.tool._run("echo hello") + assert "hello" in result + + def test_blocked_rm_rf(self): + result = self.tool._run("rm -rf /") + assert "not in the allowed list" in result + + def test_blocked_dd(self): + result = self.tool._run("dd if=/dev/zero of=/dev/null") + assert "not in the allowed list" in result + + def test_unknown_command_blocked(self): + result = self.tool._run("some_evil_script.sh") + assert "not in the allowed list" in result + + +# ── FileOperationTool path safety ──────────────────────────────────────────── + +class TestFileOperationTool: + def setup_method(self): + self.tool = FileOperationTool() + + def test_write_read_roundtrip(self): + path = f"{os.getcwd()}/_test_write.txt" + try: + write = self.tool._run("write", path, content="hello world") + assert "Successfully" in write + read = self.tool._run("read", path) + assert read == "hello world" + finally: + if os.path.exists(path): + os.unlink(path) + + def test_block_outside_cwd(self): + result = self.tool._run("write", "/tmp/evil.txt", content="bad") + assert "not allowed" in result + + def test_invalid_action(self): + result = self.tool._run("delete", "/tmp/x") + assert "Invalid action" in result + + +# ── WebRequestTool ─────────────────────────────────────────────────────────── + +class TestWebRequestTool: + def setup_method(self): + self.tool = WebRequestTool() + + def test_bad_url_returns_error(self): + result = self.tool._run("https://nonexistent.example.test/foo") + assert result.startswith("Error")