# -*- coding: utf-8 -*- import traceback import unittest import os import os.path import re import tempfile import yaml import passlib.hash import string import StringIO import copy import tempfile import shutil from nose.plugins.skip import SkipTest from mock import patch import ansible.utils import ansible.errors import ansible.constants as C import ansible.utils.template as template2 from ansible.module_utils.splitter import split_args from ansible import __version__ import sys reload(sys) sys.setdefaultencoding("utf8") class TestUtils(unittest.TestCase): def _is_fips(self): try: data = open('/proc/sys/crypto/fips_enabled').read().strip() except: return False if data != '1': return False return True def test_before_comment(self): ''' see if we can detect the part of a string before a comment. Used by INI parser in inventory ''' input = "before # comment" expected = "before " actual = ansible.utils.before_comment(input) self.assertEqual(expected, actual) input = "before \# not a comment" expected = "before # not a comment" actual = ansible.utils.before_comment(input) self.assertEqual(expected, actual) input = "" expected = "" actual = ansible.utils.before_comment(input) self.assertEqual(expected, actual) input = "#" expected = "" actual = ansible.utils.before_comment(input) self.assertEqual(expected, actual) ##################################### ### check_conditional tests def test_check_conditional_jinja2_literals(self): # see http://jinja.pocoo.org/docs/templates/#literals # none self.assertEqual(ansible.utils.check_conditional( None, '/', {}), True) self.assertEqual(ansible.utils.check_conditional( '', '/', {}), True) # list self.assertEqual(ansible.utils.check_conditional( ['true'], '/', {}), True) self.assertEqual(ansible.utils.check_conditional( ['false'], '/', {}), False) # non basestring or list self.assertEqual(ansible.utils.check_conditional( {}, '/', {}), {}) # boolean self.assertEqual(ansible.utils.check_conditional( 'true', '/', {}), True) self.assertEqual(ansible.utils.check_conditional( 'false', '/', {}), False) self.assertEqual(ansible.utils.check_conditional( 'True', '/', {}), True) self.assertEqual(ansible.utils.check_conditional( 'False', '/', {}), False) # integer self.assertEqual(ansible.utils.check_conditional( '1', '/', {}), True) self.assertEqual(ansible.utils.check_conditional( '0', '/', {}), False) # string, beware, a string is truthy unless empty self.assertEqual(ansible.utils.check_conditional( '"yes"', '/', {}), True) self.assertEqual(ansible.utils.check_conditional( '"no"', '/', {}), True) self.assertEqual(ansible.utils.check_conditional( '""', '/', {}), False) def test_check_conditional_jinja2_variable_literals(self): # see http://jinja.pocoo.org/docs/templates/#literals # boolean self.assertEqual(ansible.utils.check_conditional( 'var', '/', {'var': 'True'}), True) self.assertEqual(ansible.utils.check_conditional( 'var', '/', {'var': 'true'}), True) self.assertEqual(ansible.utils.check_conditional( 'var', '/', {'var': 'False'}), False) self.assertEqual(ansible.utils.check_conditional( 'var', '/', {'var': 'false'}), False) # integer self.assertEqual(ansible.utils.check_conditional( 'var', '/', {'var': '1'}), True) self.assertEqual(ansible.utils.check_conditional( 'var', '/', {'var': 1}), True) self.assertEqual(ansible.utils.check_conditional( 'var', '/', {'var': '0'}), False) self.assertEqual(ansible.utils.check_conditional( 'var', '/', {'var': 0}), False) # string, beware, a string is truthy unless empty self.assertEqual(ansible.utils.check_conditional( 'var', '/', {'var': '"yes"'}), True) self.assertEqual(ansible.utils.check_conditional( 'var', '/', {'var': '"no"'}), True) self.assertEqual(ansible.utils.check_conditional( 'var', '/', {'var': '""'}), False) # Python boolean in Jinja2 expression self.assertEqual(ansible.utils.check_conditional( 'var', '/', {'var': True}), True) self.assertEqual(ansible.utils.check_conditional( 'var', '/', {'var': False}), False) def test_check_conditional_jinja2_expression(self): self.assertEqual(ansible.utils.check_conditional( '1 == 1', '/', {}), True) self.assertEqual(ansible.utils.check_conditional( 'bar == 42', '/', {'bar': 42}), True) self.assertEqual(ansible.utils.check_conditional( 'bar != 42', '/', {'bar': 42}), False) def test_check_conditional_jinja2_expression_in_variable(self): self.assertEqual(ansible.utils.check_conditional( 'var', '/', {'var': '1 == 1'}), True) self.assertEqual(ansible.utils.check_conditional( 'var', '/', {'var': 'bar == 42', 'bar': 42}), True) self.assertEqual(ansible.utils.check_conditional( 'var', '/', {'var': 'bar != 42', 'bar': 42}), False) def test_check_conditional_jinja2_unicode(self): self.assertEqual(ansible.utils.check_conditional( u'"\u00df"', '/', {}), True) self.assertEqual(ansible.utils.check_conditional( u'var == "\u00df"', '/', {'var': u'\u00df'}), True) ##################################### ### key-value parsing def test_parse_kv_basic(self): self.assertEqual(ansible.utils.parse_kv('a=simple b="with space" c="this=that"'), {'a': 'simple', 'b': 'with space', 'c': 'this=that'}) self.assertEqual(ansible.utils.parse_kv('msg=АБВГД'), {'msg': 'АБВГД'}) def test_jsonify(self): self.assertEqual(ansible.utils.jsonify(None), '{}') self.assertEqual(ansible.utils.jsonify(dict(foo='bar', baz=['qux'])), '{"baz": ["qux"], "foo": "bar"}') expected = u'{"baz":["qux"],"foo":"bar"}' self.assertEqual("".join(ansible.utils.jsonify(dict(foo='bar', baz=['qux']), format=True).split()), expected) def test_is_failed(self): self.assertEqual(ansible.utils.is_failed(dict(rc=0)), False) self.assertEqual(ansible.utils.is_failed(dict(rc=1)), True) self.assertEqual(ansible.utils.is_failed(dict()), False) self.assertEqual(ansible.utils.is_failed(dict(failed=False)), False) self.assertEqual(ansible.utils.is_failed(dict(failed=True)), True) self.assertEqual(ansible.utils.is_failed(dict(failed='True')), True) self.assertEqual(ansible.utils.is_failed(dict(failed='true')), True) def test_is_changed(self): self.assertEqual(ansible.utils.is_changed(dict()), False) self.assertEqual(ansible.utils.is_changed(dict(changed=False)), False) self.assertEqual(ansible.utils.is_changed(dict(changed=True)), True) self.assertEqual(ansible.utils.is_changed(dict(changed='True')), True) self.assertEqual(ansible.utils.is_changed(dict(changed='true')), True) def test_path_dwim(self): self.assertEqual(ansible.utils.path_dwim(None, __file__), __file__) self.assertEqual(ansible.utils.path_dwim(None, '~'), os.path.expanduser('~')) self.assertEqual(ansible.utils.path_dwim(None, 'TestUtils.py'), __file__.rstrip('c')) def test_path_dwim_relative(self): self.assertEqual(ansible.utils.path_dwim_relative(__file__, 'units', 'TestUtils.py', os.path.dirname(os.path.dirname(__file__))), __file__.rstrip('c')) def test_json_loads(self): self.assertEqual(ansible.utils.json_loads('{"foo": "bar"}'), dict(foo='bar')) def test_parse_json(self): # leading junk self.assertEqual(ansible.utils.parse_json('ansible\n{"foo": "bar"}'), dict(foo="bar")) # No closing quotation try: rc = ansible.utils.parse_json('foo=bar "') print rc except ValueError: pass else: traceback.print_exc() raise AssertionError('Incorrect exception, expected ValueError') # Failed to parse try: ansible.utils.parse_json('{') except ValueError: pass else: raise AssertionError('Incorrect exception, expected ValueError') def test_parse_yaml(self): #json self.assertEqual(ansible.utils.parse_yaml('{"foo": "bar"}'), dict(foo='bar')) # broken json try: ansible.utils.parse_yaml('{') except ansible.errors.AnsibleError: pass else: raise AssertionError # broken json with path_hint try: ansible.utils.parse_yaml('{', path_hint='foo') except ansible.errors.AnsibleError: pass else: raise AssertionError # yaml with front-matter self.assertEqual(ansible.utils.parse_yaml("---\nfoo: bar"), dict(foo='bar')) # yaml no front-matter self.assertEqual(ansible.utils.parse_yaml('foo: bar'), dict(foo='bar')) # yaml indented first line (See #6348) self.assertEqual(ansible.utils.parse_yaml(' - foo: bar\n baz: qux'), [dict(foo='bar', baz='qux')]) def test_process_common_errors(self): # no quote self.assertTrue('YAML thought it' in ansible.utils.process_common_errors('', 'foo: {{bar}}', 6)) # extra colon self.assertTrue('an extra unquoted colon' in ansible.utils.process_common_errors('', 'foo: bar:', 8)) # match self.assertTrue('same kind of quote' in ansible.utils.process_common_errors('', 'foo: "{{bar}}"baz', 6)) self.assertTrue('same kind of quote' in ansible.utils.process_common_errors('', "foo: '{{bar}}'baz", 6)) # unbalanced self.assertTrue('We could be wrong' in ansible.utils.process_common_errors('', 'foo: "bad" "wolf"', 6)) self.assertTrue('We could be wrong' in ansible.utils.process_common_errors('', "foo: 'bad' 'wolf'", 6)) def test_process_yaml_error(self): data = 'foo: bar\n baz: qux' try: ansible.utils.parse_yaml(data) except yaml.YAMLError, exc: try: ansible.utils.process_yaml_error(exc, data, __file__) except ansible.errors.AnsibleYAMLValidationFailed, e: self.assertTrue('Syntax Error while loading' in str(e)) else: raise AssertionError('Incorrect exception, expected AnsibleYAMLValidationFailed') data = 'foo: bar\n baz: {{qux}}' try: ansible.utils.parse_yaml(data) except yaml.YAMLError, exc: try: ansible.utils.process_yaml_error(exc, data, __file__) except ansible.errors.AnsibleYAMLValidationFailed, e: self.assertTrue('Syntax Error while loading' in str(e)) else: raise AssertionError('Incorrect exception, expected AnsibleYAMLValidationFailed') data = '\xFF' try: ansible.utils.parse_yaml(data) except yaml.YAMLError, exc: try: ansible.utils.process_yaml_error(exc, data, __file__) except ansible.errors.AnsibleYAMLValidationFailed, e: self.assertTrue('Check over' in str(e)) else: raise AssertionError('Incorrect exception, expected AnsibleYAMLValidationFailed') data = '\xFF' try: ansible.utils.parse_yaml(data) except yaml.YAMLError, exc: try: ansible.utils.process_yaml_error(exc, data, None) except ansible.errors.AnsibleYAMLValidationFailed, e: self.assertTrue('Could not parse YAML.' in str(e)) else: raise AssertionError('Incorrect exception, expected AnsibleYAMLValidationFailed') def test_parse_yaml_from_file(self): test = os.path.join(os.path.dirname(__file__), 'inventory_test_data', 'common_vars.yml') encrypted = os.path.join(os.path.dirname(__file__), 'inventory_test_data', 'encrypted.yml') broken = os.path.join(os.path.dirname(__file__), 'inventory_test_data', 'broken.yml') try: ansible.utils.parse_yaml_from_file(os.path.dirname(__file__)) except ansible.errors.AnsibleError: pass else: raise AssertionError('Incorrect exception, expected AnsibleError') self.assertEqual(ansible.utils.parse_yaml_from_file(test), yaml.safe_load(open(test))) self.assertEqual(ansible.utils.parse_yaml_from_file(encrypted, 'ansible'), dict(foo='bar')) try: ansible.utils.parse_yaml_from_file(broken) except ansible.errors.AnsibleYAMLValidationFailed, e: self.assertTrue('Syntax Error while loading' in str(e)) else: raise AssertionError('Incorrect exception, expected AnsibleYAMLValidationFailed') def test_merge_hash(self): self.assertEqual(ansible.utils.merge_hash(dict(foo='bar', baz='qux'), dict(foo='baz')), dict(foo='baz', baz='qux')) self.assertEqual(ansible.utils.merge_hash(dict(foo=dict(bar='baz')), dict(foo=dict(bar='qux'))), dict(foo=dict(bar='qux'))) def test_md5s(self): if self._is_fips(): raise SkipTest('MD5 unavailable on FIPs enabled systems') self.assertEqual(ansible.utils.md5s('ansible'), '640c8a5376aa12fa15cf02130ce239a6') # Need a test that causes UnicodeEncodeError See 4221 def test_md5(self): if self._is_fips(): raise SkipTest('MD5 unavailable on FIPs enabled systems') self.assertEqual(ansible.utils.md5(os.path.join(os.path.dirname(__file__), 'ansible.cfg')), 'fb7b5b90ea63f04bde33e804b6fad42c') self.assertEqual(ansible.utils.md5(os.path.join(os.path.dirname(__file__), 'ansible.cf')), None) def test_checksum_s(self): self.assertEqual(ansible.utils.checksum_s('ansible'), 'bef45157a43c9e5f469d188810814a4a8ab9f2ed') # Need a test that causes UnicodeEncodeError See 4221 def test_checksum(self): self.assertEqual(ansible.utils.checksum(os.path.join(os.path.dirname(__file__), 'ansible.cfg')), '658b67c8ac7595adde7048425ff1f9aba270721a') self.assertEqual(ansible.utils.checksum(os.path.join(os.path.dirname(__file__), 'ansible.cf')), None) def test_default(self): self.assertEqual(ansible.utils.default(None, lambda: {}), {}) self.assertEqual(ansible.utils.default(dict(foo='bar'), lambda: {}), dict(foo='bar')) def test__gitinfo(self): # this fails if not run from git clone # self.assertEqual('last updated' in ansible.utils._gitinfo()) # missing test for git submodule # missing test outside of git clone pass def test_version(self): version = ansible.utils.version('ansible') self.assertTrue(version.startswith('ansible %s' % __version__)) # this fails if not run from git clone # self.assertEqual('last updated' in version) def test_getch(self): # figure out how to test this pass def test_sanitize_output(self): self.assertEqual(ansible.utils.sanitize_output('password=foo'), 'password=VALUE_HIDDEN') self.assertEqual(ansible.utils.sanitize_output('foo=user:pass@foo/whatever'), 'foo=user:********@foo/whatever') self.assertEqual(ansible.utils.sanitize_output('foo=http://username:pass@wherever/foo'), 'foo=http://username:********@wherever/foo') self.assertEqual(ansible.utils.sanitize_output('foo=http://wherever/foo'), 'foo=http://wherever/foo') def test_increment_debug(self): ansible.utils.VERBOSITY = 0 ansible.utils.increment_debug(None, None, None, None) self.assertEqual(ansible.utils.VERBOSITY, 1) def test_base_parser(self): output = ansible.utils.base_parser(output_opts=True) self.assertTrue(output.has_option('--one-line') and output.has_option('--tree')) runas = ansible.utils.base_parser(runas_opts=True) for opt in ['--sudo', '--sudo-user', '--user', '--su', '--su-user']: self.assertTrue(runas.has_option(opt)) async = ansible.utils.base_parser(async_opts=True) self.assertTrue(async.has_option('--poll') and async.has_option('--background')) connect = ansible.utils.base_parser(connect_opts=True) self.assertTrue(connect.has_option('--connection')) subset = ansible.utils.base_parser(subset_opts=True) self.assertTrue(subset.has_option('--limit')) check = ansible.utils.base_parser(check_opts=True) self.assertTrue(check.has_option('--check')) diff = ansible.utils.base_parser(diff_opts=True) self.assertTrue(diff.has_option('--diff')) def test_do_encrypt(self): salt_chars = string.ascii_letters + string.digits + './' salt = ansible.utils.random_password(length=8, chars=salt_chars) hash = ansible.utils.do_encrypt('ansible', 'sha256_crypt', salt=salt) self.assertTrue(passlib.hash.sha256_crypt.verify('ansible', hash)) hash = ansible.utils.do_encrypt('ansible', 'sha256_crypt') self.assertTrue(passlib.hash.sha256_crypt.verify('ansible', hash)) try: ansible.utils.do_encrypt('ansible', 'ansible') except ansible.errors.AnsibleError: pass else: raise AssertionError('Incorrect exception, expected AnsibleError') def test_do_encrypt_md5(self): if self._is_fips(): raise SkipTest('MD5 unavailable on FIPS systems') hash = ansible.utils.do_encrypt('ansible', 'md5_crypt', salt_size=4) self.assertTrue(passlib.hash.md5_crypt.verify('ansible', hash)) def test_last_non_blank_line(self): self.assertEqual(ansible.utils.last_non_blank_line('a\n\nb\n\nc'), 'c') self.assertEqual(ansible.utils.last_non_blank_line(''), '') def test_filter_leading_non_json_lines(self): self.assertEqual(ansible.utils.filter_leading_non_json_lines('a\nb\nansible!\n{"foo": "bar"}'), '{"foo": "bar"}\n') self.assertEqual(ansible.utils.filter_leading_non_json_lines('a\nb\nansible!\n["foo", "bar"]'), '["foo", "bar"]\n') def test_boolean(self): self.assertEqual(ansible.utils.boolean("true"), True) self.assertEqual(ansible.utils.boolean("True"), True) self.assertEqual(ansible.utils.boolean("TRUE"), True) self.assertEqual(ansible.utils.boolean("t"), True) self.assertEqual(ansible.utils.boolean("T"), True) self.assertEqual(ansible.utils.boolean("Y"), True) self.assertEqual(ansible.utils.boolean("y"), True) self.assertEqual(ansible.utils.boolean("1"), True) self.assertEqual(ansible.utils.boolean(1), True) self.assertEqual(ansible.utils.boolean("false"), False) self.assertEqual(ansible.utils.boolean("False"), False) self.assertEqual(ansible.utils.boolean("0"), False) self.assertEqual(ansible.utils.boolean(0), False) self.assertEqual(ansible.utils.boolean("foo"), False) def test_make_sudo_cmd(self): cmd = ansible.utils.make_sudo_cmd(C.DEFAULT_SUDO_EXE, 'root', '/bin/sh', '/bin/ls') self.assertTrue(isinstance(cmd, tuple)) self.assertEqual(len(cmd), 3) self.assertTrue('-u root' in cmd[0]) self.assertTrue('-p "[sudo via ansible, key=' in cmd[0] and cmd[1].startswith('[sudo via ansible, key')) self.assertTrue('echo BECOME-SUCCESS-' in cmd[0] and cmd[2].startswith('BECOME-SUCCESS-')) self.assertTrue('sudo -k' in cmd[0]) def test_make_su_cmd(self): cmd = ansible.utils.make_su_cmd('root', '/bin/sh', '/bin/ls') self.assertTrue(isinstance(cmd, tuple)) self.assertEqual(len(cmd), 3) self.assertTrue('root -c "/bin/sh' in cmd[0] or ' root -c /bin/sh' in cmd[0]) self.assertTrue('echo BECOME-SUCCESS-' in cmd[0] and cmd[2].startswith('BECOME-SUCCESS-')) def test_to_unicode(self): uni = ansible.utils.unicode.to_unicode(u'ansible') self.assertTrue(isinstance(uni, unicode)) self.assertEqual(uni, u'ansible') none = ansible.utils.unicode.to_unicode(None, nonstring='passthru') self.assertTrue(isinstance(none, type(None))) self.assertTrue(none is None) utf8 = ansible.utils.unicode.to_unicode('ansible') self.assertTrue(isinstance(utf8, unicode)) self.assertEqual(utf8, u'ansible') def test_is_list_of_strings(self): self.assertEqual(ansible.utils.is_list_of_strings(['foo', 'bar', u'baz']), True) self.assertEqual(ansible.utils.is_list_of_strings(['foo', 'bar', True]), False) self.assertEqual(ansible.utils.is_list_of_strings(['one', 2, 'three']), False) def test_contains_vars(self): self.assertTrue(ansible.utils.contains_vars('{{foo}}')) self.assertTrue(ansible.utils.contains_vars('$foo')) self.assertFalse(ansible.utils.contains_vars('foo')) def test_safe_eval(self): # Not basestring self.assertEqual(ansible.utils.safe_eval(len), len) self.assertEqual(ansible.utils.safe_eval(1), 1) self.assertEqual(ansible.utils.safe_eval(len, include_exceptions=True), (len, None)) self.assertEqual(ansible.utils.safe_eval(1, include_exceptions=True), (1, None)) # module self.assertEqual(ansible.utils.safe_eval('foo.bar('), 'foo.bar(') self.assertEqual(ansible.utils.safe_eval('foo.bar(', include_exceptions=True), ('foo.bar(', None)) # import self.assertEqual(ansible.utils.safe_eval('import foo'), 'import foo') self.assertEqual(ansible.utils.safe_eval('import foo', include_exceptions=True), ('import foo', None)) # valid simple eval self.assertEqual(ansible.utils.safe_eval('True'), True) self.assertEqual(ansible.utils.safe_eval('True', include_exceptions=True), (True, None)) # valid eval with lookup self.assertEqual(ansible.utils.safe_eval('foo + bar', dict(foo=1, bar=2)), 3) self.assertEqual(ansible.utils.safe_eval('foo + bar', dict(foo=1, bar=2), include_exceptions=True), (3, None)) # invalid eval self.assertEqual(ansible.utils.safe_eval('foo'), 'foo') nameerror = ansible.utils.safe_eval('foo', include_exceptions=True) self.assertTrue(isinstance(nameerror, tuple)) self.assertEqual(nameerror[0], 'foo') self.assertTrue(isinstance(nameerror[1], NameError)) def test_listify_lookup_plugin_terms(self): basedir = os.path.dirname(__file__) # Straight lookups #self.assertEqual(ansible.utils.listify_lookup_plugin_terms('things', basedir, dict(things=[])), []) #self.assertEqual(ansible.utils.listify_lookup_plugin_terms('things', basedir, dict(things=['one', 'two'])), ['one', 'two']) def test_deprecated(self): sys_stderr = sys.stderr sys.stderr = StringIO.StringIO() ansible.utils.deprecated('Ack!', '0.0') out = sys.stderr.getvalue() self.assertTrue('0.0' in out) self.assertTrue('[DEPRECATION WARNING]' in out) sys.stderr = StringIO.StringIO() ansible.utils.deprecated('Ack!', None) out = sys.stderr.getvalue() self.assertTrue('0.0' not in out) self.assertTrue('[DEPRECATION WARNING]' in out) sys.stderr = StringIO.StringIO() warnings = C.DEPRECATION_WARNINGS C.DEPRECATION_WARNINGS = False ansible.utils.deprecated('Ack!', None) out = sys.stderr.getvalue() self.assertTrue(not out) C.DEPRECATION_WARNINGS = warnings sys.stderr = sys_stderr try: ansible.utils.deprecated('Ack!', '0.0', True) except ansible.errors.AnsibleError, e: self.assertTrue('0.0' not in str(e)) self.assertTrue('[DEPRECATED]' in str(e)) else: raise AssertionError("Incorrect exception, expected AnsibleError") def test_warning(self): sys_stderr = sys.stderr sys.stderr = StringIO.StringIO() ansible.utils.warning('ANSIBLE') out = sys.stderr.getvalue() sys.stderr = sys_stderr self.assertTrue('[WARNING]: ANSIBLE' in out) def test_combine_vars(self): one = {'foo': {'bar': True}, 'baz': {'one': 'qux'}} two = {'baz': {'two': 'qux'}} replace = {'baz': {'two': 'qux'}, 'foo': {'bar': True}} merge = {'baz': {'two': 'qux', 'one': 'qux'}, 'foo': {'bar': True}} C.DEFAULT_HASH_BEHAVIOUR = 'replace' self.assertEqual(ansible.utils.combine_vars(one, two), replace) C.DEFAULT_HASH_BEHAVIOUR = 'merge' self.assertEqual(ansible.utils.combine_vars(one, two), merge) def test_err(self): sys_stderr = sys.stderr sys.stderr = StringIO.StringIO() ansible.utils.err('ANSIBLE') out = sys.stderr.getvalue() sys.stderr = sys_stderr self.assertEqual(out, 'ANSIBLE\n') def test_exit(self): sys_stderr = sys.stderr sys.stderr = StringIO.StringIO() try: ansible.utils.exit('ansible') except SystemExit, e: self.assertEqual(e.code, 1) self.assertEqual(sys.stderr.getvalue(), 'ansible\n') else: raise AssertionError('Incorrect exception, expected SystemExit') finally: sys.stderr = sys_stderr def test_unfrackpath(self): os.environ['TEST_ROOT'] = os.path.dirname(os.path.dirname(__file__)) self.assertEqual(ansible.utils.unfrackpath('$TEST_ROOT/units/../units/TestUtils.py'), __file__.rstrip('c')) def test_is_executable(self): self.assertEqual(ansible.utils.is_executable(__file__), 0) bin_ansible = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), 'bin', 'ansible') self.assertNotEqual(ansible.utils.is_executable(bin_ansible), 0) def test_get_diff(self): standard = dict( before_header='foo', after_header='bar', before='fooo', after='foo' ) standard_expected = """--- before: foo +++ after: bar @@ -1 +1 @@ -fooo+foo""" # workaround py26 and py27 difflib differences standard_expected = """-fooo+foo""" diff = ansible.utils.get_diff(standard) diff = diff.split('\n') del diff[0] del diff[0] del diff[0] diff = '\n'.join(diff) self.assertEqual(diff, unicode(standard_expected)) def test_split_args(self): # split_args is a smarter shlex.split for the needs of the way ansible uses it def _split_info(input, desired, actual): print "SENT: ", input print "WANT: ", desired print "GOT: ", actual def _test_combo(input, desired): actual = split_args(input) _split_info(input, desired, actual) assert actual == desired # trivial splitting _test_combo('a b=c d=f', ['a', 'b=c', 'd=f' ]) # mixed quotes _test_combo('a b=\'c\' d="e" f=\'g\'', ['a', "b='c'", 'd="e"', "f='g'" ]) # with spaces # FIXME: this fails, commenting out only for now # _test_combo('a "\'one two three\'"', ['a', "'one two three'" ]) # TODO: ... # jinja2 preservation _test_combo('a {{ y }} z', ['a', '{{ y }}', 'z' ]) # jinja2 preservation with spaces and filters and other hard things _test_combo( 'a {{ x | filter(\'moo\', \'param\') }} z {{ chicken }} "waffles"', ['a', "{{ x | filter('moo', 'param') }}", 'z', '{{ chicken }}', '"waffles"'] ) # invalid quote detection self.assertRaises(Exception, split_args, 'hey I started a quote"') self.assertRaises(Exception, split_args, 'hey I started a\' quote') # jinja2 loop blocks with lots of complexity _test_combo( # in memory of neighbors cat # we preserve line breaks unless a line continuation character precedes them 'a {% if x %} y {%else %} {{meow}} {% endif %} "cookie\nchip" \\\ndone\nand done', ['a', '{% if x %}', 'y', '{%else %}', '{{meow}}', '{% endif %}', '"cookie\nchip"', 'done\n', 'and', 'done'] ) # test space preservation within quotes _test_combo( 'content="1 2 3 4 " foo=bar', ['content="1 2 3 4 "', 'foo=bar'] ) # invalid jinja2 nesting detection # invalid quote nesting detection def test_clean_data(self): # clean data removes jinja2 tags from data self.assertEqual( ansible.utils._clean_data('this is a normal string', from_remote=True), 'this is a normal string' ) self.assertEqual( ansible.utils._clean_data('this string has a {{variable}}', from_remote=True), 'this string has a {#variable#}' ) self.assertEqual( ansible.utils._clean_data('this string {{has}} two {{variables}} in it', from_remote=True), 'this string {#has#} two {#variables#} in it' ) self.assertEqual( ansible.utils._clean_data('this string has a {{variable with a\nnewline}}', from_remote=True), 'this string has a {#variable with a\nnewline#}' ) self.assertEqual( ansible.utils._clean_data('this string is from inventory {{variable}}', from_inventory=True), 'this string is from inventory {{variable}}' ) self.assertEqual( ansible.utils._clean_data('this string is from inventory too but uses lookup {{lookup("foo","bar")}}', from_inventory=True), 'this string is from inventory too but uses lookup {#lookup("foo","bar")#}' ) self.assertEqual( ansible.utils._clean_data('this string has JSON in it: {"foo":{"bar":{"baz":"oops"}}}', from_remote=True), 'this string has JSON in it: {"foo":{"bar":{"baz":"oops"}}}' ) self.assertEqual( ansible.utils._clean_data('this string contains unicode: ¢ £ ¤ ¥', from_remote=True), 'this string contains unicode: ¢ £ ¤ ¥' ) def test_censor_unlogged_data(self): ''' used by the no_log attribute ''' input = dict( password='sekrit', rc=12, failed=True, changed=False, skipped=True, msg='moo', ) data = ansible.utils.censor_unlogged_data(input) assert 'password' not in data assert 'rc' in data assert 'failed' in data assert 'changed' in data assert 'skipped' in data assert 'msg' not in data assert data['censored'] == 'results hidden due to no_log parameter' def test_repo_url_to_role_name(self): tests = [("http://git.example.com/repos/repo.git", "repo"), ("ssh://git@git.example.com:repos/role-name", "role-name"), ("ssh://git@git.example.com:repos/role-name,v0.1", "role-name"), ("directory/role/is/installed/in", "directory/role/is/installed/in")] for (url, result) in tests: self.assertEqual(ansible.utils.repo_url_to_role_name(url), result) def test_role_spec_parse(self): tests = [ ( "git+http://git.example.com/repos/repo.git,v1.0", { 'scm': 'git', 'src': 'http://git.example.com/repos/repo.git', 'version': 'v1.0', 'name': 'repo' } ), ( "http://repo.example.com/download/tarfile.tar.gz", { 'scm': None, 'src': 'http://repo.example.com/download/tarfile.tar.gz', 'version': '', 'name': 'tarfile' } ), ( "http://repo.example.com/download/tarfile.tar.gz,,nicename", { 'scm': None, 'src': 'http://repo.example.com/download/tarfile.tar.gz', 'version': '', 'name': 'nicename' } ), ( "git+http://git.example.com/repos/repo.git,v1.0,awesome", { 'scm': 'git', 'src': 'http://git.example.com/repos/repo.git', 'version': 'v1.0', 'name': 'awesome' } ), ( # test that http://github URLs are assumed git+http:// unless they end in .tar.gz "http://github.com/ansible/fakerole/fake", { 'scm' : 'git', 'src' : 'http://github.com/ansible/fakerole/fake', 'version' : 'master', 'name' : 'fake' } ), ( # test that http://github URLs are assumed git+http:// unless they end in .tar.gz "http://github.com/ansible/fakerole/fake/archive/master.tar.gz", { 'scm' : None, 'src' : 'http://github.com/ansible/fakerole/fake/archive/master.tar.gz', 'version' : '', 'name' : 'master' } ) ] for (spec, result) in tests: self.assertEqual(ansible.utils.role_spec_parse(spec), result) def test_role_yaml_parse(self): tests = ( ( # Old style { 'role': 'debops.elasticsearch', 'name': 'elks' }, { 'role': 'debops.elasticsearch', 'name': 'elks', 'scm': None, 'src': 'debops.elasticsearch', 'version': '', } ), ( { 'role': 'debops.elasticsearch,1.0,elks', 'my_param': 'foo' }, { 'role': 'debops.elasticsearch,1.0,elks', 'name': 'elks', 'scm': None, 'src': 'debops.elasticsearch', 'version': '1.0', 'my_param': 'foo', } ), ( { 'role': 'debops.elasticsearch,1.0', 'my_param': 'foo' }, { 'role': 'debops.elasticsearch,1.0', 'name': 'debops.elasticsearch', 'scm': None, 'src': 'debops.elasticsearch', 'version': '1.0', 'my_param': 'foo', } ), # New style ( { 'src': 'debops.elasticsearch', 'name': 'elks', 'my_param': 'foo' }, { 'name': 'elks', 'scm': None, 'src': 'debops.elasticsearch', 'version': '', 'my_param': 'foo' } ), ) for (role, result) in tests: self.assertEqual(ansible.utils.role_yaml_parse(role), result) @patch('ansible.utils.plugins.module_finder._get_paths') def test_find_plugin(self, mock_get_paths): tmp_path = tempfile.mkdtemp() mock_get_paths.return_value = [tmp_path,] right_module_1 = 'module.py' right_module_2 = 'module_without_extension' wrong_module_1 = 'folder' wrong_module_2 = 'inexistent' path_right_module_1 = os.path.join(tmp_path, right_module_1) path_right_module_2 = os.path.join(tmp_path, right_module_2) path_wrong_module_1 = os.path.join(tmp_path, wrong_module_1) open(path_right_module_1, 'w').close() open(path_right_module_2, 'w').close() os.mkdir(path_wrong_module_1) self.assertEqual(ansible.utils.plugins.module_finder.find_plugin(right_module_1), path_right_module_1) self.assertEqual(ansible.utils.plugins.module_finder.find_plugin(right_module_2), path_right_module_2) self.assertEqual(ansible.utils.plugins.module_finder.find_plugin(wrong_module_1), None) self.assertEqual(ansible.utils.plugins.module_finder.find_plugin(wrong_module_2), None) shutil.rmtree(tmp_path)