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

Port urls.py to python3 and other byte vs text fixes (#16124)

Browse Source 
* Port urls.py to python3

Fixes (largely normalizing byte vs text strings) for python3

* Rework what we do with attributes that aren't set already.

* Comments
pull/16135/head
Toshio Kuratomi 9 years ago
parent 434c949d03
commit 5a3493be5f
9 changed files with 152 additions and 89 deletions
Show Stats Download Patch File Download Diff File

42
lib/ansible/executor/module_common.py
Unescape Escape View File

@ -339,7 +339,8 @@ class ModuleDepFinder(ast.NodeVisitor):
# import ansible.module_utils.MODLIB[.MODLIBn] [as asname]
for alias in (a for a in node.names if a.name.startswith('ansible.module_utils.')):
py_mod = alias.name[self.IMPORT_PREFIX_SIZE:]
self.submodules.add((py_mod,))
py_mod = tuple(py_mod.split('.'))
self.submodules.add(py_mod)
self.generic_visit(node)
def visit_ImportFrom(self, node):
@ -409,17 +410,25 @@ def recursive_finder(name, data, py_module_names, py_module_cache, zf):
# (Have to exclude them a second time once the paths are processed)
for py_module_name in finder.submodules.difference(py_module_names):
module_info = None
# Check whether either the last or the second to last identifier is
# a module name
for idx in (1, 2):
if len(py_module_name) < idx:
break
try:
module_info = imp.find_module(py_module_name[-idx],
[os.path.join(_SNIPPET_PATH, *py_module_name[:-idx])])
break
except ImportError:
continue
if py_module_name[0] == 'six':
# Special case the python six library because it messes up the
# import process in an incompatible way
module_info = imp.find_module('six', [_SNIPPET_PATH])
py_module_name = ('six',)
idx = 0
else:
# Check whether either the last or the second to last identifier is
# a module name
for idx in (1, 2):
if len(py_module_name) < idx:
break
try:
module_info = imp.find_module(py_module_name[-idx],
[os.path.join(_SNIPPET_PATH, *py_module_name[:-idx])])
break
except ImportError:
continue
# Could not find the module. Construct a helpful error message.
if module_info is None:
@ -534,7 +543,7 @@ def _find_snippet_imports(module_name, module_data, module_path, module_args, ta
if module_substyle == 'python':
params = dict(ANSIBLE_MODULE_ARGS=module_args,)
python_repred_params = to_bytes(repr(json.dumps(params)), errors='strict')
python_repred_params = repr(json.dumps(params))
try:
compression_method = getattr(zipfile, module_compression)
@ -592,7 +601,7 @@ def _find_snippet_imports(module_name, module_data, module_path, module_args, ta
# be a better place to run this
os.mkdir(lookup_path)
display.debug('ZIPLOADER: Writing module')
with open(cached_module_filename + '-part', 'w') as f:
with open(cached_module_filename + '-part', 'wb') as f:
f.write(zipdata)
# Rename the file into its final position in the cache so
@ -613,6 +622,7 @@ def _find_snippet_imports(module_name, module_data, module_path, module_args, ta
raise AnsibleError('A different worker process failed to create module file. Look at traceback for that process for debugging information.')
# Fool the check later... I think we should just remove the check
py_module_names.add(('basic',))
zipdata = to_unicode(zipdata, errors='strict')
shebang, interpreter = _get_shebang(u'/usr/bin/python', task_vars)
if shebang is None:
@ -725,7 +735,7 @@ def modify_module(module_name, module_path, module_args, task_vars=dict(), modul
(module_data, module_style, shebang) = _find_snippet_imports(module_name, module_data, module_path, module_args, task_vars, module_compression)
if module_style == 'binary':
return (module_data, module_style, shebang)
return (module_data, module_style, to_unicode(shebang, nonstring='passthru'))
elif shebang is None:
lines = module_data.split(b"\n", 1)
if lines[0].startswith(b"#!"):
@ -748,4 +758,4 @@ def modify_module(module_name, module_path, module_args, task_vars=dict(), modul
else:
shebang = to_bytes(shebang, errors='strict')
return (module_data, module_style, shebang)
return (module_data, module_style, to_unicode(shebang, nonstring='passthru'))

53
lib/ansible/module_utils/basic.py
Unescape Escape View File

@ -148,6 +148,8 @@ except ImportError:
print('\n{"msg": "SyntaxError: probably due to installed simplejson being for a different python version", "failed": true}')
sys.exit(1)
from ansible.module_utils.six import PY2, PY3, b, binary_type, text_type, string_types
HAVE_SELINUX=False
try:
import selinux
@ -540,11 +542,11 @@ def _load_params():
fd.close()
else:
buffer = sys.argv[1]
if sys.version_info >= (3,):
if PY3:
buffer = buffer.encode('utf-8', errors='surrogateescape')
# default case, read from stdin
else:
if sys.version_info < (3,):
if PY2:
buffer = sys.stdin.read()
else:
buffer = sys.stdin.buffer.read()
@ -557,7 +559,7 @@ def _load_params():
print('\n{"msg": "Error: Module unable to decode valid JSON on stdin. Unable to figure out what parameters were passed", "failed": true}')
sys.exit(1)
if sys.version_info < (3,):
if PY2:
params = json_dict_unicode_to_bytes(params)
try:
@ -1567,7 +1569,7 @@ class AnsibleModule(object):
# TODO: surrogateescape is a danger here on Py3
journal_msg = remove_values(msg, self.no_log_values)
if sys.version_info >= (3,):
if PY3:
syslog_msg = journal_msg
else:
syslog_msg = journal_msg.encode('utf-8', 'replace')
@ -1967,9 +1969,13 @@ class AnsibleModule(object):
shell = True
elif isinstance(args, basestring) and use_unsafe_shell:
shell = True
elif isinstance(args, basestring):
if isinstance(args, unicode):
elif isinstance(args, string_types):
# On python2.6 and below, shlex has problems with text type
# On python3, shlex needs a text type.
if PY2 and isinstance(args, text_type):
args = args.encode('utf-8')
elif PY3 and isinstance(args, binary_type):
args = args.decode('utf-8', errors='surrogateescape')
args = shlex.split(args)
else:
msg = "Argument 'args' to run_command must be list or string"
@ -1977,6 +1983,11 @@ class AnsibleModule(object):
prompt_re = None
if prompt_regex:
if isinstance(prompt_regex, text_type):
if PY3:
prompt_regex = prompt_regex.encode('utf-8', errors='surrogateescape')
elif PY2:
prompt_regex = prompt_regex.encode('utf-8')
try:
prompt_re = re.compile(prompt_regex, re.MULTILINE)
except re.error:
@ -2020,15 +2031,15 @@ class AnsibleModule(object):
# create a printable version of the command for use
# in reporting later, which strips out things like
# passwords from the args list
if isinstance(args, basestring):
if isinstance(args, unicode):
b_args = args.encode('utf-8')
else:
b_args = args
to_clean_args = shlex.split(b_args)
del b_args
to_clean_args = args
if PY2:
if isinstance(args, text_type):
to_clean_args = args.encode('utf-8')
else:
to_clean_args = args
if isinstance(args, binary_type):
to_clean_args = args.decode('utf-8', errors='replace')
if isinstance(args, (text_type, binary_type)):
to_clean_args = shlex.split(to_clean_args)
clean_args = []
is_passwd = False
@ -2087,13 +2098,19 @@ class AnsibleModule(object):
# the communication logic here is essentially taken from that
# of the _communicate() function in ssh.py
stdout = ''
stderr = ''
stdout = b('')
stderr = b('')
rpipes = [cmd.stdout, cmd.stderr]
if data:
if not binary_data:
data += '\n'
if isinstance(data, text_type):
if PY3:
errors = 'surrogateescape'
else:
errors = 'strict'
data = data.encode('utf-8', errors=errors)
cmd.stdin.write(data)
cmd.stdin.close()
@ -2102,12 +2119,12 @@ class AnsibleModule(object):
if cmd.stdout in rfd:
dat = os.read(cmd.stdout.fileno(), 9000)
stdout += dat
if dat == '':
if dat == b(''):
rpipes.remove(cmd.stdout)
if cmd.stderr in rfd:
dat = os.read(cmd.stderr.fileno(), 9000)
stderr += dat
if dat == '':
if dat == b(''):
rpipes.remove(cmd.stderr)
# if we're checking for prompts, do it now
if prompt_re:

107
lib/ansible/module_utils/urls.py
Unescape Escape View File

@ -98,14 +98,20 @@ except ImportError:
# Python 3
import http.client as httplib
import ansible.module_utils.six.moves.urllib.request as urllib_request
import ansible.module_utils.six.moves.urllib.error as urllib_error
try:
import urllib2
HAS_URLLIB2 = True
except:
HAS_URLLIB2 = False
# python3
import urllib.request as urllib_request
from urllib.request import AbstractHTTPHandler
except ImportError:
# python2
import urllib2 as urllib_request
from urllib2 import AbstractHTTPHandler
try:
import urlparse
from ansible.module_utils.six.moves.urllib.parse import urlparse, urlunparse
HAS_URLPARSE = True
except:
HAS_URLPARSE = False
@ -140,7 +146,8 @@ if HAS_SSL:
PROTOCOL = ssl.PROTOCOL_TLSv1
if not HAS_SSLCONTEXT and HAS_SSL:
try:
import ctypes, ctypes.util
import ctypes
import ctypes.util
except ImportError:
# python 2.4 (likely rhel5 which doesn't have tls1.1 support in its openssl)
pass
@ -159,7 +166,6 @@ if not HAS_SSLCONTEXT and HAS_SSL:
del libssl
HAS_MATCH_HOSTNAME = True
try:
from ssl import match_hostname, CertificateError
@ -308,18 +314,22 @@ zKPZsZ2miVGclicJHzm5q080b1p/sZtuKIEZk6vZqEg=
# Exceptions
#
class ConnectionError(Exception):
"""Failed to connect to the server"""
pass
class ProxyError(ConnectionError):
"""Failure to connect because of a proxy"""
pass
class SSLValidationError(ConnectionError):
"""Failure to connect due to SSL validation failing"""
pass
class NoSSLError(SSLValidationError):
"""Needed to connect to an HTTPS url but no ssl library available to verify the certificate"""
pass
@ -327,7 +337,7 @@ class NoSSLError(SSLValidationError):
# Some environments (Google Compute Engine's CoreOS deploys) do not compile
# against openssl and thus do not have any HTTPS support.
CustomHTTPSConnection = CustomHTTPSHandler = None
if hasattr(httplib, 'HTTPSConnection') and hasattr(urllib2, 'HTTPSHandler'):
if hasattr(httplib, 'HTTPSConnection') and hasattr(urllib_request, 'HTTPSHandler'):
class CustomHTTPSConnection(httplib.HTTPSConnection):
def __init__(self, *args, **kwargs):
httplib.HTTPSConnection.__init__(self, *args, **kwargs)
@ -355,16 +365,18 @@ if hasattr(httplib, 'HTTPSConnection') and hasattr(urllib2, 'HTTPSHandler'):
if HAS_SSLCONTEXT:
self.sock = self.context.wrap_socket(sock, server_hostname=server_hostname)
elif HAS_URLLIB3_SNI_SUPPORT:
self.sock = ssl_wrap_socket(sock, keyfile=self.key_file, cert_reqs=ssl.CERT_NONE, certfile=self.cert_file, ssl_version=PROTOCOL, server_hostname=server_hostname)
self.sock = ssl_wrap_socket(sock, keyfile=self.key_file, cert_reqs=ssl.CERT_NONE, certfile=self.cert_file, ssl_version=PROTOCOL,
server_hostname=server_hostname)
else:
self.sock = ssl.wrap_socket(sock, keyfile=self.key_file, certfile=self.cert_file, ssl_version=PROTOCOL)
class CustomHTTPSHandler(urllib2.HTTPSHandler):
class CustomHTTPSHandler(urllib_request.HTTPSHandler):
def https_open(self, req):
return self.do_open(CustomHTTPSConnection, req)
https_request = urllib2.AbstractHTTPHandler.do_request_
https_request = AbstractHTTPHandler.do_request_
def generic_urlparse(parts):
'''
@ -424,7 +436,8 @@ def generic_urlparse(parts):
generic_parts['port'] = None
return generic_parts
class RequestWithMethod(urllib2.Request):
class RequestWithMethod(urllib_request.Request):
'''
Workaround for using DELETE/PUT/etc with urllib2
Originally contained in library/net_infrastructure/dnsmadeeasy
@ -434,13 +447,13 @@ class RequestWithMethod(urllib2.Request):
if headers is None:
headers = {}
self._method = method.upper()
urllib2.Request.__init__(self, url, data, headers)
urllib_request.Request.__init__(self, url, data, headers)
def get_method(self):
if self._method:
return self._method
else:
return urllib2.Request.get_method(self)
return urllib_request.Request.get_method(self)
def RedirectHandlerFactory(follow_redirects=None, validate_certs=True):
@ -450,7 +463,7 @@ def RedirectHandlerFactory(follow_redirects=None, validate_certs=True):
where ``open_url`` or ``fetch_url`` are used multiple times in a module.
"""
class RedirectHandler(urllib2.HTTPRedirectHandler):
class RedirectHandler(urllib_request.HTTPRedirectHandler):
"""This is an implementation of a RedirectHandler to match the
functionality provided by httplib2. It will utilize the value of
``follow_redirects`` that is passed into ``RedirectHandlerFactory``
@ -460,12 +473,12 @@ def RedirectHandlerFactory(follow_redirects=None, validate_certs=True):
def redirect_request(self, req, fp, code, msg, hdrs, newurl):
handler = maybe_add_ssl_handler(newurl, validate_certs)
if handler:
urllib2._opener.add_handler(handler)
urllib_request._opener.add_handler(handler)
if follow_redirects == 'urllib2':
return urllib2.HTTPRedirectHandler.redirect_request(self, req, fp, code, msg, hdrs, newurl)
return urllib_request.HTTPRedirectHandler.redirect_request(self, req, fp, code, msg, hdrs, newurl)
elif follow_redirects in ['no', 'none', False]:
raise urllib2.HTTPError(newurl, code, msg, hdrs, fp)
raise urllib_error.HTTPError(newurl, code, msg, hdrs, fp)
do_redirect = False
if follow_redirects in ['all', 'yes', True]:
@ -481,13 +494,12 @@ def RedirectHandlerFactory(follow_redirects=None, validate_certs=True):
newheaders = dict((k,v) for k,v in req.headers.items()
if k.lower() not in ("content-length", "content-type")
)
return urllib2.Request(newurl,
headers=newheaders,
origin_req_host=req.get_origin_req_host(),
unverifiable=True)
return urllib_request.Request(newurl,
headers=newheaders,
origin_req_host=req.get_origin_req_host(),
unverifiable=True)
else:
raise urllib2.HTTPError(req.get_full_url(), code, msg, hdrs,
fp)
raise urllib_error.HTTPError(req.get_full_url(), code, msg, hdrs, fp)
return RedirectHandler
@ -519,7 +531,7 @@ def build_ssl_validation_error(hostname, port, paths):
raise SSLValidationError(' '.join(msg) % (hostname, port, ", ".join(paths)))
class SSLValidationHandler(urllib2.BaseHandler):
class SSLValidationHandler(urllib_request.BaseHandler):
'''
A custom handler class for SSL validation.
@ -606,7 +618,7 @@ class SSLValidationHandler(urllib2.BaseHandler):
env_no_proxy = os.environ.get('no_proxy')
if env_no_proxy:
env_no_proxy = env_no_proxy.split(',')
netloc = urlparse.urlparse(url).netloc
netloc = urlparse(url).netloc
for host in env_no_proxy:
if netloc.endswith(host) or netloc.split(':')[0].endswith(host):
@ -637,7 +649,7 @@ class SSLValidationHandler(urllib2.BaseHandler):
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
if https_proxy:
proxy_parts = generic_urlparse(urlparse.urlparse(https_proxy))
proxy_parts = generic_urlparse(urlparse(https_proxy))
port = proxy_parts.get('port') or 443
s.connect((proxy_parts.get('hostname'), port))
if proxy_parts.get('scheme') == 'http':
@ -690,13 +702,15 @@ class SSLValidationHandler(urllib2.BaseHandler):
https_request = http_request
def maybe_add_ssl_handler(url, validate_certs):
# FIXME: change the following to use the generic_urlparse function
# to remove the indexed references for 'parsed'
parsed = urlparse.urlparse(url)
parsed = urlparse(url)
if parsed[0] == 'https' and validate_certs:
if not HAS_SSL:
raise NoSSLError('SSL validation is not available in your version of python. You can use validate_certs=False, however this is unsafe and not recommended')
raise NoSSLError('SSL validation is not available in your version of python. You can use validate_certs=False,'
' however this is unsafe and not recommended')
# do the cert validation
netloc = parsed[1]
@ -712,13 +726,15 @@ def maybe_add_ssl_handler(url, validate_certs):
# add it to the list of handlers
return SSLValidationHandler(hostname, port)
# Rewrite of fetch_url to not require the module environment
def open_url(url, data=None, headers=None, method=None, use_proxy=True,
force=False, last_mod_time=None, timeout=10, validate_certs=True,
url_username=None, url_password=None, http_agent=None,
force_basic_auth=False, follow_redirects='urllib2'):
'''
Fetches a file from an HTTP/FTP server using urllib2
Does not require the module environment
'''
handlers = []
ssl_handler = maybe_add_ssl_handler(url, validate_certs)
@ -727,7 +743,7 @@ def open_url(url, data=None, headers=None, method=None, use_proxy=True,
# FIXME: change the following to use the generic_urlparse function
# to remove the indexed references for 'parsed'
parsed = urlparse.urlparse(url)
parsed = urlparse(url)
if parsed[0] != 'ftp':
username = url_username
@ -749,10 +765,10 @@ def open_url(url, data=None, headers=None, method=None, use_proxy=True,
parsed[1] = netloc
# reconstruct url without credentials
url = urlparse.urlunparse(parsed)
url = urlunparse(parsed)
if username and not force_basic_auth:
passman = urllib2.HTTPPasswordMgrWithDefaultRealm()
passman = urllib_request.HTTPPasswordMgrWithDefaultRealm()
# this creates a password manager
passman.add_password(None, netloc, username, password)
@ -760,7 +776,7 @@ def open_url(url, data=None, headers=None, method=None, use_proxy=True,
# because we have put None at the start it will always
# use this username/password combination for urls
# for which `theurl` is a super-url
authhandler = urllib2.HTTPBasicAuthHandler(passman)
authhandler = urllib_request.HTTPBasicAuthHandler(passman)
# create the AuthHandler
handlers.append(authhandler)
@ -781,7 +797,7 @@ def open_url(url, data=None, headers=None, method=None, use_proxy=True,
headers["Authorization"] = basic_auth_header(username, password)
if not use_proxy:
proxyhandler = urllib2.ProxyHandler({})
proxyhandler = urllib_request.ProxyHandler({})
handlers.append(proxyhandler)
if HAS_SSLCONTEXT and not validate_certs:
@ -791,7 +807,7 @@ def open_url(url, data=None, headers=None, method=None, use_proxy=True,
context.options |= ssl.OP_NO_SSLv3
context.verify_mode = ssl.CERT_NONE
context.check_hostname = False
handlers.append(urllib2.HTTPSHandler(context=context))
handlers.append(urllib_request.HTTPSHandler(context=context))
# pre-2.6 versions of python cannot use the custom https
# handler, since the socket class is lacking create_connection.
@ -801,15 +817,15 @@ def open_url(url, data=None, headers=None, method=None, use_proxy=True,
handlers.append(RedirectHandlerFactory(follow_redirects, validate_certs))
opener = urllib2.build_opener(*handlers)
urllib2.install_opener(opener)
opener = urllib_request.build_opener(*handlers)
urllib_request.install_opener(opener)
if method:
if method.upper() not in ('OPTIONS','GET','HEAD','POST','PUT','DELETE','TRACE','CONNECT','PATCH'):
raise ConnectionError('invalid HTTP request method; %s' % method.upper())
request = RequestWithMethod(url, method.upper(), data)
else:
request = urllib2.Request(url, data)
request = urllib_request.Request(url, data)
# add the custom agent header, to help prevent issues
# with sites that block the default urllib agent string
@ -836,16 +852,18 @@ def open_url(url, data=None, headers=None, method=None, use_proxy=True,
# have a timeout parameter
urlopen_args.append(timeout)
r = urllib2.urlopen(*urlopen_args)
r = urllib_request.urlopen(*urlopen_args)
return r
#
# Module-related functions
#
def basic_auth_header(username, password):
return "Basic %s" % base64.b64encode("%s:%s" % (username, password))
def url_argument_spec():
'''
Creates an argument spec that can be used with any module
@ -863,15 +881,14 @@ def url_argument_spec():
)
def fetch_url(module, url, data=None, headers=None, method=None,
use_proxy=True, force=False, last_mod_time=None, timeout=10):
'''
Fetches a file from an HTTP/FTP server using urllib2. Requires the module environment
'''
if not HAS_URLLIB2:
module.fail_json(msg='urllib2 is not installed')
elif not HAS_URLPARSE:
if not HAS_URLPARSE:
module.fail_json(msg='urlparse is not installed')
# Get validate_certs from the module params
@ -904,7 +921,7 @@ def fetch_url(module, url, data=None, headers=None, method=None,
except (ConnectionError, ValueError):
e = get_exception()
module.fail_json(msg=str(e))
except urllib2.HTTPError:
except urllib_error.HTTPError:
e = get_exception()
try:
body = e.read()
@ -912,7 +929,7 @@ def fetch_url(module, url, data=None, headers=None, method=None,
body = ''
info.update(dict(msg=str(e), body=body, **e.info()))
info['status'] = e.code
except urllib2.URLError:
except urllib_error.URLError:
e = get_exception()
code = int(getattr(e, 'code', -1))
info.update(dict(msg="Request failed: %s" % str(e), status=code))

26
lib/ansible/playbook/base.py
Unescape Escape View File

@ -112,10 +112,30 @@ class Base:
if hasattr(self, method):
return getattr(self, method)()
value = self._attributes[prop_name]
if value is None and hasattr(self, '_get_parent_attribute'):
# value_found is here because we think that value needs to be changed
# in the future. self._attributes[prop_name] will return None
# sometimes, apparently if it's not explicitly set in the playbook.
# This would seem to make None a sentinel value. However, the user
# could set the attribute to None explicitly (via !!nil) which will
# not be recognized because it's being used as a sentinel. And
# sometimes _attributes[prop_name] throws a KeyError so None doesn't
# always mean that prop_name was not set. To work around these
# issues, value_found is here so that if value's behaviour is changed
# in the future, things can still be made to work.
try:
value = self._attributes[prop_name]
value_found = True
except KeyError:
value = None
value_found = False
if (value is None or not value_found) and hasattr(self, '_get_parent_attribute'):
value = self._get_parent_attribute(prop_name)
return value
value_found = True
if value_found:
return value
raise AttributeError("'%s' object has no attribute '%s'" % (self.__class__.__name__, prop_name))
@staticmethod
def _generic_s(prop_name, self, value):

2
lib/ansible/plugins/action/__init__.py
Unescape Escape View File

@ -276,7 +276,7 @@ class ActionBase(with_metaclass(ABCMeta, object)):
data = jsonify(data)
afd, afile = tempfile.mkstemp()
afo = os.fdopen(afd, 'w')
afo = os.fdopen(afd, 'wb')
try:
data = to_bytes(data, errors='strict')
afo.write(data)

2
test/code-smell/use-compat-six.sh
Unescape Escape View File

@ -3,7 +3,7 @@
# Do we want to check dynamic inventory, bin, etc?
BASEDIR=${1-"lib"}
SIX_USERS=$(find "$BASEDIR" -name '*.py' -exec grep -wH six \{\} \;|grep import |grep -v ansible.compat|grep -v ansible.module_utils)
SIX_USERS=$(find "$BASEDIR" -name '*.py' -exec grep -wH six \{\} \;|grep import |grep -v ansible.compat| grep -v ansible.module_utils.six)
if test -n "$SIX_USERS" ; then
printf "$SIX_USERS"
exit 1

4
test/units/module_utils/basic/test_run_command.py
Unescape Escape View File

@ -168,13 +168,13 @@ class TestAnsibleModuleRunCommand(unittest.TestCase):
def test_text_stdin(self):
(rc, stdout, stderr) = self.module.run_command('/bin/foo', data='hello world')
self.assertEqual(self.cmd.stdin.getvalue(), 'hello world\n')
self.assertEqual(self.cmd.stdin.getvalue(), b'hello world\n')
def test_ascii_stdout(self):
self.cmd_out[sentinel.stdout] = BytesIO(b'hello')
(rc, stdout, stderr) = self.module.run_command('/bin/cat hello.txt')
self.assertEqual(rc, 0)
self.assertEqual(stdout, 'hello')
self.assertEqual(stdout, b'hello')
def test_utf8_output(self):
self.cmd_out[sentinel.stdout] = BytesIO(u'Žarn§'.encode('utf-8'))

3
test/units/module_utils/test_basic.py
Unescape Escape View File

@ -114,7 +114,7 @@ class TestModuleUtilsBasic(unittest.TestCase):
# from ansible.module_utils import basic
@patch.object(builtins, '__import__')
@unittest.skipIf(sys.version_info[0] >= 3, "Python 3 is not supported on targets (yet)")
@unittest.skipIf(sys.version_info[0] >= 3, "literal_eval is available in every version of Python3")
def test_module_utils_basic_import_literal_eval(self, mock_import):
def _mock_import(name, *args, **kwargs):
try:
@ -270,7 +270,6 @@ class TestModuleUtilsBasic(unittest.TestCase):
with patch('os.path.realpath', return_value='/path/to/foo/'):
self.assertEqual(get_module_path(), '/path/to/foo')
@unittest.skipIf(sys.version_info[0] >= 3, "Python 3 is not supported on targets (yet)")
def test_module_utils_basic_ansible_module_creation(self):
from ansible.module_utils import basic

2
test/units/plugins/action/test_action.py
Unescape Escape View File

@ -219,7 +219,7 @@ class TestActionBase(unittest.TestCase):
mock_connection.module_implementation_preferences = ('',)
(style, shebang, data, path) = action_base._configure_module(mock_task.action, mock_task.args)
self.assertEqual(style, "new")
self.assertEqual(shebang, b"#!/usr/bin/python")
self.assertEqual(shebang, u"#!/usr/bin/python")
# test module not found
self.assertRaises(AnsibleError, action_base._configure_module, 'badmodule', mock_task.args)

Write Preview
Loading…
Cancel
Save