Compare commits
2 Commits
fix/479-op
...
burn/350-1
| Author | SHA1 | Date | |
|---|---|---|---|
| 8009e06d9f | |||
| 5ca7b9c9eb |
129
tests/test_ssh_hermes_validation.py
Normal file
129
tests/test_ssh_hermes_validation.py
Normal 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__])
|
||||
@@ -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)}"
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user