|
| 1 | +import os |
| 2 | +import json |
| 3 | +import pytest |
| 4 | +from requests_oauthlib import OAuth2Session |
| 5 | + |
| 6 | +from nypl_py_utils import Oauth2ApiClient |
| 7 | +# from requests.exceptions import ConnectTimeout |
| 8 | + |
| 9 | +_TOKEN_RESPONSE = { |
| 10 | + 'access_token': 'super-secret-token', |
| 11 | + 'expires_in': 3600, |
| 12 | + 'token_type': 'Bearer', |
| 13 | + 'scope': ['offline_access', 'openid', 'login:staff', 'admin'], |
| 14 | + 'id_token': 'super-secret-token', |
| 15 | + 'expires_at': 1677599823.3180869 |
| 16 | +} |
| 17 | + |
| 18 | +BASE_URL = 'https://example.com/api/v0.1' |
| 19 | + |
| 20 | + |
| 21 | +class TestOauth2ApiClient: |
| 22 | + |
| 23 | + @pytest.fixture |
| 24 | + def test_instance(self, requests_mock): |
| 25 | + token_url = 'https://oauth.example.com/oauth/token' |
| 26 | + requests_mock.post(token_url, text=json.dumps(_TOKEN_RESPONSE)) |
| 27 | + |
| 28 | + return Oauth2ApiClient(base_url=BASE_URL, |
| 29 | + token_url=token_url, |
| 30 | + client_id='clientid', |
| 31 | + client_secret='clientsecret' |
| 32 | + ) |
| 33 | + |
| 34 | + def test_uses_env_vars(self): |
| 35 | + env = { |
| 36 | + 'NYPL_API_CLIENT_ID': 'env client id', |
| 37 | + 'NYPL_API_CLIENT_SECRET': 'env client secret', |
| 38 | + 'NYPL_API_TOKEN_URL': 'env token url', |
| 39 | + 'NYPL_API_BASE_URL': 'env base url' |
| 40 | + } |
| 41 | + for key, value in env.items(): |
| 42 | + os.environ[key] = value |
| 43 | + |
| 44 | + client = Oauth2ApiClient() |
| 45 | + assert client.client_id == 'env client id' |
| 46 | + assert client.client_secret == 'env client secret' |
| 47 | + assert client.token_url == 'env token url' |
| 48 | + assert client.base_url == 'env base url' |
| 49 | + |
| 50 | + for key, value in env.items(): |
| 51 | + os.environ[key] = '' |
| 52 | + |
| 53 | + def test_generate_access_token(self, test_instance): |
| 54 | + test_instance._generate_access_token() |
| 55 | + assert test_instance.token['access_token']\ |
| 56 | + == _TOKEN_RESPONSE['access_token'] |
| 57 | + |
| 58 | + def test_create_oauth_client(self, test_instance): |
| 59 | + test_instance._create_oauth_client() |
| 60 | + assert type(test_instance.oauth_client) is OAuth2Session |
| 61 | + |
| 62 | + def test_do_http_method(self, requests_mock, test_instance): |
| 63 | + requests_mock.get(f'{BASE_URL}/foo', json={'foo': 'bar'}) |
| 64 | + resp = test_instance._do_http_method('GET', 'foo') |
| 65 | + assert resp == {'foo': 'bar'} |
| 66 | + |
| 67 | + def test_token_expiration(self, requests_mock, test_instance): |
| 68 | + api_get_mock = requests_mock.get(f'{BASE_URL}/foo', |
| 69 | + json={'foo': 'bar'}) |
| 70 | + |
| 71 | + # Perform first request to auto-authenticate: |
| 72 | + resp = test_instance._do_http_method('GET', 'foo') |
| 73 | + assert api_get_mock.request_history[0]._request\ |
| 74 | + .headers['Authorization'] == 'Bearer super-secret-token' |
| 75 | + |
| 76 | + # Emulate token expiration: |
| 77 | + test_instance.token['expires_at'] = 0 |
| 78 | + |
| 79 | + token_post_mock = requests_mock.post( |
| 80 | + test_instance.token_url, |
| 81 | + text=json.dumps(_TOKEN_RESPONSE) |
| 82 | + ) |
| 83 | + |
| 84 | + # Perform second request, which should detect token expiration and |
| 85 | + # re-authenticate: |
| 86 | + resp = test_instance._do_http_method('GET', 'foo') |
| 87 | + |
| 88 | + assert token_post_mock.called is True |
| 89 | + assert api_get_mock.request_history[1]._request\ |
| 90 | + .headers['Authorization'] == 'Bearer super-secret-token' |
| 91 | + assert resp == {'foo': 'bar'} |
0 commit comments