mirror of
https://github.com/ansible-collections/community.general.git
synced 2024-09-14 20:13:21 +02:00
6feeb16636
We run into some problems because tar --diff will take into account the file ownership and fail if they don't match. The real-world implication of this is that we could be doing more unarchives then we need to be doing.
444 lines
No EOL
15 KiB
Python
444 lines
No EOL
15 KiB
Python
|
|
# tests are fairly 'live' (but safe to run)
|
|
# setup authorized_keys for logged in user such
|
|
# that the user can log in as themselves before running tests
|
|
|
|
import unittest
|
|
import getpass
|
|
import ansible.playbook
|
|
import ansible.utils as utils
|
|
import ansible.callbacks as ans_callbacks
|
|
import os
|
|
import shutil
|
|
import ansible.constants as C
|
|
|
|
EVENTS = []
|
|
|
|
class TestCallbacks(object):
|
|
# using same callbacks class for both runner and playbook
|
|
|
|
def __init__(self):
|
|
pass
|
|
|
|
def set_playbook(self, playbook):
|
|
self.playbook = playbook
|
|
|
|
def on_no_hosts_remaining(self):
|
|
pass
|
|
|
|
def on_start(self):
|
|
EVENTS.append('start')
|
|
|
|
def on_skipped(self, host, item=None):
|
|
EVENTS.append([ 'skipped', [ host ]])
|
|
|
|
def on_import_for_host(self, host, filename):
|
|
EVENTS.append([ 'import', [ host, filename ]])
|
|
|
|
def on_error(self, host, msg):
|
|
EVENTS.append([ 'stderr', [ host, msg ]])
|
|
|
|
def on_not_import_for_host(self, host, missing_filename):
|
|
pass
|
|
|
|
def on_notify(self, host, handler):
|
|
EVENTS.append([ 'notify', [ host, handler ]])
|
|
|
|
def on_task_start(self, name, is_conditional):
|
|
EVENTS.append([ 'task start', [ name, is_conditional ]])
|
|
|
|
def on_failed(self, host, results, ignore_errors):
|
|
EVENTS.append([ 'failed', [ host, results, ignore_errors ]])
|
|
|
|
def on_ok(self, host, result):
|
|
# delete certain info from host_result to make test comparisons easier
|
|
host_result = result.copy()
|
|
for k in [ 'ansible_job_id', 'results_file', 'md5sum', 'delta', 'start', 'end' ]:
|
|
if k in host_result:
|
|
del host_result[k]
|
|
for k in host_result.keys():
|
|
if k.startswith('facter_') or k.startswith('ohai_'):
|
|
del host_result[k]
|
|
EVENTS.append([ 'ok', [ host, host_result ]])
|
|
|
|
def on_play_start(self, pattern):
|
|
EVENTS.append([ 'play start', [ pattern ]])
|
|
|
|
def on_async_ok(self, host, res, jid):
|
|
EVENTS.append([ 'async ok', [ host ]])
|
|
|
|
def on_async_poll(self, host, res, jid, clock):
|
|
EVENTS.append([ 'async poll', [ host ]])
|
|
|
|
def on_async_failed(self, host, res, jid):
|
|
EVENTS.append([ 'async failed', [ host ]])
|
|
|
|
def on_unreachable(self, host, msg):
|
|
EVENTS.append([ 'failed/dark', [ host, msg ]])
|
|
|
|
def on_setup(self):
|
|
pass
|
|
|
|
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 = {
|
|
"localhost": {
|
|
"changed": 65,
|
|
"failures": 0,
|
|
"ok": 73,
|
|
"skipped": 0,
|
|
"unreachable": 0
|
|
}
|
|
}
|
|
|
|
assert utils.jsonify(expected, format=True) == utils.jsonify(actual,format=True)
|
|
|
|
|
|
# Disabled for now as there are permissions issues that happen if you are not the owner that created files
|
|
# in the archive.
|
|
# def test_unarchive(self):
|
|
# pb = 'test/playbook-unarchive.yml'
|
|
# actual = self._run(pb)
|
|
#
|
|
# expected = {
|
|
# "localhost": {
|
|
# "changed": 29,
|
|
# "failures": 0,
|
|
# "ok": 33,
|
|
# "skipped": 12,
|
|
# "unreachable": 0
|
|
# }
|
|
# }
|
|
#
|
|
# 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)
|
|
|
|
print "**ACTUAL**"
|
|
print actual_lines
|
|
|
|
print "**EXPECTED**"
|
|
print expected_lines
|
|
|
|
assert actual_lines == expected_lines |