Feat io tests

.coveragerc

@@ -5,8 +5,13 @@ omit =
 	*dev*
 	*docs*
 	*tutorials*
-	gpm_api/tests/*
+	gpm_api/bucket/*
 	gpm_api/cli/*
+	gpm_api/encoding/*
+	gpm_api/etc/*
+	gpm_api/retrieval/*
+	gpm_api/tests/*
+	gpm_api/_version.py
 
 [report]
 exclude_lines =

gpm_api/io/ges_disc.py

@@ -41,7 +41,7 @@ def _get_href_value(input_string):
     return href_value
 
 
-def _get_gesc_disc_list_path(url):
+def _get_ges_disc_list_path(url):
     # Retrieve url content
     # - If it returns something, means url is correct
     wget_output = _get_ges_disc_url_content(url)
@@ -179,7 +179,7 @@ def _get_gesdisc_file_list(url_product_dir, product, date, version, verbose=True
         Default is False. Whether to specify when data are not available for a specific date.
     """
     try:
-        filepaths = _get_gesc_disc_list_path(url_product_dir)
+        filepaths = _get_ges_disc_list_path(url_product_dir)
     except Exception as e:
         # If url not exist, raise an error
         if "was not found on the GES DISC server" in str(e):

gpm_api/tests/test_io/test_checks.py

@@ -1,4 +1,4 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 """
 Created on Mon Jul 17 15:41:14 2023
@@ -240,29 +240,30 @@
 
 
 def test_check_product_version(
+    check,  # For non-failing asserts
     product_info: Dict[str, Any],
     versions: List[int],
 ) -> None:
     """Test check_product_version()"""
 
-    # Check valid versions
     for product, info in product_info.items():
+        # Check valid versions
         valid_versions = info.get("available_versions", [])
 
         for version in valid_versions:
-            assert checks.check_product_version(version, product) == version
+            with check:
+                assert checks.check_product_version(version, product) == version
 
-    # Check last version return if None
-    for product, info in product_info.items():
+        # Check last version return if None
         last_version = info.get("available_versions", [])[-1]
-        assert checks.check_product_version(None, product) == last_version
+        with check:
+            assert checks.check_product_version(None, product) == last_version
 
-    # Check invalid versions
-    for product, info in product_info.items():
+        # Check invalid versions
         invalid_versions = list(set(versions) - set(info.get("available_versions", [])))
 
         for version in invalid_versions:
-            with pytest.raises(ValueError):
+            with check.raises(ValueError):
                 checks.check_product_version(version, product)
 
 
@@ -579,6 +580,7 @@
 
 
 def test_check_valid_time_request(
+    check,  # For non-failing asserts
     product_info: Dict[str, Any],
 ) -> None:
     """Test check_valid_time_request()"""
@@ -596,14 +598,14 @@
             # Check invalid start time
             start_time = valid_start_time - datetime.timedelta(days=1)
             end_time = valid_start_time + datetime.timedelta(days=1)
-            with pytest.raises(ValueError):
+            with check.raises(ValueError):
                 checks.check_valid_time_request(start_time, end_time, product)
 
         # Check invalid end time
         if valid_end_time is not None:
             start_time = valid_end_time - datetime.timedelta(days=1)
             end_time = valid_end_time + datetime.timedelta(days=1)
-            with pytest.raises(ValueError):
+            with check.raises(ValueError):
                 checks.check_valid_time_request(start_time, end_time, product)
 
 

gpm_api/tests/test_io/test_download.py

@@ -163,64 +163,65 @@
 
 
 def test_download_data(
+    check,  # For non-failing asserts
     products: List[str],
     product_types: List[str],
     remote_filepaths: Dict[str, Dict[str, Any]],
     mocker: MockerFixture,
     versions: List[str],
 ):
     """Test download_data function
 
     This test is somewhat redundant considering it is testing methods
     bundled in another functions which need to be turned off in order to
     test this function. However, it is useful to have a test that checks
     the entire download process.
 
     It may be useful as boilerplate to increase the number of tests here in the
     future.
     """
 
     mocker.patch.object(dl, "_download_files", autospec=True, return_value=[])
     mocker.patch.object(dl, "_download_daily_data", autospec=True, return_value=([], versions))
     mocker.patch.object(dl, "run", autospec=True, return_value=None)
     from gpm_api.io import info, pps
 
     mocker.patch.object(
         info,
         "_get_info_from_filename",
         autospec=True,
         return_value={
             "product": "2A-CLIM",
             "product_type": "CLIM",
             "start_time": datetime.datetime(2022, 9, 7, 12, 0, 0),
             "end_time": datetime.datetime(2022, 9, 7, 13, 0, 0),
             "version": "V07A",
             "satellite": "GPM",
             "granule_id": "2A-CLIM.GPM.GMI.GPROF2021v1.20150301-S121433-E134706.005708.V07A.HDF5",
         },
     )
     mocker.patch.object(
         find,
         "find_daily_filepaths",
         autospec=True,
         return_value=(remote_filepaths.keys(), versions),
     )
 
     # Assume files pass file integrity check by mocking return as empty
-    for product in products:
-        for product_type in product_types:
-            if product in available_products(product_type=product_type):
-                start_time = get_product_start_time(product)
-                if start_time is None:
-                    continue
-                res = dl.download_archive(
-                    product=product,
-                    start_time=start_time,
-                    end_time=start_time + datetime.timedelta(hours=1),
-                    product_type=product_type,
-                )
-
-        assert res is None  # Assume data is downloaded
+    for product_type in product_types:
+        for product in available_products(product_type=product_type):
+            start_time = get_product_start_time(product)
+            if start_time is None:
+                continue
+            res = dl.download_archive(
+                product=product,
+                start_time=start_time,
+                end_time=start_time + datetime.timedelta(hours=1),
+                product_type=product_type,
+            )
+
+        with check:
+            assert res is None  # Assume data is downloaded
 
 
 def test_download_daily_data_private(
@@ -318,7 +319,7 @@
     """Test convert_pps_to_local_filepaths function
 
     Parameters
     """
     # TODO: WRONG REDO !
     assert dl.get_fpaths_from_fnames(
         filepaths=[

gpm_api/tests/test_io/test_filter.py

@@ -50,107 +50,126 @@
         )
 
 
-def test_filter_filepaths(
-    remote_filepaths: Dict[str, Dict[str, Any]],
-    products: Dict[str, Dict[str, Any]],
-    mocker: MockerFixture,
-) -> None:
+class TestFilterFilepaths:
     """Test filter filepaths"""
 
     product = "2A-DPR"
 
-    # Test year filtering
-    # Count and assert 2019 paths
-    count_2019 = 0
-    for remote_filepath, info_dict in remote_filepaths.items():
-        if (
-            info_dict["year"] == 2019
-            and info_dict["product"] == product
-            and info_dict["version"] == 7
-        ):
-            count_2019 += 1
-
-    res = filter.filter_filepaths(
-        filepaths=list(remote_filepaths.keys()),
-        product=product,
-        start_time=datetime.datetime(2019, 1, 1),
-        end_time=datetime.datetime(2019, 12, 31, 23, 59, 59),
-        version=7,
-    )
-
-    assert len(res) == count_2019
+    def test_year_filtering(
+        self,
+        remote_filepaths: Dict[str, Dict[str, Any]],
+    ) -> None:
+        # Count and assert 2019 paths
+        count_2019 = 0
+        for remote_filepath, info_dict in remote_filepaths.items():
+            if (
+                info_dict["year"] == 2019
+                and info_dict["product"] == self.product
+                and info_dict["version"] == 7
+            ):
+                count_2019 += 1
+
+        res = filter.filter_filepaths(
+            filepaths=list(remote_filepaths.keys()),
+            product=self.product,
+            start_time=datetime.datetime(2019, 1, 1),
+            end_time=datetime.datetime(2019, 12, 31, 23, 59, 59),
+            version=7,
+        )
 
-    # Test None filepaths
-    res = filter.filter_filepaths(
-        filepaths=None,
-        product=product,
-        start_time=datetime.datetime(2019, 1, 1),
-        end_time=datetime.datetime(2019, 12, 31, 23, 59, 59),
-        version=7,
-    )
-    assert res == []
+        assert len(res) == count_2019
 
-    # Test empty filepath list
-    res = filter.filter_filepaths(
-        filepaths=[],
-        product=product,
-        start_time=datetime.datetime(2019, 1, 1),
-        end_time=datetime.datetime(2019, 12, 31, 23, 59, 59),
-        version=7,
-    )
-    assert res == []
+    def test_none_filepath(
+        self,
+        remote_filepaths: Dict[str, Dict[str, Any]],
+    ) -> None:
+        res = filter.filter_filepaths(
+            filepaths=None,
+            product=self.product,
+            start_time=datetime.datetime(2019, 1, 1),
+            end_time=datetime.datetime(2019, 12, 31, 23, 59, 59),
+            version=7,
+        )
+        assert res == []
 
-    # Test empty start time
-    count_until_2019 = 0
-    for remote_filepath, info_dict in remote_filepaths.items():
-        if info_dict["year"] == 2019 and info_dict["product"] == product:
-            count_until_2019 += 1
-    res = filter.filter_filepaths(
-        filepaths=list(remote_filepaths.keys()),
-        product=product,
-        start_time=None,
-        end_time=datetime.datetime(2019, 12, 31, 23, 59, 59),
-        version=7,
-    )
+    def test_empty_filepath_list(
+        self,
+        remote_filepaths: Dict[str, Dict[str, Any]],
+    ) -> None:
+        res = filter.filter_filepaths(
+            filepaths=[],
+            product=self.product,
+            start_time=datetime.datetime(2019, 1, 1),
+            end_time=datetime.datetime(2019, 12, 31, 23, 59, 59),
+            version=7,
+        )
+        assert res == []
 
-    assert len(res) == count_until_2019
+    def test_empty_start_time(
+        self,
+        remote_filepaths: Dict[str, Dict[str, Any]],
+    ) -> None:
+        count_until_2019 = 0
+        for remote_filepath, info_dict in remote_filepaths.items():
+            if info_dict["year"] == 2019 and info_dict["product"] == self.product:
+                count_until_2019 += 1
+        res = filter.filter_filepaths(
+            filepaths=list(remote_filepaths.keys()),
+            product=self.product,
+            start_time=None,
+            end_time=datetime.datetime(2019, 12, 31, 23, 59, 59),
+            version=7,
+        )
 
-    # Test empty end time (Error as time given (datetime.datetime.now())
-    # requires date to be less than now() in supportive
-    # function checks.check_start_end_time)
-    count_from_2019 = 0
-    for remote_filepath, info_dict in remote_filepaths.items():
-        if info_dict["year"] >= 2019 and info_dict["product"] == product:
-            count_from_2019 += 1
+        assert len(res) == count_until_2019
 
-    res = filter.filter_filepaths(
-        filepaths=list(remote_filepaths.keys()),
-        product=product,
-        start_time=datetime.datetime(2019, 1, 1),
-        end_time=None,
-        version=7,
-    )
-    assert len(res) == count_from_2019
+    def test_empty_end_time(
+        self,
+        remote_filepaths: Dict[str, Dict[str, Any]],
+    ) -> None:
+        """Test empty end time (Error as time given (datetime.datetime.now())
+        requires date to be less than now() in supportive
+        function checks.check_start_end_time)"""
 
-    # Test unmatched version
-    res = filter.filter_filepaths(
-        filepaths=list(remote_filepaths.keys()),
-        product=product,
-        start_time=datetime.datetime(2019, 1, 1),
-        end_time=datetime.datetime(2019, 12, 31, 23, 59, 59),
-        version=0,
-    )
-    assert res == []
+        count_from_2019 = 0
+        for remote_filepath, info_dict in remote_filepaths.items():
+            if info_dict["year"] >= 2019 and info_dict["product"] == self.product:
+                count_from_2019 += 1
+
+        res = filter.filter_filepaths(
+            filepaths=list(remote_filepaths.keys()),
+            product=self.product,
+            start_time=datetime.datetime(2019, 1, 1),
+            end_time=None,
+            version=7,
+        )
+        assert len(res) == count_from_2019
+
+    def test_unmatched_version(
+        self,
+        remote_filepaths: Dict[str, Dict[str, Any]],
+    ) -> None:
+        res = filter.filter_filepaths(
+            filepaths=list(remote_filepaths.keys()),
+            product=self.product,
+            start_time=datetime.datetime(2019, 1, 1),
+            end_time=datetime.datetime(2019, 12, 31, 23, 59, 59),
+            version=0,
+        )
+        assert res == []
 
-    # Test unmatched product
-    res = filter.filter_filepaths(
-        filepaths=list(remote_filepaths.keys()),
-        product="1A-GMI",
-        start_time=datetime.datetime(2019, 1, 1),
-        end_time=datetime.datetime(2019, 12, 31, 23, 59, 59),
-        version=7,
-    )
-    assert res == []
+    def test_unmatched_product(
+        self,
+        remote_filepaths: Dict[str, Dict[str, Any]],
+    ) -> None:
+        res = filter.filter_filepaths(
+            filepaths=list(remote_filepaths.keys()),
+            product="1A-GMI",
+            start_time=datetime.datetime(2019, 1, 1),
+            end_time=datetime.datetime(2019, 12, 31, 23, 59, 59),
+            version=7,
+        )
+        assert res == []
 
 
 def test_filter_by_time(
@@ -217,19 +236,19 @@
         end_time=None,
     )
 
     # Test granule starting on previous day
     count_previous_day = 0
     for remote_filepath, info_dict in remote_filepaths.items():
         if info_dict["start_time"].day != info_dict["end_time"].day:
             count_previous_day += 1
 
     res = filter.filter_by_time(
         filepaths=list(remote_filepaths.keys()),
         start_time=datetime.datetime(2020, 7, 6, 0, 0, 20),
         end_time=datetime.datetime(2020, 7, 6, 0, 0, 30),
     )
 
     assert len(res) == count_previous_day
 
 
 def test_filter_by_product(