|
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