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