# -*- coding: utf-8 -*- """ Tests for salt.utils.jinja """ from __future__ import absolute_import, print_function, unicode_literals import ast import copy import datetime import os import pprint import re import tempfile import pytest import salt.config import salt.loader # dateutils is needed so that the strftime jinja filter is loaded import salt.utils.dateutils # pylint: disable=unused-import import salt.utils.files import salt.utils.json import salt.utils.stringutils import salt.utils.yaml from jinja2 import DictLoader, Environment, exceptions from salt.exceptions import SaltRenderError from salt.ext import six from salt.ext.six.moves import builtins from salt.utils.decorators.jinja import JinjaFilter from salt.utils.jinja import ( SaltCacheLoader, SerializerExtension, ensure_sequence_filter, tojson, ) from salt.utils.odict import OrderedDict from salt.utils.templates import JINJA, render_jinja_tmpl from tests.support.case import ModuleCase from tests.support.mock import MagicMock, Mock, patch from tests.support.runtests import RUNTIME_VARS from tests.support.unit import TestCase, skipIf try: import timelib # pylint: disable=W0611 HAS_TIMELIB = True except ImportError: HAS_TIMELIB = False BLINESEP = salt.utils.stringutils.to_bytes(os.linesep) class JinjaTestCase(TestCase): def test_tojson(self): """ Test the tojson filter for those using Jinja < 2.9. Non-ascii unicode content should be dumped with ensure_ascii=True. """ data = {"Non-ascii words": ["süß", "спам", "яйца"]} result = tojson(data) expected = ( '{"Non-ascii words": ["s\\u00fc\\u00df", ' '"\\u0441\\u043f\\u0430\\u043c", ' '"\\u044f\\u0439\\u0446\\u0430"]}' ) assert result == expected, result class MockFileClient(object): """ Does not download files but records any file request for testing """ def __init__(self, loader=None): if loader: loader._file_client = self self.requests = [] def get_file(self, template, dest="", makedirs=False, saltenv="base"): self.requests.append( {"path": template, "dest": dest, "makedirs": makedirs, "saltenv": saltenv} ) def _setup_test_dir(src_dir, test_dir): os.makedirs(test_dir) salt.utils.files.recursive_copy(src_dir, test_dir) filename = os.path.join(test_dir, "non_ascii") with salt.utils.files.fopen(filename, "wb") as fp: fp.write(b"Assun\xc3\xa7\xc3\xa3o" + BLINESEP) filename = os.path.join(test_dir, "hello_simple") with salt.utils.files.fopen(filename, "wb") as fp: fp.write(b"world" + BLINESEP) filename = os.path.join(test_dir, "hello_import") lines = [ r"{% from 'macro' import mymacro -%}", r"{% from 'macro' import mymacro -%}", r"{{ mymacro('Hey') ~ mymacro(a|default('a'), b|default('b')) }}", ] with salt.utils.files.fopen(filename, "wb") as fp: for line in lines: fp.write(line.encode("utf-8") + BLINESEP) class TestSaltCacheLoader(TestCase): def setUp(self): self.tempdir = tempfile.mkdtemp() self.template_dir = os.path.join(self.tempdir, "files", "test") _setup_test_dir( os.path.join(RUNTIME_VARS.BASE_FILES, "templates"), self.template_dir ) self.opts = { "file_buffer_size": 1048576, "cachedir": self.tempdir, "file_roots": {"test": [self.template_dir]}, "pillar_roots": {"test": [self.template_dir]}, "extension_modules": os.path.join( os.path.dirname(os.path.abspath(__file__)), "extmods" ), } super(TestSaltCacheLoader, self).setUp() def tearDown(self): salt.utils.files.rm_rf(self.tempdir) def test_searchpath(self): """ The searchpath is based on the cachedir option and the saltenv parameter """ tmp = tempfile.gettempdir() opts = copy.deepcopy(self.opts) opts.update({"cachedir": tmp}) loader = self.get_loader(opts=opts, saltenv="test") assert loader.searchpath == [os.path.join(tmp, "files", "test")] def test_mockclient(self): """ A MockFileClient is used that records all file requests normally sent to the master. """ loader = self.get_loader(opts=self.opts, saltenv="test") res = loader.get_source(None, "hello_simple") assert len(res) == 3 # res[0] on Windows is unicode and use os.linesep so it works cross OS self.assertEqual(six.text_type(res[0]), "world" + os.linesep) tmpl_dir = os.path.join(self.template_dir, "hello_simple") self.assertEqual(res[1], tmpl_dir) assert res[2](), "Template up to date?" assert loader._file_client.requests self.assertEqual(loader._file_client.requests[0]["path"], "salt://hello_simple") def get_loader(self, opts=None, saltenv="base"): """ Now that we instantiate the client in the __init__, we need to mock it """ if opts is None: opts = self.opts with patch.object(SaltCacheLoader, "file_client", Mock()): loader = SaltCacheLoader(opts, saltenv) self.addCleanup(setattr, SaltCacheLoader, "_cached_client", None) # Create a mock file client and attach it to the loader MockFileClient(loader) return loader def get_test_saltenv(self): """ Setup a simple jinja test environment """ loader = self.get_loader(saltenv="test") jinja = Environment(loader=loader) return loader._file_client, jinja def test_import(self): """ You can import and use macros from other files """ fc, jinja = self.get_test_saltenv() result = jinja.get_template("hello_import").render() self.assertEqual(result, "Hey world !a b !") assert len(fc.requests) == 2 self.assertEqual(fc.requests[0]["path"], "salt://hello_import") self.assertEqual(fc.requests[1]["path"], "salt://macro") def test_relative_import(self): """ You can import using relative paths issue-13889 """ fc, jinja = self.get_test_saltenv() tmpl = jinja.get_template(os.path.join("relative", "rhello")) result = tmpl.render() self.assertEqual(result, "Hey world !a b !") assert len(fc.requests) == 3 self.assertEqual( fc.requests[0]["path"], os.path.join("salt://relative", "rhello") ) self.assertEqual( fc.requests[1]["path"], os.path.join("salt://relative", "rmacro") ) self.assertEqual(fc.requests[2]["path"], "salt://macro") # This must fail when rendered: attempts to import from outside file root template = jinja.get_template("relative/rescape") self.assertRaises(exceptions.TemplateNotFound, template.render) def test_include(self): """ You can also include a template that imports and uses macros """ fc, jinja = self.get_test_saltenv() result = jinja.get_template("hello_include").render() self.assertEqual(result, "Hey world !a b !") assert len(fc.requests) == 3 self.assertEqual(fc.requests[0]["path"], "salt://hello_include") self.assertEqual(fc.requests[1]["path"], "salt://hello_import") self.assertEqual(fc.requests[2]["path"], "salt://macro") def test_include_context(self): """ Context variables are passes to the included template by default. """ _, jinja = self.get_test_saltenv() result = jinja.get_template("hello_include").render(a="Hi", b="Salt") self.assertEqual(result, "Hey world !Hi Salt !") def test_cached_file_client(self): """ Multiple instantiations of SaltCacheLoader use the cached file client """ with patch("salt.transport.client.ReqChannel.factory", Mock()): loader_a = SaltCacheLoader(self.opts) loader_b = SaltCacheLoader(self.opts) assert loader_a._file_client is loader_b._file_client def test_file_client_kwarg(self): """ A file client can be passed to SaltCacheLoader overriding the any cached file client """ mfc = MockFileClient() loader = SaltCacheLoader(self.opts, _file_client=mfc) assert loader._file_client is mfc def test_cache_loader_shutdown(self): """ The shudown method can be called without raising an exception when the file_client does not have a destroy method """ mfc = MockFileClient() assert not hasattr(mfc, "destroy") loader = SaltCacheLoader(self.opts, _file_client=mfc) assert loader._file_client is mfc # Shutdown method should not raise any exceptions loader.shutdown() class TestGetTemplate(TestCase): def setUp(self): self.tempdir = tempfile.mkdtemp() self.template_dir = os.path.join(self.tempdir, "files", "test") _setup_test_dir( os.path.join(RUNTIME_VARS.BASE_FILES, "templates"), self.template_dir ) self.local_opts = { "file_buffer_size": 1048576, "cachedir": self.tempdir, "file_client": "local", "file_ignore_regex": None, "file_ignore_glob": None, "file_roots": {"test": [self.template_dir]}, "pillar_roots": {"test": [self.template_dir]}, "fileserver_backend": ["roots"], "hash_type": "md5", "extension_modules": os.path.join( os.path.dirname(os.path.abspath(__file__)), "extmods" ), } self.local_salt = {} super(TestGetTemplate, self).setUp() def tearDown(self): salt.utils.files.rm_rf(self.tempdir) def test_fallback(self): """ A Template with a filesystem loader is returned as fallback if the file is not contained in the searchpath """ fn_ = os.path.join(self.template_dir, "hello_simple") with salt.utils.files.fopen(fn_) as fp_: out = render_jinja_tmpl( salt.utils.stringutils.to_unicode(fp_.read()), dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(out, "world" + os.linesep) def test_fallback_noloader(self): """ A Template with a filesystem loader is returned as fallback if the file is not contained in the searchpath """ filename = os.path.join(self.template_dir, "hello_import") with salt.utils.files.fopen(filename) as fp_: out = render_jinja_tmpl( salt.utils.stringutils.to_unicode(fp_.read()), dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(out, "Hey world !a b !" + os.linesep) def test_saltenv(self): """ If the template is within the searchpath it can import, include and extend other templates. The initial template is expected to be already cached get_template does not request it from the master again. """ fc = MockFileClient() with patch.object(SaltCacheLoader, "file_client", MagicMock(return_value=fc)): filename = os.path.join(self.template_dir, "hello_import") with salt.utils.files.fopen(filename) as fp_: out = render_jinja_tmpl( salt.utils.stringutils.to_unicode(fp_.read()), dict( opts={ "cachedir": self.tempdir, "file_client": "remote", "file_roots": self.local_opts["file_roots"], "pillar_roots": self.local_opts["pillar_roots"], }, a="Hi", b="Salt", saltenv="test", salt=self.local_salt, ), ) self.assertEqual(out, "Hey world !Hi Salt !" + os.linesep) self.assertEqual(fc.requests[0]["path"], "salt://macro") def test_macro_additional_log_for_generalexc(self): """ If we failed in a macro because of e.g. a TypeError, get more output from trace. """ expected = r"""Jinja error:.*division.* .*macrogeneral\(2\): --- \{% macro mymacro\(\) -%\} \{\{ 1/0 \}\} <====================== \{%- endmacro %\} ---.*""" filename = os.path.join(self.template_dir, "hello_import_generalerror") fc = MockFileClient() with patch.object(SaltCacheLoader, "file_client", MagicMock(return_value=fc)): with salt.utils.files.fopen(filename) as fp_: self.assertRaisesRegex( SaltRenderError, expected, render_jinja_tmpl, salt.utils.stringutils.to_unicode(fp_.read()), dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) def test_macro_additional_log_for_undefined(self): """ If we failed in a macro because of undefined variables, get more output from trace. """ expected = r"""Jinja variable 'b' is undefined .*macroundefined\(2\): --- \{% macro mymacro\(\) -%\} \{\{b.greetee\}\} <-- error is here <====================== \{%- endmacro %\} ---""" filename = os.path.join(self.template_dir, "hello_import_undefined") fc = MockFileClient() with patch.object(SaltCacheLoader, "file_client", MagicMock(return_value=fc)): with salt.utils.files.fopen(filename) as fp_: self.assertRaisesRegex( SaltRenderError, expected, render_jinja_tmpl, salt.utils.stringutils.to_unicode(fp_.read()), dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) def test_macro_additional_log_syntaxerror(self): """ If we failed in a macro, get more output from trace. """ expected = r"""Jinja syntax error: expected token .*end.*got '-'.* .*macroerror\(2\): --- # macro \{% macro mymacro\(greeting, greetee='world'\) -\} <-- error is here <====================== \{\{ greeting ~ ' ' ~ greetee \}\} ! \{%- endmacro %\} ---.*""" filename = os.path.join(self.template_dir, "hello_import_error") fc = MockFileClient() with patch.object(SaltCacheLoader, "file_client", MagicMock(return_value=fc)): with salt.utils.files.fopen(filename) as fp_: self.assertRaisesRegex( SaltRenderError, expected, render_jinja_tmpl, salt.utils.stringutils.to_unicode(fp_.read()), dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) def test_non_ascii_encoding(self): fc = MockFileClient() with patch.object(SaltCacheLoader, "file_client", MagicMock(return_value=fc)): filename = os.path.join(self.template_dir, "hello_import") with salt.utils.files.fopen(filename) as fp_: out = render_jinja_tmpl( salt.utils.stringutils.to_unicode(fp_.read()), dict( opts={ "cachedir": self.tempdir, "file_client": "remote", "file_roots": self.local_opts["file_roots"], "pillar_roots": self.local_opts["pillar_roots"], }, a="Hi", b="Sàlt", saltenv="test", salt=self.local_salt, ), ) self.assertEqual( out, salt.utils.stringutils.to_unicode("Hey world !Hi Sàlt !" + os.linesep), ) self.assertEqual(fc.requests[0]["path"], "salt://macro") filename = os.path.join(self.template_dir, "non_ascii") with salt.utils.files.fopen(filename, "rb") as fp_: out = render_jinja_tmpl( salt.utils.stringutils.to_unicode(fp_.read(), "utf-8"), dict( opts={ "cachedir": self.tempdir, "file_client": "remote", "file_roots": self.local_opts["file_roots"], "pillar_roots": self.local_opts["pillar_roots"], }, a="Hi", b="Sàlt", saltenv="test", salt=self.local_salt, ), ) self.assertEqual("Assunção" + os.linesep, out) self.assertEqual(fc.requests[0]["path"], "salt://macro") @skipIf(HAS_TIMELIB is False, "The `timelib` library is not installed.") def test_strftime(self): response = render_jinja_tmpl( '{{ "2002/12/25"|strftime }}', dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(response, "2002-12-25") objects = ( datetime.datetime(2002, 12, 25, 12, 00, 00, 00), "2002/12/25", 1040814000, "1040814000", ) for object in objects: response = render_jinja_tmpl( "{{ object|strftime }}", dict( object=object, opts=self.local_opts, saltenv="test", salt=self.local_salt, ), ) self.assertEqual(response, "2002-12-25") response = render_jinja_tmpl( '{{ object|strftime("%b %d, %Y") }}', dict( object=object, opts=self.local_opts, saltenv="test", salt=self.local_salt, ), ) self.assertEqual(response, "Dec 25, 2002") response = render_jinja_tmpl( '{{ object|strftime("%y") }}', dict( object=object, opts=self.local_opts, saltenv="test", salt=self.local_salt, ), ) self.assertEqual(response, "02") def test_non_ascii(self): fn = os.path.join(self.template_dir, "non_ascii") out = JINJA(fn, opts=self.local_opts, saltenv="test", salt=self.local_salt) with salt.utils.files.fopen(out["data"], "rb") as fp: result = salt.utils.stringutils.to_unicode(fp.read(), "utf-8") self.assertEqual( salt.utils.stringutils.to_unicode("Assunção" + os.linesep), result ) def test_get_context_has_enough_context(self): template = "1\n2\n3\n4\n5\n6\n7\n8\n9\na\nb\nc\nd\ne\nf" context = salt.utils.stringutils.get_context(template, 8) expected = "---\n[...]\n3\n4\n5\n6\n7\n8\n9\na\nb\nc\nd\n[...]\n---" self.assertEqual(expected, context) def test_get_context_at_top_of_file(self): template = "1\n2\n3\n4\n5\n6\n7\n8\n9\na\nb\nc\nd\ne\nf" context = salt.utils.stringutils.get_context(template, 1) expected = "---\n1\n2\n3\n4\n5\n6\n[...]\n---" self.assertEqual(expected, context) def test_get_context_at_bottom_of_file(self): template = "1\n2\n3\n4\n5\n6\n7\n8\n9\na\nb\nc\nd\ne\nf" context = salt.utils.stringutils.get_context(template, 15) expected = "---\n[...]\na\nb\nc\nd\ne\nf\n---" self.assertEqual(expected, context) def test_get_context_2_context_lines(self): template = "1\n2\n3\n4\n5\n6\n7\n8\n9\na\nb\nc\nd\ne\nf" context = salt.utils.stringutils.get_context(template, 8, num_lines=2) expected = "---\n[...]\n6\n7\n8\n9\na\n[...]\n---" self.assertEqual(expected, context) def test_get_context_with_marker(self): template = "1\n2\n3\n4\n5\n6\n7\n8\n9\na\nb\nc\nd\ne\nf" context = salt.utils.stringutils.get_context( template, 8, num_lines=2, marker=" <---" ) expected = "---\n[...]\n6\n7\n8 <---\n9\na\n[...]\n---" self.assertEqual(expected, context) def test_render_with_syntax_error(self): template = "hello\n\n{{ bad\n\nfoo" expected = r".*---\nhello\n\n{{ bad\n\nfoo <======================\n---" self.assertRaisesRegex( SaltRenderError, expected, render_jinja_tmpl, template, dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) @skipIf(six.PY3, "Not applicable to Python 3") def test_render_with_unicode_syntax_error(self): with patch.object(builtins, "__salt_system_encoding__", "utf-8"): template = "hello\n\n{{ bad\n\nfoo한" expected = r".*---\nhello\n\n{{ bad\n\nfoo\xed\x95\x9c <======================\n---" self.assertRaisesRegex( SaltRenderError, expected, render_jinja_tmpl, template, dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) def test_render_with_utf8_syntax_error(self): with patch.object(builtins, "__salt_system_encoding__", "utf-8"): template = "hello\n\n{{ bad\n\nfoo한" expected = salt.utils.stringutils.to_str( r".*---\nhello\n\n{{ bad\n\nfoo한 <======================\n---" ) self.assertRaisesRegex( SaltRenderError, expected, render_jinja_tmpl, template, dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) def test_render_with_undefined_variable(self): template = "hello\n\n{{ foo }}\n\nfoo" expected = r"Jinja variable \'foo\' is undefined" self.assertRaisesRegex( SaltRenderError, expected, render_jinja_tmpl, template, dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) def test_render_with_undefined_variable_utf8(self): template = "hello\xed\x95\x9c\n\n{{ foo }}\n\nfoo" expected = r"Jinja variable \'foo\' is undefined" self.assertRaisesRegex( SaltRenderError, expected, render_jinja_tmpl, template, dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) def test_render_with_undefined_variable_unicode(self): template = "hello한\n\n{{ foo }}\n\nfoo" expected = r"Jinja variable \'foo\' is undefined" self.assertRaisesRegex( SaltRenderError, expected, render_jinja_tmpl, template, dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) class TestJinjaDefaultOptions(TestCase): def __init__(self, *args, **kws): TestCase.__init__(self, *args, **kws) self.local_opts = { "cachedir": os.path.join(RUNTIME_VARS.TMP, "jinja-template-cache"), "file_buffer_size": 1048576, "file_client": "local", "file_ignore_regex": None, "file_ignore_glob": None, "file_roots": { "test": [os.path.join(RUNTIME_VARS.BASE_FILES, "templates")] }, "pillar_roots": { "test": [os.path.join(RUNTIME_VARS.BASE_FILES, "templates")] }, "fileserver_backend": ["roots"], "hash_type": "md5", "extension_modules": os.path.join( os.path.dirname(os.path.abspath(__file__)), "extmods" ), "jinja_env": {"line_comment_prefix": "##", "line_statement_prefix": "%"}, } self.local_salt = { "myvar": "zero", "mylist": [0, 1, 2, 3], } def test_comment_prefix(self): template = """ %- set myvar = 'one' ## ignored comment 1 {{- myvar -}} {%- set myvar = 'two' %} ## ignored comment 2 {{- myvar }} ## ignored comment 3 %- if myvar == 'two': %- set myvar = 'three' %- endif {{- myvar -}} """ rendered = render_jinja_tmpl( template, dict(opts=self.local_opts, saltenv="test", salt=self.local_salt) ) self.assertEqual(rendered, "onetwothree") def test_statement_prefix(self): template = """ {%- set mylist = ['1', '2', '3'] %} %- set mylist = ['one', 'two', 'three'] %- for item in mylist: {{- item }} %- endfor """ rendered = render_jinja_tmpl( template, dict(opts=self.local_opts, saltenv="test", salt=self.local_salt) ) self.assertEqual(rendered, "onetwothree") class TestCustomExtensions(TestCase): def __init__(self, *args, **kws): super(TestCustomExtensions, self).__init__(*args, **kws) self.local_opts = { "cachedir": os.path.join(RUNTIME_VARS.TMP, "jinja-template-cache"), "file_buffer_size": 1048576, "file_client": "local", "file_ignore_regex": None, "file_ignore_glob": None, "file_roots": { "test": [os.path.join(RUNTIME_VARS.BASE_FILES, "templates")] }, "pillar_roots": { "test": [os.path.join(RUNTIME_VARS.BASE_FILES, "templates")] }, "fileserver_backend": ["roots"], "hash_type": "md5", "extension_modules": os.path.join( os.path.dirname(os.path.abspath(__file__)), "extmods" ), } self.local_salt = { # 'dns.A': dnsutil.A, # 'dns.AAAA': dnsutil.AAAA, # 'file.exists': filemod.file_exists, # 'file.basename': filemod.basename, # 'file.dirname': filemod.dirname } def test_regex_escape(self): dataset = "foo?:.*/\\bar" env = Environment(extensions=[SerializerExtension]) env.filters.update(JinjaFilter.salt_jinja_filters) rendered = env.from_string("{{ dataset|regex_escape }}").render(dataset=dataset) self.assertEqual(rendered, re.escape(dataset)) def test_unique_string(self): dataset = "foo" unique = set(dataset) env = Environment(extensions=[SerializerExtension]) env.filters.update(JinjaFilter.salt_jinja_filters) if six.PY3: rendered = ( env.from_string("{{ dataset|unique }}") .render(dataset=dataset) .strip("'{}") .split("', '") ) self.assertEqual(sorted(rendered), sorted(list(unique))) else: rendered = env.from_string("{{ dataset|unique }}").render(dataset=dataset) self.assertEqual(rendered, "{0}".format(unique)) def test_unique_tuple(self): dataset = ("foo", "foo", "bar") unique = set(dataset) env = Environment(extensions=[SerializerExtension]) env.filters.update(JinjaFilter.salt_jinja_filters) if six.PY3: rendered = ( env.from_string("{{ dataset|unique }}") .render(dataset=dataset) .strip("'{}") .split("', '") ) self.assertEqual(sorted(rendered), sorted(list(unique))) else: rendered = env.from_string("{{ dataset|unique }}").render(dataset=dataset) self.assertEqual(rendered, "{0}".format(unique)) def test_unique_list(self): dataset = ["foo", "foo", "bar"] unique = ["foo", "bar"] env = Environment(extensions=[SerializerExtension]) env.filters.update(JinjaFilter.salt_jinja_filters) if six.PY3: rendered = ( env.from_string("{{ dataset|unique }}") .render(dataset=dataset) .strip("'[]") .split("', '") ) self.assertEqual(rendered, unique) else: rendered = env.from_string("{{ dataset|unique }}").render(dataset=dataset) self.assertEqual(rendered, "{0}".format(unique)) def test_serialize_json(self): dataset = {"foo": True, "bar": 42, "baz": [1, 2, 3], "qux": 2.0} env = Environment(extensions=[SerializerExtension]) rendered = env.from_string("{{ dataset|json }}").render(dataset=dataset) self.assertEqual(dataset, salt.utils.json.loads(rendered)) def test_serialize_yaml(self): dataset = { "foo": True, "bar": 42, "baz": [1, 2, 3], "qux": 2.0, "spam": OrderedDict([("foo", OrderedDict([("bar", "baz"), ("qux", 42)]))]), } env = Environment(extensions=[SerializerExtension]) rendered = env.from_string("{{ dataset|yaml }}").render(dataset=dataset) self.assertEqual(dataset, salt.utils.yaml.safe_load(rendered)) def test_serialize_yaml_str(self): dataset = "str value" env = Environment(extensions=[SerializerExtension]) rendered = env.from_string("{{ dataset|yaml }}").render(dataset=dataset) self.assertEqual(dataset, rendered) def test_serialize_yaml_unicode(self): dataset = "str value" env = Environment(extensions=[SerializerExtension]) rendered = env.from_string("{{ dataset|yaml }}").render(dataset=dataset) if six.PY3: self.assertEqual("str value", rendered) else: # Due to a bug in the equality handler, this check needs to be split # up into several different assertions. We need to check that the various # string segments are present in the rendered value, as well as the # type of the rendered variable (should be unicode, which is the same as # six.text_type). This should cover all use cases but also allow the test # to pass on CentOS 6 running Python 2.7. self.assertIn("str value", rendered) self.assertIsInstance(rendered, six.text_type) def test_serialize_python(self): dataset = {"foo": True, "bar": 42, "baz": [1, 2, 3], "qux": 2.0} env = Environment(extensions=[SerializerExtension]) rendered = env.from_string("{{ dataset|python }}").render(dataset=dataset) self.assertEqual(rendered, pprint.pformat(dataset)) def test_load_yaml(self): env = Environment(extensions=[SerializerExtension]) rendered = env.from_string( '{% set document = "{foo: it works}"|load_yaml %}{{ document.foo }}' ).render() self.assertEqual(rendered, "it works") rendered = env.from_string( "{% set document = document|load_yaml %}" "{{ document.foo }}" ).render(document="{foo: it works}") self.assertEqual(rendered, "it works") with self.assertRaises((TypeError, exceptions.TemplateRuntimeError)): env.from_string( "{% set document = document|load_yaml %}" "{{ document.foo }}" ).render(document={"foo": "it works"}) def test_load_tag(self): env = Environment(extensions=[SerializerExtension]) source = ( "{{ bar }}, " + "{% load_yaml as docu %}{foo: it works, {{ bar }}: baz}{% endload %}" + "{{ docu.foo }}" ) rendered = env.from_string(source).render(bar="barred") self.assertEqual(rendered, "barred, it works") source = ( '{{ bar }}, {% load_json as docu %}{"foo": "it works", "{{ bar }}": "baz"}{% endload %}' + "{{ docu.foo }}" ) rendered = env.from_string(source).render(bar="barred") self.assertEqual(rendered, "barred, it works") with self.assertRaises(exceptions.TemplateSyntaxError): env.from_string( "{% load_yamle as document %}{foo, bar: it works}{% endload %}" ).render() with self.assertRaises(exceptions.TemplateRuntimeError): env.from_string( "{% load_json as document %}{foo, bar: it works}{% endload %}" ).render() def test_load_json(self): env = Environment(extensions=[SerializerExtension]) rendered = env.from_string( '{% set document = \'{"foo": "it works"}\'|load_json %}' "{{ document.foo }}" ).render() self.assertEqual(rendered, "it works") rendered = env.from_string( "{% set document = document|load_json %}" "{{ document.foo }}" ).render(document='{"foo": "it works"}') self.assertEqual(rendered, "it works") # bad quotes with self.assertRaises(exceptions.TemplateRuntimeError): env.from_string("{{ document|load_json }}").render( document="{'foo': 'it works'}" ) # not a string with self.assertRaises(exceptions.TemplateRuntimeError): env.from_string("{{ document|load_json }}").render( document={"foo": "it works"} ) def test_load_yaml_template(self): loader = DictLoader({"foo": '{bar: "my god is blue", foo: [1, 2, 3]}'}) env = Environment(extensions=[SerializerExtension], loader=loader) rendered = env.from_string( '{% import_yaml "foo" as doc %}{{ doc.bar }}' ).render() self.assertEqual(rendered, "my god is blue") with self.assertRaises(exceptions.TemplateNotFound): env.from_string('{% import_yaml "does not exists" as doc %}').render() def test_load_json_template(self): loader = DictLoader({"foo": '{"bar": "my god is blue", "foo": [1, 2, 3]}'}) env = Environment(extensions=[SerializerExtension], loader=loader) rendered = env.from_string( '{% import_json "foo" as doc %}{{ doc.bar }}' ).render() self.assertEqual(rendered, "my god is blue") with self.assertRaises(exceptions.TemplateNotFound): env.from_string('{% import_json "does not exists" as doc %}').render() def test_load_text_template(self): loader = DictLoader({"foo": "Foo!"}) env = Environment(extensions=[SerializerExtension], loader=loader) rendered = env.from_string('{% import_text "foo" as doc %}{{ doc }}').render() self.assertEqual(rendered, "Foo!") with self.assertRaises(exceptions.TemplateNotFound): env.from_string('{% import_text "does not exists" as doc %}').render() def test_catalog(self): loader = DictLoader( { "doc1": '{bar: "my god is blue"}', "doc2": '{% import_yaml "doc1" as local2 %} never exported', "doc3": '{% load_yaml as local3 %}{"foo": "it works"}{% endload %} me neither', "main1": '{% from "doc2" import local2 %}{{ local2.bar }}', "main2": '{% from "doc3" import local3 %}{{ local3.foo }}', "main3": """ {% import "doc2" as imported2 %} {% import "doc3" as imported3 %} {{ imported2.local2.bar }} """, "main4": """ {% import "doc2" as imported2 %} {% import "doc3" as imported3 %} {{ imported3.local3.foo }} """, "main5": """ {% from "doc2" import local2 as imported2 %} {% from "doc3" import local3 as imported3 %} {{ imported2.bar }} """, "main6": """ {% from "doc2" import local2 as imported2 %} {% from "doc3" import local3 as imported3 %} {{ imported3.foo }} """, } ) env = Environment(extensions=[SerializerExtension], loader=loader) rendered = env.get_template("main1").render() self.assertEqual(rendered, "my god is blue") rendered = env.get_template("main2").render() self.assertEqual(rendered, "it works") rendered = env.get_template("main3").render().strip() self.assertEqual(rendered, "my god is blue") rendered = env.get_template("main4").render().strip() self.assertEqual(rendered, "it works") rendered = env.get_template("main5").render().strip() self.assertEqual(rendered, "my god is blue") rendered = env.get_template("main6").render().strip() self.assertEqual(rendered, "it works") def test_nested_structures(self): env = Environment(extensions=[SerializerExtension]) rendered = env.from_string("{{ data }}").render(data="foo") self.assertEqual(rendered, "foo") data = OrderedDict([("foo", OrderedDict([("bar", "baz"), ("qux", 42)]))]) rendered = env.from_string("{{ data }}").render(data=data) self.assertEqual( rendered, "{u'foo': {u'bar': u'baz', u'qux': 42}}" if six.PY2 else "{'foo': {'bar': 'baz', 'qux': 42}}", ) rendered = env.from_string("{{ data }}").render( data=[OrderedDict(foo="bar",), OrderedDict(baz=42,)] ) self.assertEqual( rendered, "[{'foo': u'bar'}, {'baz': 42}]" if six.PY2 else "[{'foo': 'bar'}, {'baz': 42}]", ) def test_set_dict_key_value(self): """ Test the `set_dict_key_value` Jinja filter. """ rendered = render_jinja_tmpl( "{{ {} | set_dict_key_value('foo:bar:baz', 42) }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "{'foo': {'bar': {'baz': 42}}}") rendered = render_jinja_tmpl( "{{ {} | set_dict_key_value('foo.bar.baz', 42, delimiter='.') }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "{'foo': {'bar': {'baz': 42}}}") def test_update_dict_key_value(self): """ Test the `update_dict_key_value` Jinja filter. """ # Use OrderedDicts to avoid random key-order-switches in the rendered string. expected = OrderedDict( [("bar", OrderedDict([("baz", OrderedDict([("qux", 1), ("quux", 3)]))]))] ) dataset = OrderedDict( [("bar", OrderedDict([("baz", OrderedDict([("qux", 1)]))]))] ) dataset_exp = OrderedDict([("quux", 3)]) rendered = render_jinja_tmpl( "{{ foo | update_dict_key_value('bar:baz', exp) }}", dict( foo=dataset, exp=dataset_exp, opts=self.local_opts, saltenv="test", salt=self.local_salt, ), ) self.assertEqual( rendered, "{u'bar': {u'baz': {u'qux': 1, u'quux': 3}}}" if six.PY2 else "{'bar': {'baz': {'qux': 1, 'quux': 3}}}", ) # Test incorrect usage for update_with in [42, "foo", [42]]: template = "{{ {} | update_dict_key_value('bar:baz', update_with) }}" expected = r"Cannot update {} with a {}.".format( type({}), type(update_with) ) self.assertRaisesRegex( SaltRenderError, expected, render_jinja_tmpl, template, dict( update_with=update_with, opts=self.local_opts, saltenv="test", salt=self.local_salt, ), ) def test_append_dict_key_value(self): """ Test the `append_dict_key_value` Jinja filter. """ rendered = render_jinja_tmpl( "{{ {} | append_dict_key_value('foo:bar:baz', 42) }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "{'foo': {'bar': {'baz': [42]}}}") rendered = render_jinja_tmpl( "{{ foo | append_dict_key_value('bar:baz', 42) }}", dict( foo={"bar": {"baz": [1, 2]}}, opts=self.local_opts, saltenv="test", salt=self.local_salt, ), ) self.assertEqual( rendered, "{u'bar': {u'baz': [1, 2, 42]}}" if six.PY2 else "{'bar': {'baz': [1, 2, 42]}}", ) def test_extend_dict_key_value(self): """ Test the `extend_dict_key_value` Jinja filter. """ rendered = render_jinja_tmpl( "{{ {} | extend_dict_key_value('foo:bar:baz', [42]) }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "{'foo': {'bar': {'baz': [42]}}}") rendered = render_jinja_tmpl( "{{ foo | extend_dict_key_value('bar:baz', [42, 43]) }}", dict( foo={"bar": {"baz": [1, 2]}}, opts=self.local_opts, saltenv="test", salt=self.local_salt, ), ) self.assertEqual( rendered, "{u'bar': {u'baz': [1, 2, 42, 43]}}" if six.PY2 else "{'bar': {'baz': [1, 2, 42, 43]}}", ) # Edge cases rendered = render_jinja_tmpl( "{{ {} | extend_dict_key_value('foo:bar:baz', 'quux') }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "{'foo': {'bar': {'baz': ['q', 'u', 'u', 'x']}}}") # Beware! When supplying a dict, the list gets extended with the dict coerced to a list, # which will only contain the keys of the dict. rendered = render_jinja_tmpl( "{{ {} | extend_dict_key_value('foo:bar:baz', {'foo': 'bar'}) }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "{'foo': {'bar': {'baz': ['foo']}}}") # Test incorrect usage template = "{{ {} | extend_dict_key_value('bar:baz', 42) }}" expected = r"Cannot extend {} with a {}.".format(type([]), type(42)) self.assertRaisesRegex( SaltRenderError, expected, render_jinja_tmpl, template, dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) def test_sequence(self): env = Environment() env.filters["sequence"] = ensure_sequence_filter rendered = env.from_string("{{ data | sequence | length }}").render(data="foo") self.assertEqual(rendered, "1") rendered = env.from_string("{{ data | sequence | length }}").render( data=["foo", "bar"] ) self.assertEqual(rendered, "2") rendered = env.from_string("{{ data | sequence | length }}").render( data=("foo", "bar") ) self.assertEqual(rendered, "2") rendered = env.from_string("{{ data | sequence | length }}").render( data=set(["foo", "bar"]) ) self.assertEqual(rendered, "2") rendered = env.from_string("{{ data | sequence | length }}").render( data={"foo": "bar"} ) self.assertEqual(rendered, "1") def test_camel_to_snake_case(self): """ Test the `to_snake_case` Jinja filter. """ rendered = render_jinja_tmpl( "{{ 'abcdEfghhIjkLmnoP' | to_snake_case }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "abcd_efghh_ijk_lmno_p") def test_snake_to_camel_case(self): """ Test the `to_camelcase` Jinja filter. """ rendered = render_jinja_tmpl( "{{ 'the_fox_jumped_over_the_lazy_dog' | to_camelcase }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "theFoxJumpedOverTheLazyDog") rendered = render_jinja_tmpl( "{{ 'the_fox_jumped_over_the_lazy_dog' | to_camelcase(uppercamel=True) }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "TheFoxJumpedOverTheLazyDog") def test_is_ip(self): """ Test the `is_ip` Jinja filter. """ rendered = render_jinja_tmpl( "{{ '192.168.0.1' | is_ip }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "True") rendered = render_jinja_tmpl( "{{ 'FE80::' | is_ip }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "True") rendered = render_jinja_tmpl( "{{ 'random' | is_ip }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "False") def test_is_ipv4(self): """ Test the `is_ipv4` Jinja filter. """ rendered = render_jinja_tmpl( "{{ '192.168.0.1' | is_ipv4 }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "True") rendered = render_jinja_tmpl( "{{ 'FE80::' | is_ipv4 }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "False") rendered = render_jinja_tmpl( "{{ 'random' | is_ipv4 }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "False") def test_is_ipv6(self): """ Test the `is_ipv6` Jinja filter. """ rendered = render_jinja_tmpl( "{{ '192.168.0.1' | is_ipv6 }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "False") rendered = render_jinja_tmpl( "{{ 'fe80::20d:b9ff:fe01:ea8%eth0' | is_ipv6 }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "True") rendered = render_jinja_tmpl( "{{ 'FE80::' | is_ipv6 }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "True") rendered = render_jinja_tmpl( "{{ 'random' | is_ipv6 }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "False") def test_ipaddr(self): """ Test the `ipaddr` Jinja filter. """ rendered = render_jinja_tmpl( "{{ '::' | ipaddr }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "::") rendered = render_jinja_tmpl( "{{ '192.168.0.1' | ipaddr }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "192.168.0.1") # provides a list with valid IP addresses only rendered = render_jinja_tmpl( "{{ ['192.168.0.1', '172.17.17.1', 'foo', 'bar', '::'] | ipaddr | join(', ') }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "192.168.0.1, 172.17.17.1, ::") # return only multicast addresses rendered = render_jinja_tmpl( "{{ ['224.0.0.1', 'FF01::1', '::'] | ipaddr(options='multicast') | join(', ') }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "224.0.0.1, ff01::1") def test_ipv4(self): """ Test the `ipv4` Jinja filter. """ rendered = render_jinja_tmpl( "{{ '192.168.0.1' | ipv4 }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "192.168.0.1") rendered = render_jinja_tmpl( "{{ ['192.168.0.1', '172.17.17.1'] | ipv4 | join(', ')}}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "192.168.0.1, 172.17.17.1") rendered = render_jinja_tmpl( "{{ 'fe80::' | ipv4 }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "None") rendered = render_jinja_tmpl( "{{ 'random' | ipv4 }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "None") rendered = render_jinja_tmpl( "{{ '192.168.0.1' | ipv4(options='lo') }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "None") rendered = render_jinja_tmpl( "{{ '127.0.0.1' | ipv4(options='lo') }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "127.0.0.1") def test_ipv6(self): """ Test the `ipv6` Jinja filter. """ rendered = render_jinja_tmpl( "{{ '192.168.0.1' | ipv6 }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "None") rendered = render_jinja_tmpl( "{{ 'random' | ipv6 }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "None") # returns the standard format value rendered = render_jinja_tmpl( "{{ 'FE80:0:0::0' | ipv6 }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "fe80::") # fe80:: is link local therefore will be returned rendered = render_jinja_tmpl( "{{ 'fe80::' | ipv6(options='ll') }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "fe80::") # fe80:: is not loopback rendered = render_jinja_tmpl( "{{ 'fe80::' | ipv6(options='lo') }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "None") # returns only IPv6 addresses in the list rendered = render_jinja_tmpl( "{{ ['fe80::', '192.168.0.1'] | ipv6 | join(', ') }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "fe80::") rendered = render_jinja_tmpl( "{{ ['fe80::', '::'] | ipv6 | join(', ') }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "fe80::, ::") def test_network_hosts(self): """ Test the `network_hosts` Jinja filter. """ rendered = render_jinja_tmpl( "{{ '192.168.0.1/30' | network_hosts | join(', ') }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "192.168.0.1, 192.168.0.2") def test_network_size(self): """ Test the `network_size` Jinja filter. """ rendered = render_jinja_tmpl( "{{ '192.168.0.1' | network_size }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "1") rendered = render_jinja_tmpl( "{{ '192.168.0.1/8' | network_size }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "16777216") @pytest.mark.flaky(max_runs=4) @pytest.mark.slow_test(seconds=1) # Test takes >0.1 and <=1 seconds def test_http_query(self): """ Test the `http_query` Jinja filter. """ for backend in ("requests", "tornado", "urllib2"): rendered = render_jinja_tmpl( "{{ 'http://icanhazip.com' | http_query(backend='" + backend + "') }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertIsInstance( rendered, six.text_type, "Failed with backend: {}".format(backend) ) dict_reply = ast.literal_eval(rendered) self.assertIsInstance( dict_reply, dict, "Failed with backend: {}".format(backend) ) self.assertIsInstance( dict_reply["body"], six.string_types, "Failed with backend: {}".format(backend), ) def test_to_bool(self): """ Test the `to_bool` Jinja filter. """ rendered = render_jinja_tmpl( "{{ 1 | to_bool }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "True") rendered = render_jinja_tmpl( "{{ 'True' | to_bool }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "True") rendered = render_jinja_tmpl( "{{ 0 | to_bool }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "False") rendered = render_jinja_tmpl( "{{ 'Yes' | to_bool }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "True") def test_quote(self): """ Test the `quote` Jinja filter. """ rendered = render_jinja_tmpl( "{{ 'random' | quote }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "random") def test_regex_search(self): """ Test the `regex_search` Jinja filter. """ rendered = render_jinja_tmpl( "{{ 'abcdefabcdef' | regex_search('BC(.*)', ignorecase=True) }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual( rendered, "('defabcdef',)" ) # because search looks only at the beginning def test_regex_match(self): """ Test the `regex_match` Jinja filter. """ rendered = render_jinja_tmpl( "{{ 'abcdefabcdef' | regex_match('BC(.*)', ignorecase=True)}}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "None") def test_regex_replace(self): """ Test the `regex_replace` Jinja filter. """ rendered = render_jinja_tmpl( r"{{ 'lets replace spaces' | regex_replace('\s+', '__') }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "lets__replace__spaces") def test_uuid(self): """ Test the `uuid` Jinja filter. """ rendered = render_jinja_tmpl( "{{ 'random' | uuid }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "3652b285-26ad-588e-a5dc-c2ee65edc804") def test_min(self): """ Test the `min` Jinja filter. """ rendered = render_jinja_tmpl( "{{ [1, 2, 3] | min }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "1") def test_max(self): """ Test the `max` Jinja filter. """ rendered = render_jinja_tmpl( "{{ [1, 2, 3] | max }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "3") def test_avg(self): """ Test the `avg` Jinja filter. """ rendered = render_jinja_tmpl( "{{ [1, 2, 3] | avg }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "2.0") def test_union(self): """ Test the `union` Jinja filter. """ rendered = render_jinja_tmpl( "{{ [1, 2, 3] | union([2, 3, 4]) | join(', ') }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "1, 2, 3, 4") def test_intersect(self): """ Test the `intersect` Jinja filter. """ rendered = render_jinja_tmpl( "{{ [1, 2, 3] | intersect([2, 3, 4]) | join(', ') }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "2, 3") def test_difference(self): """ Test the `difference` Jinja filter. """ rendered = render_jinja_tmpl( "{{ [1, 2, 3] | difference([2, 3, 4]) | join(', ') }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "1") def test_symmetric_difference(self): """ Test the `symmetric_difference` Jinja filter. """ rendered = render_jinja_tmpl( "{{ [1, 2, 3] | symmetric_difference([2, 3, 4]) | join(', ') }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "1, 4") def test_md5(self): """ Test the `md5` Jinja filter. """ rendered = render_jinja_tmpl( "{{ 'random' | md5 }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "7ddf32e17a6ac5ce04a8ecbf782ca509") def test_sha256(self): """ Test the `sha256` Jinja filter. """ rendered = render_jinja_tmpl( "{{ 'random' | sha256 }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual( rendered, "a441b15fe9a3cf56661190a0b93b9dec7d04127288cc87250967cf3b52894d11" ) def test_sha512(self): """ Test the `sha512` Jinja filter. """ rendered = render_jinja_tmpl( "{{ 'random' | sha512 }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual( rendered, six.text_type( ( "811a90e1c8e86c7b4c0eef5b2c0bf0ec1b19c4b1b5a242e6455be93787cb473cb7bc" "9b0fdeb960d00d5c6881c2094dd63c5c900ce9057255e2a4e271fc25fef1" ) ), ) def test_hmac(self): """ Test the `hmac` Jinja filter. """ rendered = render_jinja_tmpl( "{{ 'random' | hmac('secret', 'blah') }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "False") rendered = render_jinja_tmpl( ( "{{ 'get salted' | " "hmac('shared secret', 'eBWf9bstXg+NiP5AOwppB5HMvZiYMPzEM9W5YMm/AmQ=') }}" ), dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "True") def test_base64_encode(self): """ Test the `base64_encode` Jinja filter. """ rendered = render_jinja_tmpl( "{{ 'random' | base64_encode }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "cmFuZG9t") def test_base64_decode(self): """ Test the `base64_decode` Jinja filter. """ rendered = render_jinja_tmpl( "{{ 'cmFuZG9t' | base64_decode }}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "random") def test_json_query(self): """ Test the `json_query` Jinja filter. """ rendered = render_jinja_tmpl( "{{ [1, 2, 3] | json_query('[1]')}}", dict(opts=self.local_opts, saltenv="test", salt=self.local_salt), ) self.assertEqual(rendered, "2") # def test_print(self): # env = Environment(extensions=[SerializerExtension]) # source = '{% import_yaml "toto.foo" as docu %}' # name, filename = None, '' # parsed = env._parse(source, name, filename) # print parsed # print # compiled = env._generate(parsed, name, filename) # print compiled # return class TestDotNotationLookup(ModuleCase): """ Tests to call Salt functions via Jinja with various lookup syntaxes """ def setUp(self): functions = { "mocktest.ping": lambda: True, "mockgrains.get": lambda x: "jerry", } minion_opts = salt.config.minion_config( os.path.join(RUNTIME_VARS.TMP_CONF_DIR, "minion") ) render = salt.loader.render(minion_opts, functions) self.jinja = render.get("jinja") def tearDown(self): del self.jinja def render(self, tmpl_str, context=None): return self.jinja(tmpl_str, context=context or {}, argline="-s").read() def test_normlookup(self): """ Sanity-check the normal dictionary-lookup syntax for our stub function """ tmpl_str = """Hello, {{ salt['mocktest.ping']() }}.""" with patch.object(SaltCacheLoader, "file_client", Mock()): ret = self.render(tmpl_str) self.assertEqual(ret, "Hello, True.") def test_dotlookup(self): """ Check calling a stub function using awesome dot-notation """ tmpl_str = """Hello, {{ salt.mocktest.ping() }}.""" with patch.object(SaltCacheLoader, "file_client", Mock()): ret = self.render(tmpl_str) self.assertEqual(ret, "Hello, True.") def test_shadowed_dict_method(self): """ Check calling a stub function with a name that shadows a ``dict`` method name """ tmpl_str = """Hello, {{ salt.mockgrains.get('id') }}.""" with patch.object(SaltCacheLoader, "file_client", Mock()): ret = self.render(tmpl_str) self.assertEqual(ret, "Hello, jerry.")