Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_op_line_numbers(self):
inventory = make_inventory()
state = State(inventory, Config())
connect_all(state)
# Add op to both hosts
add_op(state, server.shell, 'echo "hi"')
# Add op to just the second host - using the pseudo modules such that
# it replicates a deploy file.
pseudo_state.set(state)
pseudo_host.set(inventory['anotherhost'])
first_pseudo_hash = server.user('anotherhost_user').hash
first_pseudo_call_line = getframeinfo(currentframe()).lineno - 1
# Add op to just the first host - using the pseudo modules such that
# it replicates a deploy file.
pseudo_state.set(state)
pseudo_host.set(inventory['somehost'])
second_pseudo_hash = server.user('somehost_user').hash
second_pseudo_call_line = getframeinfo(currentframe()).lineno - 1
pseudo_state.reset()
pseudo_host.reset()
# Ensure there are two ops
op_order = state.get_op_order()
self.assertEqual(len(op_order), 3)
def _exception(name, e, always_dump=False):
print()
if pseudo_host.isset():
sys.stderr.write('--> [{0}]: {1}: '.format(
colored(pseudo_host.name, attrs=['bold']),
colored(name, 'red', attrs=['bold'])
))
else:
sys.stderr.write('--> {0}: '.format(colored(name, 'red', attrs=['bold'])))
if e:
logger.warning(e)
if arguments.get('debug') or always_dump:
dump_trace(sys.exc_info())
_exit(1)
return self._module is not None
# The current deploy state
sys.modules['pyinfra.pseudo_state'] = sys.modules['pyinfra.state'] = \
pyinfra.pseudo_state = pyinfra.state = \
PseudoModule()
# The current deploy inventory
sys.modules['pyinfra.pseudo_inventory'] = sys.modules['pyinfra.inventory'] = \
pyinfra.pseudo_inventory = pyinfra.inventory = \
PseudoModule()
# The current target host
sys.modules['pyinfra.pseudo_host'] = sys.modules['pyinfra.host'] = \
pyinfra.pseudo_host = pyinfra.host = \
PseudoModule()
directory.
Args:
hosts (string, list): group name or list of hosts to limit this include to
when (bool): indicate whether to trigger operations in this include
'''
if not pyinfra.is_cli:
raise PyinfraError('local.include is only available in CLI mode.')
if not when:
return
if hosts is not False:
hosts = ensure_host_list(hosts, inventory=pseudo_state.inventory)
if pseudo_host not in hosts:
return
if pseudo_state.deploy_dir:
filename = path.join(pseudo_state.deploy_dir, filename)
frameinfo = get_caller_frameinfo()
logger.debug('Including local file: {0}'.format(filename))
try:
# Fixes a circular import because `pyinfra.local` is really a CLI
# only thing (so should be `pyinfra_cli.local`). It is kept here
# to maintain backwards compatability and the nicer public import
# (ideally users never need to import from `pyinfra_cli`).
from pyinfra_cli.config import extract_file_config
def decorated_func(*args, **kwargs):
# Prepare state/host
#
# If we're in CLI mode, there's no state/host passed down, we need to
# use the global "pseudo" modules.
if len(args) < 2 or not (
isinstance(args[0], (State, PseudoModule))
and isinstance(args[1], (Host, PseudoModule))
):
state = pseudo_state._module
host = pseudo_host._module
if state.in_op:
raise PyinfraError((
'Nested operation called without state/host: {0} ({1})'
).format(op_name, _get_call_location()))
if state.in_deploy:
raise PyinfraError((
'Nested deploy operation called without state/host: {0} ({1})'
).format(op_name, _get_call_location()))
# Otherwise (API mode) we just trim off the commands
else:
args_copy = list(args)
state, host = args[0], args[1]
args = args_copy[2:]
def load_deploy_file(state, filename):
# Copy the inventory hosts (some might be removed during deploy)
hosts = list(state.inventory)
for host in hosts:
# Don't load for anything within our (top level, --limit) limit
if (
isinstance(state.limit_hosts, list)
and host not in state.limit_hosts
):
continue
pseudo_host.set(host)
exec_file(filename)
logger.info('{0} {1} {2}'.format(
host.print_prefix,
click.style('Ready:', 'green'),
click.style(filename, bold=True),
))
# Remove any pseudo host
pseudo_host.reset()
)):
deploy_dir = inventory_path
# Set a fake state/host/inventory
pseudo_state.set(FakeState())
pseudo_host.set(FakeHost())
pseudo_inventory.set(FakeInventory())
# Load up any config.py from the filesystem
config = load_config(deploy_dir)
# Load any hooks/config from the deploy file
load_deploy_config(arguments['deploy'], config)
# Unset fake state/host/inventory
pseudo_host.reset()
pseudo_state.reset()
pseudo_inventory.reset()
# Arg based config overrides
if arguments['sudo']:
config.SUDO = True
if arguments['sudo_user']:
config.SUDO_USER = arguments['sudo_user']
if arguments['su_user']:
config.SU_USER = arguments['su_user']
if arguments['parallel']:
config.PARALLEL = arguments['parallel']
if arguments['fail_percent'] is not None:
def load_deploy_file(state, filename):
for host in state.inventory:
pseudo_host.set(host)
exec_file(filename)
state.ready_host(host)
logger.info('{0} {1} {2}'.format(
'[{}]'.format(click.style(host.name, bold=True)),
click.style('Ready:', 'green'),
click.style(filename, bold=True),
))
# Remove any pseudo host
pseudo_host.reset()
# Un-ready the hosts - this is so that any hooks or callbacks during the deploy
# can still use facts as expected.
state.ready_hosts = set()
def load_deploy_file(state, filename):
for host in state.inventory:
pseudo_host.set(host)
exec_file(filename)
state.ready_host(host)
logger.info('{0} {1} {2}'.format(
'[{}]'.format(click.style(host.name, bold=True)),
click.style('Ready:', 'green'),
click.style(filename, bold=True),
))
# Remove any pseudo host
pseudo_host.reset()
# Un-ready the hosts - this is so that any hooks or callbacks during the deploy
# can still use facts as expected.
and host not in state.limit_hosts
):
continue
pseudo_host.set(host)
exec_file(filename)
logger.info('{0} {1} {2}'.format(
host.print_prefix,
click.style('Ready:', 'green'),
click.style(filename, bold=True),
))
# Remove any pseudo host
pseudo_host.reset()