archive
/
ansible
mirror of https://github.com/ansible/ansible.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Updates for the unarchive module and action_plugin.

Browse Source 
There is a bit going on with the changes here. Most of the changes are cleanup of files so that they line up with the standard files.

PR #5136 was merged into the current devel and brought up to working order. A few bug fixes had to be done to get the code to test correctly. Thanks out to @pib!

Issue #5431 was not able to be confirmed as it behaved as expected with a sudo user.

Tests were added via a playbook with archive files to verify functionality.

All tests fire clean including custom playbooks across multiple linux and solaris systems.
pull/5969/head
Richard C Isaacson 11 years ago
parent e6c9705058
commit cb7c2b7524
7 changed files with 459 additions and 359 deletions
Show Stats Download Patch File Download Diff File

8
lib/ansible/runner/action_plugins/unarchive.py
Unescape Escape View File

@ -28,11 +28,9 @@ from ansible.runner.return_data import ReturnData
import sys
reload(sys)
sys.setdefaultencoding("utf8")
#import base64
#import stat
#import tempfile
import pipes
class ActionModule(object):
TRANSFERS_FILES = True
@ -53,7 +51,7 @@ class ActionModule(object):
copy = utils.boolean(options.get('copy', 'yes'))
if source is None or dest is None:
result=dict(failed=True, msg="src (or content) and dest are required")
result = dict(failed=True, msg="src (or content) and dest are required")
return ReturnData(conn=conn, result=result)
source = template.template(self.runner.basedir, source, inject)
@ -65,7 +63,7 @@ class ActionModule(object):
remote_md5 = self.runner._remote_md5(conn, tmp, dest)
if remote_md5 != '3':
result = dict(failed=True, msg="dest must be an existing dir")
result = dict(failed=True, msg="dest must be an existing dir", rc=remote_md5)
return ReturnData(conn=conn, result=result)
if copy:

63
library/files/unarchive
Unescape Escape View File

@ -65,35 +65,37 @@ EXAMPLES = '''
- unarchive: src=foo.tgz dest=/var/lib/foo
'''
import os
# class to handle .zip files
class _zipfile(object):
def __init__(self,src,dest,module):
def __init__(self, src, dest, module):
self.src = src
self.dest = dest
self.module = module
def is_unarchived(self):
return dict(bool = False)
return dict(unarchived=False)
def unarchive(self):
cmd = 'unzip -o "%s" -d "%s"' % (self.src,self.dest)
cmd = 'unzip -o "%s" -d "%s"' % (self.src, self.dest)
rc, out, err = self.module.run_command(cmd)
return dict(cmd = cmd, rc=rc, out=out, err=err)
return dict(cmd=cmd, rc=rc, out=out, err=err)
def can_handle_archive(self):
cmd = 'unzip -l "%s"' % (self.src)
cmd = 'unzip -l "%s"' % self.src
rc, out, err = self.module.run_command(cmd)
if rc == 0:
return True
return False
# class to handle gzipped tar files
class _tgzfile(object):
def __init__(self,src,dest,module):
def __init__(self, src, dest, module):
self.src = src
self.dest = dest
self.module = module
@ -102,56 +104,62 @@ class _tgzfile(object):
def is_unarchived(self):
dirof = os.path.dirname(self.dest)
destbase = os.path.basename(self.dest)
cmd = 'tar -v -C "%s" --diff -%sf "%s"' % (self.dest, self.zipflag,self.src)
cmd = 'tar -v -C "%s" --diff -%sf "%s"' % (self.dest, self.zipflag, self.src)
rc, out, err = self.module.run_command(cmd)
bool = (rc == 0)
return dict( bool = bool, rc = rc , out = out, err = err, cmd = cmd)
unarchived = (rc == 0)
return dict(unarchived=unarchived, rc=rc, out=out, err=err, cmd=cmd)
def unarchive(self):
cmd = 'tar -C "%s" -x%sf "%s"' % (self.dest,self.zipflag,self.src)
cmd = 'tar -C "%s" -x%sf "%s"' % (self.dest, self.zipflag, self.src)
rc, out, err = self.module.run_command(cmd)
return dict(cmd = cmd, rc=rc, out=out, err=err)
return dict(cmd=cmd, rc=rc, out=out, err=err)
def can_handle_archive(self):
cmd = 'tar -t%sf "%s"' % (self.zipflag,self.src)
cmd = 'tar -t%sf "%s"' % (self.zipflag, self.src)
rc, out, err = self.module.run_command(cmd)
if rc == 0:
return True
if len(out.splitlines(True)) > 0:
return True
return False
# class to handle tar files that aren't compressed
class _tarfile(_tgzfile):
def __init__(self,src,dest,module):
def __init__(self, src, dest, module):
self.src = src
self.dest = dest
self.module = module
self.zipflag = ''
# class to handle bzip2 compressed tar files
class _tarbzip(_tgzfile):
def __init__(self,src,dest,module):
def __init__(self, src, dest, module):
self.src = src
self.dest = dest
self.module = module
self.zipflag = 'j'
# class to handle xz compressed tar files
class _tarxz(_tgzfile):
def __init__(self,src,dest,module):
def __init__(self, src, dest, module):
self.src = src
self.dest = dest
self.module = module
self.zipflag = 'J'
# try handlers in order and return the one that works or bail if none work
def pick_handler(src,dest,module):
def pick_handler(src, dest, module):
handlers = [_tgzfile, _zipfile, _tarfile, _tarbzip, _tarxz]
for handler in handlers:
obj = handler(src,dest,module)
obj = handler(src, dest, module)
if obj.can_handle_archive():
return obj
raise RuntimeError('Failed to find handler to unarchive "%s"' % src)
def main():
module = AnsibleModule(
# not checking because of daisy chain to file module
@ -166,37 +174,36 @@ def main():
src = os.path.expanduser(module.params['src'])
dest = os.path.expanduser(module.params['dest'])
copy = module.params['copy']
# did tar file arrive?
if not os.path.exists(src):
if copy:
module.fail_json(msg="Source '%s' failed to transfer" % (src))
module.fail_json(msg="Source '%s' failed to transfer" % src)
else:
module.fail_json(msg="Source '%s' does not exist" % (src))
module.fail_json(msg="Source '%s' does not exist" % src)
if not os.access(src, os.R_OK):
module.fail_json(msg="Source '%s' not readable" % (src))
module.fail_json(msg="Source '%s' not readable" % src)
# is dest OK to recieve tar file?
# is dest OK to receive tar file?
if not os.path.exists(os.path.dirname(dest)):
module.fail_json(msg="Destination directory '%s' does not exist" % (os.path.dirname(dest)))
if not os.access(os.path.dirname(dest), os.W_OK):
module.fail_json(msg="Destination '%s' not writable" % (os.path.dirname(dest)))
handler = pick_handler(src,dest,module)
handler = pick_handler(src, dest, module)
res_args = dict( handler=handler.__class__.__name__, dest = dest, src = src )
res_args = dict(handler=handler.__class__.__name__, dest=dest, src=src)
# do we need to do unpack?
namelist = ['bool','rc','out','err','cmd']
res_args['check_results'] = handler.is_unarchived()
if res_args['check_results']['bool']:
if res_args['check_results']['unarchived']:
res_args['changed'] = False
module.exit_json(**res_args)
# do the unpack
try:
results = handler.unarchive()
#results = (src,dest,module)
except IOError:
module.fail_json(msg="failed to unpack %s to %s" % (src, dest))

669
test/TestPlayBook.py
Unescape Escape View File

@ -82,322 +82,322 @@ class TestCallbacks(object):
def on_no_hosts(self):
pass
class TestPlaybook(unittest.TestCase):
def setUp(self):
self.user = getpass.getuser()
self.cwd = os.getcwd()
self.test_dir = os.path.join(self.cwd, 'test')
self.stage_dir = self._prepare_stage_dir()
if os.path.exists('/tmp/ansible_test_data_copy.out'):
os.unlink('/tmp/ansible_test_data_copy.out')
if os.path.exists('/tmp/ansible_test_data_template.out'):
os.unlink('/tmp/ansible_test_data_template.out')
if os.path.exists('/tmp/ansible_test_messages.out'):
os.unlink('/tmp/ansible_test_messages.out')
if os.path.exists('/tmp/ansible_test_role_messages.out'):
os.unlink('/tmp/ansible_test_role_messages.out')
def _prepare_stage_dir(self):
stage_path = os.path.join(self.test_dir, 'test_data')
if os.path.exists(stage_path):
shutil.rmtree(stage_path, ignore_errors=False)
assert not os.path.exists(stage_path)
os.makedirs(stage_path)
assert os.path.exists(stage_path)
return stage_path
def _get_test_file(self, filename):
# get a file inside the test input directory
filename = os.path.join(self.test_dir, filename)
assert os.path.exists(filename)
return filename
def _get_stage_file(self, filename):
# get a file inside the test output directory
filename = os.path.join(self.stage_dir, filename)
return filename
def _run(self, test_playbook, host_list='test/ansible_hosts',
extra_vars=None):
''' run a module and get the localhost results '''
# This ensures tests are independent of eachother
global EVENTS
ansible.playbook.SETUP_CACHE.clear()
EVENTS = []
self.test_callbacks = TestCallbacks()
self.playbook = ansible.playbook.PlayBook(
playbook = test_playbook,
host_list = host_list,
module_path = 'library/',
forks = 1,
timeout = 5,
remote_user = self.user,
remote_pass = None,
extra_vars = extra_vars,
stats = ans_callbacks.AggregateStats(),
callbacks = self.test_callbacks,
runner_callbacks = self.test_callbacks
)
result = self.playbook.run()
return result
def test_playbook_vars(self):
test_callbacks = TestCallbacks()
playbook = ansible.playbook.PlayBook(
playbook=os.path.join(self.test_dir, 'test_playbook_vars', 'playbook.yml'),
host_list='test/test_playbook_vars/hosts',
stats=ans_callbacks.AggregateStats(),
callbacks=test_callbacks,
runner_callbacks=test_callbacks
)
playbook.run()
def _test_playbook_undefined_vars(self, playbook, fail_on_undefined):
# save DEFAULT_UNDEFINED_VAR_BEHAVIOR so we can restore it in the end of the test
saved_undefined_var_behavior = C.DEFAULT_UNDEFINED_VAR_BEHAVIOR
C.DEFAULT_UNDEFINED_VAR_BEHAVIOR = fail_on_undefined
test_callbacks = TestCallbacks()
playbook = ansible.playbook.PlayBook(
playbook=os.path.join(self.test_dir, 'test_playbook_undefined_vars', playbook),
host_list='test/test_playbook_undefined_vars/hosts',
stats=ans_callbacks.AggregateStats(),
callbacks=test_callbacks,
runner_callbacks=test_callbacks
)
actual = playbook.run()
C.DEFAULT_UNDEFINED_VAR_BEHAVIOR = saved_undefined_var_behavior
# if different, this will output to screen
print "**ACTUAL**"
print utils.jsonify(actual, format=True)
expected = {
"localhost": {
"changed": 0,
"failures": 0,
"ok": int(not fail_on_undefined) + 1,
"skipped": 0,
"unreachable": int(fail_on_undefined)
}
}
print "**EXPECTED**"
print utils.jsonify(expected, format=True)
assert utils.jsonify(expected, format=True) == utils.jsonify(actual, format=True)
#def test_playbook_undefined_vars1_ignore(self):
# self._test_playbook_undefined_vars('playbook1.yml', False)
#def test_playbook_undefined_vars1_fail(self):
# self._test_playbook_undefined_vars('playbook1.yml', True)
#def test_playbook_undefined_vars2_ignore(self):
# self._test_playbook_undefined_vars('playbook2.yml', False)
#def test_playbook_undefined_vars2_fail(self):
# self._test_playbook_undefined_vars('playbook2.yml', True)
def test_yaml_hosts_list(self):
# Make sure playbooks support hosts: [host1, host2]
# TODO: Actually run the play on more than one host
test_callbacks = TestCallbacks()
playbook = ansible.playbook.PlayBook(
playbook=os.path.join(self.test_dir, 'hosts_list.yml'),
host_list='test/ansible_hosts',
stats=ans_callbacks.AggregateStats(),
callbacks=test_callbacks,
runner_callbacks=test_callbacks
)
play = ansible.playbook.Play(playbook, playbook.playbook[0], os.getcwd())
assert play.hosts == ';'.join(('host1', 'host2', 'host3'))
def test_playbook_hash_replace(self):
# save default hash behavior so we can restore it in the end of the test
saved_hash_behavior = C.DEFAULT_HASH_BEHAVIOUR
C.DEFAULT_HASH_BEHAVIOUR = "replace"
test_callbacks = TestCallbacks()
playbook = ansible.playbook.PlayBook(
playbook=os.path.join(self.test_dir, 'test_hash_behavior', 'playbook.yml'),
host_list='test/ansible_hosts',
stats=ans_callbacks.AggregateStats(),
callbacks=test_callbacks,
runner_callbacks=test_callbacks
)
playbook.run()
filename = '/tmp/ansible_test_messages.out'
expected_lines = [
"goodbye: Goodbye World!"
]
self._compare_file_output(filename, expected_lines)
filename = '/tmp/ansible_test_role_messages.out'
expected_lines = [
"inside_a_role: Indeed!"
]
self._compare_file_output(filename, expected_lines)
# restore default hash behavior
C.DEFAULT_HASH_BEHAVIOUR = saved_hash_behavior
def test_playbook_hash_merge(self):
# save default hash behavior so we can restore it in the end of the test
saved_hash_behavior = C.DEFAULT_HASH_BEHAVIOUR
C.DEFAULT_HASH_BEHAVIOUR = "merge"
test_callbacks = TestCallbacks()
playbook = ansible.playbook.PlayBook(
playbook=os.path.join(self.test_dir, 'test_hash_behavior', 'playbook.yml'),
host_list='test/ansible_hosts',
stats=ans_callbacks.AggregateStats(),
callbacks=test_callbacks,
runner_callbacks=test_callbacks
)
playbook.run()
filename = '/tmp/ansible_test_messages.out'
expected_lines = [
"goodbye: Goodbye World!",
"hello: Hello World!"
]
self._compare_file_output(filename, expected_lines)
filename = '/tmp/ansible_test_role_messages.out'
expected_lines = [
"goodbye: Goodbye World!",
"hello: Hello World!",
"inside_a_role: Indeed!"
]
self._compare_file_output(filename, expected_lines)
# restore default hash behavior
C.DEFAULT_HASH_BEHAVIOUR = saved_hash_behavior
def test_playbook_ignore_errors(self):
test_callbacks = TestCallbacks()
playbook = ansible.playbook.PlayBook(
playbook=os.path.join(self.test_dir, 'playbook-ignore-errors.yml'),
host_list='test/ansible_hosts',
stats=ans_callbacks.AggregateStats(),
callbacks=test_callbacks,
runner_callbacks=test_callbacks
)
actual = playbook.run()
# if different, this will output to screen
print "**ACTUAL**"
print utils.jsonify(actual, format=True)
expected = {
"localhost": {
"changed": 1,
"failures": 1,
"ok": 1,
"skipped": 0,
"unreachable": 0
}
}
print "**EXPECTED**"
print utils.jsonify(expected, format=True)
assert utils.jsonify(expected, format=True) == utils.jsonify(actual,format=True)
def test_playbook_changed_when(self):
test_callbacks = TestCallbacks()
playbook = ansible.playbook.PlayBook(
playbook=os.path.join(self.test_dir, 'playbook-changed_when.yml'),
host_list='test/ansible_hosts',
stats=ans_callbacks.AggregateStats(),
callbacks=test_callbacks,
runner_callbacks=test_callbacks
)
actual = playbook.run()
# if different, this will output to screen
print "**ACTUAL**"
print utils.jsonify(actual, format=True)
expected = {
"localhost": {
"changed": 3,
"failures": 0,
"ok": 6,
"skipped": 0,
"unreachable": 0
}
}
print "**EXPECTED**"
print utils.jsonify(expected, format=True)
assert utils.jsonify(expected, format=True) == utils.jsonify(actual,format=True)
def test_playbook_failed_when(self):
test_callbacks = TestCallbacks()
playbook = ansible.playbook.PlayBook(
playbook=os.path.join(self.test_dir, 'playbook-failed_when.yml'),
host_list='test/ansible_hosts',
stats=ans_callbacks.AggregateStats(),
callbacks=test_callbacks,
runner_callbacks=test_callbacks
)
actual = playbook.run()
# if different, this will output to screen
print "**ACTUAL**"
print utils.jsonify(actual, format=True)
expected = {
"localhost": {
"changed": 2,
"failures": 1,
"ok": 2,
"skipped": 0,
"unreachable": 0
}
}
print "**EXPECTED**"
print utils.jsonify(expected, format=True)
assert utils.jsonify(expected, format=True) == utils.jsonify(actual,format=True)
def test_playbook_always_run(self):
test_callbacks = TestCallbacks()
playbook = ansible.playbook.PlayBook(
playbook=os.path.join(self.test_dir, 'playbook-always-run.yml'),
host_list='test/ansible_hosts',
stats=ans_callbacks.AggregateStats(),
callbacks=test_callbacks,
runner_callbacks=test_callbacks,
check=True
)
actual = playbook.run()
# if different, this will output to screen
print "**ACTUAL**"
print utils.jsonify(actual, format=True)
expected = {
"localhost": {
"changed": 4,
"failures": 0,
"ok": 4,
"skipped": 8,
"unreachable": 0
}
}
print "**EXPECTED**"
print utils.jsonify(expected, format=True)
assert utils.jsonify(expected, format=True) == utils.jsonify(actual,format=True)
def test_recursive_copy(self):
pb = 'test/playbook-recursive-copy.yml'
actual = self._run(pb)
expected = {
def setUp(self):
self.user = getpass.getuser()
self.cwd = os.getcwd()
self.test_dir = os.path.join(self.cwd, 'test')
self.stage_dir = self._prepare_stage_dir()
if os.path.exists('/tmp/ansible_test_data_copy.out'):
os.unlink('/tmp/ansible_test_data_copy.out')
if os.path.exists('/tmp/ansible_test_data_template.out'):
os.unlink('/tmp/ansible_test_data_template.out')
if os.path.exists('/tmp/ansible_test_messages.out'):
os.unlink('/tmp/ansible_test_messages.out')
if os.path.exists('/tmp/ansible_test_role_messages.out'):
os.unlink('/tmp/ansible_test_role_messages.out')
def _prepare_stage_dir(self):
stage_path = os.path.join(self.test_dir, 'test_data')
if os.path.exists(stage_path):
shutil.rmtree(stage_path, ignore_errors=False)
assert not os.path.exists(stage_path)
os.makedirs(stage_path)
assert os.path.exists(stage_path)
return stage_path
def _get_test_file(self, filename):
# get a file inside the test input directory
filename = os.path.join(self.test_dir, filename)
assert os.path.exists(filename)
return filename
def _get_stage_file(self, filename):
# get a file inside the test output directory
filename = os.path.join(self.stage_dir, filename)
return filename
def _run(self, test_playbook, host_list='test/ansible_hosts', extra_vars=None):
''' run a module and get the localhost results '''
# This ensures tests are independent of eachother
global EVENTS
ansible.playbook.SETUP_CACHE.clear()
EVENTS = []
self.test_callbacks = TestCallbacks()
self.playbook = ansible.playbook.PlayBook(
playbook = test_playbook,
host_list = host_list,
module_path = 'library/',
forks = 1,
timeout = 5,
remote_user = self.user,
remote_pass = None,
extra_vars = extra_vars,
stats = ans_callbacks.AggregateStats(),
callbacks = self.test_callbacks,
runner_callbacks = self.test_callbacks
)
result = self.playbook.run()
return result
def test_playbook_vars(self):
test_callbacks = TestCallbacks()
playbook = ansible.playbook.PlayBook(
playbook=os.path.join(self.test_dir, 'test_playbook_vars', 'playbook.yml'),
host_list='test/test_playbook_vars/hosts',
stats=ans_callbacks.AggregateStats(),
callbacks=test_callbacks,
runner_callbacks=test_callbacks
)
playbook.run()
def _test_playbook_undefined_vars(self, playbook, fail_on_undefined):
# save DEFAULT_UNDEFINED_VAR_BEHAVIOR so we can restore it in the end of the test
saved_undefined_var_behavior = C.DEFAULT_UNDEFINED_VAR_BEHAVIOR
C.DEFAULT_UNDEFINED_VAR_BEHAVIOR = fail_on_undefined
test_callbacks = TestCallbacks()
playbook = ansible.playbook.PlayBook(
playbook=os.path.join(self.test_dir, 'test_playbook_undefined_vars', playbook),
host_list='test/test_playbook_undefined_vars/hosts',
stats=ans_callbacks.AggregateStats(),
callbacks=test_callbacks,
runner_callbacks=test_callbacks
)
actual = playbook.run()
C.DEFAULT_UNDEFINED_VAR_BEHAVIOR = saved_undefined_var_behavior
# if different, this will output to screen
print "**ACTUAL**"
print utils.jsonify(actual, format=True)
expected = {
"localhost": {
"changed": 0,
"failures": 0,
"ok": int(not fail_on_undefined) + 1,
"skipped": 0,
"unreachable": int(fail_on_undefined)
}
}
print "**EXPECTED**"
print utils.jsonify(expected, format=True)
assert utils.jsonify(expected, format=True) == utils.jsonify(actual, format=True)
#def test_playbook_undefined_vars1_ignore(self):
# self._test_playbook_undefined_vars('playbook1.yml', False)
#def test_playbook_undefined_vars1_fail(self):
# self._test_playbook_undefined_vars('playbook1.yml', True)
#def test_playbook_undefined_vars2_ignore(self):
# self._test_playbook_undefined_vars('playbook2.yml', False)
#def test_playbook_undefined_vars2_fail(self):
# self._test_playbook_undefined_vars('playbook2.yml', True)
def test_yaml_hosts_list(self):
# Make sure playbooks support hosts: [host1, host2]
# TODO: Actually run the play on more than one host
test_callbacks = TestCallbacks()
playbook = ansible.playbook.PlayBook(
playbook=os.path.join(self.test_dir, 'hosts_list.yml'),
host_list='test/ansible_hosts',
stats=ans_callbacks.AggregateStats(),
callbacks=test_callbacks,
runner_callbacks=test_callbacks
)
play = ansible.playbook.Play(playbook, playbook.playbook[0], os.getcwd())
assert play.hosts == ';'.join(('host1', 'host2', 'host3'))
def test_playbook_hash_replace(self):
# save default hash behavior so we can restore it in the end of the test
saved_hash_behavior = C.DEFAULT_HASH_BEHAVIOUR
C.DEFAULT_HASH_BEHAVIOUR = "replace"
test_callbacks = TestCallbacks()
playbook = ansible.playbook.PlayBook(
playbook=os.path.join(self.test_dir, 'test_hash_behavior', 'playbook.yml'),
host_list='test/ansible_hosts',
stats=ans_callbacks.AggregateStats(),
callbacks=test_callbacks,
runner_callbacks=test_callbacks
)
playbook.run()
filename = '/tmp/ansible_test_messages.out'
expected_lines = [
"goodbye: Goodbye World!"
]
self._compare_file_output(filename, expected_lines)
filename = '/tmp/ansible_test_role_messages.out'
expected_lines = [
"inside_a_role: Indeed!"
]
self._compare_file_output(filename, expected_lines)
# restore default hash behavior
C.DEFAULT_HASH_BEHAVIOUR = saved_hash_behavior
def test_playbook_hash_merge(self):
# save default hash behavior so we can restore it in the end of the test
saved_hash_behavior = C.DEFAULT_HASH_BEHAVIOUR
C.DEFAULT_HASH_BEHAVIOUR = "merge"
test_callbacks = TestCallbacks()
playbook = ansible.playbook.PlayBook(
playbook=os.path.join(self.test_dir, 'test_hash_behavior', 'playbook.yml'),
host_list='test/ansible_hosts',
stats=ans_callbacks.AggregateStats(),
callbacks=test_callbacks,
runner_callbacks=test_callbacks
)
playbook.run()
filename = '/tmp/ansible_test_messages.out'
expected_lines = [
"goodbye: Goodbye World!",
"hello: Hello World!"
]
self._compare_file_output(filename, expected_lines)
filename = '/tmp/ansible_test_role_messages.out'
expected_lines = [
"goodbye: Goodbye World!",
"hello: Hello World!",
"inside_a_role: Indeed!"
]
self._compare_file_output(filename, expected_lines)
# restore default hash behavior
C.DEFAULT_HASH_BEHAVIOUR = saved_hash_behavior
def test_playbook_ignore_errors(self):
test_callbacks = TestCallbacks()
playbook = ansible.playbook.PlayBook(
playbook=os.path.join(self.test_dir, 'playbook-ignore-errors.yml'),
host_list='test/ansible_hosts',
stats=ans_callbacks.AggregateStats(),
callbacks=test_callbacks,
runner_callbacks=test_callbacks
)
actual = playbook.run()
# if different, this will output to screen
print "**ACTUAL**"
print utils.jsonify(actual, format=True)
expected = {
"localhost": {
"changed": 1,
"failures": 1,
"ok": 1,
"skipped": 0,
"unreachable": 0
}
}
print "**EXPECTED**"
print utils.jsonify(expected, format=True)
assert utils.jsonify(expected, format=True) == utils.jsonify(actual,format=True)
def test_playbook_changed_when(self):
test_callbacks = TestCallbacks()
playbook = ansible.playbook.PlayBook(
playbook=os.path.join(self.test_dir, 'playbook-changed_when.yml'),
host_list='test/ansible_hosts',
stats=ans_callbacks.AggregateStats(),
callbacks=test_callbacks,
runner_callbacks=test_callbacks
)
actual = playbook.run()
# if different, this will output to screen
print "**ACTUAL**"
print utils.jsonify(actual, format=True)
expected = {
"localhost": {
"changed": 3,
"failures": 0,
"ok": 6,
"skipped": 0,
"unreachable": 0
}
}
print "**EXPECTED**"
print utils.jsonify(expected, format=True)
assert utils.jsonify(expected, format=True) == utils.jsonify(actual,format=True)
def test_playbook_failed_when(self):
test_callbacks = TestCallbacks()
playbook = ansible.playbook.PlayBook(
playbook=os.path.join(self.test_dir, 'playbook-failed_when.yml'),
host_list='test/ansible_hosts',
stats=ans_callbacks.AggregateStats(),
callbacks=test_callbacks,
runner_callbacks=test_callbacks
)
actual = playbook.run()
# if different, this will output to screen
print "**ACTUAL**"
print utils.jsonify(actual, format=True)
expected = {
"localhost": {
"changed": 2,
"failures": 1,
"ok": 2,
"skipped": 0,
"unreachable": 0
}
}
print "**EXPECTED**"
print utils.jsonify(expected, format=True)
assert utils.jsonify(expected, format=True) == utils.jsonify(actual,format=True)
def test_playbook_always_run(self):
test_callbacks = TestCallbacks()
playbook = ansible.playbook.PlayBook(
playbook=os.path.join(self.test_dir, 'playbook-always-run.yml'),
host_list='test/ansible_hosts',
stats=ans_callbacks.AggregateStats(),
callbacks=test_callbacks,
runner_callbacks=test_callbacks,
check=True
)
actual = playbook.run()
# if different, this will output to screen
print "**ACTUAL**"
print utils.jsonify(actual, format=True)
expected = {
"localhost": {
"changed": 4,
"failures": 0,
"ok": 4,
"skipped": 8,
"unreachable": 0
}
}
print "**EXPECTED**"
print utils.jsonify(expected, format=True)
assert utils.jsonify(expected, format=True) == utils.jsonify(actual,format=True)
def test_recursive_copy(self):
pb = 'test/playbook-recursive-copy.yml'
actual = self._run(pb)
expected = {
"localhost": {
"changed": 65,
"failures": 0,
@ -405,21 +405,38 @@ class TestPlaybook(unittest.TestCase):
"skipped": 0,
"unreachable": 0
}
}
}
assert utils.jsonify(expected, format=True) == utils.jsonify(actual,format=True)
def test_unarchive(self):
pb = 'test/playbook-unarchive.yml'
actual = self._run(pb)
expected = {
"localhost": {
"changed": 41,
"failures": 0,
"ok": 45,
"skipped": 0,
"unreachable": 0
}
}
assert utils.jsonify(expected, format=True) == utils.jsonify(actual,format=True)
assert utils.jsonify(expected, format=True) == utils.jsonify(actual,format=True)
def _compare_file_output(self, filename, expected_lines):
actual_lines = []
with open(filename) as f:
actual_lines = [l.strip() for l in f.readlines()]
actual_lines = sorted(actual_lines)
def _compare_file_output(self, filename, expected_lines):
actual_lines = []
with open(filename) as f:
actual_lines = [l.strip() for l in f.readlines()]
actual_lines = sorted(actual_lines)
print "**ACTUAL**"
print actual_lines
print "**ACTUAL**"
print actual_lines
print "**EXPECTED**"
print expected_lines
print "**EXPECTED**"
print expected_lines
assert actual_lines == expected_lines
assert actual_lines == expected_lines

78
test/playbook-unarchive.yml
Unescape Escape View File

@ -0,0 +1,78 @@
---
# To run me manually, use: -i "localhost,"
- hosts: localhost
connection: local
gather_facts: no
vars:
- testdir: /tmp/ansible-unarchive
- filesdir: test_unarchive/files
tasks:
- name: "Simple tar unarchive."
command: rm -rf {{testdir}}
- file: state=directory dest={{testdir}}
- unarchive: src={{filesdir}}/test.tar dest={{testdir}}
register: res
- command: test -f {{testdir}}/foo
- command: test "{{res.changed}}" = "True"
- unarchive: src={{filesdir}}/test.tar dest={{testdir}}
register: res
- command: test "{{res.changed}}" = "False"
- name: "Simple tar.gz unarchive."
command: rm -rf {{testdir}}
- file: state=directory dest={{testdir}}
- unarchive: src={{filesdir}}/test.tar.gz dest={{testdir}}
register: res
- command: test -f {{testdir}}/foo
- command: test "{{res.changed}}" = "True"
- unarchive: src={{filesdir}}/test.tar.gz dest={{testdir}}
register: res
- command: test "{{res.changed}}" = "False"
- name: "Simple zip unarchive."
command: rm -rf {{testdir}}
- file: state=directory dest={{testdir}}
- unarchive: src={{filesdir}}/test.zip dest={{testdir}}
register: res
- command: test -f {{testdir}}/foo
- command: test "{{res.changed}}" = "True"
- unarchive: src={{filesdir}}/test.zip dest={{testdir}}
register: res
- command: test "{{res.changed}}" = "True"
- name: "Unarchive a local tar file."
command : rm -rf {{testdir}}
- file: state=directory dest={{testdir}}
- copy: src={{filesdir}}/test.tar dest={{testdir}}
- unarchive: src={{testdir}}/test.tar dest={{testdir}}
register: res
- command: test -f {{testdir}}/foo
- command: test "{{res.changed}}" = "True"
- unarchive: src={{testdir}}/test.tar dest={{testdir}}
register: res
- command: test "{{res.changed}}" = "False"
- name: "Unarchive a local tar.gz file."
command : rm -rf {{testdir}}
- file: state=directory dest={{testdir}}
- copy: src={{filesdir}}/test.tar.gz dest={{testdir}}
- unarchive: src={{testdir}}/test.tar.gz dest={{testdir}}
register: res
- command: test -f {{testdir}}/foo
- command: test "{{res.changed}}" = "True"
- unarchive: src={{testdir}}/test.tar.gz dest={{testdir}}
register: res
- command: test "{{res.changed}}" = "False"
- name: "Unarchive a local zip file."
command : rm -rf {{testdir}}
- file: state=directory dest={{testdir}}
- copy: src={{filesdir}}/test.zip dest={{testdir}}
- unarchive: src={{testdir}}/test.zip dest={{testdir}}
register: res
- command: test -f {{testdir}}/foo
- command: test "{{res.changed}}" = "True"
- unarchive: src={{testdir}}/test.zip dest={{testdir}}
register: res
- command: test "{{res.changed}}" = "True"

BIN
test/test_unarchive/files/test.tar
View File

Binary file not shown.

BIN
test/test_unarchive/files/test.tar.gz
View File

Binary file not shown.

BIN
test/test_unarchive/files/test.zip
View File

Binary file not shown.
Write Preview
Loading…
Cancel
Save