Das 2008/get edl token from launchpad token (#17)

* added tests for cmr.search

* working with all tests

* deleted comments

* updated changelog, typos, misc comments etc

* misc docstring

* added union

* Updated with requested changes

* Made remaining changes, deleted extraneous test and updated test

* updated with suggested changes

* updated test with new import

* because spelling is hard

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
 

VERSION

@@ -1 +1 @@
-2.1.2
+2.2.0

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)

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: <token>'
-          - An header with an EDL bearer token: 'Authorization: Bearer <token>'
+          - A header with an EDL bearer token: 'Authorization: Bearer <token>'
 
         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: <token>'
-          - An header with an EDL bearer token: 'Authorization: Bearer <token>'
+          - A header with an EDL bearer token: 'Authorization: Bearer <token>'
         * 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:
+          <Launchpad token>
+        * 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: <token>'
+          - A 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 = (
+            f'Bearer {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
 
 
@@ -45,16 +45,19 @@ 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)
 
     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: