mirror of https://github.com/ansible/ansible.git
Consolidate boolean/mk_boolean conversion functions into a single location
Consolidate the module_utils, constants, and config functions that convert values into booleans into a single function in module_utils. Port code to use the module_utils.validate.convert_bool.boolean function isntead of mk_boolean.pull/21336/merge
Toshio Kuratomi
7 years ago
parent
f9c60e1a82
commit
ff22528b07
@ -0,0 +1,26 @@
|
||||
# Copyright: 2017, Ansible Project
|
||||
# License: GNU General Public License v3 or later (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt )
|
||||
|
||||
from ansible.module_utils.six import binary_type, text_type
|
||||
from ansible.module_utils._text import to_text
|
||||
|
||||
|
||||
BOOLEANS_TRUE = frozenset(('y', 'yes', 'on', '1', 'true', 't', 1, 1.0, True))
|
||||
BOOLEANS_FALSE = frozenset(('n', 'no', 'off', '0', 'false', 'f', 0, 0.0, False))
|
||||
BOOLEANS = BOOLEANS_TRUE.union(BOOLEANS_FALSE)
|
||||
|
||||
|
||||
def boolean(value, strict=True):
|
||||
if isinstance(value, bool):
|
||||
return value
|
||||
|
||||
normalized_value = value
|
||||
if isinstance(value, (text_type, binary_type)):
|
||||
normalized_value = to_text(value, errors='surrogate_or_strict').lower()
|
||||
|
||||
if normalized_value in BOOLEANS_TRUE:
|
||||
return True
|
||||
elif normalized_value in BOOLEANS_FALSE or not strict:
|
||||
return False
|
||||
|
||||
raise TypeError('%s is not a valid boolean. Valid booleans include: %s' % (to_text(value), ', '.join(repr(i) for i in BOOLEANS)))
|
||||
@ -1 +0,0 @@
|
||||
injector.py
|
||||
@ -0,0 +1,256 @@
|
||||
#!/usr/bin/env python
|
||||
"""Interpreter and code coverage injector for use with ansible-test.
|
||||
|
||||
The injector serves two main purposes:
|
||||
|
||||
1) Control the python interpreter used to run test tools and ansible code.
|
||||
2) Provide optional code coverage analysis of ansible code.
|
||||
|
||||
The injector is executed one of two ways:
|
||||
|
||||
1) On the controller via a symbolic link such as ansible or pytest.
|
||||
This is accomplished by prepending the injector directory to the PATH by ansible-test.
|
||||
|
||||
2) As the python interpreter when running ansible modules.
|
||||
This is only supported when connecting to the local host.
|
||||
Otherwise set the ANSIBLE_TEST_REMOTE_INTERPRETER environment variable.
|
||||
It can be empty to auto-detect the python interpreter on the remote host.
|
||||
If not empty it will be used to set ansible_python_interpreter.
|
||||
|
||||
NOTE: Running ansible-test with the --tox option or inside a virtual environment
|
||||
may prevent the injector from working for tests which use connection
|
||||
types other than local, or which use become, due to lack of permissions
|
||||
to access the interpreter for the virtual environment.
|
||||
"""
|
||||
|
||||
from __future__ import absolute_import, print_function
|
||||
|
||||
import errno
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import pipes
|
||||
import logging
|
||||
import getpass
|
||||
|
||||
logger = logging.getLogger('injector') # pylint: disable=locally-disabled, invalid-name
|
||||
# pylint: disable=locally-disabled, invalid-name
|
||||
config = None # type: InjectorConfig
|
||||
|
||||
|
||||
class InjectorConfig(object):
|
||||
"""Mandatory configuration."""
|
||||
def __init__(self, config_path):
|
||||
"""Initialize config."""
|
||||
with open(config_path) as config_fd:
|
||||
_config = json.load(config_fd)
|
||||
|
||||
self.python_interpreter = _config['python_interpreter']
|
||||
self.coverage_file = _config['coverage_file']
|
||||
|
||||
# Read from the environment instead of config since it needs to be changed by integration test scripts.
|
||||
# It also does not need to flow from the controller to the remote. It is only used on the controller.
|
||||
self.remote_interpreter = os.environ.get('ANSIBLE_TEST_REMOTE_INTERPRETER', None)
|
||||
|
||||
self.arguments = [to_text(c) for c in sys.argv]
|
||||
|
||||
|
||||
def to_text(value):
|
||||
"""
|
||||
:type value: str | None
|
||||
:rtype: str | None
|
||||
"""
|
||||
if value is None:
|
||||
return None
|
||||
|
||||
if isinstance(value, bytes):
|
||||
return value.decode('utf-8')
|
||||
|
||||
return u'%s' % value
|
||||
|
||||
|
||||
def main():
|
||||
"""Main entry point."""
|
||||
global config # pylint: disable=locally-disabled, global-statement
|
||||
|
||||
formatter = logging.Formatter('%(asctime)s %(process)d %(levelname)s %(message)s')
|
||||
log_name = 'ansible-test-coverage.%s.log' % getpass.getuser()
|
||||
self_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
|
||||
handler = logging.FileHandler(os.path.join('/tmp', log_name))
|
||||
handler.setFormatter(formatter)
|
||||
logger.addHandler(handler)
|
||||
|
||||
handler = logging.FileHandler(os.path.abspath(os.path.join(self_dir, '..', 'logs', log_name)))
|
||||
handler.setFormatter(formatter)
|
||||
logger.addHandler(handler)
|
||||
|
||||
logger.setLevel(logging.DEBUG)
|
||||
|
||||
try:
|
||||
logger.debug('Self: %s', __file__)
|
||||
|
||||
config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'injector.json')
|
||||
|
||||
try:
|
||||
config = InjectorConfig(config_path)
|
||||
except IOError:
|
||||
logger.exception('Error reading config: %s', config_path)
|
||||
exit('No injector config found. Set ANSIBLE_TEST_REMOTE_INTERPRETER if the test is not connecting to the local host.')
|
||||
|
||||
logger.debug('Arguments: %s', ' '.join(pipes.quote(c) for c in config.arguments))
|
||||
logger.debug('Python interpreter: %s', config.python_interpreter)
|
||||
logger.debug('Remote interpreter: %s', config.remote_interpreter)
|
||||
logger.debug('Coverage file: %s', config.coverage_file)
|
||||
|
||||
require_cwd = False
|
||||
|
||||
if os.path.basename(__file__) == 'injector.py':
|
||||
if config.coverage_file:
|
||||
args, env, require_cwd = cover()
|
||||
else:
|
||||
args, env = runner()
|
||||
else:
|
||||
args, env = injector()
|
||||
|
||||
logger.debug('Run command: %s', ' '.join(pipes.quote(c) for c in args))
|
||||
|
||||
altered_cwd = False
|
||||
|
||||
try:
|
||||
cwd = os.getcwd()
|
||||
except OSError as ex:
|
||||
# some platforms, such as OS X, may not allow querying the working directory when using become to drop privileges
|
||||
if ex.errno != errno.EACCES:
|
||||
raise
|
||||
if require_cwd:
|
||||
# make sure the program we execute can determine the working directory if it's required
|
||||
cwd = '/'
|
||||
os.chdir(cwd)
|
||||
altered_cwd = True
|
||||
else:
|
||||
cwd = None
|
||||
|
||||
logger.debug('Working directory: %s%s', cwd or '?', ' (altered)' if altered_cwd else '')
|
||||
|
||||
for key in sorted(env.keys()):
|
||||
logger.debug('%s=%s', key, env[key])
|
||||
|
||||
os.execvpe(args[0], args, env)
|
||||
except Exception as ex:
|
||||
logger.fatal(ex)
|
||||
raise
|
||||
|
||||
|
||||
def injector():
|
||||
"""
|
||||
:rtype: list[str], dict[str, str]
|
||||
"""
|
||||
command = os.path.basename(__file__)
|
||||
executable = find_executable(command)
|
||||
|
||||
if config.coverage_file:
|
||||
args, env = coverage_command()
|
||||
else:
|
||||
args, env = [config.python_interpreter], os.environ.copy()
|
||||
|
||||
args += [executable]
|
||||
|
||||
if command in ('ansible', 'ansible-playbook', 'ansible-pull'):
|
||||
if config.remote_interpreter is None:
|
||||
interpreter = os.path.join(os.path.dirname(__file__), 'injector.py')
|
||||
elif config.remote_interpreter == '':
|
||||
interpreter = None
|
||||
else:
|
||||
interpreter = config.remote_interpreter
|
||||
|
||||
if interpreter:
|
||||
args += ['--extra-vars', 'ansible_python_interpreter=' + interpreter]
|
||||
|
||||
args += config.arguments[1:]
|
||||
|
||||
return args, env
|
||||
|
||||
|
||||
def runner():
|
||||
"""
|
||||
:rtype: list[str], dict[str, str]
|
||||
"""
|
||||
args, env = [config.python_interpreter], os.environ.copy()
|
||||
|
||||
args += config.arguments[1:]
|
||||
|
||||
return args, env
|
||||
|
||||
|
||||
def cover():
|
||||
"""
|
||||
:rtype: list[str], dict[str, str], bool
|
||||
"""
|
||||
if len(config.arguments) > 1:
|
||||
executable = config.arguments[1]
|
||||
else:
|
||||
executable = ''
|
||||
|
||||
require_cwd = False
|
||||
|
||||
if os.path.basename(executable).startswith('ansible_module_'):
|
||||
args, env = coverage_command()
|
||||
# coverage requires knowing the working directory
|
||||
require_cwd = True
|
||||
else:
|
||||
args, env = [config.python_interpreter], os.environ.copy()
|
||||
|
||||
args += config.arguments[1:]
|
||||
|
||||
return args, env, require_cwd
|
||||
|
||||
|
||||
def coverage_command():
|
||||
"""
|
||||
:rtype: list[str], dict[str, str]
|
||||
"""
|
||||
self_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
|
||||
args = [
|
||||
config.python_interpreter,
|
||||
'-m',
|
||||
'coverage.__main__',
|
||||
'run',
|
||||
'--rcfile',
|
||||
os.path.join(self_dir, '.coveragerc'),
|
||||
]
|
||||
|
||||
env = os.environ.copy()
|
||||
env['COVERAGE_FILE'] = config.coverage_file
|
||||
|
||||
return args, env
|
||||
|
||||
|
||||
def find_executable(executable):
|
||||
"""
|
||||
:type executable: str
|
||||
:rtype: str
|
||||
"""
|
||||
self = os.path.abspath(__file__)
|
||||
path = os.environ.get('PATH', os.defpath)
|
||||
seen_dirs = set()
|
||||
|
||||
for path_dir in path.split(os.pathsep):
|
||||
if path_dir in seen_dirs:
|
||||
continue
|
||||
|
||||
seen_dirs.add(path_dir)
|
||||
candidate = os.path.abspath(os.path.join(path_dir, executable))
|
||||
|
||||
if candidate == self:
|
||||
continue
|
||||
|
||||
if os.path.exists(candidate) and os.access(candidate, os.F_OK | os.X_OK):
|
||||
return candidate
|
||||
|
||||
raise Exception('Executable "%s" not found in path: %s' % (executable, path))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@ -0,0 +1,60 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# Copyright: (c) 2017 Ansible Project
|
||||
# License: GNU General Public License v3 or later (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt )
|
||||
|
||||
# Make coding more python3-ish
|
||||
from __future__ import (absolute_import, division)
|
||||
__metaclass__ = type
|
||||
|
||||
import pytest
|
||||
|
||||
from ansible.module_utils.parsing.convert_bool import boolean
|
||||
|
||||
|
||||
class TestBoolean:
|
||||
def test_bools(self):
|
||||
assert boolean(True) is True
|
||||
assert boolean(False) is False
|
||||
|
||||
def test_none(self):
|
||||
with pytest.raises(TypeError):
|
||||
assert boolean(None, strict=True) is False
|
||||
assert boolean(None, strict=False) is False
|
||||
|
||||
def test_numbers(self):
|
||||
assert boolean(1) is True
|
||||
assert boolean(0) is False
|
||||
assert boolean(0.0) is False
|
||||
|
||||
# Current boolean() doesn't consider these to be true values
|
||||
# def test_other_numbers(self):
|
||||
# assert boolean(2) is True
|
||||
# assert boolean(-1) is True
|
||||
# assert boolean(0.1) is True
|
||||
|
||||
def test_strings(self):
|
||||
assert boolean("true") is True
|
||||
assert boolean("TRUE") is True
|
||||
assert boolean("t") is True
|
||||
assert boolean("yes") is True
|
||||
assert boolean("y") is True
|
||||
assert boolean("on") is True
|
||||
|
||||
def test_junk_values_nonstrict(self):
|
||||
assert boolean("flibbity", strict=False) is False
|
||||
assert boolean(42, strict=False) is False
|
||||
assert boolean(42.0, strict=False) is False
|
||||
assert boolean(object(), strict=False) is False
|
||||
|
||||
def test_junk_values_strict(self):
|
||||
with pytest.raises(TypeError):
|
||||
assert boolean("flibbity", strict=True)is False
|
||||
|
||||
with pytest.raises(TypeError):
|
||||
assert boolean(42, strict=True)is False
|
||||
|
||||
with pytest.raises(TypeError):
|
||||
assert boolean(42.0, strict=True)is False
|
||||
|
||||
with pytest.raises(TypeError):
|
||||
assert boolean(object(), strict=True)is False
|
||||
Loading…
Reference in New Issue