diff --git a/CHANGELOG.md b/CHANGELOG.md index ce93967..2d71c33 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,13 @@ +## v2.2.0 +### 2023-11-30 + +This version of `earthdata-varinfo` updates `varinfo.cmr_search` to include +functionality to get a users EDL token given a LaunchPad token with +`get_edl_token_from_launchpad` and `get_edl_token_header`. +`get_edl_token_from_launchpad` returns a users EDL token given a LaunchPad +token and CMR environment and `get_edl_token_header` returns the appropriate header +prefix for each respective token. + ## v2.1.2 ### 2023-11-14 diff --git a/VERSION b/VERSION index eca07e4..ccbccc3 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -2.1.2 +2.2.0 diff --git a/tests/unit/test_cmr_search.py b/tests/unit/test_cmr_search.py index 84d52f6..a384a34 100644 --- a/tests/unit/test_cmr_search.py +++ b/tests/unit/test_cmr_search.py @@ -6,15 +6,19 @@ from cmr import (GranuleQuery, CMR_UAT) import requests -from requests.exceptions import HTTPError +from requests.exceptions import HTTPError, Timeout from varinfo.cmr_search import (get_granules, get_granule_link, - download_granule) + download_granule, + get_edl_token_from_launchpad, + get_edl_token_header, urs_token_endpoints) + from varinfo.exceptions import (CMRQueryException, MissingGranuleDownloadLinks, MissingPositionalArguments, GranuleDownloadException, - DirectoryCreationException) + DirectoryCreationException, + GetEdlTokenException) class TestQuery(TestCase): @@ -399,3 +403,80 @@ def test_requests_error(self, mock_requests_get): mock_requests_get.return_value.side_effect = HTTPError('Wrong HTTP') with self.assertRaises(GranuleDownloadException): download_granule(link, auth_header=self.bearer_token_header) + + + @patch('requests.post') + def test_get_edl_token_from_launchpad(self, mock_requests_post): + ''' Check if `get_edl_token_from_launchpad` is called with + expected parameters and if its response contains + the expected content. + ''' + # Mock the `request.post` call + mock_response = Mock(spec=requests.Response) + mock_response.json.return_value = {'access_token': 'edl-token'} + mock_requests_post.return_value = mock_response + + # Input parameters + urs_uat_edl_token_endpoint = urs_token_endpoints.get(CMR_UAT) + edl_token_from_launchpad_response = get_edl_token_from_launchpad( + self.launchpad_token_header, CMR_UAT) + + self.assertEqual(edl_token_from_launchpad_response, 'edl-token') + + mock_requests_post.assert_called_once_with( + url=urs_uat_edl_token_endpoint, + data=f'token={self.launchpad_token_header}', + timeout=10) + + + @patch('requests.post') + def test_bad_edl_token_from_launchpad_response(self, mock_requests_post): + ''' Check if the `get_edl_token_from_launchpad_response` contains + the expected content for unsuccessful response. + ''' + # Create `mock_requests_post` for a unsuccessful request + # and set its return_value + mock_response = Mock(spec=requests.Response) + mock_response.status_code = 400 + mock_response.raise_for_status.side_effect = HTTPError() + mock_requests_post.return_value = mock_response + + with self.assertRaises(GetEdlTokenException): + get_edl_token_from_launchpad(self.launchpad_token_header, CMR_UAT) + + + @patch('requests.post', side_effect=Timeout('Request timed out')) + def test_request_exception(self, mock_requests_post): + ''' Check if `GetEdlTokenException` is raised if `requests.post` + fails. + ''' + with self.assertRaises(GetEdlTokenException) as context_manager: + get_edl_token_from_launchpad(self.launchpad_token_header, CMR_UAT) + + self.assertEqual(str(context_manager.exception), + str(GetEdlTokenException('Request timed out'))) + + + @patch('requests.post') + def test_get_edl_token_header_with_launchpad(self, mock_requests_post): + ''' Test if an EDL token and its appropriate header is returned given + a LaunchPad token. + ''' + # Create successful mock response + mock_response = Mock(spec=requests.Response) + mock_response.ok = True + mock_response.json.return_value = {'access_token': 'edl-token'} + mock_requests_post.return_value = mock_response + + test_bearer_token = get_edl_token_header(self.launchpad_token_header, + CMR_UAT) + self.assertEqual(test_bearer_token, 'Bearer edl-token') + + + def test_get_edl_token_header_with_edl_token(self): + ''' Test if an EDL token is entered with its "Bearer" header prefix. + If it is the same EDL token is returned. + ''' + test_bearer_token = get_edl_token_header(self.bearer_token_header, + CMR_UAT) + self.assertEqual(test_bearer_token, self.bearer_token_header) diff --git a/varinfo/cmr_search.py b/varinfo/cmr_search.py index 8790869..07dd70c 100644 --- a/varinfo/cmr_search.py +++ b/varinfo/cmr_search.py @@ -2,7 +2,8 @@ and get a granule download URL. With the granule download URL and the `requests` library a granule is downloaded via https and saved locally. ''' -from typing import Literal + +from typing import Literal, Union import os.path from cmr import GranuleQuery, CMR_OPS, CMR_SIT, CMR_UAT @@ -12,10 +13,15 @@ MissingGranuleDownloadLinks, MissingPositionalArguments, GranuleDownloadException, - DirectoryCreationException) + DirectoryCreationException, + GetEdlTokenException) CmrEnvType = Literal[CMR_OPS, CMR_UAT, CMR_SIT] +urs_token_endpoints = { + CMR_OPS: 'https://urs.earthdata.nasa.gov/api/nams/edl_user_token', + CMR_UAT: 'https://uat.urs.earthdata.nasa.gov/api/nams/edl_user_token', + CMR_SIT: 'https://sit.urs.earthdata.nasa.gov/api/nams/edl_user_token'} def get_granules(concept_id: str = None, @@ -37,7 +43,7 @@ def get_granules(concept_id: str = None, * cmr_env/mode: CMR environments (OPS, UAT, and SIT) * auth_header: Authorization HTTP header, either: - A header with a LaunchPad token: 'Authorization: ' - - An header with an EDL bearer token: 'Authorization: Bearer ' + - A header with an EDL bearer token: 'Authorization: Bearer ' For a successful search response concept_id or short_name, version and provider must be entered along with a bearer_token. @@ -106,7 +112,7 @@ def download_granule(granule_link: str, * granule_link: granule download URL. * auth_header: Authorization HTTP header, either: - A header with a LaunchPad token: 'Authorization: ' - - An header with an EDL bearer token: 'Authorization: Bearer ' + - A header with an EDL bearer token: 'Authorization: Bearer ' * out_directory: path to save downloaded granule (the default is the current directory). ''' @@ -134,3 +140,40 @@ def download_granule(granule_link: str, # Custom exception for error from `requests.get` raise GranuleDownloadException( str(requests_exception)) from requests_exception + + +def get_edl_token_from_launchpad(launchpad_token: str, + cmr_env: CmrEnvType) -> Union[str, None]: + ''' Retrieve an EDL token given a LaunchPad token. + * launchpad_token: A LaunchPad token with no header prefixes: + + * cmr_env/mode: CMR environments (OPS, UAT, and SIT) + ''' + url_urs_endpoint = urs_token_endpoints.get(cmr_env) + try: + response = requests.post(url=url_urs_endpoint, + data=f'token={launchpad_token}', + timeout=10) + response.raise_for_status() + + except Exception as requests_exception: + raise GetEdlTokenException( + str(requests_exception)) from requests_exception + + return response.json().get('access_token') + + +def get_edl_token_header(auth_header: str, cmr_env: CmrEnvType) -> str: + ''' Helper function for getting the header for an EDL token. + * auth_header: Authorization HTTP header, either: + - A header with a LaunchPad token: 'Authorization: ' + - A header with an EDL bearer token: 'Authorization: Bearer ' + * cmr_env/mode: CMR environments (OPS, UAT, and SIT) + ''' + if 'Bearer ' not in auth_header: + edl_auth_header = ( + f'Bearer {get_edl_token_from_launchpad(auth_header, cmr_env)}' + ) + else: + edl_auth_header = auth_header + return edl_auth_header diff --git a/varinfo/exceptions.py b/varinfo/exceptions.py index b8eb024..bc59fba 100644 --- a/varinfo/exceptions.py +++ b/varinfo/exceptions.py @@ -76,12 +76,12 @@ def __init__(self, cmr_exception_message): class MissingPositionalArguments(CustomError): ''' This exception is raised when a function is missing a required - positonal argument. + positional argument. ''' - def __init__(self, positonal_argument): + def __init__(self, positional_argument): super().__init__('MissingPositionalArguments', - f'Missing positional argument: {positonal_argument}') + f'Missing positional argument: {positional_argument}') class MissingGranuleDownloadLinks(CustomError): @@ -113,3 +113,13 @@ def __init__(self, directory_creation_exception_message): super().__init__('DirectoryCreationException', 'directory creation failed with the following error: ' f'{directory_creation_exception_message}') + + +class GetEdlTokenException(CustomError): + ''' This exception is raised when `requests.post` fails to get an + EDL token given a LaunchPad token + ''' + def __init__(self, get_edl_token_exception_message): + super().__init__('GetEdlTokenException', + 'requests module failed with the following error: ' + f'{get_edl_token_exception_message}') diff --git a/varinfo/generate_umm_var.py b/varinfo/generate_umm_var.py index cf36d8a..dbcee2f 100644 --- a/varinfo/generate_umm_var.py +++ b/varinfo/generate_umm_var.py @@ -17,7 +17,7 @@ from varinfo import VarInfoFromNetCDF4 from varinfo.cmr_search import (CmrEnvType, download_granule, get_granule_link, - get_granules) + get_granules, get_edl_token_header) from varinfo.umm_var import get_all_umm_var, publish_all_umm_var @@ -45,8 +45,11 @@ def generate_collection_umm_var(collection_concept_id: str, Note - if attempting to publish to CMR, a LaunchPad token must be used. """ + # Get an EDL token and its header given a LaunchPad token + auth_header_edl_token = get_edl_token_header(auth_header, cmr_env) + granule_response = get_granules(collection_concept_id, cmr_env=cmr_env, - auth_header=auth_header) + auth_header=auth_header_edl_token) # Get the data download URL for the most recent granule (NetCDF-4 file) granule_link = get_granule_link(granule_response) @@ -54,7 +57,7 @@ def generate_collection_umm_var(collection_concept_id: str, with TemporaryDirectory() as temp_dir: # Download file to runtime environment local_granule = download_granule(granule_link, - auth_header, + auth_header_edl_token, out_directory=temp_dir) # Parse the granule with VarInfo to map all variables and relations: