Enable PT (flake8-pytest-style) ruleset and fix violations.

pyproject.toml

@@ -211,13 +211,9 @@ ignore = [
   "D417",
   "E501",
   "FIX002",
-  "PT006",
-  "PT007",
-  "PT009",
-  "PT011",
-  "PT012",
-  "PT017",
-  "PT027",
+  "PGH003",
+  "S101",
+  "S108",
   "S113",
   "TD002",
   "TD003",

tests/integration/test_api.py

@@ -67,7 +67,7 @@
 
 
 @pytest.mark.parametrize(
-    "system,nasa_response,expected",
+    ("system", "nasa_response", "expected"),
     [
         (
             PROD,
@@ -261,7 +261,7 @@ def test_download_deferred_failure(tmp_path: Path):
         # With "deferred" exceptions, pqdm catches all exceptions, then at the end
         # raises a single generic Exception, passing the sequence of caught exceptions
         # as arguments to the Exception constructor.
-        pytest.raises(Exception) as exc_info,
+        pytest.raises(Exception) as exc_info,  # noqa: PT011
     ):
         earthaccess.download(
             results,

tests/integration/test_auth.py

@@ -60,7 +60,7 @@ def test_auth_can_fetch_s3_credentials(daac):
         assert "accessKeyId" in credentials
 
 
-@pytest.mark.parametrize("location", ({"daac": "podaac"}, {"provider": "pocloud"}))
+@pytest.mark.parametrize("location", [({"daac": "podaac"}, {"provider": "pocloud"})])
 def test_get_s3_credentials_lowercase_location(location):
     earthaccess.login(strategy="environment")
     creds = earthaccess.get_s3_credentials(**location)
@@ -72,7 +72,7 @@ def test_get_s3_credentials_lowercase_location(location):
     )
 
 
-@pytest.mark.parametrize("location", ({"daac": "podaac"}, {"provider": "pocloud"}))
+@pytest.mark.parametrize("location", [({"daac": "podaac"}, {"provider": "pocloud"})])
 def test_get_s3_filesystem_lowercase_location(location):
     earthaccess.login(strategy="environment")
     fs = earthaccess.get_s3_filesystem(**location)

tests/integration/test_services.py

@@ -36,16 +36,16 @@ def test_services(self):
             dataset["umm"]["ShortName"]: dataset.services() for dataset in datasets
         }
 
-        self.assertEqual(next(iter(dataset_services.keys())), "MUR-JPL-L4-GLOB-v4.1")
-        self.assertEqual(
+        assert next(iter(dataset_services.keys())) == "MUR-JPL-L4-GLOB-v4.1"
+        assert (
             dataset_services["MUR-JPL-L4-GLOB-v4.1"]["S2606110201-XYZ_PROV"][0]["umm"][
                 "LongName"
-            ],
-            "Harmony GDAL Adapter (HGA)",
+            ]
+            == "Harmony GDAL Adapter (HGA)"
         )
-        self.assertEqual(
+        assert (
             dataset_services["MUR-JPL-L4-GLOB-v4.1"]["S2839491596-XYZ_PROV"][0]["umm"][
                 "URL"
-            ]["Description"],
-            "https://harmony.earthdata.nasa.gov",
+            ]["Description"]
+            == "https://harmony.earthdata.nasa.gov"
         )

tests/unit/test_auth.py

@@ -36,16 +36,16 @@ def test_auth_gets_proper_credentials(self, user_input, user_password):
 
         # Test
         auth = Auth()
-        self.assertEqual(auth.authenticated, False)
+        assert auth.authenticated is False
         auth.login(strategy="interactive")
-        self.assertEqual(auth.authenticated, True)
-        self.assertEqual(auth.token, json_response)
+        assert auth.authenticated is True
+        assert auth.token == json_response
 
         # test that we are creating a session with the proper headers
         session = auth.get_session()
         headers = session.headers
-        self.assertTrue("User-Agent" in headers)
-        self.assertTrue("earthaccess" in headers["User-Agent"])
+        assert "User-Agent" in headers
+        assert "earthaccess" in headers["User-Agent"]
 
     @responses.activate
     @mock.patch("getpass.getpass")
@@ -71,9 +71,9 @@ def test_auth_can_create_proper_credentials(self, user_input, user_password):
         # Test
         auth = Auth()
         auth.login(strategy="interactive")
-        self.assertEqual(auth.authenticated, True)
-        self.assertEqual(auth.password, "password")
-        self.assertEqual(auth.token, json_response)
+        assert auth.authenticated is True
+        assert auth.password == "password"
+        assert auth.token == json_response
 
     @responses.activate
     @mock.patch.dict(os.environ, {"EARTHDATA_TOKEN": "ABCDEFGHIJKLMNOPQ"})
@@ -83,8 +83,8 @@ def test_auth_can_parse_existing_user_token(self):
         # Test
         auth = Auth()
         auth.login(strategy="environment")
-        self.assertEqual(auth.authenticated, True)
-        self.assertEqual(auth.token, json_response)
+        assert auth.authenticated is True
+        assert auth.token == json_response
 
     @responses.activate
     @mock.patch("getpass.getpass")
@@ -111,4 +111,4 @@ def test_auth_fails_for_wrong_credentials(self, user_input, user_password):
         with pytest.raises(LoginAttemptFailure):
             auth.login(strategy="interactive")
 
-        self.assertEqual(auth.authenticated, False)
+        assert auth.authenticated is False

tests/unit/test_collection_queries.py

@@ -68,15 +68,15 @@ def test_querybuilder_can_handle_has_granules():
     assert "has_granules" not in query.params
 
 
-@pytest.mark.parametrize("start,end,expected", valid_single_dates)
+@pytest.mark.parametrize(("start", "end", "expected"), valid_single_dates)
 def test_query_can_parse_single_dates(start, end, expected):
     query = DataCollections().temporal(start, end)
     assert query.params["temporal"][0] == expected
 
 
-@pytest.mark.parametrize("start,end,expected", invalid_single_dates)
+@pytest.mark.parametrize(("start", "end", "expected"), invalid_single_dates)
 def test_query_can_handle_invalid_dates(start, end, expected):  # noqa: ARG001
     query = DataCollections()
     assert "temporal" not in query.params
-    with pytest.raises(ValueError):
+    with pytest.raises(ValueError):  # noqa: PT011
         query.temporal(start, end)

tests/unit/test_geo_interface.py

@@ -277,5 +277,5 @@ def test_geo_interface(test_case: dict[str, object]):
 def test_missing_horizontal_spatial_domain_raises():
     granule = DataGranule({"umm": {"SpatialExtent": {"Orbit": {}}}})
 
-    with pytest.raises(ValueError):
+    with pytest.raises(ValueError):  # noqa: PT011
         _ = granule.__geo_interface__

tests/unit/test_granule_queries.py

@@ -39,21 +39,21 @@
 ]
 
 
-@pytest.mark.parametrize("start,end,expected", valid_single_dates)
+@pytest.mark.parametrize(("start", "end", "expected"), valid_single_dates)
 def test_query_can_parse_single_dates(start, end, expected):
     granules = DataGranules().short_name("MODIS").temporal(start, end)
     assert granules.params["temporal"][0] == expected
 
 
-@pytest.mark.parametrize("start,end,expected", invalid_single_dates)
+@pytest.mark.parametrize(("start", "end", "expected"), invalid_single_dates)
 def test_query_can_handle_invalid_dates(start, end, expected):  # noqa: ARG001
     granules = DataGranules().short_name("MODIS")
     assert "temporal" not in granules.params
-    with pytest.raises(ValueError):
+    with pytest.raises(ValueError):  # noqa: PT011
         granules.temporal(start, end)
 
 
-@pytest.mark.parametrize("bbox,expected", bbox_queries)
+@pytest.mark.parametrize(("bbox", "expected"), bbox_queries)
 def test_query_handles_bbox(bbox, expected):
     granules = DataGranules().short_name("MODIS").bounding_box(*bbox)
     assert ("bounding_box" in granules.params) == expected

tests/unit/test_results.py

@@ -139,9 +139,9 @@ def test_get_more_than_2000(self):
         granules = earthaccess.search_data(short_name="MOD02QKM", count=3000)
 
         # Assert that we performed one 'hits' search and two 'results' search queries
-        self.assertEqual(len(self.cassette), 3)
-        self.assertEqual(len(granules), 4000)
-        self.assertTrue(unique_results(granules))
+        assert len(self.cassette) == 3
+        assert len(granules) == 4000
+        assert unique_results(granules)
 
     def test_get(self):
         """If we execute a get with no arguments then we expect
@@ -151,9 +151,9 @@ def test_get(self):
         granules = earthaccess.search_data(short_name="MOD02QKM", count=2000)
 
         # Assert that we performed one 'hits' search and one 'results' search queries
-        self.assertEqual(len(self.cassette), 2)
-        self.assertEqual(len(granules), 2000)
-        self.assertTrue(unique_results(granules))
+        assert len(self.cassette) == 2
+        assert len(granules) == 2000
+        assert unique_results(granules)
 
     def test_get_all_less_than_2k(self):
         """If we execute a get_all then we expect multiple
@@ -166,9 +166,9 @@ def test_get_all_less_than_2k(self):
         )
 
         # Assert that we performed a hits query and one search results query
