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_state_hosts_limit(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 first host
with state.hosts('test_group'):
add_op(
state, server.user,
'somehost_user',
)
# Now, also limited but set hosts to the non-limited hosts, which
# should mean this operation applies to no hosts.
add_op(
state, server.user,
'somehost_user',
hosts=inventory.get_host('anotherhost'),
)
# Ensure there are three ops
self.assertEqual(len(state.get_op_order()), 3)
# Ensure somehost has two ops and anotherhost only has the one
self.assertEqual(len(state.ops[inventory.get_host('somehost')]), 2)
def test_ignore_errors_op_fail(self):
inventory = make_inventory()
state = State(inventory, Config())
connect_all(state)
add_op(state, server.shell, 'echo "hi"', ignore_errors=True)
with patch('pyinfra.api.connectors.ssh.run_shell_command') as fake_run_command:
fake_channel = FakeChannel(1)
fake_run_command.return_value = (
False,
FakeBuffer('', fake_channel),
)
# This should run OK
run_ops(state)
somehost = inventory.get_host('somehost')
# Ensure the op was added to results
self.assertEqual(state.results[somehost]['ops'], 1)
self.assertEqual(state.results[somehost]['error_ops'], 1)
def test_setup_op_and_json_args(self):
commands = ('server.user', '[["one", "two"], {"hello": "world"}]')
self.assertEqual(
get_operation_and_args(commands),
(
server.user,
(['one', 'two'], {'hello': 'world'}),
),
from pyinfra import host, local, state
from pyinfra.modules import files, server
server.shell(
{'First main operation'},
'echo first_main_op',
)
# Create some conditional branches
if host.name == 'somehost':
server.shell(
{'Second main operation'},
'echo second_main_op',
)
elif host.name == 'anotherhost':
local.include('tasks/a_task.py')
# Include the whole file again, but for all hosts
local.include('tasks/a_task.py')
args=args,
)
print_facts(fact_data)
_exit()
# Prepare the deploy!
#
# Execute a raw command with server.shell
if command == 'exec':
# Print the output of the command
state.print_output = True
add_op(
state, server.shell,
' '.join(operations),
)
# Deploy files(s)
elif command == 'deploy':
print()
print('--> Preparing operations...')
# Number of "steps" to make = number of files * number of hosts
for i, filename in enumerate(operations):
logger.info('Loading: {0}'.format(click.style(filename, bold=True)))
state.current_op_file = i
load_deploy_file(state, filename)
# Operation w/optional args
elif command == 'op':
from pyinfra import host
from pyinfra.modules import files, server
# Executing as the SSH user (vagrant):
#
# Generate files from local jinja2 templates
files.template(
{'Generate/upload templates/template.txt.j2'},
'templates/template.txt.j2',
'/home/vagrant/template.txt',
)
server.shell(
{'Execute some shell commands'},
[
'echo "Shell command"',
'echo "My hostname is {{ host.fact.hostname }}"',
],
)
# and scripts
server.script(
{'Run the files/test.sh script'},
'files/test.sh',
)
# Copy local files to remote host
files.put(
{'Upload files/file.txt'},
'files/file.txt',
from pyinfra.modules import apt, pip, pkg, server, yum
# Global flag - this applies to all operations in this file!
SUDO = True
# Only apply to hosts in the `bsd` group
if 'bsd' in host.groups:
# OpenBSD packages?
pkg.packages(
{'Install Python, Pip & Git with pkg_add'},
['py-pip', 'git'],
)
# add_pkg does not automatically do this
server.shell(
{'Symlink pip to pip2.7'},
'ln -sf /usr/local/bin/pip2.7 /usr/local/bin/pip',
)
# Work with facts about the remote host
if host.fact.linux_name in ('Ubuntu', 'Debian'):
apt.packages(
{'Install Pip & Git with apt'},
['git', 'python-pip'],
update=True,
cache_time=3600,
)
elif host.fact.linux_name in ('CentOS', 'Fedora'):
if host.fact.linux_name == 'CentOS':
# pyinfra
# File: example/facts.py
# Desc: example deploy script to print all facts
from pyinfra.modules import server
# Get details about a directory
server.directory('/etc')
server.directory('/etc/missing')
# Get details about a file
server.file('/etc/issue')
server.file('/var/log/syslog')
# Get/print all facts (which excludes directories & files, hence above)
print server.all_facts()
mode=777,
)
# and sync directories
files.sync(
{'Sync the files directory'},
'files',
'/home/vagrant/example_files',
delete=True,
)
# Executing with sudo
#
# Ensure the state of a user
server.user(
{'Ensure pyinfra user exists'},
'pyinfra',
shell='/bin/sh',
# Global arguments available in all operations, for the full list see:
# https://pyinfra.readthedocs.io/page/deploys.html#global-arguments
sudo=True,
)
# And groups
server.group(
{'Ensure pyinfra2 group exists'}, # use a set as the first arg to set the op name
'pyinfra2',
sudo=True,
)
To test the file is converted correctly:
pyinfra compile deploy_branches.py
And to execute it:
pyinfra @vagrant deploy_branches.py
'''
from pyinfra import host
from pyinfra.modules import server
SUDO = True
if 'debian' in host.groups:
server.shell({'DEBIAN-1'}, 'echo DEBIAN-1')
server.shell({'DEBIAN-2'}, 'echo DEBIAN-2')
if 'bsd' in host.groups:
server.shell({'BSD-1'}, 'echo BSD-1')
server.shell({'BSD-2'}, 'echo BSD-2')
for i in range(3, 5):
server.shell({'BSD-{0}'.format(i)}, 'echo BSD-{0}'.format(i))
server.shell({'EVERYONE-1'}, 'echo EVERYONE-1')
if 'bsd' in host.groups:
server.shell({'BSD_GROUP'}, 'echo BSD_GROUP')