mrml-python 0.2.3

Python wrapping on MRML
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156

import concurrent.futures

import mrml
import os
import pytest
import sys


PARTIALS_PATH = os.path.join(os.getcwd(), 'resources', 'partials')


def test_concurrent_simple_template():
    def worker():
        result = mrml.to_html("<mjml></mjml>")
        assert result.content.startswith("<!doctype html>")
        assert len(result.warnings) == 0
        return True

    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        futures = [executor.submit(worker) for _ in range(100)]
        results = [f.result() for f in futures]
        assert all(results)


def test_concurrent_memory_loader():
    parser_options = mrml.ParserOptions(
        include_loader=mrml.memory_loader(
            {
                "hello-world.mjml": "<mj-text>Hello World!</mj-text>",
            }
        )
    )

    def worker():
        result = mrml.to_html(
            '<mjml><mj-body><mj-include path="hello-world.mjml" /></mj-body></mjml>',
            parser_options=parser_options,
        )
        assert result.content.startswith("<!doctype html>")
        assert len(result.warnings) == 0
        return True

    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        futures = [executor.submit(worker) for _ in range(100)]
        results = [f.result() for f in futures]
        assert all(results)


@pytest.mark.skipif(sys.platform == "win32", reason="Unable to load from local path on windows")
def test_concurrent_local_loader():
    parser_options = mrml.ParserOptions(
        include_loader=mrml.local_loader(PARTIALS_PATH)
    )

    def worker():
        result = mrml.to_html(
            '<mjml><mj-body><mj-include path="file:///hello-world.mjml" /></mj-body></mjml>',
            parser_options=parser_options,
        )
        assert result.content.startswith("<!doctype html>")
        assert len(result.warnings) == 0
        return True

    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        futures = [executor.submit(worker) for _ in range(100)]
        results = [f.result() for f in futures]
        assert all(results)


def test_concurrent_http_loader():
    http_loader = mrml.http_loader(
        mode=mrml.HttpIncludeLoaderOptionsMode.Allow,
        list=set(["https://gist.githubusercontent.com"]),
    )
    parser_options = mrml.ParserOptions(include_loader=http_loader)

    def worker():
        result = mrml.to_html(
            """<mjml>
      <mj-body>
        <mj-include
          path="https://gist.githubusercontent.com/jdrouet/b0ac80fa08a3e7262bd4c94fc8865a87/raw/ec8771f4804a6c38427ed2a9f5937e11ec2b8c27/hello-world.mjml"
        />
      </mj-body>
    </mjml>""",
            parser_options=parser_options,
        )
        assert result.content.startswith("<!doctype html>")
        assert len(result.warnings) == 0
        return True

    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        futures = [
            executor.submit(worker) for _ in range(20)
        ]  # Fewer iterations for HTTP test
        results = [f.result() for f in futures]
        assert all(results)


@pytest.mark.skipif(sys.platform == "win32", reason="Unable to load from local path on windows")
def test_concurrent_mixed_operations():
    """Test different MRML operations running concurrently"""
    memory_parser_options = mrml.ParserOptions(
        include_loader=mrml.memory_loader(
            {
                "hello-world.mjml": "<mj-text>Hello World!</mj-text>",
            }
        )
    )

    local_parser_options = mrml.ParserOptions(
        include_loader=mrml.local_loader(PARTIALS_PATH)
    )

    def worker_simple():
        result = mrml.to_html("<mjml></mjml>")
        assert result.content.startswith("<!doctype html>")
        return "simple"

    def worker_memory():
        result = mrml.to_html(
            '<mjml><mj-body><mj-include path="hello-world.mjml" /></mj-body></mjml>',
            parser_options=memory_parser_options,
        )
        assert result.content.startswith("<!doctype html>")
        return "memory"

    def worker_local():
        result = mrml.to_html(
            '<mjml><mj-body><mj-include path="file:///hello-world.mjml" /></mj-body></mjml>',
            parser_options=local_parser_options,
        )
        assert result.content.startswith("<!doctype html>")
        return "local"

    workers = [worker_simple, worker_memory, worker_local]

    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        task_count = 100
        futures = []
        for _ in range(task_count):  # 30 total operations
            for worker in workers:
                futures.append(executor.submit(worker))

        results = [f.result() for f in futures]
        assert len(results) == task_count * len(workers)
        assert all(r in ["simple", "memory", "local"] for r in results)


def test_render_options_thread_safety():
    """Test mutation throws AttributeError"""
    render_options = mrml.RenderOptions(disable_comments=True)
    assert render_options.disable_comments
    assert render_options.social_icon_origin is None
    with pytest.raises(AttributeError) as _:
        render_options.disable_comments = False