374e265fcb
* configurable oauth request payload * support interpolation for dictionaries that are not new subcomponents * rewrite a declarative oauth authenticator that performs interpolation at runtime * formatting * whatever i don't know why factory gets flagged w/ the newline change * we java now * remove duplicate oauth * add some comments * parse time properly from string interpolation * move declarative oauth to its own package in declarative module * add changelog info
91 lines
3.1 KiB
Python
91 lines
3.1 KiB
Python
#
|
|
# Copyright (c) 2022 Airbyte, Inc., all rights reserved.
|
|
#
|
|
|
|
import logging
|
|
|
|
import pendulum
|
|
import requests
|
|
from airbyte_cdk.sources.declarative.auth import DeclarativeOauth2Authenticator
|
|
from requests import Response
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
resp = Response()
|
|
|
|
config = {
|
|
"refresh_endpoint": "refresh_end",
|
|
"client_id": "some_client_id",
|
|
"client_secret": "some_client_secret",
|
|
"refresh_token": "some_refresh_token",
|
|
"token_expiry_date": pendulum.now().subtract(days=2).to_rfc3339_string(),
|
|
"custom_field": "in_outbound_request",
|
|
"another_field": "exists_in_body",
|
|
}
|
|
|
|
|
|
class TestOauth2Authenticator:
|
|
"""
|
|
Test class for OAuth2Authenticator.
|
|
"""
|
|
|
|
def test_refresh_request_body(self):
|
|
"""
|
|
Request body should match given configuration.
|
|
"""
|
|
scopes = ["scope1", "scope2"]
|
|
oauth = DeclarativeOauth2Authenticator(
|
|
token_refresh_endpoint="{{ config['refresh_endpoint'] }}",
|
|
client_id="{{ config['client_id'] }}",
|
|
client_secret="{{ config['client_secret'] }}",
|
|
refresh_token="{{ config['refresh_token'] }}",
|
|
config=config,
|
|
scopes=["scope1", "scope2"],
|
|
token_expiry_date="{{ config['token_expiry_date'] }}",
|
|
refresh_request_body={
|
|
"custom_field": "{{ config['custom_field'] }}",
|
|
"another_field": "{{ config['another_field'] }}",
|
|
"scopes": ["no_override"],
|
|
},
|
|
)
|
|
body = oauth.get_refresh_request_body()
|
|
expected = {
|
|
"grant_type": "refresh_token",
|
|
"client_id": "some_client_id",
|
|
"client_secret": "some_client_secret",
|
|
"refresh_token": "some_refresh_token",
|
|
"scopes": scopes,
|
|
"custom_field": "in_outbound_request",
|
|
"another_field": "exists_in_body",
|
|
}
|
|
assert body == expected
|
|
|
|
def test_refresh_access_token(self, mocker):
|
|
oauth = DeclarativeOauth2Authenticator(
|
|
token_refresh_endpoint="{{ config['refresh_endpoint'] }}",
|
|
client_id="{{ config['client_id'] }}",
|
|
client_secret="{{ config['client_secret'] }}",
|
|
refresh_token="{{ config['refresh_token'] }}",
|
|
config=config,
|
|
scopes=["scope1", "scope2"],
|
|
token_expiry_date="{{ config['token_expiry_date'] }}",
|
|
refresh_request_body={
|
|
"custom_field": "{{ config['custom_field'] }}",
|
|
"another_field": "{{ config['another_field'] }}",
|
|
"scopes": ["no_override"],
|
|
},
|
|
)
|
|
|
|
resp.status_code = 200
|
|
mocker.patch.object(resp, "json", return_value={"access_token": "access_token", "expires_in": 1000})
|
|
mocker.patch.object(requests, "request", side_effect=mock_request, autospec=True)
|
|
token = oauth.refresh_access_token()
|
|
|
|
assert ("access_token", 1000) == token
|
|
|
|
|
|
def mock_request(method, url, data):
|
|
if url == "refresh_end":
|
|
return resp
|
|
raise Exception(f"Error while refreshing access token with request: {method}, {url}, {data}")
|