Skip to content

Commit

Permalink
Added suggestions and a couple other things for Pylint
Browse files Browse the repository at this point in the history
  • Loading branch information
eni-awowale committed Aug 25, 2023
1 parent 9572b9c commit dbeddd1
Show file tree
Hide file tree
Showing 3 changed files with 61 additions and 60 deletions.
48 changes: 21 additions & 27 deletions tests/unit/test_cmr_search.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from shutil import rmtree
from tempfile import mkdtemp
from os.path import exists
from unittest import TestCase
from unittest.mock import patch, Mock

Expand All @@ -9,8 +10,11 @@

from varinfo.cmr_search import (get_granules, get_granule_link,
download_granule)
from varinfo.exceptions import (CMRQueryException, MissingGranuleDownloadLinks,
MissingPositionalArguments, GranuleDownloadException)
from varinfo.exceptions import (CMRQueryException,
MissingGranuleDownloadLinks,
MissingPositionalArguments,
GranuleDownloadException,
DirectoryCreationException)


class TestQuery(TestCase):
Expand All @@ -21,7 +25,8 @@ def setUp(self):
self.output_dir = mkdtemp()

def tearDown(self):
rmtree(self.output_dir)
if exists(self.output_dir):
rmtree(self.output_dir)

@patch('varinfo.cmr_search.GranuleQuery', spec=GranuleQuery)
def test_with_concept_id(self, granule_query_mock):
Expand Down Expand Up @@ -260,25 +265,12 @@ def test_exceptions_granule_link(self):

def test_download_gran_exceptions(self):
''' Check if exceptions are raised for incorrect input parameters
'''

with self.assertRaises(MissingPositionalArguments):
# Input parameter, token, is missing
download_granule(granule_link='https://foo.gov/example.nc4')

with self.assertRaises(NotADirectoryError):
# Input parameter, `out_directory` should be a real directory
download_granule('https://foo.gov/example.nc4',
'/Usr/foo/dir',
'foo')

with self.assertRaises(MissingPositionalArguments):
# Missing argument, `granule_link`
download_granule(token='foo')

with self.assertRaises(ValueError):
# `granule_link` should be str or bytes like not int
download_granule(granule_link=0, token='foo')
'''
with self.assertRaises(DirectoryCreationException):
# Input parameter, `out_directory` is a path that does not exist
download_granule(granule_link='https://foo.gov/example.nc4',
token='foo',
out_directory='/Usr/foo/dir')

with self.assertRaises(GranuleDownloadException):
# Not a real https link
Expand Down Expand Up @@ -308,12 +300,14 @@ def test_download_granule(self, mock_requests_get):
mock_response = self._mock_requests(content=mock_content)
# Set the return_value of `mock_requests_get` to mock_response
mock_requests_get.return_value = mock_response
file_path = download_granule(link, self.output_dir, token='foo')
file_path = download_granule(link,
token="foo",
out_directory=self.output_dir)
# Check if `download_granule` was called once with expected parameters
mock_requests_get.assert_called_once_with(link, headers=
{'Authorization':
f'Bearer {"foo"}'},
timeout=10)
mock_requests_get.assert_called_once_with(
link,
headers={'Authorization': 'Bearer foo'},
timeout=10)
# Check if download file contains expected content from `requests.get`
with self.subTest('Test if the downloaded file contains '
'expected content'):
Expand Down
61 changes: 29 additions & 32 deletions varinfo/cmr_search.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,8 +8,11 @@
from cmr import GranuleQuery, CMR_OPS
import requests

from varinfo.exceptions import (CMRQueryException, MissingGranuleDownloadLinks,
MissingPositionalArguments, GranuleDownloadException)
from varinfo.exceptions import (CMRQueryException,
MissingGranuleDownloadLinks,
MissingPositionalArguments,
GranuleDownloadException,
DirectoryCreationException)


def get_granules(concept_id: str = None,
Expand Down Expand Up @@ -60,7 +63,7 @@ def get_granules(concept_id: str = None,

except Exception as cmr_exception:
# Use custom exception, CMRQueryException, if CMR fails
raise CMRQueryException(str(cmr_exception))
raise CMRQueryException(str(cmr_exception)) from cmr_exception

# Check if granule_response is an empty list
if len(granule_response) == 0:
Expand All @@ -81,38 +84,32 @@ def get_granule_link(granule_response: list) -> str:
return granule_link


def download_granule(granule_link: str = None,
out_directory: str = os.getcwd(),
token: str = None) -> str:
def download_granule(granule_link: str,
token: str,
out_directory: str = os.getcwd()) -> str:
''' Use the requests module to download data via https.
* granule_link: granule download URL.
* token: Earthdata Login (EDL) bearer_token
* out_directory: path to save downloaded granule
(the default is the current directory).
* token: Earthdata Login (EDL) bearer_token
'''
if granule_link is not None: # Check if input parameter was entered
if isinstance(granule_link, str): # Check if granule link is a string
# Check if `out_directory` is a directory and create out_filename
if os.path.isdir(out_directory):
out_filename = out_directory + '/' + os.path.basename(granule_link)
else:
raise NotADirectoryError('Input directory '
'"out_directory" is not a directory')
else:
raise ValueError('Not a string')
else:
raise MissingPositionalArguments('granule_link')
# Create `out_directory` if it does not exist and create out_filename
if not os.path.isdir(out_directory):
try:
os.mkdir(out_directory)
except Exception as os_exception:
raise DirectoryCreationException(
str(os_exception)) from os_exception
out_filename = os.path.join(out_directory, os.path.basename(granule_link))

if token is not None:
try: # Write content of data to out_filename and return response
response = requests.get(granule_link,
headers={'Authorization': f'Bearer {token}'},
timeout=10)
with open(out_filename, 'wb') as file_download:
file_download.write(response.content)
return out_filename
except Exception as requests_exception:
# Custom exception for error from `requests.get`
raise GranuleDownloadException(str(requests_exception))
else:
raise MissingPositionalArguments('token')
try: # Write content of data to out_filename and return response
response = requests.get(granule_link,
headers={'Authorization': f'Bearer {token}'},
timeout=10)
with open(out_filename, 'wb') as file_download:
file_download.write(response.content)
return out_filename
except Exception as requests_exception:
# Custom exception for error from `requests.get`
raise GranuleDownloadException(
str(requests_exception)) from requests_exception
12 changes: 11 additions & 1 deletion varinfo/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -95,10 +95,20 @@ def __init__(self, download_link):


class GranuleDownloadException(CustomError):
''' This exception is raised when a query to CMR fails.
''' This exception is raised when the requests modules fails.
'''

def __init__(self, granule_download_exception_message):
super().__init__('GranuleDownloadException',
'requests module failed with the following error: '
f'{granule_download_exception_message}')


class DirectoryCreationException(CustomError):
''' This exception is raised when creating a directory fails.
'''

def __init__(self, directory_creation_exception_message):
super().__init__('DirectoryCreationException',
'directory creation failed with the following error: '
f'{directory_creation_exception_message}')

0 comments on commit dbeddd1

Please sign in to comment.