Das 2008/get edl token from launchpad token

CHANGELOG.md

@@ -1,3 +1,13 @@
+## v2.1.3
+### 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 the each respective token.
+
 ## v2.1.2
 ### 2023-11-14
 

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,
+                                UrsEnvsEdlTokenEndpoints)
 from varinfo.exceptions import (CMRQueryException,
                                 MissingGranuleDownloadLinks,
                                 MissingPositionalArguments,
                                 GranuleDownloadException,
-                                DirectoryCreationException)
+                                DirectoryCreationException,
+                                GetEdlTokenException)
 
 
 class TestQuery(TestCase):
@@ -399,3 +403,108 @@ 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.
+        '''
+        # Set `mock_response`, the return_value of `mock_response`,
+        # and the return_value of `mock_requests_post` to mock_response
+        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 = UrsEnvsEdlTokenEndpoints.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')
+
+        # Check if `get_edl_token_from_launchpad` was called once
+        # with expected parameters.
+        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_success_edl_token_from_launchpad_response(self,
+                                                       mock_requests_post):
+        ''' Check if the `get_edl_token_from_launchpad` response contains
+            the expected content for a successful response.
+        '''
+        # Create `mock_requests_post` for a successful request
+        # and set its return_value
+        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 successful request
+        successful_response = get_edl_token_from_launchpad(
+            self.launchpad_token_header, CMR_UAT)
+
+        self.assertEqual(successful_response, 'edl-token')
+
+    @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(response=mock_response)
+        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=Exception('Request timed out'))
+    def test_request_exception(self, mock_requests_post):
+        ''' Check if `GetEdlTokenException` is raised if `requests.post`
+            fails.
+        '''
+        # Create `mock_requests_post` for a unsuccessful request
+        # and set its return_value and side effect
+        mock_response = Mock(spec=requests.Response)
+        mock_response.status_code = 400
+        mock_response.raise_for_status.side_effect = Timeout(response=mock_response)
+        mock_requests_post.return_value = mock_response
+
+        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)

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,18 @@
                                 MissingGranuleDownloadLinks,
                                 MissingPositionalArguments,
                                 GranuleDownloadException,
-                                DirectoryCreationException)
+                                DirectoryCreationException,
+                                GetEdlTokenException)
 
 
 CmrEnvType = Literal[CMR_OPS, CMR_UAT, CMR_SIT]
+UrsEdlTokenEndpoints = [
+    'https://urs.earthdata.nasa.gov/api/nams/edl_user_token',
+    'https://uat.urs.earthdata.nasa.gov/api/nams/edl_user_token',
+    'https://sit.urs.earthdata.nasa.gov/api/nams/edl_user_token']
+
+UrsEnvsEdlTokenEndpoints = dict(zip([CMR_OPS, CMR_UAT, CMR_SIT],
+                                    UrsEdlTokenEndpoints))
 
 
 def get_granules(concept_id: str = None,
@@ -134,3 +143,41 @@ 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:
+            <Launchpad token>
+        * cmr_env/mode: CMR environments (OPS, UAT, and SIT)
+    '''
+    url_urs_endpoint = UrsEnvsEdlTokenEndpoints.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: <token>'
+            - An header with an EDL bearer token:
+                'Authorization: Bearer <token>'
+        * cmr_env/mode: CMR environments (OPS, UAT, and SIT)
+    '''
+    if 'Bearer ' not in auth_header:
+        edl_auth_header = ('Bearer '
+                           f'''{get_edl_token_from_launchpad(auth_header,
+                           cmr_env)}''')
+    else:
+        edl_auth_header = auth_header
+    return edl_auth_header

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}')

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
 
 
@@ -27,7 +27,7 @@
 
 
 def generate_collection_umm_var(collection_concept_id: str,
-                                auth_header: str,
+                                launchpad_token: str,
                                 cmr_env: CmrEnvType = CMR_UAT,
                                 publish: bool = False) -> UmmVarReturnType:
     """ Run all the of the functions for downloading and publishing
@@ -36,25 +36,28 @@ def generate_collection_umm_var(collection_concept_id: str,
         * collection_concept_id: Concept ID for collection that variables have
           been generated for.
         * cmr_env: URLs for CMR environments (OPS, UAT, and SIT)
-        * auth_header: Authorization HTTP header, containing a LaunchPad
-          token: 'Authorization: <token>'
+        * launchpad_token: A LaunchPad token with no header prefixes:
+            <Launchpad token>
         * publish: Optional argument determining whether to publish the
           generated UMM-Var records to the indicated CMR instance. Defaults to
           False.
 
         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(launchpad_token, 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)
 
     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:
@@ -69,7 +72,7 @@ def generate_collection_umm_var(collection_concept_id: str,
         # error messages returned from CMR.
         publication_response = publish_all_umm_var(collection_concept_id,
                                                    all_umm_var_records,
-                                                   auth_header, cmr_env)
+                                                   launchpad_token, cmr_env)
 
         # Produce a list indicating publication information for all variables.
         # Variables that were successfully published will have a list element