jlengrand/landsat-util
1
0
Fork 0
mirror of https://github.com/Skyai-io/landsat-util.git synced 2026-03-10 09:01:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #169 from developmentseed/develop

Browse Source 
v0.12.2
This commit is contained in:
Scisco
2016-03-25 12:52:04 -04:00
parent 14f14ec902 18120db50f
commit cd100f67ef
10 changed files with 195 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
CHANGES.txt
View File

@@ -1,6 +1,11 @@
Changes
=======
0.12.2 (2016-02-18)
------------------
- Fix for #167
- Fix for #145
0.12.0 (2016-02-18)
------------------
- Add USGS download fallback closes #89

2
landsat/__init__.py
View File

@@ -1 +1 @@
__version__ = '0.12.1'
__version__ = '0.12.2'

166
landsat/downloader.py
View File

@@ -58,42 +58,49 @@ class Downloader(VerbosityMixin):
"""
if isinstance(scenes, list):
output = {}
files = []
for scene in scenes:
# If bands are provided the image is from 2015 or later use Amazon
self.scene_interpreter(scene)
if (bands and int(scene[12]) > 4):
if isinstance(bands, list):
# Create a folder to download the specific bands into
path = check_create_folder(join(self.download_dir, scene))
try:
# Always grab MTL.txt if bands are specified
if 'BQA' not in bands:
bands.append('QA')
# for all scenes if bands provided, first check AWS, if the bands exist
# download them, otherwise use Google and then USGS.
try:
# if bands are not provided, directly go to Goodle and then USGS
if not isinstance(bands, list):
raise RemoteFileDoesntExist
files.append(self.amazon_s3(scene, bands))
if 'MTL' not in bands:
bands.append('MTL')
except RemoteFileDoesntExist:
try:
files.append(self.google_storage(scene, self.download_dir))
except RemoteFileDoesntExist:
files.append(self.usgs_eros(scene, self.download_dir))
for band in bands:
self.amazon_s3(scene, band, path)
output[scene] = 'aws'
except RemoteFileDoesntExist:
self.google_storage(scene, self.download_dir)
output[scene] = 'google'
else:
raise Exception('Expected bands list')
else:
self.google_storage(scene, self.download_dir)
output[scene] = 'google'
return output
return files
else:
raise Exception('Expected sceneIDs list')
def usgs_eros(self, scene, path):
""" Downloads the image from USGS """
# download from usgs if login information is provided
if self.usgs_user and self.usgs_pass:
try:
api_key = api.login(self.usgs_user, self.usgs_pass)
except USGSError as e:
error_tree = ElementTree.fromstring(str(e.message))
error_text = error_tree.find("SOAP-ENV:Body/SOAP-ENV:Fault/faultstring", api.NAMESPACES).text
raise USGSInventoryAccessMissing(error_text)
download_url = api.download('LANDSAT_8', 'EE', [scene], api_key=api_key)
if download_url:
self.output('Source: USGS EarthExplorer', normal=True, arrow=True)
return self.fetch(download_url[0], path)
raise RemoteFileDoesntExist('%s is not available on AWS S3, Google or USGS Earth Explorer' % scene)
raise RemoteFileDoesntExist('%s is not available on AWS S3 or Google Storage' % scene)
def google_storage(self, scene, path):
"""
Google Storage Downloader.
@@ -110,67 +117,49 @@ class Downloader(VerbosityMixin):
:returns:
Boolean
"""
sat = self.scene_interpreter(scene)
filename = scene + '.tar.bz'
sat = self.scene_interpreter(scene)
url = self.google_storage_url(sat)
if self.remote_file_exists(url):
return self.fetch(url, path, filename)
self.remote_file_exists(url)
else:
# download from usgs if login information is provided
if self.usgs_user and self.usgs_pass:
try:
api_key = api.login(self.usgs_user, self.usgs_pass)
except USGSError as e:
error_tree = ElementTree.fromstring(str(e.message))
error_text = error_tree.find("SOAP-ENV:Body/SOAP-ENV:Fault/faultstring", api.NAMESPACES).text
raise USGSInventoryAccessMissing(error_text)
self.output('Source: Google Storge', normal=True, arrow=True)
return self.fetch(url, path)
download_url = api.download('LANDSAT_8', 'EE', [scene], api_key=api_key)
if download_url:
return self.fetch(download_url[0], path, filename)
raise RemoteFileDoesntExist('%s is not available on AWS S3, Google or USGS Earth Explorer' % filename)
raise RemoteFileDoesntExist('%s is not available on AWS S3 or Google Storage' % filename)
def amazon_s3(self, scene, band, path):
def amazon_s3(self, scene, bands):
"""
Amazon S3 downloader
:param scene:
The scene ID.
:type scene:
String
:param band:
The band number.
:type band:
String, Integer
:param path:
The directory path to where the image should be stored
:type path:
String
:returns:
Boolean
"""
sat = self.scene_interpreter(scene)
if band != 'MTL':
filename = '%s_B%s.TIF' % (scene, band)
else:
filename = '%s_%s.txt' % (scene, band)
url = self.amazon_s3_url(sat, filename)
# Always grab MTL.txt and QA band if bands are specified
if 'BQA' not in bands:
bands.append('QA')
if self.remote_file_exists(url):
return self.fetch(url, path, filename)
if 'MTL' not in bands:
bands.append('MTL')
else:
raise RemoteFileDoesntExist('%s is not available on Amazon S3' % filename)
urls = []
def fetch(self, url, path, filename):
for band in bands:
# get url for the band
url = self.amazon_s3_url(sat, band)
# make sure it exist
self.remote_file_exists(url)
urls.append(url)
# create folder
path = check_create_folder(join(self.download_dir, scene))
self.output('Source: AWS S3', normal=True, arrow=True)
for url in urls:
self.fetch(url, path)
return path
def fetch(self, url, path):
""" Downloads the given url.
:param url:
@@ -190,18 +179,26 @@ class Downloader(VerbosityMixin):
Boolean
"""
segments = url.split('/')
filename = segments[-1]
# remove query parameters from the filename
filename = filename.split('?')[0]
self.output('Downloading: %s' % filename, normal=True, arrow=True)
# print(join(path, filename))
# raise Exception
if exists(join(path, filename)):
size = getsize(join(path, filename))
if size == self.get_remote_file_size(url):
self.output('%s already exists on your system' % filename, normal=True, color='green', indent=1)
return False
fetch(url, path)
else:
fetch(url, path)
self.output('stored at %s' % path, normal=True, color='green', indent=1)
return True
return join(path, filename)
def google_storage_url(self, sat):
"""
@@ -218,7 +215,7 @@ class Downloader(VerbosityMixin):
filename = sat['scene'] + '.tar.bz'
return url_builder([self.google, sat['sat'], sat['path'], sat['row'], filename])
def amazon_s3_url(self, sat, filename):
def amazon_s3_url(self, sat, band):
"""
Return an amazon s3 url the contains the scene and band provided.
@@ -234,6 +231,11 @@ class Downloader(VerbosityMixin):
:returns:
(String) The URL to a S3 file
"""
if band != 'MTL':
filename = '%s_B%s.TIF' % (sat['scene'], band)
else:
filename = '%s_%s.txt' % (sat['scene'], band)
return url_builder([self.s3, sat['sat'], sat['path'], sat['row'], sat['scene'], filename])
def remote_file_exists(self, url):
@@ -249,10 +251,8 @@ class Downloader(VerbosityMixin):
"""
status = requests.head(url).status_code
if status == 200:
return True
else:
return False
if status != 200:
raise RemoteFileDoesntExist
def get_remote_file_size(self, url):
""" Gets the filesize of a remote file.

