|
|
|
@ -25,136 +25,134 @@ pytestmark = pytest.mark.skipif(
|
|
|
|
|
reason='Signals require the blinker library.' |
|
|
|
|
) |
|
|
|
|
|
|
|
|
|
class TestSignals(TestFlask): |
|
|
|
|
|
|
|
|
|
def test_template_rendered(self): |
|
|
|
|
app = flask.Flask(__name__) |
|
|
|
|
|
|
|
|
|
@app.route('/') |
|
|
|
|
def index(): |
|
|
|
|
return flask.render_template('simple_template.html', whiskey=42) |
|
|
|
|
|
|
|
|
|
recorded = [] |
|
|
|
|
|
|
|
|
|
def record(sender, template, context): |
|
|
|
|
recorded.append((template, context)) |
|
|
|
|
|
|
|
|
|
flask.template_rendered.connect(record, app) |
|
|
|
|
try: |
|
|
|
|
app.test_client().get('/') |
|
|
|
|
assert len(recorded) == 1 |
|
|
|
|
template, context = recorded[0] |
|
|
|
|
assert template.name == 'simple_template.html' |
|
|
|
|
assert context['whiskey'] == 42 |
|
|
|
|
finally: |
|
|
|
|
flask.template_rendered.disconnect(record, app) |
|
|
|
|
|
|
|
|
|
def test_request_signals(self): |
|
|
|
|
app = flask.Flask(__name__) |
|
|
|
|
calls = [] |
|
|
|
|
|
|
|
|
|
def before_request_signal(sender): |
|
|
|
|
calls.append('before-signal') |
|
|
|
|
|
|
|
|
|
def after_request_signal(sender, response): |
|
|
|
|
assert response.data == b'stuff' |
|
|
|
|
calls.append('after-signal') |
|
|
|
|
|
|
|
|
|
@app.before_request |
|
|
|
|
def before_request_handler(): |
|
|
|
|
calls.append('before-handler') |
|
|
|
|
|
|
|
|
|
@app.after_request |
|
|
|
|
def after_request_handler(response): |
|
|
|
|
calls.append('after-handler') |
|
|
|
|
response.data = 'stuff' |
|
|
|
|
return response |
|
|
|
|
|
|
|
|
|
@app.route('/') |
|
|
|
|
def index(): |
|
|
|
|
calls.append('handler') |
|
|
|
|
return 'ignored anyway' |
|
|
|
|
|
|
|
|
|
flask.request_started.connect(before_request_signal, app) |
|
|
|
|
flask.request_finished.connect(after_request_signal, app) |
|
|
|
|
|
|
|
|
|
try: |
|
|
|
|
rv = app.test_client().get('/') |
|
|
|
|
assert rv.data == b'stuff' |
|
|
|
|
|
|
|
|
|
assert calls == ['before-signal', 'before-handler', 'handler', |
|
|
|
|
'after-handler', 'after-signal'] |
|
|
|
|
finally: |
|
|
|
|
flask.request_started.disconnect(before_request_signal, app) |
|
|
|
|
flask.request_finished.disconnect(after_request_signal, app) |
|
|
|
|
|
|
|
|
|
def test_request_exception_signal(self): |
|
|
|
|
app = flask.Flask(__name__) |
|
|
|
|
recorded = [] |
|
|
|
|
|
|
|
|
|
@app.route('/') |
|
|
|
|
def index(): |
|
|
|
|
1 // 0 |
|
|
|
|
|
|
|
|
|
def record(sender, exception): |
|
|
|
|
recorded.append(exception) |
|
|
|
|
|
|
|
|
|
flask.got_request_exception.connect(record, app) |
|
|
|
|
try: |
|
|
|
|
assert app.test_client().get('/').status_code == 500 |
|
|
|
|
def test_template_rendered(): |
|
|
|
|
app = flask.Flask(__name__) |
|
|
|
|
|
|
|
|
|
@app.route('/') |
|
|
|
|
def index(): |
|
|
|
|
return flask.render_template('simple_template.html', whiskey=42) |
|
|
|
|
|
|
|
|
|
recorded = [] |
|
|
|
|
|
|
|
|
|
def record(sender, template, context): |
|
|
|
|
recorded.append((template, context)) |
|
|
|
|
|
|
|
|
|
flask.template_rendered.connect(record, app) |
|
|
|
|
try: |
|
|
|
|
app.test_client().get('/') |
|
|
|
|
assert len(recorded) == 1 |
|
|
|
|
template, context = recorded[0] |
|
|
|
|
assert template.name == 'simple_template.html' |
|
|
|
|
assert context['whiskey'] == 42 |
|
|
|
|
finally: |
|
|
|
|
flask.template_rendered.disconnect(record, app) |
|
|
|
|
|
|
|
|
|
def test_request_signals(): |
|
|
|
|
app = flask.Flask(__name__) |
|
|
|
|
calls = [] |
|
|
|
|
|
|
|
|
|
def before_request_signal(sender): |
|
|
|
|
calls.append('before-signal') |
|
|
|
|
|
|
|
|
|
def after_request_signal(sender, response): |
|
|
|
|
assert response.data == b'stuff' |
|
|
|
|
calls.append('after-signal') |
|
|
|
|
|
|
|
|
|
@app.before_request |
|
|
|
|
def before_request_handler(): |
|
|
|
|
calls.append('before-handler') |
|
|
|
|
|
|
|
|
|
@app.after_request |
|
|
|
|
def after_request_handler(response): |
|
|
|
|
calls.append('after-handler') |
|
|
|
|
response.data = 'stuff' |
|
|
|
|
return response |
|
|
|
|
|
|
|
|
|
@app.route('/') |
|
|
|
|
def index(): |
|
|
|
|
calls.append('handler') |
|
|
|
|
return 'ignored anyway' |
|
|
|
|
|
|
|
|
|
flask.request_started.connect(before_request_signal, app) |
|
|
|
|
flask.request_finished.connect(after_request_signal, app) |
|
|
|
|
|
|
|
|
|
try: |
|
|
|
|
rv = app.test_client().get('/') |
|
|
|
|
assert rv.data == b'stuff' |
|
|
|
|
|
|
|
|
|
assert calls == ['before-signal', 'before-handler', 'handler', |
|
|
|
|
'after-handler', 'after-signal'] |
|
|
|
|
finally: |
|
|
|
|
flask.request_started.disconnect(before_request_signal, app) |
|
|
|
|
flask.request_finished.disconnect(after_request_signal, app) |
|
|
|
|
|
|
|
|
|
def test_request_exception_signal(): |
|
|
|
|
app = flask.Flask(__name__) |
|
|
|
|
recorded = [] |
|
|
|
|
|
|
|
|
|
@app.route('/') |
|
|
|
|
def index(): |
|
|
|
|
1 // 0 |
|
|
|
|
|
|
|
|
|
def record(sender, exception): |
|
|
|
|
recorded.append(exception) |
|
|
|
|
|
|
|
|
|
flask.got_request_exception.connect(record, app) |
|
|
|
|
try: |
|
|
|
|
assert app.test_client().get('/').status_code == 500 |
|
|
|
|
assert len(recorded) == 1 |
|
|
|
|
assert isinstance(recorded[0], ZeroDivisionError) |
|
|
|
|
finally: |
|
|
|
|
flask.got_request_exception.disconnect(record, app) |
|
|
|
|
|
|
|
|
|
def test_appcontext_signals(): |
|
|
|
|
app = flask.Flask(__name__) |
|
|
|
|
recorded = [] |
|
|
|
|
|
|
|
|
|
def record_push(sender, **kwargs): |
|
|
|
|
recorded.append('push') |
|
|
|
|
|
|
|
|
|
def record_pop(sender, **kwargs): |
|
|
|
|
recorded.append('pop') |
|
|
|
|
|
|
|
|
|
@app.route('/') |
|
|
|
|
def index(): |
|
|
|
|
return 'Hello' |
|
|
|
|
|
|
|
|
|
flask.appcontext_pushed.connect(record_push, app) |
|
|
|
|
flask.appcontext_popped.connect(record_pop, app) |
|
|
|
|
try: |
|
|
|
|
with app.test_client() as c: |
|
|
|
|
rv = c.get('/') |
|
|
|
|
assert rv.data == b'Hello' |
|
|
|
|
assert recorded == ['push'] |
|
|
|
|
assert recorded == ['push', 'pop'] |
|
|
|
|
finally: |
|
|
|
|
flask.appcontext_pushed.disconnect(record_push, app) |
|
|
|
|
flask.appcontext_popped.disconnect(record_pop, app) |
|
|
|
|
|
|
|
|
|
def test_flash_signal(): |
|
|
|
|
app = flask.Flask(__name__) |
|
|
|
|
app.config['SECRET_KEY'] = 'secret' |
|
|
|
|
|
|
|
|
|
@app.route('/') |
|
|
|
|
def index(): |
|
|
|
|
flask.flash('This is a flash message', category='notice') |
|
|
|
|
return flask.redirect('/other') |
|
|
|
|
|
|
|
|
|
recorded = [] |
|
|
|
|
|
|
|
|
|
def record(sender, message, category): |
|
|
|
|
recorded.append((message, category)) |
|
|
|
|
|
|
|
|
|
flask.message_flashed.connect(record, app) |
|
|
|
|
try: |
|
|
|
|
client = app.test_client() |
|
|
|
|
with client.session_transaction(): |
|
|
|
|
client.get('/') |
|
|
|
|
assert len(recorded) == 1 |
|
|
|
|
assert isinstance(recorded[0], ZeroDivisionError) |
|
|
|
|
finally: |
|
|
|
|
flask.got_request_exception.disconnect(record, app) |
|
|
|
|
|
|
|
|
|
def test_appcontext_signals(self): |
|
|
|
|
app = flask.Flask(__name__) |
|
|
|
|
recorded = [] |
|
|
|
|
|
|
|
|
|
def record_push(sender, **kwargs): |
|
|
|
|
recorded.append('push') |
|
|
|
|
|
|
|
|
|
def record_pop(sender, **kwargs): |
|
|
|
|
recorded.append('pop') |
|
|
|
|
|
|
|
|
|
@app.route('/') |
|
|
|
|
def index(): |
|
|
|
|
return 'Hello' |
|
|
|
|
|
|
|
|
|
flask.appcontext_pushed.connect(record_push, app) |
|
|
|
|
flask.appcontext_popped.connect(record_pop, app) |
|
|
|
|
try: |
|
|
|
|
with app.test_client() as c: |
|
|
|
|
rv = c.get('/') |
|
|
|
|
assert rv.data == b'Hello' |
|
|
|
|
assert recorded == ['push'] |
|
|
|
|
assert recorded == ['push', 'pop'] |
|
|
|
|
finally: |
|
|
|
|
flask.appcontext_pushed.disconnect(record_push, app) |
|
|
|
|
flask.appcontext_popped.disconnect(record_pop, app) |
|
|
|
|
|
|
|
|
|
def test_flash_signal(self): |
|
|
|
|
app = flask.Flask(__name__) |
|
|
|
|
app.config['SECRET_KEY'] = 'secret' |
|
|
|
|
|
|
|
|
|
@app.route('/') |
|
|
|
|
def index(): |
|
|
|
|
flask.flash('This is a flash message', category='notice') |
|
|
|
|
return flask.redirect('/other') |
|
|
|
|
|
|
|
|
|
recorded = [] |
|
|
|
|
|
|
|
|
|
def record(sender, message, category): |
|
|
|
|
recorded.append((message, category)) |
|
|
|
|
|
|
|
|
|
flask.message_flashed.connect(record, app) |
|
|
|
|
try: |
|
|
|
|
client = app.test_client() |
|
|
|
|
with client.session_transaction(): |
|
|
|
|
client.get('/') |
|
|
|
|
assert len(recorded) == 1 |
|
|
|
|
message, category = recorded[0] |
|
|
|
|
assert message == 'This is a flash message' |
|
|
|
|
assert category == 'notice' |
|
|
|
|
finally: |
|
|
|
|
flask.message_flashed.disconnect(record, app) |
|
|
|
|
message, category = recorded[0] |
|
|
|
|
assert message == 'This is a flash message' |
|
|
|
|
assert category == 'notice' |
|
|
|
|
finally: |
|
|
|
|
flask.message_flashed.disconnect(record, app) |
|
|
|
|