Skip to content

Commit b885a43

Browse files
committed
Implement an IDP simulator for tests
1 parent 8af36da commit b885a43

File tree

5 files changed

+63
-115
lines changed

5 files changed

+63
-115
lines changed

app.py

Lines changed: 0 additions & 46 deletions
This file was deleted.

main.py

Lines changed: 0 additions & 66 deletions
This file was deleted.

tests/idp/__init__.py

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
import urllib.parse
2+
3+
from fastapi import APIRouter
4+
from fastapi import FastAPI
5+
from fastapi import Request
6+
from oauthlib.oauth2 import Server
7+
from starlette.responses import RedirectResponse
8+
from starlette.responses import Response
9+
10+
from .backend import TestOAuth2
11+
from .validator import TestValidator
12+
13+
14+
def get_idp():
15+
application = FastAPI()
16+
app_router = APIRouter()
17+
oauth2_server = Server(TestValidator())
18+
19+
@app_router.get("/oauth/authorization")
20+
async def authorization(request: Request):
21+
uri = str(request.url)
22+
http_method = request.method
23+
headers = dict(request.headers)
24+
body_bytes = await request.body()
25+
body = body_bytes.decode("utf-8")
26+
27+
scopes, credentials = oauth2_server.validate_authorization_request(uri, http_method, body, headers)
28+
uri = "http://idp/oauth/authorization?" + urllib.parse.urlencode({"scopes": ','.join(scopes), **credentials})
29+
headers, body, status_code = oauth2_server.create_authorization_response(uri, http_method, body, headers)
30+
31+
if status_code == 302:
32+
location = headers.get('Location', '')
33+
return RedirectResponse(location, headers=headers, status_code=status_code)
34+
35+
return Response(content=body, status_code=status_code)
36+
37+
@app_router.post("/oauth/token")
38+
async def token(request: Request):
39+
uri = str(request.url)
40+
http_method = request.method
41+
headers = dict(request.headers)
42+
body_bytes = await request.body()
43+
body = body_bytes.decode("utf-8")
44+
45+
headers, body, status_code = oauth2_server.create_token_response(uri, http_method, body, headers, {})
46+
47+
return Response(content=body, headers=headers, status_code=status_code)
48+
49+
application.include_router(app_router)
50+
return application

tests/idp/backend.py

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
from social_core.backends.oauth import BaseOAuth2
2+
3+
4+
class TestOAuth2(BaseOAuth2):
5+
name = "test"
6+
AUTHORIZATION_URL = "http://idp/oauth/authorization"
7+
ACCESS_TOKEN_URL = "http://idp/oauth/token"
8+
ACCESS_TOKEN_METHOD = "POST"
9+
REDIRECT_STATE = False
10+
STATE_PARAMETER = True
11+
SEND_USER_AGENT = True

validator.py renamed to tests/idp/validator.py

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,7 @@
22
from oauthlib.oauth2 import RequestValidator
33

44

5-
class MyRequestValidator(RequestValidator):
6-
5+
class TestValidator(RequestValidator):
76
def validate_client_id(self, client_id, request, *args, **kwargs):
87
return True
98

@@ -17,7 +16,7 @@ def get_default_scopes(self, client_id, request, *args, **kwargs):
1716
return []
1817

1918
def authenticate_client(self, request, *args, **kwargs):
20-
request.client = Client(client_id="my_client", access_token="my_token")
19+
request.client = Client(client_id="")
2120
return True
2221

2322
def confirm_redirect_uri(self, client_id, code, redirect_uri, client, request, *args, **kwargs):

0 commit comments

Comments
 (0)