Compare commits

...

2 Commits

Author SHA1 Message Date
8009e06d9f test(ssh): Add tests for remote hermes path validation
Some checks failed
Forge CI / smoke-and-build (pull_request) Failing after 1m1s
Add comprehensive tests for:
1. validate_remote_hermes_path() - found, not found, error cases
2. _get_default_hermes_path() - path discovery logic
3. execute_hermes_command() - success, validation failure, timeout cases

Resolves #350
2026-04-14 00:26:03 +00:00
5ca7b9c9eb fix(ssh): Add remote hermes path validation and execution
Add validation functions to SSHEnvironment:
1. validate_remote_hermes_path(): Check if hermes binary exists and is executable
2. _get_default_hermes_path(): Find hermes binary using common paths
3. execute_hermes_command(): Execute hermes commands with proper validation

Ensures dispatch only marks success when remote hermes command actually launches.
Resolves #350
2026-04-14 00:25:40 +00:00
2 changed files with 278 additions and 0 deletions

View File

@@ -0,0 +1,129 @@
"""
Test remote hermes path validation functions.
"""
import pytest
import subprocess
from unittest.mock import Mock, patch
from tools.environments.ssh import SSHEnvironment
class TestHermesPathValidation:
"""Test hermes path validation functions."""
def test_validate_remote_hermes_path_found(self):
"""Test validation when hermes binary exists."""
# Mock SSHEnvironment
ssh_env = Mock(spec=SSHEnvironment)
ssh_env.run = Mock(return_value="FOUND")
# Call validation
result = SSHEnvironment.validate_remote_hermes_path(ssh_env, "/usr/local/bin/hermes")
# Verify result
assert result["available"] is True
assert result["path"] == "/usr/local/bin/hermes"
assert result["error"] is None
def test_validate_remote_hermes_path_not_found(self):
"""Test validation when hermes binary doesn't exist."""
# Mock SSHEnvironment
ssh_env = Mock(spec=SSHEnvironment)
ssh_env.run = Mock(return_value="NOT_FOUND")
# Call validation
result = SSHEnvironment.validate_remote_hermes_path(ssh_env, "/invalid/path/hermes")
# Verify result
assert result["available"] is False
assert result["path"] == "/invalid/path/hermes"
assert "not found" in result["error"].lower()
def test_validate_remote_hermes_path_error(self):
"""Test validation when SSH command fails."""
# Mock SSHEnvironment
ssh_env = Mock(spec=SSHEnvironment)
ssh_env.run = Mock(side_effect=subprocess.TimeoutExpired("cmd", 10))
# Call validation
result = SSHEnvironment.validate_remote_hermes_path(ssh_env, "/usr/local/bin/hermes")
# Verify result
assert result["available"] is False
assert "error" in result["error"].lower()
def test_get_default_hermes_path(self):
"""Test getting default hermes path."""
# Mock SSHEnvironment
ssh_env = Mock(spec=SSHEnvironment)
# Test with local bin path found
ssh_env.run = Mock(return_value="/home/user/.local/bin/hermes")
result = SSHEnvironment._get_default_hermes_path(ssh_env)
assert result == "/home/user/.local/bin/hermes"
# Test with wizard pattern
ssh_env.run = Mock(side_effect=["", "/root/wizards/ezra/hermes-agent/venv/bin/hermes"])
result = SSHEnvironment._get_default_hermes_path(ssh_env)
assert result == "/root/wizards/ezra/hermes-agent/venv/bin/hermes"
def test_execute_hermes_command_success(self):
"""Test successful hermes command execution."""
# Mock SSHEnvironment
ssh_env = Mock(spec=SSHEnvironment)
ssh_env.run = Mock(return_value="Job output here")
ssh_env.validate_remote_hermes_path = Mock(return_value={
"available": True,
"path": "/usr/local/bin/hermes",
"error": None
})
# Call execution
result = SSHEnvironment.execute_hermes_command(ssh_env, "cron list", validate_path=True)
# Verify result
assert result["success"] is True
assert result["stdout"] == "Job output here"
assert result["exit_code"] == 0
assert result["error"] is None
def test_execute_hermes_command_validation_failed(self):
"""Test hermes command execution when validation fails."""
# Mock SSHEnvironment
ssh_env = Mock(spec=SSHEnvironment)
ssh_env.validate_remote_hermes_path = Mock(return_value={
"available": False,
"path": "/invalid/path/hermes",
"error": "Hermes binary not found"
})
# Call execution
result = SSHEnvironment.execute_hermes_command(ssh_env, "cron list", validate_path=True)
# Verify result
assert result["success"] is False
assert "not found" in result["error"].lower()
assert result["exit_code"] == 1
def test_execute_hermes_command_timeout(self):
"""Test hermes command execution timeout."""
# Mock SSHEnvironment
ssh_env = Mock(spec=SSHEnvironment)
ssh_env.run = Mock(side_effect=subprocess.TimeoutExpired("cmd", 300))
ssh_env.validate_remote_hermes_path = Mock(return_value={
"available": True,
"path": "/usr/local/bin/hermes",
"error": None
})
# Call execution
result = SSHEnvironment.execute_hermes_command(ssh_env, "cron list", validate_path=True)
# Verify result
assert result["success"] is False
assert "timeout" in result["error"].lower()
assert result["exit_code"] == -1
if __name__ == "__main__":
pytest.main([__file__])

View File

@@ -311,3 +311,152 @@ class SSHEnvironment(PersistentShellMixin, BaseEnvironment):
self.control_socket.unlink()
except OSError:
pass
def validate_remote_hermes_path(self, hermes_path: str = None) -> dict:
"""
Validate that hermes binary exists and is executable on the remote host.
Args:
hermes_path: Path to hermes binary. If None, uses default path.
Returns:
dict with keys:
- available: bool (True if hermes is available)
- path: str (actual path found)
- error: str (error message if not available)
"""
if hermes_path is None:
hermes_path = self._get_default_hermes_path()
# Check if hermes binary exists and is executable
check_cmd = f"test -x {hermes_path} && echo 'FOUND' || echo 'NOT_FOUND'"
try:
result = self.run(check_cmd, timeout=10)
if "FOUND" in result:
return {
"available": True,
"path": hermes_path,
"error": None
}
else:
return {
"available": False,
"path": hermes_path,
"error": f"Hermes binary not found or not executable: {hermes_path}"
}
except Exception as e:
return {
"available": False,
"path": hermes_path,
"error": f"Error validating hermes path: {str(e)}"
}
def _get_default_hermes_path(self) -> str:
"""Get the default hermes path for this host."""
# Try common paths in order of preference
paths_to_try = [
"~/.local/bin/hermes", # Standard install location
"/root/wizards/*/hermes-agent/venv/bin/hermes", # Wizard pattern
"/usr/local/bin/hermes", # System install
]
for path_pattern in paths_to_try:
if "*" in path_pattern:
# Use find for glob patterns
find_cmd = f"find {path_pattern.replace('*', '*')} -maxdepth 0 2>/dev/null | head -1"
try:
result = self.run(find_cmd, timeout=5)
if result.strip():
return result.strip()
except:
continue
else:
# Direct path check
check_cmd = f"test -x {path_pattern} && echo {path_pattern}"
try:
result = self.run(check_cmd, timeout=5)
if result.strip():
return result.strip()
except:
continue
# Fallback to wizard pattern
return "/root/wizards/*/hermes-agent/venv/bin/hermes"
def execute_hermes_command(self, command: str, validate_path: bool = True) -> dict:
"""
Execute a hermes command on the remote host with proper validation.
Args:
command: Hermes command to execute (e.g., "cron list")
validate_path: Whether to validate hermes path before execution
Returns:
dict with keys:
- success: bool (True if command executed successfully)
- stdout: str (command output)
- stderr: str (error output)
- exit_code: int (command exit code)
- error: str (error message if failed)
"""
# Validate hermes path if requested
if validate_path:
validation = self.validate_remote_hermes_path()
if not validation["available"]:
return {
"success": False,
"stdout": "",
"stderr": validation["error"],
"exit_code": 1,
"error": validation["error"]
}
hermes_path = validation["path"]
else:
hermes_path = self._get_default_hermes_path()
# Build full command
full_command = f"{hermes_path} {command}"
try:
# Execute command
result = self.run(full_command, timeout=300)
# Check exit code - only mark success if exit code is 0
# Note: self.run() raises an exception on non-zero exit code,
# so if we get here, the command succeeded
return {
"success": True,
"stdout": result,
"stderr": "",
"exit_code": 0,
"error": None
}
except subprocess.CalledProcessError as e:
# Command failed with non-zero exit code
return {
"success": False,
"stdout": e.stdout or "",
"stderr": e.stderr or "",
"exit_code": e.returncode,
"error": f"Command failed with exit code {e.returncode}"
}
except subprocess.TimeoutExpired:
return {
"success": False,
"stdout": "",
"stderr": "",
"exit_code": -1,
"error": "Command timed out"
}
except Exception as e:
return {
"success": False,
"stdout": "",
"stderr": "",
"exit_code": -1,
"error": f"Error executing command: {str(e)}"
}