jinja

1
import contextlib
2
from collections import namedtuple
3

4
import pytest
5
from markupsafe import Markup
6

7
from jinja2 import Environment
8
from jinja2.async_utils import auto_aiter
9

10

11
async def make_aiter(iter):
12
    for item in iter:
13
        yield item
14

15

16
def mark_dualiter(parameter, factory):
17
    def decorator(f):
18
        return pytest.mark.parametrize(
19
            parameter, [lambda: factory(), lambda: make_aiter(factory())]
20
        )(f)
21

22
    return decorator
23

24

25
@pytest.fixture
26
def env_async():
27
    return Environment(enable_async=True)
28

29

30
@contextlib.asynccontextmanager
31
async def closing_factory():
32
    async with contextlib.AsyncExitStack() as stack:
33

34
        def closing(maybe_agen):
35
            try:
36
                aclose = maybe_agen.aclose
37
            except AttributeError:
38
                pass
39
            else:
40
                stack.push_async_callback(aclose)
41
            return maybe_agen
42

43
        yield closing
44

45

46
@mark_dualiter("foo", lambda: range(10))
47
def test_first(env_async, foo, run_async_fn):
48
    async def test():
49
        async with closing_factory() as closing:
50
            tmpl = env_async.from_string("{{ closing(foo())|first }}")
51
            return await tmpl.render_async(foo=foo, closing=closing)
52

53
    out = run_async_fn(test)
54
    assert out == "0"
55

56

57
@mark_dualiter(
58
    "items",
59
    lambda: [
60
        {"foo": 1, "bar": 2},
61
        {"foo": 2, "bar": 3},
62
        {"foo": 1, "bar": 1},
63
        {"foo": 3, "bar": 4},
64
    ],
65
)
66
def test_groupby(env_async, items):
67
    tmpl = env_async.from_string(
68
        """
69
    {%- for grouper, list in items()|groupby('foo') -%}
70
        {{ grouper }}{% for x in list %}: {{ x.foo }}, {{ x.bar }}{% endfor %}|
71
    {%- endfor %}"""
72
    )
73
    assert tmpl.render(items=items).split("|") == [
74
        "1: 1, 2: 1, 1",
75
        "2: 2, 3",
76
        "3: 3, 4",
77
        "",
78
    ]
79

80

81
@pytest.mark.parametrize(
82
    ("case_sensitive", "expect"),
83
    [
84
        (False, "a: 1, 3\nb: 2\n"),
85
        (True, "A: 3\na: 1\nb: 2\n"),
86
    ],
87
)
88
def test_groupby_case(env_async, case_sensitive, expect):
89
    tmpl = env_async.from_string(
90
        "{% for k, vs in data|groupby('k', case_sensitive=cs) %}"
91
        "{{ k }}: {{ vs|join(', ', attribute='v') }}\n"
92
        "{% endfor %}"
93
    )
94
    out = tmpl.render(
95
        data=[{"k": "a", "v": 1}, {"k": "b", "v": 2}, {"k": "A", "v": 3}],
96
        cs=case_sensitive,
97
    )
98
    assert out == expect
99

100

101
@mark_dualiter("items", lambda: [("a", 1), ("a", 2), ("b", 1)])
102
def test_groupby_tuple_index(env_async, items):
103
    tmpl = env_async.from_string(
104
        """
105
    {%- for grouper, list in items()|groupby(0) -%}
106
        {{ grouper }}{% for x in list %}:{{ x.1 }}{% endfor %}|
107
    {%- endfor %}"""
108
    )
109
    assert tmpl.render(items=items) == "a:1:2|b:1|"
110

111

112
def make_articles():
113
    Date = namedtuple("Date", "day,month,year")
114
    Article = namedtuple("Article", "title,date")
115
    return [
116
        Article("aha", Date(1, 1, 1970)),
117
        Article("interesting", Date(2, 1, 1970)),
118
        Article("really?", Date(3, 1, 1970)),
119
        Article("totally not", Date(1, 1, 1971)),
120
    ]
121

122

123
@mark_dualiter("articles", make_articles)
124
def test_groupby_multidot(env_async, articles):
125
    tmpl = env_async.from_string(
126
        """
127
    {%- for year, list in articles()|groupby('date.year') -%}
128
        {{ year }}{% for x in list %}[{{ x.title }}]{% endfor %}|
129
    {%- endfor %}"""
130
    )
131
    assert tmpl.render(articles=articles).split("|") == [
132
        "1970[aha][interesting][really?]",
133
        "1971[totally not]",
134
        "",
135
    ]
136

137

138
@mark_dualiter("int_items", lambda: [1, 2, 3])
139
def test_join_env_int(env_async, int_items):
140
    tmpl = env_async.from_string('{{ items()|join("|") }}')
141
    out = tmpl.render(items=int_items)
142
    assert out == "1|2|3"
143

144

145
@mark_dualiter("string_items", lambda: ["<foo>", Markup("<span>foo</span>")])
146
def test_join_string_list(string_items):
147
    env2 = Environment(autoescape=True, enable_async=True)
148
    tmpl = env2.from_string('{{ ["<foo>", "<span>foo</span>"|safe]|join }}')
149
    assert tmpl.render(items=string_items) == "&lt;foo&gt;<span>foo</span>"
150

151

152
def make_users():
153
    User = namedtuple("User", "username")
154
    return map(User, ["foo", "bar"])
155

156

157
@mark_dualiter("users", make_users)
158
def test_join_attribute(env_async, users):
159
    tmpl = env_async.from_string("""{{ users()|join(', ', 'username') }}""")
160
    assert tmpl.render(users=users) == "foo, bar"
161

162

