salt/tests/unit/utils/test_jinja.py

"""
Tests for salt.utils.jinja
"""

import ast
import builtins
import copy
import datetime
import os
import pprint
import random
import re
import tempfile

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, Markup, exceptions
from salt.exceptions import SaltRenderError
from salt.utils.decorators.jinja import JinjaFilter
from salt.utils.jinja import (
    SaltCacheLoader,
    SerializerExtension,
    ensure_sequence_filter,
    indent,
    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.helpers import requires_network
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 ported tojson filter. 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

    def test_indent(self):
        """
        Test the indent filter with Markup object as input. Double-quotes
        should not be URL-encoded.
        """
        data = Markup('foo:\n  "bar"')
        result = indent(data)
        expected = Markup('foo:\n      "bar"')
        assert result == expected, result


class MockFileClient:
    """
    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().setUp()

    def tearDown(self):
        salt.utils.files.rm_rf(self.tempdir)
        self.tempdir = self.template_dir = self.opts

    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(str(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().setUp()

    def tearDown(self):
        salt.utils.files.rm_rf(self.tempdir)
        self.tempdir = self.template_dir = self.local_opts = self.local_salt = None

    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),
        )

    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):
    @classmethod
    def setUpClass(cls):
        cls.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": "%"},
        }
        cls.local_salt = {
            "myvar": "zero",
            "mylist": [0, 1, 2, 3],
        }

    @classmethod
    def tearDownClass(cls):
        cls.local_opts = cls.local_salt = None

    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):
    @classmethod
    def setUpClass(cls):
        cls.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"
            ),
        }
        cls.local_salt = {
            # 'dns.A': dnsutil.A,
            # 'dns.AAAA': dnsutil.AAAA,
            # 'file.exists': filemod.file_exists,
            # 'file.basename': filemod.basename,
            # 'file.dirname': filemod.dirname
        }

    @classmethod
    def tearDownClass(cls):
        cls.local_opts = cls.local_salt = None

    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)
        rendered = (
            env.from_string("{{ dataset|unique }}")
            .render(dataset=dataset)
            .strip("'{}")
            .split("', '")
        )
        self.assertEqual(sorted(rendered), sorted(list(unique)))

    def test_unique_tuple(self):
        dataset = ("foo", "foo", "bar")
        unique = set(dataset)
        env = Environment(extensions=[SerializerExtension])
        env.filters.update(JinjaFilter.salt_jinja_filters)
        rendered = (
            env.from_string("{{ dataset|unique }}")
            .render(dataset=dataset)
            .strip("'{}")
            .split("', '")
        )
        self.assertEqual(sorted(rendered), sorted(list(unique)))

    def test_unique_list(self):
        dataset = ["foo", "foo", "bar"]
        unique = ["foo", "bar"]
        env = Environment(extensions=[SerializerExtension])
        env.filters.update(JinjaFilter.salt_jinja_filters)
        rendered = (
            env.from_string("{{ dataset|unique }}")
            .render(dataset=dataset)
            .strip("'[]")
            .split("', '")
        )
        self.assertEqual(rendered, 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)
        self.assertEqual("str value", rendered)

    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_profile(self):
        env = Environment(extensions=[SerializerExtension])

        source = (
            "{%- profile as 'profile test' %}"
            + "{% set var = 'val' %}"
            + "{%- endprofile %}"
            + "{{ var }}"
        )

        rendered = env.from_string(source).render()
        self.assertEqual(rendered, "val")

    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, "{'foo': {'bar': 'baz', 'qux': 42}}",
        )

        rendered = env.from_string("{{ data }}").render(
            data=[OrderedDict(foo="bar",), OrderedDict(baz=42,)]
        )
        self.assertEqual(
            rendered, "[{'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, "{'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, "{'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, "{'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={"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")

    @requires_network()
    def test_http_query(self):
        """
        Test the `http_query` Jinja filter.
        """
        urls = (
            # These cannot be HTTPS urls since urllib2 chokes on those
            "http://saltstack.com",
            "http://community.saltstack.com",
            "http://google.com",
            "http://duckduckgo.com",
        )
        for backend in ("requests", "tornado", "urllib2"):
            rendered = render_jinja_tmpl(
                "{{ '"
                + random.choice(urls)
                + "' | http_query(backend='"
                + backend
                + "') }}",
                dict(opts=self.local_opts, saltenv="test", salt=self.local_salt),
            )
            self.assertIsInstance(
                rendered, str, "Failed with rendered template: {}".format(rendered)
            )
            dict_reply = ast.literal_eval(rendered)
            self.assertIsInstance(
                dict_reply, dict, "Failed with rendered template: {}".format(rendered)
            )
            self.assertIn(
                "body",
                dict_reply,
                "'body' not found in request response({}). Rendered template: {!r}".format(
                    dict_reply, rendered
                ),
            )
            self.assertIsInstance(
                dict_reply["body"],
                str,
                "Failed with rendered template: {}".format(rendered),
            )

    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_method_call(self):
        """
        Test the `method_call` Jinja filter.
        """
        rendered = render_jinja_tmpl(
            "{{ 6|method_call('bit_length') }}",
            dict(opts=self.local_opts, saltenv="test", salt=self.local_salt),
        )
        self.assertEqual(rendered, "3")
        rendered = render_jinja_tmpl(
            "{{ 6.7|method_call('is_integer') }}",
            dict(opts=self.local_opts, saltenv="test", salt=self.local_salt),
        )
        self.assertEqual(rendered, "False")
        rendered = render_jinja_tmpl(
            "{{ 'absaltba'|method_call('strip', 'ab') }}",
            dict(opts=self.local_opts, saltenv="test", salt=self.local_salt),
        )
        self.assertEqual(rendered, "salt")
        rendered = render_jinja_tmpl(
            "{{ [1, 2, 1, 3, 4]|method_call('index', 1, 1, 3) }}",
            dict(opts=self.local_opts, saltenv="test", salt=self.local_salt),
        )
        self.assertEqual(rendered, "2")

        # have to use `dictsort` to keep test result deterministic
        rendered = render_jinja_tmpl(
            "{{ {}|method_call('fromkeys', ['a', 'b', 'c'], 0)|dictsort }}",
            dict(opts=self.local_opts, saltenv="test", salt=self.local_salt),
        )
        self.assertEqual(rendered, "[('a', 0), ('b', 0), ('c', 0)]")

        # missing object method test
        rendered = render_jinja_tmpl(
            "{{ 6|method_call('bit_width') }}",
            dict(opts=self.local_opts, saltenv="test", salt=self.local_salt),
        )
        self.assertEqual(rendered, "None")

    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,
            str(
                "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, '<filename>'
    #     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.")