-        self.assertEqual(len(self.cassette), 2)
-        self.assertEqual(len(granules), 163)
-        self.assertTrue(unique_results(granules))
+        assert len(self.cassette) == 2
+        assert len(granules) == 163
+        assert unique_results(granules)
 
     def test_get_all_more_than_2k(self):
         """If we execute a get_all then we expect multiple
@@ -181,16 +181,14 @@ def test_get_all_more_than_2k(self):
         )
 
         # Assert that we performed a hits query and two search results queries
-        self.assertEqual(len(self.cassette), 3)
-        self.assertEqual(
-            len(granules),
-            int(self.cassette.responses[0]["headers"]["CMR-Hits"][0]),
+        assert len(self.cassette) == 3
+        assert len(granules) == int(
+            self.cassette.responses[0]["headers"]["CMR-Hits"][0]
         )
-        self.assertEqual(
-            len(granules),
-            min(3000, int(self.cassette.responses[0]["headers"]["CMR-Hits"][0])),
+        assert len(granules) == min(
+            3000, int(self.cassette.responses[0]["headers"]["CMR-Hits"][0])
         )
-        self.assertTrue(unique_results(granules))
+        assert unique_results(granules)
 
     def test_collections_less_than_2k(self):
         """If we execute a get_all then we expect multiple
