1import pytest
2
3try:
4    import blinker
5except ImportError:
6    blinker = None
7
8import flask
9
10pytestmark = pytest.mark.skipif(
11    blinker is None, reason="Signals require the blinker library."
12)
13
14
15def test_template_rendered(app, client):
16    @app.route("/")
17    def index():
18        return flask.render_template("simple_template.html", whiskey=42)
19
20    recorded = []
21
22    def record(sender, template, context):
23        recorded.append((template, context))
24
25    flask.template_rendered.connect(record, app)
26    try:
27        client.get("/")
28        assert len(recorded) == 1
29        template, context = recorded[0]
30        assert template.name == "simple_template.html"
31        assert context["whiskey"] == 42
32    finally:
33        flask.template_rendered.disconnect(record, app)
34
35
36def test_before_render_template():
37    app = flask.Flask(__name__)
38
39    @app.route("/")
40    def index():
41        return flask.render_template("simple_template.html", whiskey=42)
42
43    recorded = []
44
45    def record(sender, template, context):
46        context["whiskey"] = 43
47        recorded.append((template, context))
48
49    flask.before_render_template.connect(record, app)
50    try:
51        rv = app.test_client().get("/")
52        assert len(recorded) == 1
53        template, context = recorded[0]
54        assert template.name == "simple_template.html"
55        assert context["whiskey"] == 43
56        assert rv.data == b"<h1>43</h1>"
57    finally:
58        flask.before_render_template.disconnect(record, app)
59
60
61def test_request_signals():
62    app = flask.Flask(__name__)
63    calls = []
64
65    def before_request_signal(sender):
66        calls.append("before-signal")
67
68    def after_request_signal(sender, response):
69        assert response.data == b"stuff"
70        calls.append("after-signal")
71
72    @app.before_request
73    def before_request_handler():
74        calls.append("before-handler")
75
76    @app.after_request
77    def after_request_handler(response):
78        calls.append("after-handler")
79        response.data = "stuff"
80        return response
81
82    @app.route("/")
83    def index():
84        calls.append("handler")
85        return "ignored anyway"
86
87    flask.request_started.connect(before_request_signal, app)
88    flask.request_finished.connect(after_request_signal, app)
89
90    try:
91        rv = app.test_client().get("/")
92        assert rv.data == b"stuff"
93
94        assert calls == [
95            "before-signal",
96            "before-handler",
97            "handler",
98            "after-handler",
99            "after-signal",
100        ]
101    finally:
102        flask.request_started.disconnect(before_request_signal, app)
103        flask.request_finished.disconnect(after_request_signal, app)
104
105
106def test_request_exception_signal():
107    app = flask.Flask(__name__)
108    recorded = []
109
110    @app.route("/")
111    def index():
112        1 // 0
113
114    def record(sender, exception):
115        recorded.append(exception)
116
117    flask.got_request_exception.connect(record, app)
118    try:
119        assert app.test_client().get("/").status_code == 500
120        assert len(recorded) == 1
121        assert isinstance(recorded[0], ZeroDivisionError)
122    finally:
123        flask.got_request_exception.disconnect(record, app)
124
125
126def test_appcontext_signals():
127    app = flask.Flask(__name__)
128    recorded = []
129
130    def record_push(sender, **kwargs):
131        recorded.append("push")
132
133    def record_pop(sender, **kwargs):
134        recorded.append("pop")
135
136    @app.route("/")
137    def index():
138        return "Hello"
139
140    flask.appcontext_pushed.connect(record_push, app)
141    flask.appcontext_popped.connect(record_pop, app)
142    try:
143        with app.test_client() as c:
144            rv = c.get("/")
145            assert rv.data == b"Hello"
146            assert recorded == ["push"]
147        assert recorded == ["push", "pop"]
148    finally:
149        flask.appcontext_pushed.disconnect(record_push, app)
150        flask.appcontext_popped.disconnect(record_pop, app)
151
152
153def test_flash_signal(app):
154    @app.route("/")
155    def index():
156        flask.flash("This is a flash message", category="notice")
157        return flask.redirect("/other")
158
159    recorded = []
160
161    def record(sender, message, category):
162        recorded.append((message, category))
163
164    flask.message_flashed.connect(record, app)
165    try:
166        client = app.test_client()
167        with client.session_transaction():
168            client.get("/")
169            assert len(recorded) == 1
170            message, category = recorded[0]
171            assert message == "This is a flash message"
172            assert category == "notice"
173    finally:
174        flask.message_flashed.disconnect(record, app)
175
176
177def test_appcontext_tearing_down_signal():
178    app = flask.Flask(__name__)
179    recorded = []
180
181    def record_teardown(sender, **kwargs):
182        recorded.append(("tear_down", kwargs))
183
184    @app.route("/")
185    def index():
186        1 // 0
187
188    flask.appcontext_tearing_down.connect(record_teardown, app)
189    try:
190        with app.test_client() as c:
191            rv = c.get("/")
192            assert rv.status_code == 500
193            assert recorded == []
194        assert recorded == [("tear_down", {"exc": None})]
195    finally:
196        flask.appcontext_tearing_down.disconnect(record_teardown, app)
197