Test Jinja filters

This commit is contained in:
Mircea Ulinic 2016-12-01 14:05:00 +00:00
parent 6f13962478
commit a345e39c5a
2 changed files with 319 additions and 18 deletions

View File

@ -24,6 +24,7 @@ import salt.utils.http
import salt.utils.files
import salt.utils.yamlencoding
import salt.utils.locales
import salt.utils.hashutils
from salt.exceptions import (
SaltRenderError, CommandExecutionError, SaltInvocationError
)

View File

@ -3,6 +3,7 @@
# Import python libs
from __future__ import absolute_import
import os
import ast
import copy
import tempfile
import json
@ -164,6 +165,8 @@ class TestGetTemplate(TestCase):
os.path.dirname(os.path.abspath(__file__)),
'extmods'),
}
self.local_salt = {
}
def test_fallback(self):
'''
@ -174,7 +177,7 @@ class TestGetTemplate(TestCase):
with salt.utils.fopen(fn_) as fp_:
out = render_jinja_tmpl(
fp_.read(),
dict(opts=self.local_opts, saltenv='test'))
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt))
self.assertEqual(out, 'world\n')
def test_fallback_noloader(self):
@ -186,7 +189,7 @@ class TestGetTemplate(TestCase):
with salt.utils.fopen(filename) as fp_:
out = render_jinja_tmpl(
fp_.read(),
dict(opts=self.local_opts, saltenv='test'))
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt))
self.assertEqual(out, 'Hey world !a b !\n')
def test_saltenv(self):
@ -207,7 +210,7 @@ class TestGetTemplate(TestCase):
dict(opts={'cachedir': TEMPLATES_DIR, 'file_client': 'remote',
'file_roots': self.local_opts['file_roots'],
'pillar_roots': self.local_opts['pillar_roots']},
a='Hi', b='Salt', saltenv='test'))
a='Hi', b='Salt', saltenv='test', salt=self.local_salt))
self.assertEqual(out, 'Hey world !Hi Salt !\n')
self.assertEqual(fc.requests[0]['path'], 'salt://macro')
SaltCacheLoader.file_client = _fc
@ -235,7 +238,7 @@ class TestGetTemplate(TestCase):
expected,
render_jinja_tmpl,
fp_.read(),
dict(opts=self.local_opts, saltenv='test'))
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt))
SaltCacheLoader.file_client = _fc
def test_macro_additional_log_for_undefined(self):
@ -261,7 +264,7 @@ class TestGetTemplate(TestCase):
expected,
render_jinja_tmpl,
fp_.read(),
dict(opts=self.local_opts, saltenv='test'))
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt))
SaltCacheLoader.file_client = _fc
def test_macro_additional_log_syntaxerror(self):
@ -287,7 +290,7 @@ class TestGetTemplate(TestCase):
expected,
render_jinja_tmpl,
fp_.read(),
dict(opts=self.local_opts, saltenv='test'))
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt))
SaltCacheLoader.file_client = _fc
def test_non_ascii_encoding(self):
@ -302,7 +305,7 @@ class TestGetTemplate(TestCase):
dict(opts={'cachedir': TEMPLATES_DIR, 'file_client': 'remote',
'file_roots': self.local_opts['file_roots'],
'pillar_roots': self.local_opts['pillar_roots']},
a='Hi', b='Sàlt', saltenv='test'))
a='Hi', b='Sàlt', saltenv='test', salt=self.local_salt))
self.assertEqual(out, u'Hey world !Hi Sàlt !\n')
self.assertEqual(fc.requests[0]['path'], 'salt://macro')
SaltCacheLoader.file_client = _fc
@ -316,7 +319,7 @@ class TestGetTemplate(TestCase):
dict(opts={'cachedir': TEMPLATES_DIR, 'file_client': 'remote',
'file_roots': self.local_opts['file_roots'],
'pillar_roots': self.local_opts['pillar_roots']},
a='Hi', b='Sàlt', saltenv='test'))
a='Hi', b='Sàlt', saltenv='test', salt=self.local_salt))
self.assertEqual(u'Assunção\n', out)
self.assertEqual(fc.requests[0]['path'], 'salt://macro')
SaltCacheLoader.file_client = _fc
@ -324,7 +327,7 @@ class TestGetTemplate(TestCase):
@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'))
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt))
self.assertEqual(response, '2002-12-25')
objects = (
@ -336,15 +339,15 @@ class TestGetTemplate(TestCase):
for object in objects:
response = render_jinja_tmpl('{{ object|strftime }}',
dict(object=object, opts=self.local_opts, saltenv='test'))
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'))
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'))
dict(object=object, opts=self.local_opts, saltenv='test', salt=self.local_salt))
self.assertEqual(response, '02')
def test_non_ascii(self):
@ -394,7 +397,7 @@ class TestGetTemplate(TestCase):
expected,
render_jinja_tmpl,
template,
dict(opts=self.local_opts, saltenv='test')
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt)
)
@skipIf(six.PY3, 'Not applicable to Python 3: skipping.')
@ -408,7 +411,7 @@ class TestGetTemplate(TestCase):
expected,
render_jinja_tmpl,
template,
dict(opts=self.local_opts, saltenv='test')
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt)
)
builtins.__salt_system_encoding__ = encoding
@ -422,7 +425,7 @@ class TestGetTemplate(TestCase):
expected,
render_jinja_tmpl,
template,
dict(opts=self.local_opts, saltenv='test')
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt)
)
builtins.__salt_system_encoding__ = encoding
@ -434,7 +437,7 @@ class TestGetTemplate(TestCase):
expected,
render_jinja_tmpl,
template,
dict(opts=self.local_opts, saltenv='test')
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt)
)
def test_render_with_undefined_variable_utf8(self):
@ -445,7 +448,7 @@ class TestGetTemplate(TestCase):
expected,
render_jinja_tmpl,
template,
dict(opts=self.local_opts, saltenv='test')
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt)
)
def test_render_with_undefined_variable_unicode(self):
@ -456,11 +459,39 @@ class TestGetTemplate(TestCase):
expected,
render_jinja_tmpl,
template,
dict(opts=self.local_opts, saltenv='test')
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt)
)
class TestCustomExtensions(TestCase):
def __init__(self, *args, **kws):
TestCase.__init__(self, *args, **kws)
self.local_opts = {
'cachedir': TEMPLATES_DIR,
'file_client': 'local',
'file_ignore_regex': None,
'file_ignore_glob': None,
'file_roots': {
'test': [os.path.join(TEMPLATES_DIR, 'files', 'test')]
},
'pillar_roots': {
'test': [os.path.join(TEMPLATES_DIR, 'files', 'test')]
},
'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])
@ -720,6 +751,275 @@ class TestCustomExtensions(TestCase):
.render(data={'foo': 'bar'})
self.assertEqual(rendered, '1')
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, u'True')
rendered = render_jinja_tmpl("{{ 'FE80::' | is_ip }}",
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt))
self.assertEqual(rendered, u'True')
rendered = render_jinja_tmpl("{{ 'random' | is_ip }}",
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt))
self.assertEqual(rendered, u'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, u'True')
rendered = render_jinja_tmpl("{{ 'FE80::' | is_ipv4 }}",
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt))
self.assertEqual(rendered, u'False')
rendered = render_jinja_tmpl("{{ 'random' | is_ipv4 }}",
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt))
self.assertEqual(rendered, u'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, u'False')
rendered = render_jinja_tmpl("{{ 'FE80::' | is_ipv6 }}",
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt))
self.assertEqual(rendered, u'True')
rendered = render_jinja_tmpl("{{ 'random' | is_ipv6 }}",
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt))
self.assertEqual(rendered, u'False')
def test_ipaddr(self):
'''Test the `ipaddr` Jinja filter.'''
rendered = render_jinja_tmpl("{{ '192.168.0.1' | ipaddr }}",
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt))
self.assertEqual(rendered, u'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, u'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, u'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, u'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, u'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, u'None')
rendered = render_jinja_tmpl("{{ 'random' | ipv4 }}",
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt))
self.assertEqual(rendered, u'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, u'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, u'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, u'None')
rendered = render_jinja_tmpl("{{ 'random' | ipv6 }}",
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt))
self.assertEqual(rendered, u'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, u'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, u'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, u'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, u'fe80::')
rendered = render_jinja_tmpl("{{ ['fe80::', '::'] | ipv6 | join(', ') }}",
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt))
self.assertEqual(rendered, u'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, u'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, u'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, u'16777216')
def test_http_query(self):
'''Test the `http_query` Jinja filter.'''
rendered = render_jinja_tmpl("{{ 'http://www.google.com' | http_query }}",
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt))
self.assertIsInstance(rendered, six.text_type)
dict_reply = ast.literal_eval(rendered)
self.assertIsInstance(dict_reply, dict)
self.assertIsInstance(dict_reply['body'], six.string_types)
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, u'True')
rendered = render_jinja_tmpl("{{ 'True' | to_bool }}",
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt))
self.assertEqual(rendered, u'True')
rendered = render_jinja_tmpl("{{ 0 | to_bool }}",
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt))
self.assertEqual(rendered, u'False')
rendered = render_jinja_tmpl("{{ 'Yes' | to_bool }}",
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt))
self.assertEqual(rendered, u'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, u'random')
def test_regex_search(self):
'''Test the `regex_search` Jinja filter.'''
rendered = render_jinja_tmpl("{{ 'abcd' | regex_search('^(.*)BC(.*)$', ignorecase=True) | join(', ') }}",
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt))
self.assertEqual(rendered, u"a, d")
def test_regex_replace(self):
'''Test the `regex_replace` Jinja filter.'''
rendered = render_jinja_tmpl("{{ 'lets replace spaces' | regex_replace(r'\s+', '__') }}",
dict(opts=self.local_opts, saltenv='test', salt=self.local_salt))
self.assertEqual(rendered, u'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.assertIsInstance(rendered, six.text_type)
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, u'1.0')
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, u'3.0')
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, u'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, u'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, u'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, u'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, u'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, u'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, u'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, u'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, u'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, u'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, u'random')
# def test_print(self):
# env = Environment(extensions=[SerializerExtension])
# source = '{% import_yaml "toto.foo" as docu %}'