@@ -201,9 +199,9 @@ def test_collections_less_than_2k(self):
         collections = query.get(20)
 
         # Assert that we performed a single search results query
-        self.assertEqual(len(self.cassette), 1)
-        self.assertEqual(len(collections), 20)
-        self.assertTrue(unique_results(collections))
+        assert len(self.cassette) == 1
+        assert len(collections) == 20
+        assert unique_results(collections)
         self.assert_is_using_search_after(self.cassette)
 
     def test_collections_more_than_2k(self):
@@ -215,19 +213,19 @@ def test_collections_more_than_2k(self):
         collections = query.get(3000)
 
         # Assert that we performed two search results queries
-        self.assertEqual(len(self.cassette), 2)
-        self.assertEqual(len(collections), 4000)
-        self.assertTrue(unique_results(collections))
+        assert len(self.cassette) == 2
+        assert len(collections) == 4000
+        assert unique_results(collections)
         self.assert_is_using_search_after(self.cassette)
 
     def assert_is_using_search_after(self, cass):
         first_request = True
 
         for request in cass.requests:
             # Verify the page number was not used
-            self.assertTrue("page_num" not in request.uri)
+            assert "page_num" not in request.uri
             # Verify that Search After was used in all requests except first
-            self.assertEqual(first_request, "CMR-Search-After" not in request.headers)
+            assert first_request == ("CMR-Search-After" not in request.headers)
             first_request = False
 
 

tests/unit/test_services.py

@@ -32,14 +32,12 @@ def test_services(self):
         earthaccess._auth.authenticated = False
         actual = earthaccess.search_services(concept_id="S2004184019-POCLOUD")
 
-        self.assertTrue(actual[0]["umm"]["Type"] == "OPeNDAP")
-        self.assertTrue(
-            actual[0]["umm"]["ServiceOrganizations"][0]["ShortName"] == "UCAR/UNIDATA",
+        assert actual[0]["umm"]["Type"] == "OPeNDAP"
+        assert (
+            actual[0]["umm"]["ServiceOrganizations"][0]["ShortName"] == "UCAR/UNIDATA"
         )
-        self.assertTrue(
-            actual[0]["umm"]["Description"] == "Earthdata OPEnDAP in the cloud",
-        )
-        self.assertTrue(actual[0]["umm"]["LongName"] == "PO.DAAC OPeNDADP In the Cloud")
+        assert actual[0]["umm"]["Description"] == "Earthdata OPEnDAP in the cloud"
+        assert actual[0]["umm"]["LongName"] == "PO.DAAC OPeNDADP In the Cloud"
 
     def test_service_results(self):
         """Test results.DataCollection.services to return available services."""
@@ -53,18 +51,16 @@ def test_service_results(self):
         assert len(datasets) > 0
         results = datasets[0].services()
 
-        self.assertTrue(
-            results["S2004184019-POCLOUD"][0]["meta"]["provider-id"] == "POCLOUD",
-        )
-        self.assertTrue(
+        assert results["S2004184019-POCLOUD"][0]["meta"]["provider-id"] == "POCLOUD"
+        assert (
             results["S2004184019-POCLOUD"][0]["umm"]["URL"]["URLValue"]
-            == "https://opendap.earthdata.nasa.gov/",
+            == "https://opendap.earthdata.nasa.gov/"
         )
-        self.assertTrue(
+        assert (
             results["S2606110201-XYZ_PROV"][0]["umm"]["Name"]
-            == "Harmony GDAL Adapter (HGA)",
+            == "Harmony GDAL Adapter (HGA)"
         )
-        self.assertTrue(results["S2164732315-XYZ_PROV"][0]["umm"]["Type"] == "Harmony")
+        assert results["S2164732315-XYZ_PROV"][0]["umm"]["Type"] == "Harmony"
 
 
 if __name__ == "__main__":