Merge pull request #74 from release-engineering/mypy

mypy: Few simple typing fixes

Author: JAVGan

docs/ext/attr_index.py

@@ -63,6 +63,6 @@ def add_attr_index(
     lines.extend(["", ""])
 
 
-def setup(app: Sphinx):
+def setup(app: Sphinx) -> None:
     # entrypoint invoked by sphinx when extension is loaded
     app.connect("autodoc-process-docstring", add_attr_index)

docs/ext/attr_types.py

@@ -53,6 +53,6 @@ def add_attr_types(
     lines.extend(["", f":type: {type.__name__}"])
 
 
-def setup(app: Sphinx):
+def setup(app: Sphinx) -> None:
     # entrypoint invoked by sphinx when extension is loaded
     app.connect("autodoc-process-docstring", add_attr_types)

starmap_client/models.py

@@ -103,7 +103,7 @@ def from_json(cls, json: Any):
         json = cls._preprocess_json(json)
 
         args = {}
-        cls_attr = [a.name for a in cls.__attrs_attrs__ if isinstance(a, Attribute)]  # type: ignore
+        cls_attr = [a.name for a in cls.__attrs_attrs__ if isinstance(a, Attribute)]
         for a in cls_attr:
             args[a] = json.pop(a, None)
         return cls(**args)

starmap_client/session.py

@@ -124,7 +124,7 @@ def _register_starmap_endpoints(self) -> None:
         for m in methods:
             self.register_uri(m, re.compile(f"{base_url}/.*"))  # type: ignore [arg-type]
 
-    def register_uri(self, method: str, uri: str):
+    def register_uri(self, method: str, uri: str) -> None:
         """Register an URI into the ``requests_mock`` adapter.
 
         Args:

tests/test_client.py

@@ -22,7 +22,7 @@ def load_json(json_file: str) -> Any:
 
 class TestStarmapClient(TestCase):
     @pytest.fixture(autouse=True)
-    def inject_fixtures(self, caplog: LogCaptureFixture):
+    def inject_fixtures(self, caplog: LogCaptureFixture) -> None:
         self._caplog = caplog
 
     def setUp(self) -> None:
@@ -40,12 +40,14 @@ def setUp(self) -> None:
         self.mock_resp_success.status_code = 200
         self.mock_resp_not_found = mock.MagicMock()
         self.mock_resp_not_found.status_code = 404
-        self.mock_resp_not_found.raise_for_status.side_effect = HTTPError("Not found")  # type: ignore  # noqa: E501
+        self.mock_resp_not_found.raise_for_status.side_effect = HTTPError(
+            "Not found", response=self.mock_resp_not_found
+        )
 
-    def tearDown(self):
+    def tearDown(self) -> None:
         mock.patch.stopall()
 
-    def test_query_image_success_APIv2(self):
+    def test_query_image_success_APIv2(self) -> None:
         fpath = "tests/data/query_v2/query_response_container/valid_qrc1.json"
         self.mock_resp_success.json.return_value = load_json(fpath)
         self.mock_session_v2.get.return_value = self.mock_resp_success
@@ -56,19 +58,20 @@ def test_query_image_success_APIv2(self):
         self.mock_session_v2.get.assert_called_once_with("/query", params=expected_params)
         self.mock_resp_success.raise_for_status.assert_called_once()
         # Note: JSON need to be loaded twice as `from_json` pops its original data
-        self.assertEqual(res, QueryResponseContainer.from_json(load_json(fpath)))
+        assert res == QueryResponseContainer.from_json(load_json(fpath))
 
-    def test_in_memory_query_image_APIv2(self):
+    def test_in_memory_query_image_APIv2(self) -> None:
         fpath = "tests/data/query_v2/query_response_container/valid_qrc1.json"
         data = QueryResponseContainer.from_json(load_json(fpath))
         provider = InMemoryMapProviderV2(data)
 
         self.svc_v2 = StarmapClient("https://test.starmap.com", api_version="v2", provider=provider)
 
         res = self.svc_v2.query_image("product-test-1.0-1.raw.xz", workflow="stratosphere")
-        self.assertEqual(res.responses, [data.responses[0]])
+        assert res
+        assert res.responses == [data.responses[0]]
 
-    def test_in_memory_api_mismatch(self):
+    def test_in_memory_api_mismatch(self) -> None:
         fpath = "tests/data/query_v2/query_response_container/valid_qrc1.json"
         data = QueryResponseContainer.from_json(load_json(fpath))
         provider_v2 = InMemoryMapProviderV2(data)
@@ -77,7 +80,7 @@ def test_in_memory_api_mismatch(self):
         with pytest.raises(ValueError, match=err):
             StarmapClient("foo", api_version="v1", provider=provider_v2)
 
-    def test_query_image_not_found(self):
+    def test_query_image_not_found(self) -> None:
         self.mock_session_v2.get.return_value = self.mock_resp_not_found
         self.mock_session_v2.get.return_value = self.mock_resp_not_found
 
@@ -86,9 +89,9 @@ def test_query_image_not_found(self):
                 res = svc.query_image(self.image)
                 expected_msg = "Marketplace mappings not defined for {'image': '%s'}" % self.image
                 assert expected_msg in self._caplog.text
-                self.assertIsNone(res)
+                assert res is None
 
-    def test_query_image_by_name_APIv2(self):
+    def test_query_image_by_name_APIv2(self) -> None:
         fpath = "tests/data/query_v2/query_response_container/valid_qrc1.json"
         self.mock_resp_success.json.return_value = load_json(fpath)
         self.mock_session_v2.get.return_value = self.mock_resp_success
@@ -99,20 +102,21 @@ def test_query_image_by_name_APIv2(self):
         self.mock_session_v2.get.assert_called_once_with("/query", params=expected_params)
         self.mock_resp_success.raise_for_status.assert_called_once()
         # Note: JSON need to be loaded twice as `from_json` pops its original data
-        self.assertEqual(res, QueryResponseContainer.from_json(load_json(fpath)))
+        assert res == QueryResponseContainer.from_json(load_json(fpath))
 
-    def test_in_memory_query_image_by_name_APIv2(self):
+    def test_in_memory_query_image_by_name_APIv2(self) -> None:
         fpath = "tests/data/query_v2/query_response_container/valid_qrc1.json"
         data = QueryResponseContainer.from_json(load_json(fpath))
         provider = InMemoryMapProviderV2(data)
 
         self.svc_v2 = StarmapClient("https://test.starmap.com", api_version="v2", provider=provider)
 
         res = self.svc_v2.query_image_by_name(name="product-test", workflow="stratosphere")
+        assert res
         self.mock_session_v2.get.assert_not_called()
-        self.assertEqual(res.responses, [data.responses[0]])
+        assert res.responses == [data.responses[0]]
 
-    def test_query_image_by_name_version_APIv2(self):
+    def test_query_image_by_name_version_APIv2(self) -> None:
         fpath = "tests/data/query_v2/query_response_container/valid_qrc1.json"
         self.mock_resp_success.json.return_value = load_json(fpath)
         self.mock_session_v2.get.return_value = self.mock_resp_success
@@ -126,9 +130,9 @@ def test_query_image_by_name_version_APIv2(self):
         self.mock_session_v2.get.assert_called_once_with("/query", params=expected_params)
         self.mock_resp_success.raise_for_status.assert_called_once()
         # Note: JSON need to be loaded twice as `from_json` pops its original data
-        self.assertEqual(res, QueryResponseContainer.from_json(load_json(fpath)))
+        assert res == QueryResponseContainer.from_json(load_json(fpath))
 
-    def test_policies_single_page(self):
+    def test_policies_single_page(self) -> None:
         fpath = "tests/data/policy/valid_pol1.json"
         single_page = {
             "items": [load_json(fpath)],
@@ -150,14 +154,14 @@ def test_policies_single_page(self):
         # Iterate over all policies from StarmapClient property and
         # ensure each of them has a valid format.
         for p in self.svc_v2.policies:
-            self.assertEqual(p, Policy.from_json(load_json(fpath)))
+            assert p == Policy.from_json(load_json(fpath))
 
         expected_params = {"page": 1, "per_page": 1}
         self.mock_session_v2.get.assert_called_once_with("policy", params=expected_params)
         self.mock_session_v2.get.call_count == 1
         self.mock_resp_success.raise_for_status.assert_called_once()
 
-    def test_policies_multi_page(self):
+    def test_policies_multi_page(self) -> None:
         fpath = "tests/data/policy/valid_pol1.json"
         page1 = {
             "items": [load_json(fpath)],
@@ -173,16 +177,16 @@ def test_policies_multi_page(self):
             },
         }
         page2 = deepcopy(page1)
-        page2["nav"]["next"] = None
-        page2["nav"]["page"] = 2
+        page2["nav"]["next"] = None  # type: ignore[index]
+        page2["nav"]["page"] = 2  # type: ignore[index]
         self.svc_v2.POLICIES_PER_PAGE = 1
         self.mock_resp_success.json.side_effect = [page1, page2]
         self.mock_session_v2.get.return_value = self.mock_resp_success
 
         # Iterate over all policies from StarmapClient property and
         # ensure each of them has a valid format.
         for p in self.svc_v2.policies:
-            self.assertEqual(p, Policy.from_json(load_json(fpath)))
+            assert p == Policy.from_json(load_json(fpath))
 
         get_calls = [
             mock.call("policy", params={"page": 1, "per_page": 1}),
@@ -196,7 +200,7 @@ def test_policies_multi_page(self):
         self.mock_session_v2.get.call_count == 2
         self.mock_resp_success.raise_for_status.call_count == 2
 
-    def test_policies_not_found(self):
+    def test_policies_not_found(self) -> None:
         self.svc_v2.POLICIES_PER_PAGE = 1
         self.mock_session_v2.get.return_value = self.mock_resp_not_found
 
@@ -206,7 +210,7 @@ def test_policies_not_found(self):
         assert "No policies registered in StArMap." in self._caplog.text
 
     @mock.patch('starmap_client.StarmapClient.policies')
-    def test_list_policies(self, mock_policies: mock.MagicMock):
+    def test_list_policies(self, mock_policies: mock.MagicMock) -> None:
         fpath = "tests/data/policy/valid_pol1.json"
         pol = Policy.from_json(load_json(fpath))
         pol_list = [pol]
@@ -216,15 +220,15 @@ def test_list_policies(self, mock_policies: mock.MagicMock):
         self.svc_v2._policies = pol_list
         res = self.svc_v2.list_policies()
         mock_policies.__iter__.assert_not_called()
-        self.assertEqual(res, self.svc_v2._policies)
+        assert res == self.svc_v2._policies
 
         # Test uncached policies list
         self.svc_v2._policies = []
         res = self.svc_v2.list_policies()
         mock_policies.__iter__.assert_called_once()
-        self.assertEqual(res, pol_list)
+        assert res == pol_list
 
-    def test_get_policy(self):
+    def test_get_policy(self) -> None:
         fpath = "tests/data/policy/valid_pol1.json"
         self.mock_resp_success.json.return_value = load_json(fpath)
         self.mock_session_v2.get.return_value = self.mock_resp_success
@@ -234,9 +238,9 @@ def test_get_policy(self):
         self.mock_session_v2.get.assert_called_once_with("/policy/policy-id")
         self.mock_resp_success.raise_for_status.assert_called_once()
         # Note: JSON need to be loaded twice as `from_json` pops its original data
-        self.assertEqual(res, Policy.from_json(load_json(fpath)))
+        assert res == Policy.from_json(load_json(fpath))
 
-    def test_get_policy_not_found(self):
+    def test_get_policy_not_found(self) -> None:
         self.mock_session_v2.get.return_value = self.mock_resp_not_found
 
         with self._caplog.at_level(logging.ERROR):
@@ -245,7 +249,7 @@ def test_get_policy_not_found(self):
         expected_msg = "Policy not found with ID = \"policy-id\""
         assert expected_msg in self._caplog.text
 
-        self.assertIsNone(res)
+        assert res is None
 
     @mock.patch("starmap_client.StarmapClient.get_policy")
     def test_list_mappings(self, mock_get_policy: mock.MagicMock) -> None:
@@ -256,7 +260,7 @@ def test_list_mappings(self, mock_get_policy: mock.MagicMock) -> None:
         res = self.svc_v2.list_mappings(policy_id="policy-id")
 
         mock_get_policy.assert_called_once_with("policy-id")
-        self.assertEqual(res, p.mappings)
+        assert res == p.mappings
 
     @mock.patch("starmap_client.StarmapClient.get_policy")
     def test_list_mappings_not_found(self, mock_get_policy: mock.MagicMock) -> None:
@@ -265,9 +269,9 @@ def test_list_mappings_not_found(self, mock_get_policy: mock.MagicMock) -> None:
 
         mock_get_policy.assert_called_once_with("policy-id")
 
-        self.assertEqual(res, [])
+        assert res == []
 
-    def test_get_mapping(self):
+    def test_get_mapping(self) -> None:
         fpath = "tests/data/mapping/valid_map1.json"
         self.mock_resp_success.json.return_value = load_json(fpath)
         self.mock_session_v2.get.return_value = self.mock_resp_success
@@ -277,9 +281,9 @@ def test_get_mapping(self):
         self.mock_session_v2.get.assert_called_once_with("/mapping/mapping-id")
         self.mock_resp_success.raise_for_status.assert_called_once()
         # Note: JSON need to be loaded twice as `from_json` pops its original data
-        self.assertEqual(res, Mapping.from_json(load_json(fpath)))
+        assert res == Mapping.from_json(load_json(fpath))
 
-    def test_get_mapping_not_found(self):
+    def test_get_mapping_not_found(self) -> None:
         self.mock_session_v2.get.return_value = self.mock_resp_not_found
 
         with self._caplog.at_level(logging.ERROR):
@@ -288,7 +292,7 @@ def test_get_mapping_not_found(self):
         expected_msg = "Marketplace Mapping not found with ID = \"mapping-id\""
         assert expected_msg in self._caplog.text
 
-        self.assertIsNone(res)
+        assert res is None
 
     @mock.patch("starmap_client.StarmapClient.get_mapping")
     def test_list_destinations(self, mock_get_mapping: mock.MagicMock) -> None:
@@ -299,7 +303,7 @@ def test_list_destinations(self, mock_get_mapping: mock.MagicMock) -> None:
         res = self.svc_v2.list_destinations(mapping_id="mapping-id")
 
         mock_get_mapping.assert_called_once_with("mapping-id")
-        self.assertEqual(res, m.destinations)
+        assert res == m.destinations
 
     @mock.patch("starmap_client.StarmapClient.get_mapping")
     def test_list_destinations_not_found(self, mock_get_mapping: mock.MagicMock) -> None:
@@ -308,9 +312,9 @@ def test_list_destinations_not_found(self, mock_get_mapping: mock.MagicMock) ->
 
         mock_get_mapping.assert_called_once_with("mapping-id")
 
-        self.assertEqual(res, [])
+        assert res == []
 
-    def test_get_destination(self):
+    def test_get_destination(self) -> None:
         fpath = "tests/data/destination/valid_dest1.json"
         self.mock_resp_success.json.return_value = load_json(fpath)
         self.mock_session_v2.get.return_value = self.mock_resp_success
@@ -320,9 +324,9 @@ def test_get_destination(self):
         self.mock_session_v2.get.assert_called_once_with("/destination/destination-id")
         self.mock_resp_success.raise_for_status.assert_called_once()
         # Note: JSON need to be loaded twice as `from_json` pops its original data
-        self.assertEqual(res, Destination.from_json(load_json(fpath)))
+        assert res == Destination.from_json(load_json(fpath))
 
-    def test_get_destination_not_found(self):
+    def test_get_destination_not_found(self) -> None:
         self.mock_session_v2.get.return_value = self.mock_resp_not_found
 
         with self._caplog.at_level(logging.ERROR):
@@ -331,15 +335,15 @@ def test_get_destination_not_found(self):
         expected_msg = "Destination not found with ID = \"destination-id\""
         assert expected_msg in self._caplog.text
 
-        self.assertIsNone(res)
+        assert res is None
 
-    def test_client_requires_url_or_session(self):
+    def test_client_requires_url_or_session(self) -> None:
         error = "Cannot initialize the client without defining either an \"url\" or \"session\"."
         with pytest.raises(ValueError, match=error):
             StarmapClient()
 
 
-def test_offline_client():
+def test_offline_client() -> None:
     """Ensure the cient can be used offline with a local provider."""
     fpath = "tests/data/query_v2/query_response_container/valid_qrc2.json"
     qrc = QueryResponseContainer.from_json(load_json(fpath))