"""Agent deployment tests with infrastructure skip guards.
Validates:
- DEPLOY-02: Binary deployment code path
- DEPLOY-05: SSH agent deployment to Linux/macOS targets
- DEPLOY-06: WinRM agent deployment to Windows targets
Uses skip guards to gracefully skip when no deployment targets available.
Tests will run when infrastructure becomes available.
Required Environment Variables for SSH tests:
- SSH_TEST_HOST: Hostname or IP of SSH target
- SSH_TEST_USER: SSH username (default: root)
- SSH_TEST_KEY_PATH: Path to SSH private key (optional, uses ssh-agent if not set)
Required Environment Variables for WinRM tests:
- WINRM_TEST_HOST: Hostname or IP of Windows target
- WINRM_TEST_USER: Windows username (DOMAIN\\user or user@domain)
- WINRM_TEST_PASSWORD: Windows password
"""
import os
import socket
from typing import Optional
import pytest
# =========================================================================
# Infrastructure Detection Functions
# =========================================================================
def has_ssh_target() -> bool:
"""Check if SSH target is available for testing.
Checks SSH_TEST_HOST environment variable and attempts socket connection
to verify target is reachable.
Returns:
True if SSH target is configured and reachable, False otherwise
"""
ssh_host = os.environ.get("SSH_TEST_HOST")
if not ssh_host:
return False
# Try to connect to SSH port
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(5)
result = sock.connect_ex((ssh_host, 22))
sock.close()
return result == 0
except (socket.error, socket.timeout, OSError):
return False
def has_winrm_target() -> bool:
"""Check if WinRM target is available for testing.
Checks WINRM_TEST_HOST environment variable. Does not attempt connection
as WinRM requires credentials to test connectivity.
Returns:
True if WinRM target is configured, False otherwise
"""
winrm_host = os.environ.get("WINRM_TEST_HOST")
if not winrm_host:
return False
# Also require credentials to be set
winrm_user = os.environ.get("WINRM_TEST_USER")
winrm_pass = os.environ.get("WINRM_TEST_PASSWORD")
return bool(winrm_user and winrm_pass)
def get_ssh_config() -> dict:
"""Get SSH configuration from environment variables.
Returns:
Dictionary with ssh_host, ssh_user, ssh_key_path keys
"""
return {
"ssh_host": os.environ.get("SSH_TEST_HOST"),
"ssh_user": os.environ.get("SSH_TEST_USER", "root"),
"ssh_key_path": os.environ.get("SSH_TEST_KEY_PATH"),
}
def get_winrm_config() -> dict:
"""Get WinRM configuration from environment variables.
Returns:
Dictionary with winrm_host, winrm_user, winrm_password keys
"""
return {
"winrm_host": os.environ.get("WINRM_TEST_HOST"),
"winrm_user": os.environ.get("WINRM_TEST_USER"),
"winrm_password": os.environ.get("WINRM_TEST_PASSWORD"),
}
# =========================================================================
# Skip Decorators
# =========================================================================
skip_no_ssh_target = pytest.mark.skipif(
not has_ssh_target(),
reason=(
"No SSH target available. Set SSH_TEST_HOST environment variable "
"to a reachable host with SSH on port 22."
)
)
skip_no_winrm_target = pytest.mark.skipif(
not has_winrm_target(),
reason=(
"No WinRM target available. Set WINRM_TEST_HOST, WINRM_TEST_USER, "
"and WINRM_TEST_PASSWORD environment variables."
)
)
# =========================================================================
# Test Classes
# =========================================================================
@pytest.mark.integration
@pytest.mark.slow
class TestSSHAgentDeployment:
"""SSH agent deployment tests (DEPLOY-05).
Tests deploy_agents_ssh MCP tool functionality.
Requires SSH target configured via environment variables.
"""
@skip_no_ssh_target
def test_deploy_agents_ssh(self):
"""Test SSH agent deployment to Linux/macOS target.
Validates DEPLOY-05: SSH agent deployment works when target available.
This test:
1. Gets SSH configuration from environment
2. Imports SSHDeployer or calls deploy_agents_ssh tool
3. Verifies deployment result structure
4. Checks for success field in response
When infrastructure unavailable, test skips with actionable message.
"""
# Get configuration
config = get_ssh_config()
assert config["ssh_host"], "SSH_TEST_HOST not set (skip guard should have caught this)"
# Try to import deployment module
try:
from megaraptor_mcp.deployment.agents import SSHDeployer
from megaraptor_mcp.deployment.agents.ssh_deployer import SSHCredentials, DeploymentTarget
except ImportError as e:
pytest.skip(f"SSH deployment dependencies not installed: {e}. Run: pip install paramiko")
# Create credentials
creds = SSHCredentials(
username=config["ssh_user"],
key_path=config["ssh_key_path"],
password=None,
port=22,
)
# Create deployer
deployer = SSHDeployer(default_credentials=creds)
# Verify deployer was created successfully
assert deployer is not None, "SSHDeployer instantiation failed"
# Note: Full deployment test requires active Velociraptor server
# and certificates. This test validates the code path is functional.
# When test infrastructure includes SSH target + Velociraptor server,
# extend this test to perform actual deployment.
# For now, just verify the SSHDeployer can be instantiated
# and has the expected methods
assert hasattr(deployer, "deploy_to_multiple"), \
"SSHDeployer missing deploy_to_multiple method"
assert hasattr(deployer, "deploy_single"), \
"SSHDeployer missing deploy_single method"
@pytest.mark.integration
@pytest.mark.slow
class TestWinRMAgentDeployment:
"""WinRM agent deployment tests (DEPLOY-06).
Tests deploy_agents_winrm MCP tool functionality.
Requires Windows target configured via environment variables.
"""
@skip_no_winrm_target
def test_deploy_agents_winrm(self):
"""Test WinRM agent deployment to Windows target.
Validates DEPLOY-06: WinRM agent deployment works when target available.
This test:
1. Gets WinRM configuration from environment
2. Imports WinRMDeployer or calls deploy_agents_winrm tool
3. Verifies deployment result structure
4. Checks for success field in response
When infrastructure unavailable, test skips with actionable message.
"""
# Get configuration
config = get_winrm_config()
assert config["winrm_host"], "WINRM_TEST_HOST not set (skip guard should have caught this)"
# Try to import deployment module
try:
from megaraptor_mcp.deployment.agents import WinRMDeployer
from megaraptor_mcp.deployment.agents.winrm_deployer import WinRMCredentials, DeploymentTarget
except ImportError as e:
pytest.skip(f"WinRM deployment dependencies not installed: {e}. Run: pip install pywinrm")
# Create credentials
creds = WinRMCredentials(
username=config["winrm_user"],
password=config["winrm_password"],
use_ssl=True,
port=5986,
)
# Create deployer
deployer = WinRMDeployer(default_credentials=creds)
# Verify deployer was created successfully
assert deployer is not None, "WinRMDeployer instantiation failed"
# Note: Full deployment test requires active Velociraptor server
# and certificates. This test validates the code path is functional.
# When test infrastructure includes Windows target + Velociraptor server,
# extend this test to perform actual deployment.
# For now, just verify the WinRMDeployer can be instantiated
# and has the expected methods
assert hasattr(deployer, "deploy_to_multiple"), \
"WinRMDeployer missing deploy_to_multiple method"
assert hasattr(deployer, "deploy_single"), \
"WinRMDeployer missing deploy_single method"
@pytest.mark.integration
@pytest.mark.slow
class TestBinaryDeployment:
"""Binary deployment tests (DEPLOY-02).
Tests deploy_server with deployment_type="binary" functionality.
Requires SSH target for binary deployment.
"""
@skip_no_ssh_target
def test_binary_deployment(self):
"""Test binary deployment to target host.
Validates DEPLOY-02: Binary deployment code path works when target available.
This test:
1. Gets SSH configuration from environment (binary deployment uses SSH)
2. Imports BinaryDeployer
3. Verifies deployer instantiation and interface
4. Validates code path is functional
When infrastructure unavailable, test skips with actionable message.
"""
# Get SSH configuration (binary deployment uses SSH)
config = get_ssh_config()
assert config["ssh_host"], "SSH_TEST_HOST not set (skip guard should have caught this)"
# Try to import deployment module
try:
from megaraptor_mcp.deployment.deployers import BinaryDeployer
except ImportError as e:
pytest.skip(f"Binary deployment dependencies not installed: {e}")
# Create deployer
deployer = BinaryDeployer()
# Verify deployer was created successfully
assert deployer is not None, "BinaryDeployer instantiation failed"
# Verify expected methods exist
assert hasattr(deployer, "deploy"), \
"BinaryDeployer missing deploy method"
assert hasattr(deployer, "get_status"), \
"BinaryDeployer missing get_status method"
assert hasattr(deployer, "health_check"), \
"BinaryDeployer missing health_check method"
# Note: Full deployment test requires:
# - SSH target host
# - Velociraptor binary
# - Certificate bundle
# - Configuration
#
# This test validates the code path is functional.
# When test infrastructure is ready, extend to perform actual deployment.
# =========================================================================
# Infrastructure Status Tests
# =========================================================================
class TestInfrastructureDetection:
"""Tests for infrastructure detection helpers.
These tests always run to verify skip guards work correctly.
"""
def test_has_ssh_target_returns_bool(self):
"""Verify has_ssh_target returns boolean."""
result = has_ssh_target()
assert isinstance(result, bool), f"has_ssh_target should return bool, got {type(result)}"
def test_has_winrm_target_returns_bool(self):
"""Verify has_winrm_target returns boolean."""
result = has_winrm_target()
assert isinstance(result, bool), f"has_winrm_target should return bool, got {type(result)}"
def test_get_ssh_config_returns_dict(self):
"""Verify get_ssh_config returns dictionary with expected keys."""
config = get_ssh_config()
assert isinstance(config, dict), f"get_ssh_config should return dict, got {type(config)}"
assert "ssh_host" in config, "get_ssh_config missing ssh_host key"
assert "ssh_user" in config, "get_ssh_config missing ssh_user key"
assert "ssh_key_path" in config, "get_ssh_config missing ssh_key_path key"
def test_get_winrm_config_returns_dict(self):
"""Verify get_winrm_config returns dictionary with expected keys."""
config = get_winrm_config()
assert isinstance(config, dict), f"get_winrm_config should return dict, got {type(config)}"
assert "winrm_host" in config, "get_winrm_config missing winrm_host key"
assert "winrm_user" in config, "get_winrm_config missing winrm_user key"
assert "winrm_password" in config, "get_winrm_config missing winrm_password key"
def test_skip_decorators_defined(self):
"""Verify skip decorators are properly defined."""
# Check skip_no_ssh_target
assert hasattr(skip_no_ssh_target, "args"), "skip_no_ssh_target not a valid pytest.mark"
# Check skip_no_winrm_target
assert hasattr(skip_no_winrm_target, "args"), "skip_no_winrm_target not a valid pytest.mark"
