Add test for ASTER command.

Also introduce the TestData class, which has a mechanism to download
external data so that datasets too big for the repository can be
hosted elsewhere and downloaded as needed during unit test running.

tests/aster/test_commands.py

@@ -0,0 +1,26 @@
+import os
+from tempfile import TemporaryDirectory
+
+import pystac
+
+from stactools.aster.commands import create_aster_command
+from tests.utils import (TestData, CliTestCase)
+
+EXTERNAL_DATA_PATH = 'aster/AST_L1T_00301012006003619_20150512141939_7778.hdf'
+
+
+class CreateItemTest(CliTestCase):
+    def create_subcommand_functions(self):
+        return [create_aster_command]
+
+    def test_create_item(self):
+        test_path = TestData.get_external_data(EXTERNAL_DATA_PATH)
+
+        with TemporaryDirectory() as tmp_dir:
+            cmd = ['aster', 'create-item', test_path, tmp_dir]
+            self.run_command(cmd)
+
+            item_path = os.path.join(tmp_dir, os.listdir(tmp_dir)[0])
+            item = pystac.read_file(item_path)
+
+        self.assertEqual(set(item.assets.keys()), set(['hdf']))

tests/data-files/external/.gitignore

@@ -0,0 +1 @@
+*

tests/planet/test_commands.py

@@ -4,15 +4,15 @@
 import pystac
 
 from stactools.planet.commands import create_planet_command
-from tests.utils import (TestCases, CliTestCase)
+from tests.utils import (TestData, CliTestCase)
 
 
 class ConvertOrderTest(CliTestCase):
     def create_subcommand_functions(self):
         return [create_planet_command]
 
     def test_converts(self):
-        test_order_manifest = TestCases.get_path(
+        test_order_manifest = TestData.get_path(
             'data-files/planet-order/manifest.json')
 
         with TemporaryDirectory() as tmp_dir:

tests/utils/__init__.py

@@ -1,4 +1,5 @@
 # flake8: noqa
 
 from tests.utils.test_cases import TestCases
+from tests.utils.test_data import TestData
 from tests.utils.cli_test import CliTestCase

tests/utils/test_cases.py

@@ -6,6 +6,8 @@
                     SpatialExtent, MediaType, Extensions)
 from pystac.extensions.label import (LabelOverview, LabelClasses, LabelCount)
 
+from tests.utils.test_data import TestData
+
 TEST_LABEL_CATALOG = {
     'country-1': {
         'area-1-1': {
@@ -75,17 +77,17 @@ def all_test_catalogs():
     @staticmethod
     def planet_disaster():
         return pystac.read_file(
-            TestCases.get_path('data-files/planet-disaster/collection.json'))
+            TestData.get_path('data-files/planet-disaster/collection.json'))
 
     @staticmethod
     def test_case_1():
         return Catalog.from_file(
-            TestCases.get_path('data-files/catalogs/test-case-1/catalog.json'))
+            TestData.get_path('data-files/catalogs/test-case-1/catalog.json'))
 
     @staticmethod
     def test_case_2():
         return Catalog.from_file(
-            TestCases.get_path('data-files/catalogs/test-case-2/catalog.json'))
+            TestData.get_path('data-files/catalogs/test-case-2/catalog.json'))
 
     @staticmethod
     def test_case_3():
@@ -143,31 +145,31 @@ def test_case_4():
         See: https://www.drivendata.org/competitions/60/building-segmentation-disaster-resilience
         """
         return Catalog.from_file(
-            TestCases.get_path('data-files/catalogs/test-case-4/catalog.json'))
+            TestData.get_path('data-files/catalogs/test-case-4/catalog.json'))
 
     @staticmethod
     def test_case_5():
         """Based on a subset of https://cbers.stac.cloud/"""
         return Catalog.from_file(
-            TestCases.get_path('data-files/catalogs/test-case-5/catalog.json'))
+            TestData.get_path('data-files/catalogs/test-case-5/catalog.json'))
 
     @staticmethod
     def test_case_6():
         """Based on a subset of CBERS, contains a root and 4 empty children"""
         return Catalog.from_file(
-            TestCases.get_path(
+            TestData.get_path(
                 'data-files/catalogs/cbers-partial/catalog.json'))
 
     @staticmethod
     def test_case_7():
         """Test case 4 as STAC version 0.8.1"""
         return Catalog.from_file(
-            TestCases.get_path(
+            TestData.get_path(
                 'data-files/catalogs/label_catalog_0_8_1/catalog.json'))
 
     @staticmethod
     def test_case_8():
         """Planet disaster data example catalog, 1.0.0-beta.2"""
         return pystac.read_file(
-            TestCases.get_path('data-files/catalogs/'
-                               'planet-example-1.0.0-beta.2/collection.json'))
+            TestData.get_path('data-files/catalogs/'
+                              'planet-example-1.0.0-beta.2/collection.json'))

tests/utils/test_data.py

@@ -0,0 +1,52 @@
+import os
+import shutil
+from tempfile import TemporaryDirectory
+from zipfile import ZipFile
+
+import requests
+
+EXTERNAL_DATA = {
+    'aster/AST_L1T_00301012006003619_20150512141939_7778.hdf': {
+        'url':
+        ('https://ai4epublictestdata.blob.core.windows.net/'
+         'stactools/aster/AST_L1T_00301012006003619_20150512141939_7778.zip'),
+        'compress':
+        'zip'
+    }
+}
+
+
+class TestData:
+    @staticmethod
+    def get_path(rel_path):
+        return os.path.abspath(
+            os.path.join(os.path.dirname(__file__), '..', rel_path))
+
+    @staticmethod
+    def get_external_data(rel_path):
+        path = TestData.get_path(os.path.join('data-files/external', rel_path))
+        if not os.path.exists(path):
+            entry = EXTERNAL_DATA.get(rel_path)
+            if entry is None:
+                raise Exception('Path {} does not exist and there is no entry '
+                                'for external test data {}.'.format(
+                                    path, rel_path))
+
+            print('Downloading external test data {}...'.format(rel_path))
+            os.makedirs(os.path.dirname(path), exist_ok=True)
+
+            resp = requests.get(entry['url'])
+            if entry['compress'] == 'zip':
+                with TemporaryDirectory() as tmp_dir:
+                    tmp_path = os.path.join(tmp_dir, 'file.zip')
+                    with open(tmp_path, 'wb') as f:
+                        f.write(resp.content)
+                    z = ZipFile(tmp_path)
+                    name = z.namelist()[0]
+                    extracted_path = z.extract(name)
+                    shutil.move(extracted_path, path)
+            else:
+                with open(path, 'wb') as f:
+                    f.write(resp.content)
+
+        return path