new debug/test preserve context implementation

This commit is contained in:
David Lord 2022-06-29 21:02:44 -07:00
parent 3635583ce2
commit 84c722044a
No known key found for this signature in database
GPG key ID: 7A1C87E3F5BC42A8
10 changed files with 84 additions and 220 deletions

View file

@ -928,13 +928,8 @@ def test_baseexception_error_handling(app, client):
def broken_func():
raise KeyboardInterrupt()
with client:
with pytest.raises(KeyboardInterrupt):
client.get("/")
ctx = flask._request_ctx_stack.top
assert ctx.preserved
assert type(ctx._preserved_exc) is KeyboardInterrupt
with pytest.raises(KeyboardInterrupt):
client.get("/")
def test_before_request_and_routing_errors(app, client):
@ -1769,57 +1764,6 @@ def test_route_decorator_custom_endpoint(app, client):
assert client.get("/bar/123").data == b"123"
def test_preserve_only_once(app, client):
app.debug = True
@app.route("/fail")
def fail_func():
1 // 0
for _x in range(3):
with pytest.raises(ZeroDivisionError):
client.get("/fail")
assert flask._request_ctx_stack.top is not None
assert flask._app_ctx_stack.top is not None
# implicit appctx disappears too
flask._request_ctx_stack.top.pop()
assert flask._request_ctx_stack.top is None
assert flask._app_ctx_stack.top is None
def test_preserve_remembers_exception(app, client):
app.debug = True
errors = []
@app.route("/fail")
def fail_func():
1 // 0
@app.route("/success")
def success_func():
return "Okay"
@app.teardown_request
def teardown_handler(exc):
errors.append(exc)
# After this failure we did not yet call the teardown handler
with pytest.raises(ZeroDivisionError):
client.get("/fail")
assert errors == []
# But this request triggers it, and it's an error
client.get("/success")
assert len(errors) == 2
assert isinstance(errors[0], ZeroDivisionError)
# At this point another request does nothing.
client.get("/success")
assert len(errors) == 3
assert errors[1] is None
def test_get_method_on_g(app_ctx):
assert flask.g.get("x") is None
assert flask.g.get("x", 11) == 11

View file

@ -123,8 +123,7 @@ def test_request_exception_signal():
flask.got_request_exception.disconnect(record, app)
def test_appcontext_signals():
app = flask.Flask(__name__)
def test_appcontext_signals(app, client):
recorded = []
def record_push(sender, **kwargs):
@ -140,10 +139,8 @@ def test_appcontext_signals():
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"]
rv = client.get("/")
assert rv.data == b"Hello"
assert recorded == ["push", "pop"]
finally:
flask.appcontext_pushed.disconnect(record_push, app)
@ -174,12 +171,12 @@ def test_flash_signal(app):
flask.message_flashed.disconnect(record, app)
def test_appcontext_tearing_down_signal():
app = flask.Flask(__name__)
def test_appcontext_tearing_down_signal(app, client):
app.testing = False
recorded = []
def record_teardown(sender, **kwargs):
recorded.append(("tear_down", kwargs))
def record_teardown(sender, exc):
recorded.append(exc)
@app.route("/")
def index():
@ -187,10 +184,9 @@ def test_appcontext_tearing_down_signal():
flask.appcontext_tearing_down.connect(record_teardown, app)
try:
with app.test_client() as c:
rv = c.get("/")
assert rv.status_code == 500
assert recorded == []
assert recorded == [("tear_down", {"exc": None})]
rv = client.get("/")
assert rv.status_code == 500
assert len(recorded) == 1
assert isinstance(recorded[0], ZeroDivisionError)
finally:
flask.appcontext_tearing_down.disconnect(record_teardown, app)

View file

@ -187,7 +187,6 @@ def test_session_transactions(app, client):
def test_session_transactions_no_null_sessions():
app = flask.Flask(__name__)
app.testing = True
with app.test_client() as c:
with pytest.raises(RuntimeError) as e:
@ -254,29 +253,6 @@ def test_reuse_client(client):
assert client.get("/").status_code == 404
def test_test_client_calls_teardown_handlers(app, client):
called = []
@app.teardown_request
def remember(error):
called.append(error)
with client:
assert called == []
client.get("/")
assert called == []
assert called == [None]
del called[:]
with client:
assert called == []
client.get("/")
assert called == []
client.get("/")
assert called == [None]
assert called == [None, None]
def test_full_url_request(app, client):
@app.route("/action", methods=["POST"])
def action():
@ -412,13 +388,15 @@ def test_cli_custom_obj(app):
def test_client_pop_all_preserved(app, req_ctx, client):
@app.route("/")
def index():
# stream_with_context pushes a third context, preserved by client
return flask.Response(flask.stream_with_context("hello"))
# stream_with_context pushes a third context, preserved by response
return flask.stream_with_context("hello")
# req_ctx fixture pushed an initial context, not marked preserved
# req_ctx fixture pushed an initial context
with client:
# request pushes a second request context, preserved by client
client.get("/")
rv = client.get("/")
# close the response, releasing the context held by stream_with_context
rv.close()
# only req_ctx fixture should still be pushed
assert flask._request_ctx_stack.top is req_ctx