163
@mark_dualiter("items", lambda: [1, 2, 3, 4, 5])
164
def test_simple_reject(env_async, items):
165
    tmpl = env_async.from_string('{{ items()|reject("odd")|join("|") }}')
166
    assert tmpl.render(items=items) == "2|4"
167

168

169
@mark_dualiter("items", lambda: [None, False, 0, 1, 2, 3, 4, 5])
170
def test_bool_reject(env_async, items):
171
    tmpl = env_async.from_string('{{ items()|reject|join("|") }}')
172
    assert tmpl.render(items=items) == "None|False|0"
173

174

175
@mark_dualiter("items", lambda: [1, 2, 3, 4, 5])
176
def test_simple_select(env_async, items):
177
    tmpl = env_async.from_string('{{ items()|select("odd")|join("|") }}')
178
    assert tmpl.render(items=items) == "1|3|5"
179

180

181
@mark_dualiter("items", lambda: [None, False, 0, 1, 2, 3, 4, 5])
182
def test_bool_select(env_async, items):
183
    tmpl = env_async.from_string('{{ items()|select|join("|") }}')
184
    assert tmpl.render(items=items) == "1|2|3|4|5"
185

186

187
def make_users():  # type: ignore
188
    User = namedtuple("User", "name,is_active")
189
    return [
190
        User("john", True),
191
        User("jane", True),
192
        User("mike", False),
193
    ]
194

195

196
@mark_dualiter("users", make_users)
197
def test_simple_select_attr(env_async, users):
198
    tmpl = env_async.from_string(
199
        '{{ users()|selectattr("is_active")|map(attribute="name")|join("|") }}'
200
    )
201
    assert tmpl.render(users=users) == "john|jane"
202

203

204
@mark_dualiter("items", lambda: list("123"))
205
def test_simple_map(env_async, items):
206
    tmpl = env_async.from_string('{{ items()|map("int")|sum }}')
207
    assert tmpl.render(items=items) == "6"
208

209

210
def test_map_sum(env_async):  # async map + async filter
211
    tmpl = env_async.from_string('{{ [[1,2], [3], [4,5,6]]|map("sum")|list }}')
212
    assert tmpl.render() == "[3, 3, 15]"
213

214

215
@mark_dualiter("users", make_users)
216
def test_attribute_map(env_async, users):
217
    tmpl = env_async.from_string('{{ users()|map(attribute="name")|join("|") }}')
218
    assert tmpl.render(users=users) == "john|jane|mike"
219

220

221
def test_empty_map(env_async):
222
    tmpl = env_async.from_string('{{ none|map("upper")|list }}')
223
    assert tmpl.render() == "[]"
224

225

226
@mark_dualiter("items", lambda: [1, 2, 3, 4, 5, 6])
227
def test_sum(env_async, items):
228
    tmpl = env_async.from_string("""{{ items()|sum }}""")
229
    assert tmpl.render(items=items) == "21"
230

231

232
@mark_dualiter("items", lambda: [{"value": 23}, {"value": 1}, {"value": 18}])
233
def test_sum_attributes(env_async, items):
234
    tmpl = env_async.from_string("""{{ items()|sum('value') }}""")
235
    assert tmpl.render(items=items)
236

237

238
def test_sum_attributes_nested(env_async):
239
    tmpl = env_async.from_string("""{{ values|sum('real.value') }}""")
240
    assert (
241
        tmpl.render(
242
            values=[
243
                {"real": {"value": 23}},
244
                {"real": {"value": 1}},
245
                {"real": {"value": 18}},
246
            ]
247
        )
248
        == "42"
249
    )
250

251

252
def test_sum_attributes_tuple(env_async):
253
    tmpl = env_async.from_string("""{{ values.items()|sum('1') }}""")
254
    assert tmpl.render(values={"foo": 23, "bar": 1, "baz": 18}) == "42"
255

256

257
@mark_dualiter("items", lambda: range(10))
258
def test_slice(env_async, items):
259
    tmpl = env_async.from_string(
260
        "{{ items()|slice(3)|list }}|{{ items()|slice(3, 'X')|list }}"
261
    )
262
    out = tmpl.render(items=items)
263
    assert out == (
264
        "[[0, 1, 2, 3], [4, 5, 6], [7, 8, 9]]|"
265
        "[[0, 1, 2, 3], [4, 5, 6, 'X'], [7, 8, 9, 'X']]"
266
    )
267

268

269
def test_custom_async_filter(env_async, run_async_fn):
270
    async def customfilter(val):
271
        return str(val)
272

273
    async def test():
274
        env_async.filters["customfilter"] = customfilter
275
        tmpl = env_async.from_string(
276
            "{{ 'static'|customfilter }} {{ arg|customfilter }}"
277
        )
278
        return await tmpl.render_async(arg="dynamic")
279

280
    out = run_async_fn(test)
281
    assert out == "static dynamic"
282

283

284
@mark_dualiter("items", lambda: range(10))
285
def test_custom_async_iteratable_filter(env_async, items, run_async_fn):
286
    async def customfilter(iterable):
287
        items = []
288
        async for item in auto_aiter(iterable):
289
            items.append(str(item))
290
            if len(items) == 3:
291
                break
292
        return ",".join(items)
293

294
    async def test():
295
        async with closing_factory() as closing:
296
            env_async.filters["customfilter"] = customfilter
297
            tmpl = env_async.from_string(
298
                "{{ closing(items())|customfilter }} .. {{ [3, 4, 5, 6]|customfilter }}"
299
            )
300
            return await tmpl.render_async(items=items, closing=closing)
301

302
    out = run_async_fn(test)
303
    assert out == "0,1,2 .. 3,4,5"
304