Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import fnmatch
import json
import os
from six.moves import configparser
import testinfra
from testinfra.utils import cached_property
from testinfra.utils import check_ip_address
from testinfra.utils import TemporaryDirectory
__all__ = ['AnsibleRunner']
local = testinfra.get_host('local://')
def get_ansible_config():
fname = os.environ.get('ANSIBLE_CONFIG')
if not fname:
for possible in (
'ansible.cfg',
os.path.join(os.path.expanduser('~'), '.ansible.cfg'),
os.path.join('/', 'etc', 'ansible', 'ansible.cfg'),
):
if os.path.exists(possible):
fname = possible
break
config = configparser.ConfigParser()
if not fname:
return config
def get_ansible_host(config, inventory, host, ssh_config=None,
ssh_identity_file=None):
if is_empty_inventory(inventory):
if host == 'localhost':
return testinfra.get_host('local://')
return None
hostvars = inventory['_meta'].get('hostvars', {}).get(host, {})
connection = hostvars.get('ansible_connection', 'ssh')
if connection not in (
'smart', 'ssh', 'paramiko_ssh', 'local', 'docker', 'lxc', 'lxd',
):
# unhandled connection type, must use force_ansible=True
return None
connection = {
'lxd': 'lxc',
'paramiko_ssh': 'paramiko',
'smart': 'ssh',
}.get(connection, connection)
testinfra_host = hostvars.get('ansible_host', host)
user = hostvars.get('ansible_user')
password = hostvars.get('ansible_ssh_pass')
spec = '{}://'.format(connection)
# Fallback to user:pasword auth when identity file is not used
if user and password and not kwargs.get('ssh_identity_file'):
spec += '{}:{}@'.format(user, password)
elif user:
spec += '{}@'.format(user)
if check_ip_address(testinfra_host) == 6:
spec += '[' + testinfra_host + ']'
else:
spec += testinfra_host
if port:
spec += ':{}'.format(port)
return testinfra.get_host(spec, **kwargs)
def test_ssh_hostspec(hostspec, expected):
backend = testinfra.get_host(hostspec).backend
cmd, cmd_args = backend._build_ssh_command('true')
command = backend.quote(' '.join(cmd), *cmd_args)
assert command == expected
async def on_request_replied(self, req_data, gen_req, resp_or_exp):
# start promoted node
host = testinfra.get_host('ssh://persistent_node' + req_data['node_alias'][4:])
host.run('sudo systemctl start indy-node')
async def on_request_replied(self, req_data, gen_req, resp_or_exp):
# stop demoted node
host = testinfra.get_host('ssh://persistent_node' + req_data['node_alias'][4:])
host.run('sudo systemctl stop indy-node')