30
landsat/landsat.py
View File

@@ -324,8 +324,15 @@ def main(args):
s = Search()
try:
lat = float(args.lat) if args.lat else None
lon = float(args.lon) if args.lon else None
if args.lat is not None:
lat = float(args.lat)
else:
lat = None
if args.lon is not None:
lon = float(args.lon)
else:
lon = None
except ValueError:
return ["The latitude and longitude values must be valid numbers", 1]
@@ -392,23 +399,14 @@ def main(args):
if not args.bands:
bands = [4, 3, 2]
downloaded = d.download(args.scenes, bands)
files = d.download(args.scenes, bands)
if args.process:
if not args.bands:
args.bands = '432'
force_unzip = True if args.force_unzip else False
for scene, src in downloaded.iteritems():
if args.dest:
path = join(args.dest, scene)
else:
path = join(settings.DOWNLOAD_DIR, scene)
# Keep using Google if the image is before 2015
if src == 'google':
path = path + '.tar.bz'
stored = process_image(path, args.bands, False, args.pansharpen, args.ndvi, force_unzip,
for f in files:
stored = process_image(f, args.bands, False, args.pansharpen, args.ndvi, force_unzip,
args.ndvigrey, bounds=bounds)
if args.upload:
@@ -467,8 +465,8 @@ def process_image(path, bands=None, verbose=False, pansharpen=False, ndvi=False,
p = Simple(path, bands=bands, dst_path=settings.PROCESSED_IMAGE, verbose=verbose, force_unzip=force_unzip,
bounds=bounds)
except IOError:
exit("Zip file corrupted", 1)
except IOError as e:
exit(e.message, 1)
except FileDoesNotExist as e:
exit(e.message, 1)

2
landsat/search.py
View File

@@ -218,7 +218,7 @@ class Search(object):
if address:
query.append(self.address_builder(address))
elif lat and lon:
elif (lat is not None) and (lon is not None):
query.append(self.lat_lon_builder(lat, lon))
if query:

2
requirements/docker.txt
View File

@@ -8,4 +8,4 @@ boto>=2.38.0
polyline==1.1
geocoder>=1.5.1
jsonschema==2.5.1
git+git://github.com/developmentseed/usgs@develop
usgs==0.1.9

5
setup.py
View File

@@ -35,7 +35,7 @@ setup(
license='CCO',
platforms='Posix; MacOS X; Windows',
install_requires=[
'usgs2==0.2.0',
'usgs==0.1.9',
'requests==2.7.0',
'python-dateutil>=2.4.2',
'numpy>=1.9.3',
@@ -47,7 +47,8 @@ setup(
'homura>=0.1.2',
'boto>=2.38.0',
'polyline==1.1',
'geocoder>=1.5.1'
'geocoder>=1.5.1',
'matplotlib==1.5.1'
],
test_suite='nose.collector',
tests_require=test_requirements

108
tests/test_download.py
View File

@@ -19,8 +19,8 @@ class TestDownloader(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.d = Downloader()
cls.temp_folder = mkdtemp()
cls.d = Downloader(download_dir=cls.temp_folder)
cls.scene = 'LT81360082013127LGN01'
cls.scene_2 = 'LC82050312014229LGN00'
cls.scene_s3 = 'LC80010092015051LGN00'
@@ -47,63 +47,52 @@ class TestDownloader(unittest.TestCase):
# download one scene
self.d.download([self.scene])
self.assertEqual({self.scene: 'google'}, self.d.download([self.scene]))
paths = self.d.download([self.scene])
self.assertTrue(isinstance(paths, list))
self.assertEqual([self.temp_folder + '/' + self.scene + '.tar.bz'],
paths)
# download multiple scenes
self.assertEqual({self.scene: 'google', self.scene_2: 'google'}, self.d.download([self.scene, self.scene_2]))
paths = self.d.download([self.scene, self.scene_2])
test_paths = [self.temp_folder + '/' + self.scene + '.tar.bz',
self.temp_folder + '/' + self.scene_2 + '.tar.bz']
self.assertTrue(isinstance(paths, list))
self.assertEqual(test_paths, paths)
# Test if error is raised when passing scene as string instead of list
self.assertRaises(Exception, self.d.download, self.scene)
# Test when passing band list along with sceneID
self.assertEqual({self.scene_s3: 'aws', self.scene_s3_2: 'aws'},
self.d.download([self.scene_s3, self.scene_s3_2], bands=[11]))
paths = self.d.download([self.scene_s3, self.scene_s3_2], bands=[11])
test_paths = [self.temp_folder + '/' + self.scene_s3,
self.temp_folder + '/' + self.scene_s3_2]
self.assertEqual(test_paths, paths)
# Test whether passing band as string raises an exception
self.assertRaises(Exception, self.d.download, self.scene, 4)
# When passing scene as string, google storage download should be triggered
paths = self.d.download([self.scene], bands=4)
test_paths = [self.temp_folder + '/' + self.scene + '.tar.bz']
self.assertEqual(test_paths, paths)
@mock.patch('landsat.downloader.Downloader.amazon_s3')
@mock.patch('landsat.downloader.Downloader.google_storage')
def test_download_google_amazon(self, fake_google, fake_amazon):
def test_download_google_when_amazon_is_unavailable(self, fake_google):
""" Test whether google or amazon are correctly selected based on input """
fake_amazon.return_value = True
fake_google.return_value = False
# Test if google is used when an image from 2014 is passed even if bands are provided
self.d.download([self.scene], bands=[432])
fake_google.assert_called_with(self.scene, self.d.download_dir)
@mock.patch('landsat.downloader.fetch')
def test_download_amazon_when_available(self, mock_fetch):
""" Test whether google or amazon are correctly selected based on input """
mock_fetch.return_value = True
# Test if amazon is used when an image from 2015 is passed with bands
self.d.download([self.scene_s3], bands=[432])
fake_amazon.assert_called_with(self.scene_s3, 'MTL', self.d.download_dir + '/' + self.scene_s3)
@mock.patch('landsat.downloader.fetch')
def test_google_storage(self, mock_fetch):
mock_fetch.return_value = True
# If the file exist
self.assertTrue(self.d.google_storage(self.scene, self.temp_folder))
# If scene id is incorrect
self.assertRaises(IncorrectSceneId, self.d.google_storage, 'somerandomscene', self.temp_folder)
# If scene id doesn't exist
self.assertRaises(RemoteFileDoesntExist, self.d.google_storage, 'LG21360082013227LGN01',
self.temp_folder)
@mock.patch('landsat.downloader.fetch')
def test_amazon_s3(self, mock_fetch):
mock_fetch.return_value = True
scene = self.scene_s3
self.assertTrue(self.d.amazon_s3(scene, 11, self.temp_folder))
# If scene id is incorrect
self.assertRaises(IncorrectSceneId, self.d.amazon_s3, 'somerandomscene', 11, self.temp_folder)
# If scene id doesn't exist
self.assertRaises(RemoteFileDoesntExist, self.d.amazon_s3, 'LT81360082013127LGN01', 33, self.temp_folder)
paths = self.d.download([self.scene_s3], bands=[4, 3, 2])
test_paths = [self.temp_folder + '/' + self.scene_s3]
self.assertEqual(test_paths, paths)
@mock.patch('landsat.downloader.fetch')
def test_fetch(self, mock_fetch):
@@ -112,7 +101,7 @@ class TestDownloader(unittest.TestCase):
sat = self.d.scene_interpreter(self.scene)
url = self.d.google_storage_url(sat)
self.assertTrue(self.d.fetch(url, self.temp_folder, self.scene))
self.assertTrue(self.d.fetch(url, self.temp_folder))
def test_remote_file_size(self):
@@ -130,29 +119,38 @@ class TestDownloader(unittest.TestCase):
def test_amazon_s3_url(self):
sat = self.d.scene_interpreter(self.scene)
filename = '%s_B11.TIF' % self.scene
string = self.d.amazon_s3_url(sat, filename)
string = self.d.amazon_s3_url(sat, 11)
expect = os.path.join(S3_LANDSAT, 'L8/136/008/LT81360082013127LGN01/LT81360082013127LGN01_B11.TIF')
self.assertEqual(expect, string)
def test_remote_file_exist(self):
# Test a S3 url that exists
self.assertTrue(self.d.remote_file_exists(os.path.join(S3_LANDSAT, 'L8/003/017/LC80030172015001L'
'GN00/LC80030172015001LGN00_B6.TIF')))
# Exists and should return None
# Test a S3 url that doesn't exist
self.assertFalse(self.d.remote_file_exists(os.path.join(S3_LANDSAT, 'L8/003/017/LC80030172015001L'
'GN00/LC80030172015001LGN00_B34.TIF')))
self.assertIsNone(self.d.remote_file_exists(os.path.join(S3_LANDSAT, 'L8/003/017/LC80030172015001L'
'GN00/LC80030172015001LGN00_B6.TIF')))
# Test a Google Storage url that doesn't exist
self.assertFalse(self.d.remote_file_exists(os.path.join(GOOGLE_STORAGE, 'L8/003/017/LC80030172015001L'
'GN00/LC80030172015001LGN00_B6.TIF')))
# Doesn't exist and should raise errror
with self.assertRaises(RemoteFileDoesntExist):
self.d.remote_file_exists(
os.path.join(
S3_LANDSAT,
'L8/003/017/LC80030172015001LGN00/LC80030172015001LGN00_B34.TIF'
)
)
# Test a Google Storage url that exists
self.assertTrue(self.d.remote_file_exists(os.path.join(GOOGLE_STORAGE,
'L8/003/017/LC80030172015001LGN00.tar.bz')))
# Doesn't exist and should raise errror
with self.assertRaises(RemoteFileDoesntExist):
self.d.remote_file_exists(
os.path.join(
GOOGLE_STORAGE,
'L8/003/017/LC80030172015001LGN00/LC80030172015001LGN00_B6.TIF'
)
)
# Exist and shouldn't raise error
self.assertIsNone(self.d.remote_file_exists(os.path.join(GOOGLE_STORAGE,
'L8/003/017/LC80030172015001LGN00.tar.bz')))
def test_scene_interpreter(self):
# Test with correct input

52
tests/test_landsat.py
View File

@@ -132,78 +132,71 @@ class TestLandsat(unittest.TestCase):
['The SceneID provided was incorrect', 1])
@mock.patch('landsat.landsat.process_image')
@mock.patch('landsat.landsat.Downloader.download')
def test_download_process_continuous(self, mock_downloader, mock_process):
@mock.patch('landsat.downloader.fetch')
def test_download_process_continuous(self, fetch, mock_process):
"""Test download and process commands together"""
mock_downloader.return_value = {'LC80010092015051LGN00': 'aws',
'LC80010092014051LGN00': 'aws'}
fetch.return_value = True
mock_process.return_value = 'image.TIF'
args = ['download', 'LC80010092015051LGN00', 'LC80010092014051LGN00', '-b', '432', '-d', self.mock_path, '-p']
args = ['download', 'LC80010092015051LGN00', 'LC80470222014354LGN00', '-b', '432', '-d', self.mock_path, '-p']
output = landsat.main(self.parser.parse_args(args))
mock_downloader.assert_called_with(['LC80010092015051LGN00', 'LC80010092014051LGN00'], [4, 3, 2])
mock_process.assert_called_with('path/to/folder/LC80010092014051LGN00', '432',
mock_process.assert_called_with('path/to/folder/LC80470222014354LGN00', '432',
False, False, False, False, False, bounds=None)
self.assertEquals(output, ["The output is stored at image.TIF", 0])
# Call with force unzip flag
args = ['download', 'LC80010092015051LGN00', 'LC80010092014051LGN00', '-b', '432', '-d',
args = ['download', 'LC80010092015051LGN00', 'LC80470222014354LGN00', '-b', '432', '-d',
self.mock_path, '-p', '--force-unzip']
output = landsat.main(self.parser.parse_args(args))
mock_downloader.assert_called_with(['LC80010092015051LGN00', 'LC80010092014051LGN00'], [4, 3, 2])
mock_process.assert_called_with('path/to/folder/LC80010092014051LGN00', '432', False, False, False,
mock_process.assert_called_with('path/to/folder/LC80470222014354LGN00', '432', False, False, False,
True, False, bounds=None)
self.assertEquals(output, ["The output is stored at image.TIF", 0])
# Call with pansharpen
args = ['download', 'LC80010092015051LGN00', 'LC80010092014051LGN00', '-b', '432', '-d',
args = ['download', 'LC80010092015051LGN00', 'LC80470222014354LGN00', '-b', '432', '-d',
self.mock_path, '-p', '--pansharpen']
output = landsat.main(self.parser.parse_args(args))
mock_downloader.assert_called_with(['LC80010092015051LGN00', 'LC80010092014051LGN00'], [4, 3, 2, 8])
mock_process.assert_called_with('path/to/folder/LC80010092014051LGN00', '432', False, True, False,
mock_process.assert_called_with('path/to/folder/LC80470222014354LGN00', '432', False, True, False,
False, False, bounds=None)
self.assertEquals(output, ["The output is stored at image.TIF", 0])
# Call with pansharpen and clipping
args = ['download', 'LC80010092015051LGN00', 'LC80010092014051LGN00', '-b', '432', '-d',
args = ['download', 'LC80010092015051LGN00', 'LC80470222014354LGN00', '-b', '432', '-d',
self.mock_path, '-p', '--pansharpen', '--clip', '"-180,-180,0,0"']
output = landsat.main(self.parser.parse_args(args))
mock_downloader.assert_called_with(['LC80010092015051LGN00', 'LC80010092014051LGN00'], [4, 3, 2, 8])
mock_process.assert_called_with('path/to/folder/LC80010092014051LGN00', '432', False, True, False,
mock_process.assert_called_with('path/to/folder/LC80470222014354LGN00', '432', False, True, False,
False, False, bounds=[-180.0, -180.0, 0.0, 0.0])
self.assertEquals(output, ["The output is stored at image.TIF", 0])
# Call with ndvi
args = ['download', 'LC80010092015051LGN00', 'LC80010092014051LGN00', '-b', '432', '-d',
args = ['download', 'LC80010092015051LGN00', 'LC80470222014354LGN00', '-b', '432', '-d',
self.mock_path, '-p', '--ndvi']
output = landsat.main(self.parser.parse_args(args))
mock_downloader.assert_called_with(['LC80010092015051LGN00', 'LC80010092014051LGN00'], [4, 5])
mock_process.assert_called_with('path/to/folder/LC80010092014051LGN00', '432', False, False, True,
mock_process.assert_called_with('path/to/folder/LC80470222014354LGN00', '432', False, False, True,
False, False, bounds=None)
self.assertEquals(output, ["The output is stored at image.TIF", 0])
# Call with ndvigrey
args = ['download', 'LC80010092015051LGN00', 'LC80010092014051LGN00', '-b', '432', '-d',
args = ['download', 'LC80010092015051LGN00', 'LC80470222014354LGN00', '-b', '432', '-d',
self.mock_path, '-p', '--ndvigrey']
output = landsat.main(self.parser.parse_args(args))
mock_downloader.assert_called_with(['LC80010092015051LGN00', 'LC80010092014051LGN00'], [4, 5])
mock_process.assert_called_with('path/to/folder/LC80010092014051LGN00', '432', False, False, False,
mock_process.assert_called_with('path/to/folder/LC80470222014354LGN00', '432', False, False, False,
False, True, bounds=None)
self.assertEquals(output, ["The output is stored at image.TIF", 0])
@mock.patch('landsat.landsat.Uploader')
@mock.patch('landsat.landsat.process_image')
@mock.patch('landsat.landsat.Downloader.download')
def test_download_process_continuous_with_upload(self, mock_downloader, mock_process, mock_upload):
@mock.patch('landsat.downloader.fetch')
def test_download_process_continuous_with_upload(self, fetch, mock_process, mock_upload):
"""Test download and process commands together"""
mock_downloader.return_value = {'LC80010092015051LGN00': 'aws'}
fetch.return_value = True
mock_process.return_value = 'image.TIF'
mock_upload.run.return_value = True
args = ['download', 'LC80010092015051LGN00', '-b', '432', '-d', self.mock_path, '-p',
'-u', '--key', 'somekey', '--secret', 'somesecret', '--bucket', 'mybucket', '--region', 'this']
output = landsat.main(self.parser.parse_args(args))
mock_downloader.assert_called_with(['LC80010092015051LGN00'], [4, 3, 2])
# mock_downloader.assert_called_with(['LC80010092015051LGN00'], [4, 3, 2])
mock_process.assert_called_with('path/to/folder/LC80010092015051LGN00', '432', False, False, False,
False, False, bounds=None)
mock_upload.assert_called_with('somekey', 'somesecret', 'this')
@@ -211,16 +204,15 @@ class TestLandsat(unittest.TestCase):
self.assertEquals(output, ['The output is stored at image.TIF', 0])
@mock.patch('landsat.landsat.process_image')
@mock.patch('landsat.landsat.Downloader.download')
def test_download_process_continuous_with_wrong_args(self, mock_downloader, mock_process):
@mock.patch('landsat.downloader.fetch')
def test_download_process_continuous_with_wrong_args(self, fetch, mock_process):
"""Test download and process commands together"""
mock_downloader.return_value = {'LC80010092015051LGN00': 'aws'}
fetch.return_value = True
mock_process.return_value = 'image.TIF'
args = ['download', 'LC80010092015051LGN00', '-b', '432', '-d', self.mock_path, '-p',
'-u', '--region', 'whatever']
output = landsat.main(self.parser.parse_args(args))
mock_downloader.assert_called_with(['LC80010092015051LGN00'], [4, 3, 2])
mock_process.assert_called_with('path/to/folder/LC80010092015051LGN00', '432', False, False, False,
False, False, bounds=None)
self.assertEquals(output, ['Could not authenticate with AWS', 1])

12
tests/test_search.py
View File

@@ -34,6 +34,18 @@ class TestSearchHelper(unittest.TestCase):
result = self.s.search(lat=lat, lon=lon, start_date=start_date, end_date=end_date)
self.assertEqual('2015-02-06', result['results'][0]['date'])
def test_search_zero_lon(self):
# Make sure that zero coordinates are handled correctly
paths_rows = '003,003'
lon = 0.0
lat = 52.0
start_date = '2016-01-01'
end_date = '2016-01-10'
result = self.s.search(start_date=start_date, end_date=end_date,
lon=0.0, lat=52.0)
self.assertEqual('2016-01-06', result['results'][0]['date'])
def test_search_with_geojson(self):
# TEST A REGULAR SEARCH WITH KNOWN RESULT for paths and rows