mirror of
https://github.com/ansible-collections/community.general.git
synced 2024-09-14 20:13:21 +02:00
377bc31311
As documented in #2623, early variable substitution causes when_
tests to fail and possibly other side effects.
I can see the reason for this early substitution, likely introduced
in 1dfe60a6, to allow many playbook parameters to be templated.
This is a valid goal, but the recursive nature of the utils.template
function means that it goes too far.
At this point removing tasks from the list of parameters to be
substituted seems sufficient to make my tests pass. It may be the
case that other parameters should be excluded, but I suspect not.
Adding a test case. I would prefer to analyse not just the aggregate
statistics but also whether the results are as expected - I can't
see an easy way to do that with the available callbacks at present.
287 lines
9.1 KiB
Python
287 lines
9.1 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
|
|
|
|
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')
|
|
|
|
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'):
|
|
''' 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,
|
|
stats = ans_callbacks.AggregateStats(),
|
|
callbacks = self.test_callbacks,
|
|
runner_callbacks = self.test_callbacks
|
|
)
|
|
result = self.playbook.run()
|
|
print EVENTS
|
|
return result
|
|
|
|
def test_one(self):
|
|
pb = 'test/playbook1.yml'
|
|
actual = self._run(pb)
|
|
|
|
# if different, this will output to screen
|
|
print "**ACTUAL**"
|
|
print utils.jsonify(actual, format=True)
|
|
expected = {
|
|
"localhost": {
|
|
"changed": 9,
|
|
"failures": 0,
|
|
"ok": 11,
|
|
"skipped": 1,
|
|
"unreachable": 0
|
|
}
|
|
}
|
|
print "**EXPECTED**"
|
|
print utils.jsonify(expected, format=True)
|
|
|
|
assert utils.jsonify(expected, format=True) == utils.jsonify(actual,format=True)
|
|
|
|
# make sure the template module took options from the vars section
|
|
data = file('/tmp/ansible_test_data_template.out').read()
|
|
print data
|
|
assert data.find("ears") != -1, "template success"
|
|
|
|
# disabling until we have a nice way of using lookup plugins inside '{{' and '}}'
|
|
|
|
#def test_lookups(self):
|
|
# pb = os.path.join(self.test_dir, 'lookup_plugins.yml')
|
|
# actual = self._run(pb)
|
|
#
|
|
# # if different, this will output to screen
|
|
# print "**ACTUAL**"
|
|
# print utils.jsonify(actual, format=True)
|
|
# expected = {
|
|
# "localhost": {
|
|
# "changed": 16,
|
|
# "failures": 0,
|
|
# "ok": 21,
|
|
# "skipped": 1,
|
|
# "unreachable": 0
|
|
# }
|
|
# }
|
|
# print "**EXPECTED**"
|
|
# print utils.jsonify(expected, format=True)
|
|
#
|
|
# assert utils.jsonify(expected, format=True) == utils.jsonify(actual,format=True)
|
|
#
|
|
# print "len(EVENTS) = %d" % len(EVENTS)
|
|
# assert len(EVENTS) == 74
|
|
|
|
def test_includes(self):
|
|
pb = os.path.join(self.test_dir, 'playbook-includer.yml')
|
|
actual = self._run(pb)
|
|
|
|
# if different, this will output to screen
|
|
print "**ACTUAL**"
|
|
print utils.jsonify(actual, format=True)
|
|
expected = {
|
|
"localhost": {
|
|
"changed": 0,
|
|
"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_task_includes(self):
|
|
pb = os.path.join(self.test_dir, 'task-includer.yml')
|
|
actual = self._run(pb)
|
|
|
|
# if different, this will output to screen
|
|
print "**ACTUAL**"
|
|
print utils.jsonify(actual, format=True)
|
|
expected = {
|
|
"localhost": {
|
|
"changed": 0,
|
|
"failures": 0,
|
|
"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_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_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_when(self):
|
|
test_callbacks = TestCallbacks()
|
|
playbook = ansible.playbook.PlayBook(
|
|
playbook=os.path.join(self.test_dir, 'playbook-when.yml'),
|
|
host_list='test/ansible_hosts',
|
|
extra_vars={ 'external' : 'xyz', 'identity': 'identity' },
|
|
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": 0,
|
|
"failures": 0,
|
|
"ok": 3,
|
|
"skipped": 3,
|
|
"unreachable": 0
|
|
}
|
|
}
|
|
print "**EXPECTED**"
|
|
print utils.jsonify(expected, format=True)
|
|
|
|
assert utils.jsonify(expected, format=True) == utils.jsonify(actual,format=True)
|