()
| 99 | |
| 100 | @pytest.mark.asyncio |
| 101 | async def test_oauth2_authorize(): |
| 102 | oauth = OAuth() |
| 103 | transport = ASGITransport( |
| 104 | AsyncPathMapDispatch({"/token": {"body": get_bearer_token()}}) |
| 105 | ) |
| 106 | client = oauth.register( |
| 107 | "dev", |
| 108 | client_id="dev", |
| 109 | client_secret="dev", |
| 110 | api_base_url="https://resource.test/api", |
| 111 | access_token_url="https://provider.test/token", |
| 112 | authorize_url="https://provider.test/authorize", |
| 113 | client_kwargs={ |
| 114 | "transport": transport, |
| 115 | }, |
| 116 | ) |
| 117 | |
| 118 | req_scope = {"type": "http", "session": {}} |
| 119 | req = Request(req_scope) |
| 120 | resp = await client.authorize_redirect(req, "https://client.test/callback") |
| 121 | assert resp.status_code == 302 |
| 122 | url = resp.headers.get("Location") |
| 123 | assert "state=" in url |
| 124 | state = dict(url_decode(urlparse.urlparse(url).query))["state"] |
| 125 | |
| 126 | assert f"_state_dev_{state}" in req.session |
| 127 | |
| 128 | req_scope.update( |
| 129 | { |
| 130 | "path": "/", |
| 131 | "query_string": f"code=a&state={state}", |
| 132 | "session": req.session, |
| 133 | } |
| 134 | ) |
| 135 | req = Request(req_scope) |
| 136 | token = await client.authorize_access_token(req) |
| 137 | assert token["access_token"] == "a" |
| 138 | |
| 139 | |
| 140 | class _FakeAsyncCache: |
nothing calls this directly
no test coverage detected
searching dependent graphs…