openstack/validations-libs
Code Issues Proposed changes

Moving callbacks to validations-libs

Browse Source 
Callback plugins were transferred from validations-common
repository to validations-libs.

Necessary adjustments were made to the module structure,
requierements, as well as installation and documentation generator config.

Associated tests were moved as well, with removal of superfluous inheritance
and imports included.

Demonstration http server module for communication with `http_json`
callback was moved with directory structure preserved.

Signed-off-by: Jiri Podivin <jpodivin@redhat.com>
Change-Id: I31768375430a2f29da71aae8f3db3882c373ced5
This commit is contained in:
Jiri Podivin 2021-12-06 16:09:12 +01:00 committed by mbu
parent 299b7b6163
commit 9163d8bcbc
20 changed files with 2428 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
doc/source/conf.py
View File

@ -53,6 +53,7 @@ openstackdocs_bug_tag = 'documentation'
# autodoc generation is a bit aggressive and a nuisance when doing heavy
# text edit cycles.
# execute "export SPHINX_DEBUG=1" in your terminal to disable
autodoc_mock_imports = ['oslotest', 'ansible']
# Add any paths that contain templates here, relative to this directory.
templates_path = ['_templates']

1
setup.cfg
View File

@ -25,6 +25,7 @@ packages = validations_libs
data_files =
etc =
validation.cfg
share/ansible/callback_plugins = validations_libs/callback_plugins/*
[compile_catalog]
directory = validations-libs/locale

1
test-requirements.txt
View File

@ -13,4 +13,5 @@ python-subunit>=1.0.0 # Apache-2.0/BSD
stestr>=2.0.0 # Apache-2.0
testscenarios>=0.4 # Apache-2.0/BSD
testtools>=2.2.0 # MIT
oslotest>=3.2.0 # Apache-2.0
pre-commit # MIT

54
tools/http_server.py Normal file
View File

@ -0,0 +1,54 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from http.server import BaseHTTPRequestHandler, HTTPServer
import logging
class SimpleHandler(BaseHTTPRequestHandler):
def _set_headers(self):
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
def do_GET(self):
logging.info("Received GET request:\n"
"Headers: {}\n".format(str(self.headers)))
self._set_headers()
self.wfile.write("GET request: {}".format(self.path).encode('utf-8'))
def do_POST(self):
content_length = int(self.headers['Content-Length'])
data = self.rfile.read(content_length)
logging.info("Received POST request:\n"
"Headers: {}\n"
"Body: \n{}\n".format(self.headers, data.decode('utf-8')))
self._set_headers()
self.wfile.write("POST request: {}".format(self.path).encode('utf-8'))
def run(host='localhost', port=8989):
logging.basicConfig(level=logging.INFO)
http_server = HTTPServer((host, port), SimpleHandler)
logging.info("Starting http server...\n")
try:
http_server.serve_forever()
except KeyboardInterrupt:
pass
http_server.server_close()
logging.info('Stopping http server...\n')
if __name__ == '__main__':
run()

7
validations_libs/ansible.py
View File

@ -116,10 +116,13 @@ class Ansible(object):
if isinstance(envvars, dict):
env.update(envvars)
output_callback = env.get('ANSIBLE_STDOUT_CALLBACK', output_callback)
# TODO(jpodivin) Whitelist was extended with new callback names
# to prevent issues during transition period.
# The entries with 'vf_' prefix should be removed afterwards.
callback_whitelist = ','.join(filter(None, [callback_whitelist,
output_callback,
'validation_json',
'profile_tasks']))
'profile_tasks',
'vf_validation_json']))
return callback_whitelist, output_callback
def _ansible_env_var(self, output_callback, ssh_user, workdir, connection,

13
validations_libs/callback_plugins/__init__.py Normal file
View File

@ -0,0 +1,13 @@
"""
This module contains various callbacks developed to facilitate functions
of the Validation Framework.
Somewhat unorthodox naming of the callback classes is a direct result of how
ansible handles loading plugins.
The ansible determines the purpose of each plugin by looking at its class name.
As you can see in the 'https://github.com/ansible/ansible/blob/devel/lib/ansible/plugins/loader.py'
from the ansible repo, the loader uses the class names to categorize plugins.
This means that every callback plugin has to have the same class name,
and the unfortunate coder has to discern their purpose by checking
their module names.
"""

29
validations_libs/callback_plugins/vf_fail_if_no_hosts.py Normal file
View File

@ -0,0 +1,29 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import sys
from ansible.plugins.callback import CallbackBase
class CallbackModule(CallbackBase):
CALLBACK_VERSION = 2.0
CALLBACK_NAME = 'fail_if_no_hosts'
def __init__(self, display=None):
super(CallbackModule, self).__init__(display)
def v2_playbook_on_stats(self, stats):
if len(stats.processed.keys()) == 0:
sys.exit(10)

94
validations_libs/callback_plugins/vf_http_json.py Normal file
View File

@ -0,0 +1,94 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
__metaclass__ = type
DOCUMENTATION = '''
requirements:
- whitelist in configuration
short_description: sends JSON events to a HTTP server
description:
- This plugin logs ansible-playbook and ansible runs to an HTTP server in JSON format
options:
server:
description: remote server that will receive the event
env:
- name: HTTP_JSON_SERVER
default: http://localhost
ini:
- section: callback_http_json
key: http_json_server
port:
description: port on which the remote server is listening
env:
- name: HTTP_JSON_PORT
default: 8989
ini:
- section: callback_http_json
key: http_json_port
'''
import datetime
import json
import os
from urllib import request
from validations_libs.callback_plugins import vf_validation_json
url = '{}:{}'.format(os.getenv('HTTP_JSON_SERVER', 'http://localhost'),
os.getenv('HTTP_JSON_PORT', '8989'))
def http_post(data):
req = request.Request(url)
req.add_header('Content-Type', 'application/json; charset=utf-8')
json_data = json.dumps(data)
json_bytes = json_data.encode('utf-8')
req.add_header('Content-Length', len(json_bytes))
response = request.urlopen(req, json_bytes)
def current_time():
return '%sZ' % datetime.datetime.utcnow().isoformat()
class CallbackModule(vf_validation_json.CallbackModule):
CALLBACK_VERSION = 2.0
CALLBACK_TYPE = 'aggregate'
CALLBACK_NAME = 'http_json'
CALLBACK_NEEDS_WHITELIST = True
def __init__(self):
super(vf_validation_json.CallbackModule, self).__init__()
self.results = []
self.simple_results = []
self.env = {}
self.t0 = None
self.current_time = current_time()
def v2_playbook_on_stats(self, stats):
"""Display info about playbook statistics"""
hosts = sorted(stats.processed.keys())
summary = {}
for h in hosts:
s = stats.summarize(h)
summary[h] = s
http_post({
'plays': self.results,
'stats': summary,
'validation_output': self.simple_results
})

238
validations_libs/callback_plugins/vf_validation_json.py Normal file
View File

@ -0,0 +1,238 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
__metaclass__ = type
import datetime
import json
import time
import os
from functools import partial
from functools import reduce
from ansible.parsing.ajson import AnsibleJSONEncoder
from ansible.plugins.callback import CallbackBase
DOCUMENTATION = '''
callback: json
short_description: Log Ansible results on filesystem
version_added: "1.0"
description:
- This callback converts all events into a JSON file
stored in the selected validations logging directory,
as defined by the $VALIDATIONS_LOG_DIR env variable,
or the $HOME/validations by default.
type: aggregate
requirements: None
'''
VALIDATIONS_LOG_DIR = os.environ.get(
'VALIDATIONS_LOG_DIR',
os.path.expanduser('~/validations'))
def current_time():
return '%sZ' % datetime.datetime.utcnow().isoformat()
def secondsToStr(t):
def rediv(ll, b):
return list(divmod(ll[0], b)) + ll[1:]
return "%d:%02d:%02d.%03d" % tuple(
reduce(rediv, [[
t * 1000,
], 1000, 60, 60]))
class CallbackModule(CallbackBase):
CALLBACK_VERSION = 2.0
CALLBACK_TYPE = 'aggregate'
CALLBACK_NAME = 'validation_json'
CALLBACK_NEEDS_WHITELIST = True
def __init__(self, display=None):
super(CallbackModule, self).__init__(display)
self.results = []
self.simple_results = []
self.env = {}
self.start_time = None
self.current_time = current_time()
def _new_play(self, play):
return {
'play': {
'host': play.get_name(),
'validation_id': self.env['playbook_name'],
'validation_path': self.env['playbook_path'],
'id': (os.getenv('ANSIBLE_UUID') if os.getenv('ANSIBLE_UUID')
else str(play._uuid)),
'duration': {
'start': current_time()
}
},
'tasks': []
}
def _new_task(self, task):
return {
'task': {
'name': task.get_name(),
'id': str(task._uuid),
'duration': {
'start': current_time()
}
},
'hosts': {}
}
def _val_task(self, task_name):
return {
'task': {
'name': task_name,
'hosts': {}
}
}
def _val_task_host(self, task_name):
return {
'task': {
'name': task_name,
'hosts': {}
}
}
def v2_playbook_on_start(self, playbook):
self.start_time = time.time()
pl = playbook._file_name
validation_id = os.path.splitext(os.path.basename(pl))[0]
self.env = {
"playbook_name": validation_id,
"playbook_path": playbook._basedir
}
def v2_playbook_on_play_start(self, play):
self.results.append(self._new_play(play))
def v2_playbook_on_task_start(self, task, is_conditional):
self.results[-1]['tasks'].append(self._new_task(task))
def v2_playbook_on_handler_task_start(self, task):
self.results[-1]['tasks'].append(self._new_task(task))
def v2_playbook_on_stats(self, stats):
"""Display info about playbook statistics"""
hosts = sorted(stats.processed.keys())
summary = {}
for h in hosts:
s = stats.summarize(h)
summary[h] = s
output = {
'plays': self.results,
'stats': summary,
'validation_output': self.simple_results
}
log_file = "{}/{}_{}_{}.json".format(
VALIDATIONS_LOG_DIR,
(os.getenv('ANSIBLE_UUID') if os.getenv('ANSIBLE_UUID') else
self.results[0].get('play').get('id')),
self.env['playbook_name'],
self.current_time)
with open(log_file, 'w') as js:
js.write(json.dumps(output,
cls=AnsibleJSONEncoder,
indent=4,
sort_keys=True))
def _record_task_result(self, on_info, result, **kwargs):
"""This function is used as a partial to add info in a single method
"""
host = result._host
task = result._task
task_result = result._result.copy()
task_result.update(on_info)
task_result['action'] = task.action
self.results[-1]['tasks'][-1]['hosts'][host.name] = task_result
if 'failed' in task_result.keys():
self.simple_results.append(self._val_task(task.name))
self.simple_results[-1]['task']['status'] = "FAILED"
self.simple_results[-1]['task']['hosts'][host.name] = task_result
if 'warnings' in task_result.keys() and task_result.get('warnings'):
self.simple_results.append(self._val_task(task.name))
self.simple_results[-1]['task']['status'] = "WARNING"
self.simple_results[-1]['task']['hosts'][host.name] = task_result
end_time = current_time()
time_elapsed = secondsToStr(time.time() - self.start_time)
for result in self.results:
if len(result['tasks']) > 1:
result['tasks'][-1]['task']['duration']['end'] = end_time
result['play']['duration']['end'] = end_time
result['play']['duration']['time_elapsed'] = time_elapsed
def v2_playbook_on_no_hosts_matched(self):
no_match_result = self._val_task('No tasks run')
no_match_result['task']['status'] = "FAILED"
no_match_result['task']['info'] = (
"None of the hosts specified"
" were matched in the inventory file")
output = {
'plays': self.results,
'stats': {
'No host matched': {
'changed': 0,
'failures': 1,
'ignored': 0,
'ok': 0,
'rescued': 0,
'skipped': 0,
'unreachable': 0}},
'validation_output': self.simple_results + [no_match_result]
}
log_file = "{}/{}_{}_{}.json".format(
VALIDATIONS_LOG_DIR,
os.getenv(
'ANSIBLE_UUID',
self.results[0].get('play').get('id')),
self.env['playbook_name'],
self.current_time)
with open(log_file, 'w') as js:
js.write(json.dumps(output,
cls=AnsibleJSONEncoder,
indent=4,
sort_keys=True))
def __getattribute__(self, name):
"""Return ``_record_task_result`` partial with a dict
containing skipped/failed if necessary
"""
if name not in ('v2_runner_on_ok', 'v2_runner_on_failed',
'v2_runner_on_unreachable', 'v2_runner_on_skipped'):
return object.__getattribute__(self, name)
on = name.rsplit('_', 1)[1]
on_info = {}
on_info[on] = True
return partial(self._record_task_result, on_info)

203
validations_libs/callback_plugins/vf_validation_output.py Normal file
View File

@ -0,0 +1,203 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import pprint
from ansible import constants as C
from ansible.plugins.callback import CallbackBase
FAILURE_TEMPLATE = """\
Task '{}' failed:
Host: {}
Message: {}
"""
WARNING_TEMPLATE = """\
Task '{}' succeeded, but had some warnings:
Host: {}
Warnings: {}
"""
DEBUG_TEMPLATE = """\
Task: Debug
Host: {}
{}
"""
def indent(text):
'''Indent the given text by four spaces.'''
return ''.join(' {}\n'.format(line) for line in text.splitlines())
# TODO(shadower): test with async settings
class CallbackModule(CallbackBase):
CALLBACK_VERSION = 2.0
CALLBACK_TYPE = 'stdout'
CALLBACK_NAME = 'validation_output'
def __init__(self, display=None):
super(CallbackModule, self).__init__(display)
def print_failure_message(self, host_name, task_name, results,
abridged_result):
'''Print a human-readable error info from Ansible result dictionary.'''
def is_script(results):
return ('rc' in results and 'invocation' in results
and 'script' in results._task_fields['action']
and '_raw_params' in results._task_fields['args'])
display_full_results = False
if 'rc' in results and 'cmd' in results:
command = results['cmd']
# The command can be either a list or a string.
# Concat if it's a list:
if type(command) == list:
command = " ".join(results['cmd'])
message = "Command `{}` exited with code: {}".format(
command, results['rc'])
# There may be an optional message attached to the command.
# Display it:
if 'msg' in results:
message = message + ": " + results['msg']
elif is_script(results):
script_name = results['invocation']['module_args']['_raw_params']
message = "Script `{}` exited with code: {}".format(
script_name, results['rc'])
elif 'msg' in results:
message = results['msg']
else:
message = "Unknown error"
display_full_results = True
self._display.display(
FAILURE_TEMPLATE.format(task_name, host_name, message),
color=C.COLOR_ERROR)
stdout = results.get('module_stdout', results.get('stdout', ''))
if stdout:
print('stdout:')
self._display.display(indent(stdout), color=C.COLOR_ERROR)
stderr = results.get('module_stderr', results.get('stderr', ''))
if stderr:
print('stderr:')
self._display.display(indent(stderr), color=C.COLOR_ERROR)
if display_full_results:
print(
"Could not get an error message. Here is the Ansible output:")
pprint.pprint(abridged_result, indent=4)
warnings = results.get('warnings', [])
if warnings:
print("Warnings:")
for warning in warnings:
self._display.display("* %s " % warning, color=C.COLOR_WARN)
print("")
def v2_playbook_on_play_start(self, play):
pass # No need to notify that a play started
def v2_playbook_on_task_start(self, task, is_conditional):
pass # No need to notify that a task started
def v2_runner_on_ok(self, result, **kwargs):
host_name = result._host
task_name = result._task.get_name()
task_fields = result._task_fields
results = result._result # A dict of the module name etc.
self._dump_results(results)
warnings = results.get('warnings', [])
# Print only tasks that produced some warnings:
if warnings:
for warning in warnings:
warn_msg = "{}\n".format(warning)
self._display.display(WARNING_TEMPLATE.format(task_name,
host_name,
warn_msg),
color=C.COLOR_WARN)
if 'debug' in task_fields['action']:
output = ""
if 'var' in task_fields['args']:
variable = task_fields['args']['var']
value = results[variable]
output = "{}: {}".format(variable, str(value))
elif 'msg' in task_fields['args']:
output = "Message: {}".format(
task_fields['args']['msg'])
self._display.display(DEBUG_TEMPLATE.format(host_name, output),
color=C.COLOR_OK)
def v2_runner_on_failed(self, result, **kwargs):
host_name = result._host
task_name = result._task.get_name()
result_dict = result._result # A dict of the module name etc.
abridged_result = self._dump_results(result_dict)
if 'results' in result_dict:
# The task is a list of items under `results`
for item in result_dict['results']:
if item.get('failed', False):
self.print_failure_message(host_name, task_name,
item, item)
else:
# The task is a "normal" module invocation
self.print_failure_message(host_name, task_name, result_dict,
abridged_result)
def v2_runner_on_skipped(self, result, **kwargs):
pass # No need to print skipped tasks
def v2_runner_on_unreachable(self, result, **kwargs):
host_name = result._host
task_name = result._task.get_name()
results = {'msg': 'The host is unreachable.'}
self.print_failure_message(host_name, task_name, results, results)
def v2_playbook_on_stats(self, stats):
def failed(host):
_failures = stats.summarize(host).get('failures', 0) > 0
_unreachable = stats.summarize(host).get('unreachable', 0) > 0
return (_failures or _unreachable)
hosts = sorted(stats.processed.keys())
failed_hosts = [host for host in hosts if failed(host)]
if hosts:
if failed_hosts:
if len(failed_hosts) == len(hosts):
print("Failure! The validation failed for all hosts:")
for failed_host in failed_hosts:
self._display.display("* %s" % failed_host,
color=C.COLOR_ERROR)
else:
print("Failure! The validation failed for hosts:")
for failed_host in failed_hosts:
self._display.display("* %s" % failed_host,
color=C.COLOR_ERROR)
print("and passed for hosts:")
for host in [h for h in hosts if h not in failed_hosts]:
self._display.display("* %s" % host,
color=C.COLOR_OK)
else:
print("Success! The validation passed for all hosts:")
for host in hosts:
self._display.display("* %s" % host,
color=C.COLOR_OK)
else:
print("Warning! The validation did not run on any host.")

99
validations_libs/callback_plugins/vf_validation_stdout.py Normal file
View File

@ -0,0 +1,99 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
__metaclass__ = type
import datetime
import os
from functools import reduce
from ansible.plugins.callback import CallbackBase
DOCUMENTATION = '''
callback: stdout
short_description: Ansible screen output as JSON file
version_added: "1.0"
description: This callback prints simplify Ansible information to the
console.
type: stdout
requirements: None
'''
def current_time():
return '%sZ' % datetime.datetime.utcnow().isoformat()
def secondsToStr(t):
def rediv(ll, b):
return list(divmod(ll[0], b)) + ll[1:]
return "%d:%02d:%02d.%03d" % tuple(
reduce(rediv, [[
t * 1000,
], 1000, 60, 60]))
class CallbackModule(CallbackBase):
CALLBACK_VERSION = 2.0
CALLBACK_TYPE = 'stdout'
CALLBACK_NAME = 'validation_stdout'
def __init__(self, display=None):
super(CallbackModule, self).__init__(display)
self.env = {}
self.start_time = None
self.current_time = current_time()
def _new_play(self, play):
return {
'play': {
'host': play.get_name(),
'validation_id': self.env['playbook_name'],
'validation_path': self.env['playbook_path'],
'id': (os.getenv('ANSIBLE_UUID') if os.getenv('ANSIBLE_UUID')
else str(play._uuid)),
'duration': {
'start': current_time()
}
},
'tasks': []
}
def _new_task(self, task):
return {
'task': {
'name': task.get_name(),
'id': str(task._uuid),
'duration': {
'start': current_time()
}
},
'hosts': {}
}
def _val_task(self, task_name):
return {
'task': {
'name': task_name,
'hosts': {}
}
}
def _val_task_host(self, task_name):
return {
'task': {
'name': task_name,
'hosts': {}
}
}

2
validations_libs/tests/callback_plugins/__init__.py Normal file
View File

@ -0,0 +1,2 @@
"""
"""

8
validations_libs/tests/callback_plugins/fakes.py Normal file
View File

@ -0,0 +1,8 @@
"""
"""
HTTP_POST_DATA = {
'plays': "foo,bar",
'stats': "buzz",
'validation_output': "SUCCESS"
}

92
validations_libs/tests/callback_plugins/test_vf_fail_if_no_hosts.py Normal file
View File

@ -0,0 +1,92 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""
test_fail_if_no_hosts
----------------------------------
Tests for `fail_if_no_hosts` callback plugin.
"""
try:
from unittest import mock
except ImportError:
import mock
from oslotest import base
from validations_libs.callback_plugins import vf_fail_if_no_hosts
from ansible.plugins.callback import CallbackBase
class TestFailIfNoHosts(base.BaseTestCase):
def setUp(self):
super(TestFailIfNoHosts, self).setUp()
def test_callback_instantiation(self):
"""
Verifying that the CallbackModule is instantiated properly.
Test checks presence of CallbackBase in the inheritance chain,
in order to ensure that
"""
callback = vf_fail_if_no_hosts.CallbackModule()
self.assertEqual(type(callback).__mro__[1], CallbackBase)
self.assertIn('CALLBACK_NAME', dir(callback))
self.assertIn('CALLBACK_VERSION', dir(callback))
self.assertEqual(callback.CALLBACK_NAME, 'fail_if_no_hosts')
self.assertIsInstance(callback.CALLBACK_VERSION, float)
@mock.patch('sys.exit', autospec=True)
def test_callback_playbook_on_stats_no_hosts(self, mock_exit):
"""
Following test concerns stats, an instance of AggregateStats
and how it's processed by the callback.
When the v2_playbook_on_stats method of the callback is called,
a number of hosts in the stats.processed dictionary is checked.
If there are no hosts in the stats.processed dictionary,
the callback calls sys.exit.
"""
callback = vf_fail_if_no_hosts.CallbackModule()
stats = mock.MagicMock()
callback.v2_playbook_on_stats(stats)
mock_exit.assert_called_once_with(10)
@mock.patch('sys.exit', autospec=True)
def test_callback_playbook_on_stats_some_hosts(self, mock_exit):
"""
Following test concerns stats, an instance of AggregateStats
and how it's processed by the callback.
When the v2_playbook_on_stats method of the callback is called,
a number of hosts in the stats.processed dictionary is checked.
If there are hosts in the stats.processed dictionary,
sys.exit is never called.
"""
callback = vf_fail_if_no_hosts.CallbackModule()
stats = mock.MagicMock()
stats.processed = {
'system_foo': 'foo',
'system_bar': 'bar'}
callback.v2_playbook_on_stats(stats)
mock_exit.assert_not_called()

108
validations_libs/tests/callback_plugins/test_vf_http_json.py Normal file
View File

@ -0,0 +1,108 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""
test_http_json
----------------------------------
Tests for `http_json` callback plugin.
"""
import re
from oslotest import base
try:
from unittest import mock
except ImportError:
import mock
from ansible.plugins.callback import CallbackBase
from validations_libs.callback_plugins import vf_http_json
from validations_libs.tests.callback_plugins import fakes
def is_iso_time(time_string):
"""
Checks if string represents valid time in ISO format,
with the default delimiter.
Regex is somewhat convoluted, but general enough to last
at least until the 9999 AD.
Returns:
True if string matches the pattern.
False otherwise.
"""
match = re.match(
r'\d{4}-[01][0-9]-[0-3][0-9]T[0-3][0-9](:[0-5][0-9]){2}\.\d+Z',
time_string)
if match:
return True
else:
return False
class TestHttpJson(base.BaseTestCase):
def setUp(self):
super(TestHttpJson, self).setUp()
self.callback = vf_http_json.CallbackModule()
def test_callback_instantiation(self):
"""
Verifying that the CallbackModule is instantiated properly.
Test checks presence of CallbackBase in the inheritance chain,
in order to ensure that folowing tests are performed with
the correct assumptions.
"""
self.assertEqual(type(self.callback).__mro__[2], CallbackBase)
"""
Every ansible callback needs to define variable with name and version.
"""
self.assertIn('CALLBACK_NAME', dir(self.callback))
self.assertIn('CALLBACK_VERSION', dir(self.callback))
self.assertIn('CALLBACK_TYPE', dir(self.callback))
self.assertEqual(self.callback.CALLBACK_NAME, 'http_json')
self.assertIsInstance(self.callback.CALLBACK_VERSION, float)
self.assertEqual(self.callback.CALLBACK_TYPE, 'aggregate')
"""
Additionally, the 'http_json' callback performs several
other operations during instantiation.
"""
self.assertEqual(self.callback.env, {})
self.assertIsNone(self.callback.t0)
"""
Callback time sanity check only verifies general format
of the stored time to be iso format `YYYY-MM-DD HH:MM:SS.mmmmmm`
with 'T' as a separator.
For example: '2020-07-03T13:28:21.224103Z'
"""
self.assertTrue(is_iso_time(self.callback.current_time))
@mock.patch('validations_libs.callback_plugins.vf_http_json.request.urlopen', autospec=True)
@mock.patch('validations_libs.callback_plugins.vf_http_json.json.dumps', autospec=True)
@mock.patch('validations_libs.callback_plugins.vf_http_json.request.Request', autospec=True)
def test_http_post(self, mock_request, mock_json, mock_url_open):
vf_http_json.http_post(fakes.HTTP_POST_DATA)
mock_request.assert_called_once()
mock_json.assert_called_once_with(fakes.HTTP_POST_DATA)
mock_url_open.assert_called_once()

537
validations_libs/tests/callback_plugins/test_vf_validation_json.py Normal file
View File

@ -0,0 +1,537 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""
test_validation_json
----------------------------------
Tests for `validation_json` callback plugin.
"""
import re
try:
from unittest import mock
except ImportError:
import mock
from oslotest import base
from ansible.executor.stats import AggregateStats
from ansible.parsing.ajson import AnsibleJSONEncoder
from ansible.playbook import Playbook
from ansible.plugins.callback import CallbackBase
from validations_libs.callback_plugins import vf_validation_json
def is_iso_time(time_string):
"""
Checks if string represents valid time in ISO format,
with the default delimiter.
Regex is somewhat convoluted, but general enough to last
at least until the 9999 AD.
:returns:
True if string matches the pattern.
False otherwise.
"""
match = re.match(
r'\d{4}-[01][0-9]-[0-3][0-9]T[0-3][0-9](:[0-5][0-9]){2}\.\d+Z',
time_string)
if match:
return True
else:
return False
class TestValidationJson(base.BaseTestCase):
def setUp(self):
super(TestValidationJson, self).setUp()
self.module = mock.MagicMock()
def test_callback_instantiation(self):
"""
Verifying that the CallbackModule is instantiated properly.
Test checks presence of CallbackBase in the inheritance chain,
in order to ensure that folowing tests are performed with
the correct assumptions.
"""
callback = vf_validation_json.CallbackModule()
self.assertEqual(type(callback).__mro__[1], CallbackBase)
"""
Every ansible callback needs to define variable with name and version.
The validation_json plugin also defines CALLBACK_TYPE,
so we need to check it too.
"""
self.assertIn('CALLBACK_NAME', dir(callback))
self.assertIn('CALLBACK_VERSION', dir(callback))
self.assertIn('CALLBACK_TYPE', dir(callback))
self.assertEqual(callback.CALLBACK_NAME, 'validation_json')
self.assertIsInstance(callback.CALLBACK_VERSION, float)
self.assertEqual(callback.CALLBACK_TYPE, 'aggregate')
"""
Additionally, the 'validation_json' callback performs several
other operations during instantiation.
"""
self.assertEqual(callback.results, [])
self.assertEqual(callback.simple_results, [])
self.assertEqual(callback.env, {})
self.assertIsNone(callback.start_time)
"""
Callback time sanity check only verifies general format
of the stored time to be iso format `YYYY-MM-DD HH:MM:SS.mmmmmm`
with 'T' as a separator.
For example: '2020-07-03T13:28:21.224103Z'
"""
self.assertTrue(is_iso_time(callback.current_time))
@mock.patch(
'ansible.playbook.play.Play._uuid',
autospec=True,
return_value='bar')
@mock.patch(
'ansible.playbook.play.Play.get_name',
autospec=True,
return_value='foo')
@mock.patch('ansible.playbook.play.Play')
def test_new_play(self, mock_play, mock_play_name, mock_play_uuid):
"""
From the callback point of view,
both Play and Task are virtually identical.
Test involving them are therefore also very similar.
"""
callback = vf_validation_json.CallbackModule()
callback.env['playbook_name'] = 'fizz'
callback.env['playbook_path'] = 'buzz/fizz'
play_dict = callback._new_play(mock_play)
mock_play_name.assert_called_once()
mock_play_uuid.assert_called_once()
"""
Callback time sanity check only verifies general format
of the stored time to be iso format `YYYY-MM-DD HH:MM:SS.mmmmmm`
with 'T' as a separator.
For example: '2020-07-03T13:28:21.224103Z'
"""
self.assertTrue(is_iso_time(play_dict['play']['duration']['start']))
self.assertEqual('fizz', play_dict['play']['validation_id'])
self.assertEqual('buzz/fizz', play_dict['play']['validation_path'])
@mock.patch(
'ansible.playbook.task.Task._uuid',
autospec=True,
return_value='bar')
@mock.patch(
'ansible.playbook.task.Task.get_name',
autospec=True,
return_value='foo')
@mock.patch('ansible.playbook.task.Task')
def test_new_task(self, mock_task, mock_task_name, mock_task_uuid):
"""
From the callback point of view,
both Play and Task are virtually identical.
Test involving them are therefore also very similar.
"""
callback = vf_validation_json.CallbackModule()
task_dict = callback._new_task(mock_task)
mock_task_name.assert_called_once()
mock_task_uuid.assert_called_once()
"""
Callback time sanity check only verifies general format
of the stored time to be iso format `YYYY-MM-DD HH:MM:SS.mmmmmm`
with 'T' as a separator.
For example: '2020-07-03T13:28:21.224103Z'
"""
self.assertTrue(is_iso_time(task_dict['task']['duration']['start']))
def test_val_task(self):
"""
_val_task and _val_task_host methods are virtually identical.
Their tests are too.
"""
task_name = 'foo'
expected_dict = {
'task': {
'name': task_name,
'hosts': {}
}
}
callback = vf_validation_json.CallbackModule()
self.assertEqual(
expected_dict,
callback._val_task(task_name=task_name))
def test_val_task_host(self):
"""
_val_task and _val_task_host methods are virtually identical.
Their tests are too.
"""
task_name = 'foo'
expected_dict = {
'task': {
'name': task_name,
'hosts': {}
}
}
callback = vf_validation_json.CallbackModule()
self.assertEqual(
expected_dict,
callback._val_task_host(task_name=task_name))
@mock.patch('os.path.basename',
autospec=True,
return_value='foo.yaml')
@mock.patch('os.path.splitext',
autospec=True,
return_value=['foo', '.yaml'])
@mock.patch('ansible.parsing.dataloader.DataLoader', autospec=True)
def test_v2_playbook_on_start(self, mock_loader,
mock_path_splitext, mock_path_basename):
callback = vf_validation_json.CallbackModule()
dummy_playbook = Playbook(mock_loader)
dummy_playbook._basedir = '/bar'
dummy_playbook._file_name = '/bar/foo.yaml'
callback.v2_playbook_on_start(dummy_playbook)
mock_path_basename.assert_called_once_with('/bar/foo.yaml')
mock_path_splitext.assert_called_once_with('foo.yaml')
self.assertEqual('foo', callback.env['playbook_name'])
self.assertEqual('/bar', callback.env['playbook_path'])
@mock.patch(
'validations_libs.callback_plugins.vf_validation_json.CallbackModule._new_play',
autospec=True,
return_value={'play': {'host': 'foo'}})
@mock.patch('ansible.playbook.play.Play', autospec=True)
def test_v2_playbook_on_play_start(self, mock_play, mock_new_play):
callback = vf_validation_json.CallbackModule()
callback.v2_playbook_on_play_start(mock_play)
self.assertIn({'play': {'host': 'foo'}}, callback.results)
@mock.patch(
'validations_libs.callback_plugins.vf_validation_json.CallbackModule._new_task',
autospec=True,
return_value={'task': {'host': 'foo'}})
@mock.patch('ansible.playbook.task.Task', autospec=True)
def test_v2_playbook_on_task_start(self, mock_task, mock_new_task):
"""
CallbackModule methods v2_playbook_on_task_start
and v2_playbook_on_handler_task_start are virtually identical.
The only exception being is_conditional parameter
of the v2_playbook_on_task_start, which isn't used by the method
at all.
Therefore both of their tests share documentation.
In order to verify methods functionality we first append
a dummy result at the end of CallbackModule.result list.
Simple dictionary is more than sufficient.
"""
callback = vf_validation_json.CallbackModule()
callback.results.append(
{
'fizz': 'buzz',
'tasks': []
})
callback.v2_playbook_on_task_start(mock_task, False)
"""
First we verify that CallbackModule._new_task method was indeed
called with supplied arguments.
Afterwards we verify that the supplied dummy task is present
in first (and in our case only) element of CallbackModule.result list.
"""
mock_new_task.assert_called_once_with(callback, mock_task)
self.assertIn({'task': {'host': 'foo'}}, callback.results[0]['tasks'])
@mock.patch(
'validations_libs.callback_plugins.vf_validation_json.CallbackModule._new_task',
autospec=True,
return_value={'task': {'host': 'foo'}})
@mock.patch('ansible.playbook.task.Task', autospec=True)
def test_v2_playbook_on_handler_task_start(self, mock_task, mock_new_task):
"""
CallbackModule methods v2_playbook_on_task_start
and v2_playbook_on_handler_task_start are virtually identical.
The only exception being is_conditional parameter
of the v2_playbook_on_task_start, which isn't used by the method
at all.
Therefore both of their tests share documentation.
In order to verify methods functionality we first append
a dummy result at the end of CallbackModule.result list.
Simple dictionary is more than sufficient.
"""
callback = vf_validation_json.CallbackModule()
callback.results.append(
{
'fizz': 'buzz',
'tasks': []
})
callback.v2_playbook_on_handler_task_start(mock_task)
"""
First we verify that CallbackModule._new_task method was indeed
called with supplied arguments.
Afterwards we verify that the supplied dummy task is present
in first (and in our case only) element of CallbackModule.result list.
"""
mock_new_task.assert_called_once_with(callback, mock_task)
self.assertIn({'task': {'host': 'foo'}}, callback.results[0]['tasks'])
@mock.patch(
'json.dumps',
return_value='json_dump_foo',
autospec=True)
@mock.patch(
'validations_libs.callback_plugins.vf_validation_json.open',
create=True)
def test_v2_playbook_on_stats(self, mock_open,
mock_json_dumps):
results = [
{
'play': {
'id': 'fizz'
}
}
]
vf_validation_json.VALIDATIONS_LOG_DIR = '/home/foo/validations'
callback = vf_validation_json.CallbackModule()
dummy_stats = AggregateStats()
callback.results = results
callback.simple_results = results
callback.env['playbook_name'] = 'foo'
callback.current_time = 'foo-bar-fooTfoo:bar:foo.fizz'
dummy_stats.processed['foohost'] = 5
output = {
'plays': results,
'stats': {'foohost': {
'ok': 0,
'failures': 0,
'unreachable': 0,
'changed': 0,
'skipped': 0,
'rescued': 0,
'ignored': 0}},
'validation_output': results
}
log_file = "{}/{}_{}_{}.json".format(
"/home/foo/validations",
'fizz',
'foo',
'foo-bar-fooTfoo:bar:foo.fizz')
kwargs = {
'cls': AnsibleJSONEncoder,
'indent': 4,
'sort_keys': True
}
callback.v2_playbook_on_stats(dummy_stats)
mock_write = mock_open.return_value.__enter__.return_value.write
mock_open.assert_called_once_with(log_file, 'w')
mock_json_dumps.assert_called_once_with(output, **kwargs)
mock_write.assert_called_once_with('json_dump_foo')
@mock.patch(
'json.dumps',
return_value='json_dump_foo',
autospec=True)
@mock.patch(
'validations_libs.callback_plugins.vf_validation_json.open',
create=True)
def test_v2_playbook_on_no_hosts_matched(self, mock_open,
mock_json_dumps):
results = [
{
'play': {
'id': 'fizz'
}
}
]
validation_task = {
'task': {
'name': 'No tasks run',
'hosts': {}}}
vf_validation_json.VALIDATIONS_LOG_DIR = '/home/foo/validations'
callback = vf_validation_json.CallbackModule()
dummy_stats = AggregateStats()
callback.results = results
callback.simple_results = results
callback.env['playbook_name'] = 'foo'
callback.current_time = 'foo-bar-fooTfoo:bar:foo.fizz'
dummy_stats.processed['foohost'] = 5
no_match_result = validation_task
no_match_result['task']['status'] = "FAILED"
no_match_result['task']['info'] = (
"None of the hosts specified"
" were matched in the inventory file")
output = {
'plays': results,
'stats': {
'No host matched': {
'changed': 0,
'failures': 1,
'ignored': 0,
'ok': 0,
'rescued': 0,
'skipped': 0,
'unreachable': 0}},
'validation_output': results + [no_match_result]
}
log_file = "{}/{}_{}_{}.json".format(
"/home/foo/validations",
'fizz',
'foo',
'foo-bar-fooTfoo:bar:foo.fizz')
kwargs = {
'cls': AnsibleJSONEncoder,
'indent': 4,
'sort_keys': True
}
callback.v2_playbook_on_no_hosts_matched()
mock_write = mock_open.return_value.__enter__.return_value.write
mock_open.assert_called_once_with(log_file, 'w')
mock_json_dumps.assert_called_once_with(output, **kwargs)
mock_write.assert_called_once_with('json_dump_foo')
@mock.patch('time.time', return_value=99.99)
@mock.patch(
'validations_libs.callback_plugins.vf_validation_json.secondsToStr',
return_value='99.99')
def test_record_task_result(self, mock_secondsToStr, mock_time):
"""
Method CallbackModule._record_task_result works mostly with dicts
and performs few other calls. Therefore the assertions are placed
on calls to those few functions and the operations performed
with supplied MagicMock objects.
"""
mock_on_info = mock.MagicMock()
mock_result = mock.MagicMock()
"""
As we have just initialized the callback, we can't expect it to have
populated properties as the method expects.
Following lines explicitly set all necessary properties.
"""
callback_results = [
{
'play': {
'id': 'fizz',
'duration': {}
},
'tasks': [
{
'hosts': {}
}
]
}
]
callback_simple_results = [
{
'task': {
'hosts': {
}
}
}
]
callback = vf_validation_json.CallbackModule()
callback.results = callback_results
callback.simple_results = callback_simple_results
callback.start_time = 0
callback._record_task_result(mock_on_info, mock_result)
mock_time.assert_called()
mock_secondsToStr.assert_called_once_with(99.99)
"""
Asserting on set lets us check if the method accessed all expected
properties of our MagicMock, while also leaving space for
possible future expansion.
"""
self.assertGreaterEqual(set(dir(mock_result)), set(['_result', '_host', '_task']))
@mock.patch(
'validations_libs.callback_plugins.vf_validation_json.CallbackModule._record_task_result',
autospec=True)
def test_getattribute_valid_listed(self, mock_record_task_result):
"""
All of the listed attribute names are checked.
The __getattribute__ method returns a partial,
the args supplied to it are stored a tuple.
"""
listed_names = ['v2_runner_on_ok', 'v2_runner_on_failed',
'v2_runner_on_unreachable', 'v2_runner_on_skipped']
callback = vf_validation_json.CallbackModule()
for name in listed_names:
attribute = callback.__getattribute__(name)
self.assertEqual(
({name.split('_')[-1]: True},),
attribute.args)
@mock.patch(
'validations_libs.callback_plugins.vf_validation_json.CallbackModule._record_task_result',
autospec=True)
def test_getattribute_valid_unlisted(self, mock_record_task_result):
"""
Since the validation_json.CallbackModule defines it's own
__getattribute__ method, we can't use `dir` to safely check
the name of attributes individually,
as dir itself uses the __getattribute__ method.
Instead we check if the namespace of the CallbackBase class
is a subset of validation_json.CallbackModule namespace.
"""
callback = vf_validation_json.CallbackModule()
listed_names = set(dir(callback))
self.assertTrue(listed_names.issuperset(set(dir(CallbackBase))))
def test_getattribute_invalid(self):
"""
Attempting to call __getattribute__ method with invalid attribute
name should result in exception.
"""
callback = vf_validation_json.CallbackModule()
fake_names = [name + 'x' for name in [
'v2_runner_on_ok', 'v2_runner_on_failed',
'v2_runner_on_unreachable', 'v2_runner_on_skipped']]
for name in fake_names:
self.assertRaises(AttributeError, callback.__getattribute__, name)

741
validations_libs/tests/callback_plugins/test_vf_validation_output.py Normal file
View File

@ -0,0 +1,741 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""
test_validation_output
----------------------------------
Tests for `validation_output` callback plugin.
"""
try:
from unittest import mock
except ImportError:
import mock
from oslotest import base
from ansible.plugins.callback import CallbackBase
from validations_libs.callback_plugins import vf_validation_output
class MockStats(mock.MagicMock):
"""
MockStats mimics some behavior of the ansible.executor.stats.AggregateStats.
Othewise it behaves like an ordinary MagicMock
"""
summary = {}
def summarize(self, anything):
return self.summary.get(anything, self.summary)
class DummyResults(dict):
"""
DummyResults is used in tests as a substitute, mimicking the behavior
of the ansible.executor.task_results.TaskResults class.
"""
def __init__(self):
self.task_fields = {}
class TestValidationOutput(base.BaseTestCase):
def setUp(self):
super(TestValidationOutput, self).setUp()
self.module = mock.MagicMock()
def test_callback_instantiation(self):
"""
Verifying that the CallbackModule is instantiated properly.
Test checks presence of CallbackBase in the inheritance chain,
in order to ensure that folowing tests are performed with
the correct assumptions.
"""
callback = vf_validation_output.CallbackModule()
self.assertEqual(type(callback).__mro__[1], CallbackBase)
"""
Every ansible callback needs to define variable with name and version.
The validation_output plugin also defines CALLBACK_TYPE,
so we need to check it too.
"""
self.assertIn('CALLBACK_NAME', dir(callback))
self.assertIn('CALLBACK_VERSION', dir(callback))
self.assertIn('CALLBACK_TYPE', dir(callback))
self.assertEqual(callback.CALLBACK_NAME, 'validation_output')
self.assertIsInstance(callback.CALLBACK_VERSION, float)
self.assertEqual(callback.CALLBACK_TYPE, 'stdout')
@mock.patch('ansible.constants.COLOR_ERROR')
@mock.patch('ansible.constants.COLOR_WARN')
@mock.patch('pprint.pprint')
@mock.patch(
'validations_libs.callback_plugins.vf_validation_output.FAILURE_TEMPLATE',
create=True)
@mock.patch(
'ansible.utils.display.Display.display',
create=True)
def test_print_failure_message_script(self, mock_display,
mock_failure_template, mock_pprint,
mock_color_warn, mock_color_error):
"""
The test places assertions on the values of arguments passed
to the format method of the FAILURE_TEMPLATE obj, and the display
method of the ansible.utils.display.Display class.
As such it mostly deals with string manipulation, and is therefore
sensitive to localisation and formatting changes,
including the color of the output text.
"""
mock_abridged_result = mock.MagicMock()
mock_results = DummyResults()
mock_results._task_fields = {
'action': 'script',
'args': '_raw_params'
}
host_name = 'foo'
task_name = 'bar'
mock_results['results'] = [
{
'foo': 'bar',
'failed': 5
}
]
mock_results['rc'] = 'fizz'
mock_results['invocation'] = {
'module_args': {
'_raw_params': 'buzz'
},
}
callback = vf_validation_output.CallbackModule()
callback.print_failure_message(
host_name,
task_name,
mock_results,
mock_abridged_result
)
mock_failure_template.format.assert_called_once_with(
task_name,
host_name,
'Script `buzz` exited with code: fizz'
)
mock_display.assert_called_once_with(
mock_failure_template.format(),
color=mock_color_error
)
@mock.patch('ansible.constants.COLOR_ERROR')
@mock.patch('ansible.constants.COLOR_WARN')
@mock.patch('pprint.pprint')
@mock.patch(
'validations_libs.callback_plugins.vf_validation_output.FAILURE_TEMPLATE',
create=True)
@mock.patch(
'ansible.utils.display.Display.display',
create=True)
def test_print_failure_message_rc_and_cmd(self, mock_display,
mock_failure_template,
mock_pprint,
mock_color_warn,
mock_color_error):
"""
The test places assertions on the values of arguments passed
to the format method of the FAILURE_TEMPLATE obj, and the display
method of the ansible.utils.display.Display class.
As such it mostly deals with string manipulation, and is therefore
sensitive to localisation and formatting changes,
including the color of the output text.
The test assumes that both 'rc' and 'cmd' keys are present
within the results object.
"""
mock_abridged_result = mock.MagicMock()
host_name = 'foo'
task_name = 'bar'
result_dict = {
'results': [
{
'foo': 'bar',
'failed': 5
}
],
'cmd': 'fizz',
'rc': 'buzz'
}
callback = vf_validation_output.CallbackModule()
callback.print_failure_message(
host_name,
task_name,
result_dict,
mock_abridged_result
)
mock_failure_template.format.assert_called_once_with(
task_name,
host_name,
"Command `fizz` exited with code: buzz"
)
mock_display.assert_called_once_with(
mock_failure_template.format(),
color=mock_color_error
)
@mock.patch('ansible.constants.COLOR_ERROR')
@mock.patch('ansible.constants.COLOR_WARN')
@mock.patch('pprint.pprint')
@mock.patch(
'validations_libs.callback_plugins.vf_validation_output.FAILURE_TEMPLATE',
create=True)
@mock.patch(
'ansible.utils.display.Display.display',
create=True)
def test_print_failure_message_unknown_error_no_warn(self, mock_display,
mock_failure_template,
mock_pprint,
mock_color_warn,
mock_color_error):
"""
The test places assertions on the values of arguments passed
to the format method of the FAILURE_TEMPLATE obj, the display
method of the ansible.utils.display.Display class
and the pprint method.
As such it mostly deals with string manipulation, and is therefore
sensitive to localisation and formatting changes,
including the color of the output text.
Test assumes that neither pair of 'rc' and 'cmd' keys,
nor the 'msg' key, exists within the results object.
Therefore an Unknown error is assumed to have occured and
output is adjusted accordignly.
Furthermore, the test assumes that in absence of 'warnings' key,
no warnings will be passed to the display method.
"""
mock_abridged_result = mock.MagicMock()
host_name = 'foo'
task_name = 'bar'
result_dict = {
'results': [
{
'foo': 'bar',
'failed': 5
}
]
}
callback = vf_validation_output.CallbackModule()
callback.print_failure_message(
host_name,
task_name,
result_dict,
mock_abridged_result
)
mock_failure_template.format.assert_called_once_with(
task_name,
host_name,
"Unknown error"
)
mock_display.assert_called_once_with(
mock_failure_template.format(),
color=mock_color_error
)
mock_pprint.assert_called_once_with(
mock_abridged_result,
indent=4)
@mock.patch('ansible.constants.COLOR_ERROR')
@mock.patch('ansible.constants.COLOR_WARN')
@mock.patch('pprint.pprint')
@mock.patch(
'validations_libs.callback_plugins.vf_validation_output.FAILURE_TEMPLATE',
create=True)
@mock.patch(
'ansible.utils.display.Display.display',
create=True)
def test_print_failure_message_unknown_error_warn(self, mock_display,
mock_failure_template,
mock_pprint,
mock_color_warn,
mock_color_error):
"""
The test places assertions on the values of arguments passed
to the format method of the FAILURE_TEMPLATE obj, the display
method of the ansible.utils.display.Display class
and the pprint method.
As such it mostly deals with string manipulation, and is therefore
sensitive to localisation and formatting changes,
including the color of the output text.
Test assumes that neither pair of 'rc' and 'cmd' keys,
nor the 'msg' key, exists within the results object.
Therefore an Unknown error is assumed to have occured and
output is adjusted accordignly.
Furthermore, the test assumes that when the 'warnings' key is present,
the display method will be called with list entries as arguments.
"""
mock_abridged_result = mock.MagicMock()
host_name = 'foo'
task_name = 'bar'
result_dict = {
'results': [
{
'foo': 'bar',
'failed': 5
}
],
'warnings': [
'foo'
]
}
callback = vf_validation_output.CallbackModule()
callback.print_failure_message(
host_name,
task_name,
result_dict,
mock_abridged_result)
mock_failure_template.format.assert_called_once_with(
task_name,
host_name,
"Unknown error")
mock_display.assert_has_calls(
[
mock.call(
mock_failure_template.format(),
color=mock_color_error
),
mock.call(
"* foo ",
color=mock_color_warn
)
]
)
mock_pprint.assert_called_once_with(
mock_abridged_result,
indent=4)
@mock.patch('ansible.constants.COLOR_WARN')
@mock.patch(
'validations_libs.callback_plugins.vf_validation_output.WARNING_TEMPLATE',
create=True)
@mock.patch(
'validations_libs.callback_plugins.vf_validation_output.CallbackModule._dump_results',
return_value={'foo': 'bar'})
@mock.patch(
'ansible.utils.display.Display.display',
create=True)
def test_v2_runner_on_ok_warnings(self, mock_display, mock_dump_results,
mock_warn_template, mock_error_color):
"""
The test asserts on argumets passed to print_failure_message method.
In order to check the call arguments we need
initialize them before passing the mock_results to the tested method.
It is a bit hacky, but the most simple way I know how to make sure
the relevant mocks ids don't change.
If you know how to improve it, go for it.
"""
mock_results = mock.MagicMock()
result_dict = {
'results': [
{
'foo': 'bar',
'failed': 5
}
],
'warnings': [
'foo'
]
}
mock_results._result = result_dict
mock_results._host()
mock_results._task.get_name()
mock_results._task_fields()
callback = vf_validation_output.CallbackModule()
callback.v2_runner_on_ok(mock_results)
mock_dump_results.assert_called_once_with(result_dict)
mock_warn_template.format.assert_called_once_with(
mock_results._task.get_name(),
mock_results._host,
'foo\n')
mock_display.assert_called_once_with(
mock_warn_template.format(),
color=mock_error_color)
@mock.patch('ansible.constants.COLOR_OK')
@mock.patch(
'validations_libs.callback_plugins.vf_validation_output.DEBUG_TEMPLATE',
create=True)
@mock.patch(
'validations_libs.callback_plugins.vf_validation_output.CallbackModule._dump_results',
return_value={'foo': 'bar'})
@mock.patch(
'ansible.utils.display.Display.display',
create=True)
def test_v2_runner_on_ok_debug_vars(self, mock_display, mock_dump_results,
mock_debug_template, mock_ok_color):
"""
The test asserts on argumets passed to print_failure_message method.
In order to check the call arguments we need
initialize them before passing the mock_results to the tested method.
It is a bit hacky, but the most simple way I know how to make sure
the relevant mocks ids don't change.
If you know how to improve it, go for it.
"""
mock_results = mock.MagicMock()
result_dict = {
'results': [
{
'foo': 'bar',
'failed': 5
}
],
'fizz': 'buzz'
}
mock_results._result = result_dict
mock_results._host()
mock_results._task.get_name()
mock_results._task_fields = {
'action': 'debug',
'args': {'var': 'fizz'}
}
callback = vf_validation_output.CallbackModule()
callback.v2_runner_on_ok(mock_results)
mock_dump_results.assert_called_once_with(result_dict)
mock_debug_template.format.assert_called_once_with(
mock_results._host,
"fizz: buzz"
)
mock_display.assert_called_once_with(
mock_debug_template.format(),
color=mock_ok_color)
@mock.patch('ansible.constants.COLOR_OK')
@mock.patch(
'validations_libs.callback_plugins.vf_validation_output.DEBUG_TEMPLATE',
create=True)
@mock.patch(
'validations_libs.callback_plugins.vf_validation_output.CallbackModule._dump_results',
return_value={'foo': 'bar'})
@mock.patch(
'ansible.utils.display.Display.display',
create=True)
def test_v2_runner_on_ok_debug_msg(self, mock_display, mock_dump_results,
mock_debug_template, mock_ok_color):
"""
The test asserts on argumets passed to print_failure_message method.
In order to check the call arguments we need
initialize them before passing the mock_results to the tested method.
It is a bit hacky, but the most simple way I know how to make sure
the relevant mocks ids don't change.
If you know how to improve it, go for it.
"""
mock_results = mock.MagicMock()
result_dict = {
'results': [
{
'foo': 'bar',
'failed': 5
}
]
}
mock_results._result = result_dict
mock_results._host()
mock_results._task.get_name()
mock_results._task_fields = {
'action': 'debug',
'args': {'msg': 'fizz'}
}
callback = vf_validation_output.CallbackModule()
callback.v2_runner_on_ok(mock_results)
mock_dump_results.assert_called_once_with(result_dict)
mock_debug_template.format.assert_called_once_with(
mock_results._host,
"Message: fizz"
)
mock_display.assert_called_once_with(
mock_debug_template.format(),
color=mock_ok_color)
@mock.patch(
'validations_libs.callback_plugins.vf_validation_output.CallbackModule._dump_results',
return_value={'foo': 'bar'})
@mock.patch('validations_libs.callback_plugins.vf_validation_output.CallbackModule.print_failure_message')
def test_v2_runner_on_failed_one_result(self, mock_print, mock_dump_results):
"""
The test asserts on argumets passed to print_failure_message method.
In order to check the call arguments we need
initialize them before passing the mock_results to the tested method.
It is a bit hacky, but the most simple way I know how to make sure
the relevant mocks ids don't change.
If you know how to improve it, go for it.
"""
mock_results = mock.MagicMock()
result_dict = {
'results': [
{
'foo': 'bar',
'failed': 5
}
]
}
mock_results._result = result_dict
mock_results._host()
mock_results._task.get_name()
callback = vf_validation_output.CallbackModule()
callback.v2_runner_on_failed(mock_results)
mock_print.assert_called_once_with(
mock_results._host,
mock_results._task.get_name(),
{
'foo': 'bar',
'failed': 5
},
{
'foo': 'bar',
'failed': 5
}
)
@mock.patch(
'validations_libs.callback_plugins.vf_validation_output.CallbackModule._dump_results',
return_value={'foo': 'bar'})
@mock.patch('validations_libs.callback_plugins.vf_validation_output.CallbackModule.print_failure_message')
def test_v2_runner_on_failed_no_result(self, mock_print, mock_dump_results):
"""
The test asserts on argumets passed to print_failure_message method.
In order to check the call arguments we need
initialize them before passing the mock_results to the tested method.
It is a bit hacky, but the most simple way I know how to make sure
the relevant mocks ids don't change.
If you know how to improve it, go for it.
"""
mock_results = mock.MagicMock()
result_dict = {}
mock_results._result = result_dict
mock_results._host()
mock_results._task.get_name()
callback = vf_validation_output.CallbackModule()
callback.v2_runner_on_failed(mock_results)
mock_print.assert_called_once_with(
mock_results._host,
mock_results._task.get_name(),
{},
{
'foo': 'bar'
}
)
@mock.patch('validations_libs.callback_plugins.vf_validation_output.CallbackModule.print_failure_message')
def test_v2_runner_on_unreachable(self, mock_print):
"""
The test asserts on argumets passed to print_failure_message method.
In order to check the call arguments we need
initialize them before passing the mock_results to the tested method.
It is a bit hacky, but the most simple way I know how to make sure
the relevant mocks ids don't change.
If you know how to improve it, go for it.
"""
mock_results = mock.MagicMock()
results_dict = {'msg': 'The host is unreachable.'}
mock_results._host()
mock_results._task.get_name()
callback = vf_validation_output.CallbackModule()
callback.v2_runner_on_unreachable(mock_results)
mock_print.assert_called_once_with(
mock_results._host,
mock_results._task.get_name(),
results_dict,
results_dict)
@mock.patch('ansible.constants.COLOR_ERROR')
@mock.patch('ansible.constants.COLOR_OK')
@mock.patch('validations_libs.callback_plugins.vf_validation_output.print')
@mock.patch.object(CallbackBase, '_display.display', create=True)
def test_v2_playbook_on_stats_no_hosts(self, mock_display, mock_print,
mock_color_ok, mock_color_error):
"""
In case we don't supply any hosts, we expect the method not to call
display or related methods and attributes even once.
The final call to print function is not an ideal place for assertion,
as the string might get localised and/or adjusted in the future.
"""
callback = vf_validation_output.CallbackModule()
dummy_stats = mock.MagicMock()
callback.v2_playbook_on_stats(dummy_stats)
mock_color_ok.assert_not_called()
mock_color_error.assert_not_called()
mock_display.assert_not_called()
mock_print.assert_called_once()
@mock.patch('ansible.constants.COLOR_ERROR')
@mock.patch('ansible.constants.COLOR_OK')
@mock.patch('validations_libs.callback_plugins.vf_validation_output.print')
@mock.patch(
'validations_libs.callback_plugins.vf_validation_output.sorted',
return_value=['bar', 'foo'])
@mock.patch('ansible.utils.display.Display.display')
@mock.patch('ansible.plugins.callback.CallbackBase')
def test_v2_playbook_on_stats_no_fail(self, mock_callback_base,
mock_display, mock_sorted,
mock_print, mock_color_ok,
mock_color_error):
"""
When we have hosts and their state is not specified,
we expect them to be considered a `pass` and the display method
to be called with appropriate arguments.
The final call to print function is not an ideal place for assertion,
as the string might get localised and/or adjusted in the future.
"""
callback = vf_validation_output.CallbackModule()
dummy_stats = MockStats()
callback.v2_playbook_on_stats(dummy_stats)
mock_display.assert_called_with('* foo', color=mock_color_ok)
mock_print.assert_called_once()
@mock.patch('ansible.constants.COLOR_ERROR')
@mock.patch('ansible.constants.COLOR_OK')
@mock.patch('validations_libs.callback_plugins.vf_validation_output.print')
@mock.patch(
'validations_libs.callback_plugins.vf_validation_output.sorted',
return_value=['bar', 'buzz', 'fizz', 'foo'])
@mock.patch('ansible.utils.display.Display.display')
@mock.patch('ansible.plugins.callback.CallbackBase')
def test_v2_playbook_on_stats_some_fail(self, mock_callback_base,
mock_display, mock_sorted,
mock_print, mock_color_ok,
mock_color_error):
"""
When at least one host is specified as failure and/or unreachable
we expect it to be considered a `failure` and the display method
to be called with the appropriate arguments in the proper order.
The final call to print function is not an ideal place for assertion,
as the string might get localised and/or adjusted in the future.
"""
callback = vf_validation_output.CallbackModule()
dummy_stats = MockStats()
dummy_stats.summary = {
'fizz': {
'failures': 5
}
}
expected_calls = [
mock.call('* fizz', color=mock_color_error),
mock.call('* bar', color=mock_color_ok),
mock.call('* buzz', color=mock_color_ok),
mock.call('* foo', color=mock_color_ok)
]
callback.v2_playbook_on_stats(dummy_stats)
mock_display.assert_has_calls(expected_calls)
mock_print.assert_called()
@mock.patch('ansible.constants.COLOR_ERROR')
@mock.patch('ansible.constants.COLOR_OK')
@mock.patch('validations_libs.callback_plugins.vf_validation_output.print')
@mock.patch(
'validations_libs.callback_plugins.vf_validation_output.sorted',
return_value=['bar', 'buzz', 'fizz', 'foo'])
@mock.patch('ansible.utils.display.Display.display')
@mock.patch('ansible.plugins.callback.CallbackBase')
def test_v2_playbook_on_stats_all_fail(self, mock_callback_base,
mock_display, mock_sorted,
mock_print, mock_color_ok,
mock_color_error):
"""
When at all hosts are specified as failure and/or unreachable
we expect them to be considered a `failure` and the display method
to be called with the appropriate arguments in the proper order.
The final call to print function is not an ideal place for assertion,
as the string might get localised and/or adjusted in the future.
"""
callback = vf_validation_output.CallbackModule()
dummy_stats = MockStats()
dummy_stats.summary = {
'fizz': {
'failures': 5
},
'foo': {
'failures': 5
},
'bar': {
'failures': 5
},
'buzz': {
'failures': 5
}
}
expected_calls = [
mock.call('* bar', color=mock_color_error),
mock.call('* buzz', color=mock_color_error),
mock.call('* fizz', color=mock_color_error),
mock.call('* foo', color=mock_color_error)
]
callback.v2_playbook_on_stats(dummy_stats)
mock_display.assert_has_calls(expected_calls)
mock_print.assert_called()

198
validations_libs/tests/callback_plugins/test_vf_validation_stdout.py Normal file
View File

@ -0,0 +1,198 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""
test_vf_validation_stdout
----------------------------------
Tests for `vf_validation_stdout` callback plugin.
"""
import re
try:
from unittest import mock
except ImportError:
import mock
from oslotest import base
from validations_libs.callback_plugins import vf_validation_stdout
from ansible.plugins.callback import CallbackBase
def is_iso_time(time_string):
"""
Checks if string represents valid time in ISO format,
with the default delimiter.
Regex is somewhat convoluted, but general enough to last
at least until the 9999 AD.
Returns:
True if string matches the pattern.
False otherwise.
"""
match = re.match(
r'\d{4}-[01][0-9]-[0-3][0-9]T[0-3][0-9](:[0-5][0-9]){2}\.\d+Z',
time_string)
if match:
return True
else:
return False
class TestValidationStdout(base.BaseTestCase):
"""Tests of validation_stdout callback module.
"""
def setUp(self):
super(TestValidationStdout, self).setUp()
self.module = mock.MagicMock()
def test_callback_instantiation(self):
"""
Verifying that the CallbackModule is instantiated properly.
Test checks presence of CallbackBase in the inheritance chain,
in order to ensure that folowing tests are performed with
the correct assumptions.
"""
callback = vf_validation_stdout.CallbackModule()
self.assertEqual(type(callback).__mro__[1], CallbackBase)
"""
Every ansible callback needs to define variable with name and version.
"""
self.assertIn('CALLBACK_NAME', dir(callback))
self.assertIn('CALLBACK_VERSION', dir(callback))
self.assertEqual(callback.CALLBACK_NAME, 'validation_stdout')
self.assertIsInstance(callback.CALLBACK_VERSION, float)
"""
Additionally, the 'validation_stdout' callback performs several
other operations during instantiation.
"""
self.assertEqual(callback.env, {})
self.assertIsNone(callback.start_time)
"""
Callback time sanity check only verifies general format
of the stored time to be iso format `YYYY-MM-DD HH:MM:SS.mmmmmm`
with 'T' as a separator.
For example: '2020-07-03T13:28:21.224103Z'
"""
self.assertTrue(is_iso_time(callback.current_time))
@mock.patch(
'ansible.playbook.play.Play._uuid',
autospec=True,
return_value='bar')
@mock.patch(
'ansible.playbook.play.Play.get_name',
autospec=True,
return_value='foo')
@mock.patch('ansible.playbook.play.Play')
def test_new_play(self, mock_play, mock_play_name, mock_play_uuid):
"""
From the callback point of view,
both Play and Task are virtually identical.
Test involving them are therefore also very similar.
"""
callback = vf_validation_stdout.CallbackModule()
callback.env['playbook_name'] = 'fizz'
callback.env['playbook_path'] = 'buzz/fizz'
play_dict = callback._new_play(mock_play)
mock_play_name.assert_called_once()
mock_play_uuid.assert_called_once()
"""
Callback time sanity check only verifies general format
of the stored time to be iso format `YYYY-MM-DD HH:MM:SS.mmmmmm`
with 'T' as a separator.
For example: '2020-07-03T13:28:21.224103Z'
"""
self.assertTrue(is_iso_time(play_dict['play']['duration']['start']))
self.assertEqual('fizz', play_dict['play']['validation_id'])
self.assertEqual('buzz/fizz', play_dict['play']['validation_path'])
@mock.patch(
'ansible.playbook.task.Task._uuid',
autospec=True,
return_value='bar')
@mock.patch(
'ansible.playbook.task.Task.get_name',
autospec=True,
return_value='foo')
@mock.patch('ansible.playbook.task.Task')
def test_new_task(self, mock_task, mock_task_name, mock_task_uuid):
"""
From the callback point of view,
both Play and Task are virtually identical.
Test involving them are therefore also very similar.
"""
callback = vf_validation_stdout.CallbackModule()
task_dict = callback._new_task(mock_task)
mock_task_name.assert_called_once()
mock_task_uuid.assert_called_once()
"""
Callback time sanity check only verifies general format
of the stored time to be iso format `YYYY-MM-DD HH:MM:SS.mmmmmm`
with 'T' as a separator.
For example: '2020-07-03T13:28:21.224103Z'
"""
self.assertTrue(is_iso_time(task_dict['task']['duration']['start']))
def test_val_task(self):
"""
_val_task and _val_task_host methods are virtually identical.
Their tests are too.
"""
task_name = 'foo'
expected_dict = {
'task': {
'name': task_name,
'hosts': {}
}
}
callback = vf_validation_stdout.CallbackModule()
self.assertEqual(
expected_dict,
callback._val_task(task_name=task_name))
def test_val_task_host(self):
"""
_val_task and _val_task_host methods are virtually identical.
Their tests are too.
"""
task_name = 'foo'
expected_dict = {
'task': {
'name': task_name,
'hosts': {}
}
}
callback = vf_validation_stdout.CallbackModule()
self.assertEqual(
expected_dict,
callback._val_task_host(task_name=task_name))

6
validations_libs/tests/test_validation_actions.py
View File

@ -97,7 +97,7 @@ class TestValidationActions(TestCase):
'playbook_dir': '/tmp/foo',
'parallel_run': True,
'inventory': 'tmp/inventory.yaml',
'output_callback': 'validation_stdout',
'output_callback': 'vf_validation_stdout',
'callback_whitelist': None,
'quiet': True,
'extra_vars': None,
@ -157,7 +157,7 @@ class TestValidationActions(TestCase):
'playbook_dir': '/tmp/foo',
'parallel_run': True,
'inventory': 'tmp/inventory.yaml',
'output_callback': 'validation_stdout',
'output_callback': 'vf_validation_stdout',
'callback_whitelist': None,
'quiet': True,
'extra_vars': None,
@ -229,7 +229,7 @@ class TestValidationActions(TestCase):
playbook_dir='/tmp/foobar/validation-playbooks',
parallel_run=True,
inventory='tmp/inventory.yaml',
output_callback='validation_stdout',
output_callback='vf_validation_stdout',
callback_whitelist=None,
quiet=True,
extra_vars=None,

2
validations_libs/validation_actions.py
View File

@ -331,7 +331,7 @@ class ValidationActions(object):
base_dir=constants.DEFAULT_VALIDATIONS_BASEDIR,
log_path=None, python_interpreter=None, skip_list=None,
callback_whitelist=None,
output_callback='validation_stdout', ssh_user=None,
output_callback='vf_validation_stdout', ssh_user=None,
validation_config=None):
"""Run one or multiple validations by name(s), by group(s) or by
product(s)