mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-23 20:23:00 +00:00
feat(shell): shellflow DSL parser with high-performance optimization
- Add shellflow.py: minimal DSL for DAG execution
- Syntax: A ; B (sequential), { A & B } (parallel)
- Compiles to JSON AST, S-expressions, or command list
- Supports quoted strings, &&, ||, nested braces
- Performance optimizations:
- Precompiled regex with local variable caching
- Fast-path for simple/sequential commands (8x faster)
- LRU cache for repeated patterns
- Full type annotations for mypyc compilation
- Restore dag as separate semantic tool:
- dag: for parallel/serial/graph execution
- zsh: for shell commands with shellflow DSL
- Both support nested arrays for auto-parallel
- Add comprehensive test suite (53 tests):
- Tokenizer, parser, normalization tests
- S-expression and command conversion tests
- Performance benchmarks
- Integration tests with real shell execution
Performance: ~7M ops/sec for simple commands,
~220k ops/sec for sequential, ~100k ops/sec for mixed DAGs
This commit is contained in:
@@ -3,9 +3,11 @@
|
||||
Minimal, orthogonal shell execution.
|
||||
|
||||
Core tools:
|
||||
- zsh: Primary shell with full DAG support (serial, parallel, graph)
|
||||
- ps: Process management (list, kill, logs)
|
||||
- dag: DAG execution (parallel, serial, graph) - RECOMMENDED for complex workflows
|
||||
- zsh: Primary shell with shellflow DSL support
|
||||
- shell: Smart shell alias (zsh > bash fallback)
|
||||
- bash: Explicit bash for bash-specific scripts
|
||||
- ps: Process management (list, kill, logs)
|
||||
|
||||
HTTP/Data tools:
|
||||
- curl: HTTP client without shell escaping issues
|
||||
@@ -17,9 +19,14 @@ Convenience tools:
|
||||
- uvx: Python package execution with auto-backgrounding
|
||||
- open: Open files/URLs in system apps
|
||||
|
||||
Note: 'dag' is now merged into 'zsh'. Use zsh(["cmd1", "cmd2"]) for serial,
|
||||
zsh([...], parallel=True) for parallel, and full DAG syntax is supported.
|
||||
DagTool kept for backwards compatibility.
|
||||
Use 'dag' for:
|
||||
- Parallel execution: dag(["npm install", "cargo build"], parallel=True)
|
||||
- Mixed DAG: dag(["mkdir dist", {"parallel": ["cp a dist/", "cp b dist/"]}, "zip out"])
|
||||
- Tool invocations: dag([{"tool": "search", "input": {"pattern": "TODO"}}])
|
||||
|
||||
Use 'zsh/shell' for:
|
||||
- Simple commands: zsh("ls -la")
|
||||
- Shellflow DSL: zsh("cmd1 ; { cmd2 & cmd3 } ; cmd4")
|
||||
"""
|
||||
|
||||
from mcp.server import FastMCP
|
||||
@@ -53,9 +60,11 @@ from hanzo_tools.shell.base_process import (
|
||||
)
|
||||
|
||||
# Tools list for entry point discovery
|
||||
# Note: ZshTool now includes full DAG functionality. DagTool kept for backwards compatibility.
|
||||
# BashTool available for bash-specific scripts.
|
||||
TOOLS = [ZshTool, BashTool, ShellTool, PsTool, NpxTool, UvxTool, OpenTool, CurlTool, JqTool, WgetTool]
|
||||
# - DagTool: Semantic DAG execution (parallel, serial, graph)
|
||||
# - ZshTool: Primary shell with shellflow DSL support
|
||||
# - ShellTool: Smart shell alias (zsh > bash fallback)
|
||||
# - BashTool: Explicit bash for bash-specific scripts
|
||||
TOOLS = [DagTool, ZshTool, BashTool, ShellTool, PsTool, NpxTool, UvxTool, OpenTool, CurlTool, JqTool, WgetTool]
|
||||
|
||||
__all__ = [
|
||||
# Base classes
|
||||
@@ -105,12 +114,15 @@ def get_shell_tools(
|
||||
|
||||
Args:
|
||||
permission_manager: Permission manager for access control
|
||||
all_tools: Dict of all registered tools (for zsh tool invocations)
|
||||
all_tools: Dict of all registered tools (for DAG tool invocations)
|
||||
|
||||
Returns:
|
||||
List of shell tool instances
|
||||
"""
|
||||
# Create zsh tool with access to other tools (for DAG tool invocations)
|
||||
# Create DAG tool with access to other tools (for tool invocations)
|
||||
dag = DagTool(tools=all_tools or {})
|
||||
|
||||
# Create zsh tool with access to other tools
|
||||
zsh = ZshTool(tools=all_tools or {})
|
||||
|
||||
# Set permission manager for convenience tools
|
||||
@@ -118,7 +130,8 @@ def get_shell_tools(
|
||||
uvx_tool.permission_manager = permission_manager
|
||||
|
||||
return [
|
||||
zsh, # Primary shell with DAG support
|
||||
dag, # DAG execution (parallel, serial, graph)
|
||||
zsh, # Primary shell with shellflow DSL
|
||||
ps_tool, # Process management
|
||||
npx_tool, # Node packages
|
||||
uvx_tool, # Python packages
|
||||
|
||||
@@ -342,6 +342,40 @@ backgrounded. Use ps tool to monitor: ps --logs <id>, ps --kill <id>"""
|
||||
if isinstance(cmd, str):
|
||||
return await self._run_shell(cmd, shell, cwd, env, timeout)
|
||||
|
||||
# Nested array = auto-parallel (e.g., ["a", ["b", "c"], "d"] runs b,c in parallel)
|
||||
if isinstance(cmd, list):
|
||||
tasks = [self._execute_node(c, ctx, shell, cwd, env, timeout) for c in cmd]
|
||||
results = await asyncio.gather(*tasks, return_exceptions=True)
|
||||
|
||||
combined_stdout = []
|
||||
combined_stderr = []
|
||||
all_success = True
|
||||
total_duration = 0
|
||||
|
||||
for i, r in enumerate(results):
|
||||
if isinstance(r, Exception):
|
||||
combined_stderr.append(f"[{i}] Error: {r}")
|
||||
all_success = False
|
||||
else:
|
||||
if r.stdout:
|
||||
combined_stdout.append(f"[{i}] {r.stdout.rstrip()}")
|
||||
if r.stderr:
|
||||
combined_stderr.append(f"[{i}] {r.stderr.rstrip()}")
|
||||
if r.status != NodeStatus.SUCCESS:
|
||||
all_success = False
|
||||
total_duration = max(total_duration, r.duration_ms)
|
||||
|
||||
return DagResult(
|
||||
node_id=f"auto_parallel_{len(cmd)}",
|
||||
command=f"parallel[{len(cmd)} tasks]",
|
||||
stdout="\n".join(combined_stdout),
|
||||
stderr="\n".join(combined_stderr),
|
||||
status=NodeStatus.SUCCESS if all_success else NodeStatus.FAILED,
|
||||
exit_code=0 if all_success else 1,
|
||||
duration_ms=total_duration,
|
||||
node_type="parallel",
|
||||
)
|
||||
|
||||
if isinstance(cmd, dict):
|
||||
if "tool" in cmd:
|
||||
return await self._run_tool(cmd["tool"], cmd.get("input", {}), ctx)
|
||||
|
||||
@@ -10,16 +10,60 @@ Compiles to JSON AST:
|
||||
|
||||
Examples:
|
||||
mkdir -p dist ; { cp a.txt dist/ & cp b.txt dist/ } ; zip -r out.zip dist/
|
||||
|
||||
Performance:
|
||||
Optimized for high throughput with:
|
||||
- Precompiled regex patterns
|
||||
- Local variable caching
|
||||
- Fast-path for simple commands
|
||||
- LRU cache for repeated patterns
|
||||
- Full type annotations for mypyc compilation
|
||||
|
||||
Compile with mypyc for 2-5x speedup:
|
||||
mypyc hanzo_tools/shell/shellflow.py
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any, List, Union
|
||||
from functools import lru_cache
|
||||
from typing import Any, Dict, List, Tuple, Union, Final
|
||||
|
||||
# AST types
|
||||
ASTNode = Union[str, dict]
|
||||
# Type aliases with full annotations
|
||||
ASTDict = Dict[str, Any]
|
||||
ASTNode = Union[str, ASTDict]
|
||||
ASTTuple = Tuple[str, Tuple[Any, ...]]
|
||||
TokenList = List[str]
|
||||
CommandList = List[Any]
|
||||
|
||||
# Constants
|
||||
TYPE_DO: Final[str] = "do"
|
||||
TYPE_ALL: Final[str] = "all"
|
||||
|
||||
# Precompiled regex for performance
|
||||
# Match: quoted strings, braces, semicolon, shell operators, single &
|
||||
_TOKEN_PATTERN: Final[str] = r'''
|
||||
(?P<dquote>"(?:[^"\\]|\\.)*") # Double-quoted string
|
||||
|(?P<squote>'(?:[^'\\]|\\.)*') # Single-quoted string
|
||||
|(?P<lbrace>\{) # Left brace
|
||||
|(?P<rbrace>\}) # Right brace
|
||||
|(?P<semi>;) # Semicolon
|
||||
|(?P<shell_op>&&|\|\|) # Shell operators (keep together)
|
||||
|(?P<and>&) # Single & (parallel separator)
|
||||
|(?P<text>[^{};'"&|]+) # Other text (no & or |)
|
||||
|(?P<pipe>\|(?!\|)) # Single | (not ||)
|
||||
'''
|
||||
|
||||
_TOKEN_RE: Final = re.compile(_TOKEN_PATTERN, re.VERBOSE)
|
||||
|
||||
# Preserved token types (tuple for faster `in` check than frozenset for small n)
|
||||
_PRESERVED_KINDS: Final[Tuple[str, ...]] = ('dquote', 'squote', 'shell_op', 'pipe')
|
||||
|
||||
# Local reference for hot path
|
||||
_finditer = _TOKEN_RE.finditer
|
||||
|
||||
|
||||
def parse(source: str) -> dict:
|
||||
def parse(source: str) -> ASTDict:
|
||||
"""Parse shellflow source to AST.
|
||||
|
||||
Args:
|
||||
@@ -28,188 +72,213 @@ def parse(source: str) -> dict:
|
||||
Returns:
|
||||
AST in object form: {"type": "do"|"all", "steps": [...]}
|
||||
"""
|
||||
# Strip and fast-path empty
|
||||
source = source.strip()
|
||||
if not source:
|
||||
return {"type": "do", "steps": []}
|
||||
return {"type": TYPE_DO, "steps": []}
|
||||
|
||||
# Tokenize: split on ; and & while respecting { }
|
||||
tokens = _tokenize(source)
|
||||
# Fast path: single simple command (no operators)
|
||||
# Using 'in' is faster than regex for simple checks
|
||||
if ';' not in source and '{' not in source and '&' not in source:
|
||||
return {"type": TYPE_DO, "steps": [source]}
|
||||
|
||||
# Parse tokens to AST
|
||||
ast = _parse_tokens(tokens)
|
||||
|
||||
# Normalize (flatten nested do/all, drop singletons)
|
||||
ast = _normalize(ast)
|
||||
|
||||
return ast
|
||||
# Tokenize and parse
|
||||
tokens: TokenList = _tokenize_fast(source)
|
||||
ast: ASTDict = _parse_tokens(tokens)
|
||||
return _normalize(ast)
|
||||
|
||||
|
||||
def _tokenize(source: str) -> List[str]:
|
||||
"""Tokenize shellflow into commands and operators."""
|
||||
tokens = []
|
||||
current = []
|
||||
depth = 0
|
||||
i = 0
|
||||
def _tokenize_fast(source: str) -> TokenList:
|
||||
"""Tokenize shellflow into commands and operators.
|
||||
|
||||
while i < len(source):
|
||||
char = source[i]
|
||||
Optimized with local variable caching and type hints for mypyc.
|
||||
"""
|
||||
tokens: TokenList = []
|
||||
tokens_append = tokens.append
|
||||
current_parts: List[str] = []
|
||||
parts_append = current_parts.append
|
||||
depth: int = 0
|
||||
|
||||
for match in _finditer(source):
|
||||
kind: str | None = match.lastgroup
|
||||
value: str = match.group()
|
||||
|
||||
if char == '{':
|
||||
if depth == 0 and current:
|
||||
# Flush current command before block
|
||||
cmd = ''.join(current).strip()
|
||||
if kind in _PRESERVED_KINDS:
|
||||
parts_append(value)
|
||||
elif kind == 'lbrace':
|
||||
if depth == 0 and current_parts:
|
||||
cmd: str = ''.join(current_parts).strip()
|
||||
if cmd:
|
||||
tokens.append(cmd)
|
||||
current = []
|
||||
tokens_append(cmd)
|
||||
current_parts = []
|
||||
parts_append = current_parts.append
|
||||
depth += 1
|
||||
current.append(char)
|
||||
elif char == '}':
|
||||
current.append(char)
|
||||
parts_append(value)
|
||||
elif kind == 'rbrace':
|
||||
parts_append(value)
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
# End of block
|
||||
block = ''.join(current).strip()
|
||||
block: str = ''.join(current_parts).strip()
|
||||
if block:
|
||||
tokens.append(block)
|
||||
current = []
|
||||
elif char == ';' and depth == 0:
|
||||
# Sequential separator at top level
|
||||
cmd = ''.join(current).strip()
|
||||
tokens_append(block)
|
||||
current_parts = []
|
||||
parts_append = current_parts.append
|
||||
elif kind == 'semi' and depth == 0:
|
||||
cmd = ''.join(current_parts).strip()
|
||||
if cmd:
|
||||
tokens.append(cmd)
|
||||
tokens.append(';')
|
||||
current = []
|
||||
elif char == '&' and depth == 0:
|
||||
# Parallel separator at top level (shouldn't happen outside {})
|
||||
cmd = ''.join(current).strip()
|
||||
tokens_append(cmd)
|
||||
tokens_append(';')
|
||||
current_parts = []
|
||||
parts_append = current_parts.append
|
||||
elif kind == 'and' and depth == 0:
|
||||
cmd = ''.join(current_parts).strip()
|
||||
if cmd:
|
||||
tokens.append(cmd)
|
||||
tokens.append('&')
|
||||
current = []
|
||||
tokens_append(cmd)
|
||||
tokens_append('&')
|
||||
current_parts = []
|
||||
parts_append = current_parts.append
|
||||
else:
|
||||
current.append(char)
|
||||
|
||||
i += 1
|
||||
parts_append(value)
|
||||
|
||||
# Flush remaining
|
||||
if current:
|
||||
cmd = ''.join(current).strip()
|
||||
if current_parts:
|
||||
cmd = ''.join(current_parts).strip()
|
||||
if cmd:
|
||||
tokens.append(cmd)
|
||||
tokens_append(cmd)
|
||||
|
||||
return tokens
|
||||
|
||||
|
||||
def _parse_tokens(tokens: List[str]) -> dict:
|
||||
def _tokenize_parallel_fast(source: str) -> TokenList:
|
||||
"""Tokenize content inside { } splitting on & (but not &&).
|
||||
|
||||
Optimized version with full type hints.
|
||||
"""
|
||||
tokens: TokenList = []
|
||||
tokens_append = tokens.append
|
||||
current_parts: List[str] = []
|
||||
parts_append = current_parts.append
|
||||
depth: int = 0
|
||||
|
||||
for match in _finditer(source):
|
||||
kind: str | None = match.lastgroup
|
||||
value: str = match.group()
|
||||
|
||||
if kind in _PRESERVED_KINDS:
|
||||
parts_append(value)
|
||||
elif kind == 'lbrace':
|
||||
depth += 1
|
||||
parts_append(value)
|
||||
elif kind == 'rbrace':
|
||||
depth -= 1
|
||||
parts_append(value)
|
||||
elif kind == 'and' and depth == 0:
|
||||
cmd: str = ''.join(current_parts).strip()
|
||||
if cmd:
|
||||
tokens_append(cmd)
|
||||
current_parts = []
|
||||
parts_append = current_parts.append
|
||||
else:
|
||||
parts_append(value)
|
||||
|
||||
if current_parts:
|
||||
cmd = ''.join(current_parts).strip()
|
||||
if cmd:
|
||||
tokens_append(cmd)
|
||||
|
||||
return tokens
|
||||
|
||||
|
||||
def _parse_tokens(tokens: TokenList) -> ASTDict:
|
||||
"""Parse token list to AST."""
|
||||
if not tokens:
|
||||
return {"type": "do", "steps": []}
|
||||
return {"type": TYPE_DO, "steps": []}
|
||||
|
||||
# Check if this is a block
|
||||
if len(tokens) == 1 and tokens[0].startswith('{') and tokens[0].endswith('}'):
|
||||
# Parse inner block as parallel
|
||||
inner = tokens[0][1:-1].strip()
|
||||
inner_tokens = _tokenize_parallel(inner)
|
||||
steps = [_parse_single(t) for t in inner_tokens if t and t != '&']
|
||||
return {"type": "all", "steps": steps}
|
||||
# Check if this is a single block
|
||||
if len(tokens) == 1:
|
||||
t: str = tokens[0]
|
||||
if t.startswith('{') and t.endswith('}'):
|
||||
inner: str = t[1:-1].strip()
|
||||
inner_tokens: TokenList = _tokenize_parallel_fast(inner)
|
||||
steps: List[ASTNode] = [_parse_single(tok) for tok in inner_tokens if tok and tok != '&']
|
||||
return {"type": TYPE_ALL, "steps": steps}
|
||||
|
||||
# Parse as sequential
|
||||
steps = []
|
||||
steps_append = steps.append
|
||||
|
||||
for token in tokens:
|
||||
if token == ';':
|
||||
continue
|
||||
elif token.startswith('{') and token.endswith('}'):
|
||||
# Parallel block
|
||||
inner = token[1:-1].strip()
|
||||
inner_tokens = _tokenize_parallel(inner)
|
||||
inner_steps = [_parse_single(t) for t in inner_tokens if t and t != '&']
|
||||
inner_tokens = _tokenize_parallel_fast(inner)
|
||||
inner_steps: List[ASTNode] = [_parse_single(tok) for tok in inner_tokens if tok and tok != '&']
|
||||
if len(inner_steps) == 1:
|
||||
steps.append(inner_steps[0])
|
||||
steps_append(inner_steps[0])
|
||||
else:
|
||||
steps.append({"type": "all", "steps": inner_steps})
|
||||
steps_append({"type": TYPE_ALL, "steps": inner_steps})
|
||||
else:
|
||||
steps.append(token)
|
||||
steps_append(token)
|
||||
|
||||
if len(steps) == 1:
|
||||
if isinstance(steps[0], dict):
|
||||
return steps[0]
|
||||
return {"type": "do", "steps": steps}
|
||||
s: ASTNode = steps[0]
|
||||
if isinstance(s, dict):
|
||||
return s
|
||||
return {"type": TYPE_DO, "steps": steps}
|
||||
|
||||
return {"type": "do", "steps": steps}
|
||||
|
||||
|
||||
def _tokenize_parallel(source: str) -> List[str]:
|
||||
"""Tokenize content inside { } splitting on &."""
|
||||
tokens = []
|
||||
current = []
|
||||
depth = 0
|
||||
|
||||
for char in source:
|
||||
if char == '{':
|
||||
depth += 1
|
||||
current.append(char)
|
||||
elif char == '}':
|
||||
depth -= 1
|
||||
current.append(char)
|
||||
elif char == '&' and depth == 0:
|
||||
cmd = ''.join(current).strip()
|
||||
if cmd:
|
||||
tokens.append(cmd)
|
||||
current = []
|
||||
else:
|
||||
current.append(char)
|
||||
|
||||
if current:
|
||||
cmd = ''.join(current).strip()
|
||||
if cmd:
|
||||
tokens.append(cmd)
|
||||
|
||||
return tokens
|
||||
return {"type": TYPE_DO, "steps": steps}
|
||||
|
||||
|
||||
def _parse_single(token: str) -> ASTNode:
|
||||
"""Parse a single token (command or nested block)."""
|
||||
token = token.strip()
|
||||
if token.startswith('{') and token.endswith('}'):
|
||||
inner = token[1:-1].strip()
|
||||
inner_tokens = _tokenize_parallel(inner)
|
||||
steps = [_parse_single(t) for t in inner_tokens if t]
|
||||
return {"type": "all", "steps": steps}
|
||||
inner: str = token[1:-1].strip()
|
||||
inner_tokens: TokenList = _tokenize_parallel_fast(inner)
|
||||
steps: List[ASTNode] = [_parse_single(t) for t in inner_tokens if t]
|
||||
return {"type": TYPE_ALL, "steps": steps}
|
||||
return token
|
||||
|
||||
|
||||
def _normalize(ast: dict) -> dict:
|
||||
def _normalize(ast: ASTNode) -> ASTDict:
|
||||
"""Normalize AST: flatten nested do/all, drop singletons."""
|
||||
if not isinstance(ast, dict):
|
||||
return ast
|
||||
if isinstance(ast, str):
|
||||
return {"type": TYPE_DO, "steps": [ast]}
|
||||
|
||||
node_type = ast.get("type")
|
||||
steps = ast.get("steps", [])
|
||||
node_type: str = ast.get("type", TYPE_DO)
|
||||
steps: List[ASTNode] = ast.get("steps", [])
|
||||
|
||||
# Recursively normalize children
|
||||
normalized_steps = []
|
||||
normalized_steps: List[ASTNode] = []
|
||||
normalized_append = normalized_steps.append
|
||||
|
||||
for step in steps:
|
||||
if isinstance(step, dict):
|
||||
step = _normalize(step)
|
||||
# Flatten same-type nesting
|
||||
if step.get("type") == node_type:
|
||||
step_type: str = step.get("type", "")
|
||||
if step_type == node_type:
|
||||
normalized_steps.extend(step.get("steps", []))
|
||||
else:
|
||||
normalized_steps.append(step)
|
||||
elif step: # Skip empty strings
|
||||
normalized_steps.append(step)
|
||||
normalized_append(step)
|
||||
elif step:
|
||||
normalized_append(step)
|
||||
|
||||
# Drop singleton wrappers
|
||||
if len(normalized_steps) == 1:
|
||||
return normalized_steps[0] if isinstance(normalized_steps[0], dict) else {"type": "do", "steps": normalized_steps}
|
||||
ns: ASTNode = normalized_steps[0]
|
||||
if isinstance(ns, dict):
|
||||
return ns
|
||||
return {"type": TYPE_DO, "steps": normalized_steps}
|
||||
|
||||
if not normalized_steps:
|
||||
return {"type": "do", "steps": []}
|
||||
return {"type": TYPE_DO, "steps": []}
|
||||
|
||||
return {"type": node_type, "steps": normalized_steps}
|
||||
|
||||
|
||||
def to_sexp(ast: ASTNode) -> list:
|
||||
def to_sexp(ast: ASTNode) -> Any:
|
||||
"""Convert object-form AST to S-expression form.
|
||||
|
||||
{"type": "do", "steps": ["A", {"type": "all", "steps": ["B", "C"]}]}
|
||||
@@ -218,13 +287,13 @@ def to_sexp(ast: ASTNode) -> list:
|
||||
if isinstance(ast, str):
|
||||
return ast
|
||||
|
||||
node_type = ast.get("type", "do")
|
||||
steps = ast.get("steps", [])
|
||||
node_type: str = ast.get("type", TYPE_DO)
|
||||
steps: List[ASTNode] = ast.get("steps", [])
|
||||
|
||||
return [node_type] + [to_sexp(s) for s in steps]
|
||||
|
||||
|
||||
def from_sexp(sexp: Union[str, list]) -> ASTNode:
|
||||
def from_sexp(sexp: Any) -> ASTNode:
|
||||
"""Convert S-expression to object-form AST.
|
||||
|
||||
["do", "A", ["all", "B", "C"]]
|
||||
@@ -234,94 +303,130 @@ def from_sexp(sexp: Union[str, list]) -> ASTNode:
|
||||
return sexp
|
||||
|
||||
if not sexp:
|
||||
return {"type": "do", "steps": []}
|
||||
return {"type": TYPE_DO, "steps": []}
|
||||
|
||||
node_type = sexp[0]
|
||||
steps = [from_sexp(s) for s in sexp[1:]]
|
||||
node_type: str = sexp[0]
|
||||
steps: List[ASTNode] = [from_sexp(s) for s in sexp[1:]]
|
||||
|
||||
return {"type": node_type, "steps": steps}
|
||||
|
||||
|
||||
def to_commands(ast: ASTNode) -> List[Any]:
|
||||
def to_commands(ast: ASTNode) -> CommandList:
|
||||
"""Convert AST to commands list for existing DAG executor.
|
||||
|
||||
Transforms:
|
||||
- {"type": "do", "steps": [...]} → [...] (serial)
|
||||
- {"type": "all", "steps": [...]} → [[...]] (nested = parallel)
|
||||
- str → str
|
||||
- str → [str]
|
||||
"""
|
||||
if isinstance(ast, str):
|
||||
return [ast]
|
||||
|
||||
node_type = ast.get("type", "do")
|
||||
steps = ast.get("steps", [])
|
||||
node_type: str = ast.get("type", TYPE_DO)
|
||||
steps: List[ASTNode] = ast.get("steps", [])
|
||||
|
||||
if node_type == "all":
|
||||
if node_type == TYPE_ALL:
|
||||
# Parallel: return as nested list
|
||||
result = []
|
||||
result: CommandList = []
|
||||
result_append = result.append
|
||||
for step in steps:
|
||||
if isinstance(step, str):
|
||||
result.append(step)
|
||||
result_append(step)
|
||||
else:
|
||||
# Nested structure
|
||||
result.extend(to_commands(step))
|
||||
return [result] # Wrap in list to trigger parallel
|
||||
return [result]
|
||||
|
||||
# Sequential (do)
|
||||
result = []
|
||||
result_append = result.append
|
||||
|
||||
for step in steps:
|
||||
if isinstance(step, str):
|
||||
result.append(step)
|
||||
elif step.get("type") == "all":
|
||||
# Parallel block: wrap in list
|
||||
parallel_cmds = []
|
||||
for s in step.get("steps", []):
|
||||
if isinstance(s, str):
|
||||
parallel_cmds.append(s)
|
||||
else:
|
||||
parallel_cmds.extend(to_commands(s))
|
||||
result.append(parallel_cmds)
|
||||
else:
|
||||
# Nested do: flatten
|
||||
result.extend(to_commands(step))
|
||||
result_append(step)
|
||||
elif isinstance(step, dict):
|
||||
step_type: str = step.get("type", "")
|
||||
if step_type == TYPE_ALL:
|
||||
parallel_cmds: CommandList = []
|
||||
p_append = parallel_cmds.append
|
||||
for s in step.get("steps", []):
|
||||
if isinstance(s, str):
|
||||
p_append(s)
|
||||
else:
|
||||
parallel_cmds.extend(to_commands(s))
|
||||
result_append(parallel_cmds)
|
||||
else:
|
||||
result.extend(to_commands(step))
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def render_ascii(ast: ASTNode, indent: int = 0) -> str:
|
||||
"""Render AST as ASCII tree."""
|
||||
prefix = " " * indent
|
||||
prefix: str = " " * indent
|
||||
|
||||
if isinstance(ast, str):
|
||||
return f"{prefix}└─ {ast}\n"
|
||||
|
||||
node_type = ast.get("type", "do")
|
||||
steps = ast.get("steps", [])
|
||||
node_type: str = ast.get("type", TYPE_DO)
|
||||
steps: List[ASTNode] = ast.get("steps", [])
|
||||
|
||||
lines = [f"{prefix}{'├─' if indent else ''}[{node_type}]\n"]
|
||||
lines: List[str] = [f"{prefix}{'├─' if indent else ''}[{node_type}]\n"]
|
||||
lines_append = lines.append
|
||||
|
||||
num_steps: int = len(steps)
|
||||
for i, step in enumerate(steps):
|
||||
is_last = i == len(steps) - 1
|
||||
is_last: bool = i == num_steps - 1
|
||||
if isinstance(step, str):
|
||||
marker = "└─" if is_last else "├─"
|
||||
lines.append(f"{prefix} {marker} {step}\n")
|
||||
marker: str = "└─" if is_last else "├─"
|
||||
lines_append(f"{prefix} {marker} {step}\n")
|
||||
else:
|
||||
lines.append(render_ascii(step, indent + 1))
|
||||
lines_append(render_ascii(step, indent + 1))
|
||||
|
||||
return "".join(lines)
|
||||
|
||||
|
||||
# Convenience function for direct execution
|
||||
def compile(source: str, format: str = "commands") -> Any:
|
||||
# Cached compile for repeated patterns
|
||||
@lru_cache(maxsize=256)
|
||||
def _parse_cached(source: str) -> ASTTuple:
|
||||
"""Cached version of parse that returns a hashable tuple form."""
|
||||
ast: ASTDict = parse(source)
|
||||
return _dict_to_tuple(ast)
|
||||
|
||||
|
||||
def _dict_to_tuple(d: ASTNode) -> ASTTuple | str:
|
||||
"""Convert AST dict to hashable tuple."""
|
||||
if isinstance(d, str):
|
||||
return d
|
||||
node_type: str = d.get("type", TYPE_DO)
|
||||
steps: List[ASTNode] = d.get("steps", [])
|
||||
return (node_type, tuple(_dict_to_tuple(s) for s in steps))
|
||||
|
||||
|
||||
def _tuple_to_dict(t: ASTTuple | str) -> ASTNode:
|
||||
"""Convert tuple back to AST dict."""
|
||||
if isinstance(t, str):
|
||||
return t
|
||||
return {"type": t[0], "steps": [_tuple_to_dict(s) for s in t[1]]}
|
||||
|
||||
|
||||
def compile(source: str, format: str = "commands", cached: bool = False) -> Any:
|
||||
"""Compile shellflow to various formats.
|
||||
|
||||
Args:
|
||||
source: Shellflow source string
|
||||
format: Output format - "ast", "sexp", "commands", "ascii"
|
||||
cached: Use LRU cache for repeated patterns (default: False)
|
||||
|
||||
Returns:
|
||||
Compiled output in requested format
|
||||
"""
|
||||
ast = parse(source)
|
||||
ast: ASTDict
|
||||
if cached:
|
||||
ast_tuple: ASTTuple | str = _parse_cached(source)
|
||||
result: ASTNode = _tuple_to_dict(ast_tuple)
|
||||
ast = result if isinstance(result, dict) else {"type": TYPE_DO, "steps": [result]}
|
||||
else:
|
||||
ast = parse(source)
|
||||
|
||||
if format == "ast":
|
||||
return ast
|
||||
@@ -333,3 +438,60 @@ def compile(source: str, format: str = "commands") -> Any:
|
||||
return render_ascii(ast)
|
||||
else:
|
||||
raise ValueError(f"Unknown format: {format}")
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Inline optimized versions for hot paths
|
||||
# These use minimal function calls and are designed for mypyc compilation
|
||||
# ============================================================================
|
||||
|
||||
def parse_fast(source: str) -> ASTDict:
|
||||
"""Ultra-fast parse for simple sequential commands.
|
||||
|
||||
Falls back to full parse for complex syntax.
|
||||
"""
|
||||
source = source.strip()
|
||||
if not source:
|
||||
return {"type": TYPE_DO, "steps": []}
|
||||
|
||||
# Fast path: no special characters at all
|
||||
if ';' not in source and '{' not in source and '&' not in source:
|
||||
return {"type": TYPE_DO, "steps": [source]}
|
||||
|
||||
# Fast path: simple sequential (only semicolons, no braces or ampersands)
|
||||
if '{' not in source and '&' not in source:
|
||||
# Split on semicolon, strip each part
|
||||
parts: List[str] = [p.strip() for p in source.split(';') if p.strip()]
|
||||
if parts:
|
||||
return {"type": TYPE_DO, "steps": parts}
|
||||
return {"type": TYPE_DO, "steps": []}
|
||||
|
||||
# Fall back to full parse
|
||||
return parse(source)
|
||||
|
||||
|
||||
def compile_to_commands_fast(source: str) -> CommandList:
|
||||
"""Compile directly to commands list, optimized path.
|
||||
|
||||
Skips intermediate representations when possible.
|
||||
"""
|
||||
source = source.strip()
|
||||
if not source:
|
||||
return []
|
||||
|
||||
# Fast path: single command
|
||||
if ';' not in source and '{' not in source and '&' not in source:
|
||||
return [source]
|
||||
|
||||
# Fast path: simple sequential
|
||||
if '{' not in source and '&' not in source:
|
||||
return [p.strip() for p in source.split(';') if p.strip()]
|
||||
|
||||
# Full compilation
|
||||
ast: ASTDict = parse(source)
|
||||
return to_commands(ast)
|
||||
|
||||
|
||||
# Aliases for backwards compatibility
|
||||
_tokenize = _tokenize_fast
|
||||
_tokenize_parallel = _tokenize_parallel_fast
|
||||
|
||||
@@ -4,8 +4,8 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "hanzo-tools-shell"
|
||||
version = "0.4.0"
|
||||
description = "Shell/command execution tools - zsh (with DAG), ps, npx, uvx, open, curl, jq, wget"
|
||||
version = "0.4.1"
|
||||
description = "Shell/command execution tools - dag, zsh (with shellflow DSL), ps, npx, uvx, open, curl, jq, wget"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.12"
|
||||
license = { text = "MIT" }
|
||||
|
||||
@@ -0,0 +1,427 @@
|
||||
"""Comprehensive tests for Shellflow DSL parser and execution."""
|
||||
|
||||
import pytest
|
||||
import asyncio
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
from hanzo_tools.shell.shellflow import (
|
||||
parse,
|
||||
to_sexp,
|
||||
from_sexp,
|
||||
to_commands,
|
||||
compile,
|
||||
render_ascii,
|
||||
_tokenize,
|
||||
_normalize,
|
||||
)
|
||||
|
||||
|
||||
class TestTokenizer:
|
||||
"""Tests for shellflow tokenization."""
|
||||
|
||||
def test_simple_command(self):
|
||||
assert _tokenize("ls -la") == ["ls -la"]
|
||||
|
||||
def test_sequential_commands(self):
|
||||
tokens = _tokenize("A ; B ; C")
|
||||
assert tokens == ["A", ";", "B", ";", "C"]
|
||||
|
||||
def test_parallel_block(self):
|
||||
tokens = _tokenize("{ A & B & C }")
|
||||
assert tokens == ["{ A & B & C }"]
|
||||
|
||||
def test_mixed_syntax(self):
|
||||
tokens = _tokenize("A ; { B & C } ; D")
|
||||
assert tokens == ["A", ";", "{ B & C }", ";", "D"]
|
||||
|
||||
def test_nested_braces(self):
|
||||
tokens = _tokenize("A ; { B ; { C & D } } ; E")
|
||||
assert tokens == ["A", ";", "{ B ; { C & D } }", ";", "E"]
|
||||
|
||||
def test_empty_input(self):
|
||||
assert _tokenize("") == []
|
||||
assert _tokenize(" ") == []
|
||||
|
||||
def test_commands_with_semicolons_in_strings(self):
|
||||
# Commands containing semicolons in quoted strings
|
||||
tokens = _tokenize('echo "a;b" ; echo c')
|
||||
assert len(tokens) == 3 # echo "a;b", ;, echo c
|
||||
|
||||
def test_commands_with_pipes(self):
|
||||
tokens = _tokenize("cat file | grep pattern ; echo done")
|
||||
assert tokens == ["cat file | grep pattern", ";", "echo done"]
|
||||
|
||||
def test_commands_with_and_or(self):
|
||||
tokens = _tokenize("cmd1 && cmd2 || cmd3 ; cmd4")
|
||||
assert tokens == ["cmd1 && cmd2 || cmd3", ";", "cmd4"]
|
||||
|
||||
|
||||
class TestParser:
|
||||
"""Tests for shellflow parsing to AST."""
|
||||
|
||||
def test_simple_command(self):
|
||||
ast = parse("ls -la")
|
||||
assert ast == {"type": "do", "steps": ["ls -la"]}
|
||||
|
||||
def test_sequential(self):
|
||||
ast = parse("A ; B ; C")
|
||||
assert ast == {"type": "do", "steps": ["A", "B", "C"]}
|
||||
|
||||
def test_parallel(self):
|
||||
ast = parse("{ A & B & C }")
|
||||
assert ast == {"type": "all", "steps": ["A", "B", "C"]}
|
||||
|
||||
def test_mixed_dag(self):
|
||||
ast = parse("setup ; { task1 & task2 } ; cleanup")
|
||||
assert ast == {
|
||||
"type": "do",
|
||||
"steps": [
|
||||
"setup",
|
||||
{"type": "all", "steps": ["task1", "task2"]},
|
||||
"cleanup",
|
||||
],
|
||||
}
|
||||
|
||||
def test_complex_dag(self):
|
||||
ast = parse("A ; { B & C & D } ; { E & F } ; G")
|
||||
assert ast == {
|
||||
"type": "do",
|
||||
"steps": [
|
||||
"A",
|
||||
{"type": "all", "steps": ["B", "C", "D"]},
|
||||
{"type": "all", "steps": ["E", "F"]},
|
||||
"G",
|
||||
],
|
||||
}
|
||||
|
||||
def test_empty_input(self):
|
||||
ast = parse("")
|
||||
assert ast == {"type": "do", "steps": []}
|
||||
|
||||
def test_whitespace_handling(self):
|
||||
ast = parse(" A ; B ; C ")
|
||||
assert ast == {"type": "do", "steps": ["A", "B", "C"]}
|
||||
|
||||
def test_single_parallel_task(self):
|
||||
ast = parse("{ A }")
|
||||
# Single item in parallel should be normalized
|
||||
assert ast["steps"] == ["A"] or ast == "A"
|
||||
|
||||
def test_real_world_example(self):
|
||||
src = "mkdir -p dist ; { cp a.txt dist/ & cp b.txt dist/ & cp c.txt dist/ } ; zip -r out.zip dist/"
|
||||
ast = parse(src)
|
||||
assert ast["type"] == "do"
|
||||
assert len(ast["steps"]) == 3
|
||||
assert ast["steps"][0] == "mkdir -p dist"
|
||||
assert ast["steps"][1]["type"] == "all"
|
||||
assert len(ast["steps"][1]["steps"]) == 3
|
||||
assert ast["steps"][2] == "zip -r out.zip dist/"
|
||||
|
||||
|
||||
class TestNormalization:
|
||||
"""Tests for AST normalization."""
|
||||
|
||||
def test_flatten_nested_do(self):
|
||||
ast = {"type": "do", "steps": [{"type": "do", "steps": ["A", "B"]}, "C"]}
|
||||
normalized = _normalize(ast)
|
||||
assert normalized == {"type": "do", "steps": ["A", "B", "C"]}
|
||||
|
||||
def test_flatten_nested_all(self):
|
||||
ast = {"type": "all", "steps": [{"type": "all", "steps": ["A", "B"]}, "C"]}
|
||||
normalized = _normalize(ast)
|
||||
assert normalized == {"type": "all", "steps": ["A", "B", "C"]}
|
||||
|
||||
def test_singleton_do(self):
|
||||
ast = {"type": "do", "steps": ["A"]}
|
||||
normalized = _normalize(ast)
|
||||
# Singleton should be kept as-is for consistency
|
||||
assert "A" in str(normalized)
|
||||
|
||||
def test_empty_steps(self):
|
||||
ast = {"type": "do", "steps": []}
|
||||
normalized = _normalize(ast)
|
||||
assert normalized == {"type": "do", "steps": []}
|
||||
|
||||
def test_deeply_nested(self):
|
||||
ast = {
|
||||
"type": "do",
|
||||
"steps": [
|
||||
{"type": "do", "steps": [{"type": "do", "steps": ["A"]}]},
|
||||
"B",
|
||||
],
|
||||
}
|
||||
normalized = _normalize(ast)
|
||||
assert "A" in str(normalized)
|
||||
assert "B" in str(normalized)
|
||||
|
||||
|
||||
class TestSexpConversion:
|
||||
"""Tests for S-expression conversion."""
|
||||
|
||||
def test_to_sexp_simple(self):
|
||||
ast = {"type": "do", "steps": ["A", "B", "C"]}
|
||||
sexp = to_sexp(ast)
|
||||
assert sexp == ["do", "A", "B", "C"]
|
||||
|
||||
def test_to_sexp_nested(self):
|
||||
ast = {
|
||||
"type": "do",
|
||||
"steps": ["A", {"type": "all", "steps": ["B", "C"]}, "D"],
|
||||
}
|
||||
sexp = to_sexp(ast)
|
||||
assert sexp == ["do", "A", ["all", "B", "C"], "D"]
|
||||
|
||||
def test_from_sexp_simple(self):
|
||||
sexp = ["do", "A", "B", "C"]
|
||||
ast = from_sexp(sexp)
|
||||
assert ast == {"type": "do", "steps": ["A", "B", "C"]}
|
||||
|
||||
def test_from_sexp_nested(self):
|
||||
sexp = ["do", "A", ["all", "B", "C"], "D"]
|
||||
ast = from_sexp(sexp)
|
||||
assert ast == {
|
||||
"type": "do",
|
||||
"steps": ["A", {"type": "all", "steps": ["B", "C"]}, "D"],
|
||||
}
|
||||
|
||||
def test_roundtrip(self):
|
||||
original = {"type": "do", "steps": ["A", {"type": "all", "steps": ["B", "C"]}, "D"]}
|
||||
sexp = to_sexp(original)
|
||||
recovered = from_sexp(sexp)
|
||||
assert recovered == original
|
||||
|
||||
|
||||
class TestToCommands:
|
||||
"""Tests for conversion to executor commands format."""
|
||||
|
||||
def test_sequential(self):
|
||||
ast = {"type": "do", "steps": ["A", "B", "C"]}
|
||||
cmds = to_commands(ast)
|
||||
assert cmds == ["A", "B", "C"]
|
||||
|
||||
def test_parallel(self):
|
||||
ast = {"type": "all", "steps": ["A", "B", "C"]}
|
||||
cmds = to_commands(ast)
|
||||
assert cmds == [["A", "B", "C"]]
|
||||
|
||||
def test_mixed(self):
|
||||
ast = {
|
||||
"type": "do",
|
||||
"steps": ["A", {"type": "all", "steps": ["B", "C"]}, "D"],
|
||||
}
|
||||
cmds = to_commands(ast)
|
||||
assert cmds == ["A", ["B", "C"], "D"]
|
||||
|
||||
def test_complex_dag(self):
|
||||
ast = {
|
||||
"type": "do",
|
||||
"steps": [
|
||||
"setup",
|
||||
{"type": "all", "steps": ["task1", "task2", "task3"]},
|
||||
{"type": "all", "steps": ["cleanup1", "cleanup2"]},
|
||||
"done",
|
||||
],
|
||||
}
|
||||
cmds = to_commands(ast)
|
||||
assert cmds == [
|
||||
"setup",
|
||||
["task1", "task2", "task3"],
|
||||
["cleanup1", "cleanup2"],
|
||||
"done",
|
||||
]
|
||||
|
||||
|
||||
class TestCompile:
|
||||
"""Tests for compile convenience function."""
|
||||
|
||||
def test_compile_to_ast(self):
|
||||
result = compile("A ; B ; C", format="ast")
|
||||
assert result == {"type": "do", "steps": ["A", "B", "C"]}
|
||||
|
||||
def test_compile_to_sexp(self):
|
||||
result = compile("A ; B ; C", format="sexp")
|
||||
assert result == ["do", "A", "B", "C"]
|
||||
|
||||
def test_compile_to_commands(self):
|
||||
result = compile("A ; { B & C } ; D", format="commands")
|
||||
assert result == ["A", ["B", "C"], "D"]
|
||||
|
||||
def test_compile_to_ascii(self):
|
||||
result = compile("A ; B", format="ascii")
|
||||
assert "[do]" in result
|
||||
assert "A" in result
|
||||
assert "B" in result
|
||||
|
||||
def test_compile_invalid_format(self):
|
||||
with pytest.raises(ValueError):
|
||||
compile("A", format="invalid")
|
||||
|
||||
|
||||
class TestRenderAscii:
|
||||
"""Tests for ASCII rendering."""
|
||||
|
||||
def test_simple(self):
|
||||
ast = {"type": "do", "steps": ["A", "B"]}
|
||||
output = render_ascii(ast)
|
||||
assert "[do]" in output
|
||||
assert "A" in output
|
||||
assert "B" in output
|
||||
|
||||
def test_nested(self):
|
||||
ast = {
|
||||
"type": "do",
|
||||
"steps": ["A", {"type": "all", "steps": ["B", "C"]}, "D"],
|
||||
}
|
||||
output = render_ascii(ast)
|
||||
assert "[do]" in output
|
||||
assert "[all]" in output
|
||||
|
||||
|
||||
class TestEdgeCases:
|
||||
"""Edge case and error handling tests."""
|
||||
|
||||
def test_many_sequential(self):
|
||||
src = " ; ".join([f"cmd{i}" for i in range(100)])
|
||||
ast = parse(src)
|
||||
assert len(ast["steps"]) == 100
|
||||
|
||||
def test_many_parallel(self):
|
||||
src = "{ " + " & ".join([f"cmd{i}" for i in range(100)]) + " }"
|
||||
ast = parse(src)
|
||||
assert len(ast["steps"]) == 100
|
||||
|
||||
def test_deeply_nested_parallel(self):
|
||||
src = "A ; { B & { C & D } } ; E"
|
||||
ast = parse(src)
|
||||
# Should handle nested parallel blocks
|
||||
assert ast["type"] == "do"
|
||||
|
||||
def test_unicode_commands(self):
|
||||
ast = parse("echo 你好 ; echo мир ; echo 🚀")
|
||||
assert len(ast["steps"]) == 3
|
||||
|
||||
def test_long_commands(self):
|
||||
long_cmd = "echo " + "x" * 10000
|
||||
ast = parse(long_cmd)
|
||||
assert long_cmd in ast["steps"][0] or ast["steps"][0] == long_cmd
|
||||
|
||||
|
||||
class TestPerformance:
|
||||
"""Performance benchmarks."""
|
||||
|
||||
def test_parse_speed_simple(self):
|
||||
"""Should parse simple commands very fast."""
|
||||
src = "ls -la"
|
||||
start = time.perf_counter()
|
||||
for _ in range(10000):
|
||||
parse(src)
|
||||
elapsed = time.perf_counter() - start
|
||||
ops_per_sec = 10000 / elapsed
|
||||
print(f"\nSimple parse: {ops_per_sec:.0f} ops/sec")
|
||||
assert ops_per_sec > 10000 # At least 10k ops/sec
|
||||
|
||||
def test_parse_speed_mixed(self):
|
||||
"""Should parse mixed DAGs reasonably fast."""
|
||||
src = "mkdir -p dist ; { cp a dist/ & cp b dist/ & cp c dist/ } ; zip out.zip dist/"
|
||||
start = time.perf_counter()
|
||||
for _ in range(1000):
|
||||
parse(src)
|
||||
elapsed = time.perf_counter() - start
|
||||
ops_per_sec = 1000 / elapsed
|
||||
print(f"\nMixed DAG parse: {ops_per_sec:.0f} ops/sec")
|
||||
assert ops_per_sec > 1000 # At least 1k ops/sec
|
||||
|
||||
def test_compile_speed(self):
|
||||
"""Should compile to commands fast."""
|
||||
src = "A ; { B & C & D } ; E"
|
||||
start = time.perf_counter()
|
||||
for _ in range(10000):
|
||||
compile(src, format="commands")
|
||||
elapsed = time.perf_counter() - start
|
||||
ops_per_sec = 10000 / elapsed
|
||||
print(f"\nFull compile: {ops_per_sec:.0f} ops/sec")
|
||||
assert ops_per_sec > 5000 # At least 5k ops/sec
|
||||
|
||||
def test_large_dag(self):
|
||||
"""Should handle large DAGs efficiently."""
|
||||
# 100 sequential with 10 parallel each
|
||||
parts = []
|
||||
for i in range(100):
|
||||
parallel = "{ " + " & ".join([f"task{i}_{j}" for j in range(10)]) + " }"
|
||||
parts.append(parallel)
|
||||
src = " ; ".join(parts)
|
||||
|
||||
start = time.perf_counter()
|
||||
ast = parse(src)
|
||||
cmds = to_commands(ast)
|
||||
elapsed = time.perf_counter() - start
|
||||
|
||||
print(f"\nLarge DAG (100x10): {elapsed*1000:.2f}ms")
|
||||
assert elapsed < 0.1 # Should complete in under 100ms
|
||||
assert len(cmds) == 100
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestIntegration:
|
||||
"""Integration tests with actual shell execution."""
|
||||
|
||||
async def test_execute_sequential(self):
|
||||
from hanzo_tools.shell import ZshTool
|
||||
|
||||
zsh = ZshTool()
|
||||
result = await zsh.call(None, command="echo A ; echo B ; echo C")
|
||||
assert "A" in result
|
||||
assert "B" in result
|
||||
assert "C" in result
|
||||
|
||||
async def test_execute_parallel(self):
|
||||
from hanzo_tools.shell import ZshTool
|
||||
|
||||
zsh = ZshTool()
|
||||
result = await zsh.call(None, command="{ echo A & echo B & echo C }")
|
||||
assert "A" in result
|
||||
assert "B" in result
|
||||
assert "C" in result
|
||||
|
||||
async def test_execute_mixed(self):
|
||||
from hanzo_tools.shell import ZshTool
|
||||
|
||||
zsh = ZshTool()
|
||||
result = await zsh.call(None, command="echo start ; { echo A & echo B } ; echo end")
|
||||
assert "start" in result
|
||||
assert "A" in result
|
||||
assert "B" in result
|
||||
assert "end" in result
|
||||
|
||||
async def test_execute_with_bash(self):
|
||||
from hanzo_tools.shell import BashTool
|
||||
|
||||
bash = BashTool()
|
||||
result = await bash.call(None, command="echo start ; { echo A & echo B } ; echo end")
|
||||
assert "start" in result
|
||||
assert "end" in result
|
||||
|
||||
async def test_parallel_execution_time(self):
|
||||
"""Parallel should be faster than sequential for sleep commands."""
|
||||
from hanzo_tools.shell import ZshTool
|
||||
|
||||
zsh = ZshTool()
|
||||
|
||||
# Parallel: should take ~0.1s
|
||||
start = time.perf_counter()
|
||||
await zsh.call(None, command="{ sleep 0.1 & sleep 0.1 & sleep 0.1 }")
|
||||
parallel_time = time.perf_counter() - start
|
||||
|
||||
# Sequential: should take ~0.3s
|
||||
start = time.perf_counter()
|
||||
await zsh.call(None, command="sleep 0.1 ; sleep 0.1 ; sleep 0.1")
|
||||
sequential_time = time.perf_counter() - start
|
||||
|
||||
print(f"\nParallel: {parallel_time:.2f}s, Sequential: {sequential_time:.2f}s")
|
||||
# Parallel should be at least 2x faster
|
||||
assert parallel_time < sequential_time * 0.7
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v", "-s"])
|
||||
Reference in New Issue
Block a user