|
1 | 1 | import pytest
|
2 |
| -from httpx import AsyncClient |
3 | 2 | from fastapi.responses import JSONResponse
|
4 |
| -from starlette.authentication import AuthenticationError |
| 3 | +from httpx import AsyncClient |
| 4 | +from jose import jwt |
5 | 5 |
|
6 | 6 |
|
7 | 7 | @pytest.mark.anyio
|
@@ -29,59 +29,68 @@ async def test_middleware_on_logout(get_app):
|
29 | 29 | response = await client.get("/user")
|
30 | 30 | assert response.status_code == 403 # Forbidden
|
31 | 31 |
|
| 32 | + |
32 | 33 | @pytest.mark.anyio
|
33 |
| -async def test_middleware_do_not_interfer_user_errors(get_app): |
34 |
| - app=get_app() |
35 |
| - @app.get('/unexpected_error') |
| 34 | +async def test_middleware_do_not_interfere_user_errors(get_app): |
| 35 | + app = get_app() |
| 36 | + |
| 37 | + @app.get("/unexpected_error") |
36 | 38 | def my_entry_point():
|
37 |
| - undefined_id # Intended code error |
| 39 | + raise NameError # Intended code error |
38 | 40 |
|
39 | 41 | async with AsyncClient(app=app, base_url="http://test") as client:
|
40 | 42 | with pytest.raises(NameError):
|
41 | 43 | await client.get("/unexpected_error")
|
42 | 44 |
|
| 45 | + |
43 | 46 | @pytest.mark.anyio
|
44 | 47 | async def test_middleware_ignores_custom_exceptions(get_app):
|
45 |
| - class MyCustomException(Exception): pass |
46 |
| - app=get_app() |
47 |
| - @app.get('/custom_exception') |
| 48 | + class MyCustomException(Exception): |
| 49 | + pass |
| 50 | + |
| 51 | + app = get_app() |
| 52 | + |
| 53 | + @app.get("/custom_exception") |
48 | 54 | def my_entry_point():
|
49 | 55 | raise MyCustomException()
|
50 | 56 |
|
51 | 57 | async with AsyncClient(app=app, base_url="http://test") as client:
|
52 |
| - with pytest.raises(MyCustomException): |
| 58 | + with pytest.raises(MyCustomException): |
53 | 59 | await client.get("/custom_exception")
|
54 | 60 |
|
| 61 | + |
55 | 62 | @pytest.mark.anyio
|
56 | 63 | async def test_middleware_ignores_handled_custom_exceptions(get_app):
|
57 |
| - class MyHandledException(Exception): pass |
58 |
| - app=get_app() |
| 64 | + class MyHandledException(Exception): |
| 65 | + pass |
| 66 | + |
| 67 | + app = get_app() |
| 68 | + |
59 | 69 | @app.exception_handler(MyHandledException)
|
60 | 70 | async def unicorn_exception_handler(request, exc):
|
61 | 71 | return JSONResponse(
|
62 | 72 | status_code=418,
|
63 | 73 | content={"details": "I am a custom Teapot!"},
|
64 | 74 | )
|
65 | 75 |
|
66 |
| - @app.get('/handled_exception') |
| 76 | + @app.get("/handled_exception") |
67 | 77 | def my_entry_point():
|
68 | 78 | raise MyHandledException()
|
69 | 79 |
|
70 | 80 | async with AsyncClient(app=app, base_url="http://test") as client:
|
71 | 81 | response = await client.get("/handled_exception")
|
72 |
| - assert response.status_code == 418 # I am a teapot! |
| 82 | + assert response.status_code == 418 # I am a teapot! |
73 | 83 | assert response.json() == {"details": "I am a custom Teapot!"}
|
74 | 84 |
|
| 85 | + |
75 | 86 | @pytest.mark.anyio
|
76 | 87 | async def test_middleware_reports_invalid_jwt(get_app):
|
77 | 88 | async with AsyncClient(app=get_app(with_ssr=False), base_url="http://test") as client:
|
78 | 89 | await client.get("/auth") # Simulate login
|
79 | 90 | # Insert a bad token instead
|
80 |
| - from jose import jwt |
81 |
| - badtoken=jwt.encode({"bad": "token"}, 'badsecret', 'HS256') |
| 91 | + badtoken = jwt.encode({"bad": "token"}, "badsecret", "HS256") |
82 | 92 | client.cookies.update(dict(Authorization=f"Bearer: {badtoken}"))
|
83 | 93 |
|
84 | 94 | response = await client.get("/user")
|
85 |
| - assert response.status_code == 401 # Not authenticated |
| 95 | + assert response.status_code == 401 # Not authenticated |
86 | 96 | assert response.text == "Signature verification failed."
|
87 |
| - |
|
0 commit comments