WIP - fix remaining issues

Signed-off-by: David Black <[email protected]>

Author: dbaxa

atlassian_jwt_auth/contrib/tests/aiohttp/test_public_key_provider.py

@@ -9,7 +9,7 @@
     from unittest.mock import AsyncMock as CoroutineMock
     from unittest.mock import Mock
 except ImportError:
-    from asynctest import CoroutineMock, TestCase, Mock
+    from asynctest import CoroutineMock, TestCase, Mock  # type: ignore[import-untyped, no-redef]
 
 from atlassian_jwt_auth.contrib.aiohttp import HTTPSPublicKeyRetriever
 from atlassian_jwt_auth.key import PEM_FILE_TYPE
@@ -21,7 +21,7 @@
 class DummyHTTPSPublicKeyRetriever(HTTPSPublicKeyRetriever):
 
     def set_headers(self, headers) -> None:
-        self._session.get.return_value.headers.update(headers)
+        self._session.get.return_value.headers.update(headers)  # type: ignore[attr-defined]
 
     def set_text(self, text: str | bytes) -> None:
         self._session.get.return_value.text.return_value = text  # type: ignore[attr-defined]

atlassian_jwt_auth/contrib/tests/aiohttp/test_verifier.py

@@ -7,7 +7,7 @@
     from unittest.mock import AsyncMock as CoroutineMock
     from unittest.mock import Mock
 except ImportError:
-    from asynctest import TestCase, CoroutineMock
+    from asynctest import TestCase, CoroutineMock  # type: ignore[import-untyped, no-redef]
 
 from atlassian_jwt_auth.contrib.aiohttp import (HTTPSPublicKeyRetriever,
                                                 JWTAuthVerifier)
@@ -32,13 +32,13 @@ def verify_jwt(self, *args: Any, **kwargs: Any):
 class JWTAuthVerifierTestMixin(test_verifier.BaseJWTAuthVerifierTest):
     loop = None
 
-    def _setup_mock_public_key_retriever(self, pub_key_pem: str) -> Mock:
+    def _setup_mock_public_key_retriever(self, pub_key_pem: str) -> Mock:  # type: ignore[override]
         m_public_key_ret = CoroutineMock(spec=HTTPSPublicKeyRetriever)
-        m_public_key_ret.retrieve.return_value = pub_key_pem.decode()
+        m_public_key_ret.retrieve.return_value = pub_key_pem.decode()  # type: ignore[attr-defined]
         return m_public_key_ret
 
     def _setup_jwt_auth_verifier(
-            self, pub_key_pem: str, **kwargs: Any) -> SyncJWTAuthVerifier:
+            self, pub_key_pem: str, **kwargs: Any) -> SyncJWTAuthVerifier:  # type: ignore[override]
         m_public_key_ret = self._setup_mock_public_key_retriever(pub_key_pem)
         return SyncJWTAuthVerifier(m_public_key_ret, loop=self.loop, **kwargs)
 

atlassian_jwt_auth/contrib/tests/test_requests.py

@@ -17,7 +17,7 @@ class BaseRequestsTest(object):
     auth_cls: Type[BaseJWTAuth] = JWTAuth
 
     def setUp(self) -> None:
-        self._private_key_pem = self.get_new_private_key_in_pem_format()
+        self._private_key_pem = self.get_new_private_key_in_pem_format()  # type: ignore[attr-defined]
         self._public_key_pem = utils.get_public_key_pem_for_private_key_pem(
             self._private_key_pem)
 
@@ -47,15 +47,15 @@ def test_JWTAuth_make_authenticated_request(self) -> None:
             'issuer',
             'issuer/key',
             self._private_key_pem.decode(),
-            algorithm=self.algorithm)
+            algorithm=self.algorithm)  # type: ignore[attr-defined]
         auth = self.auth_cls(jwt_auth_signer, 'audience')
         self.assert_authorization_header_is_valid(auth)
 
     def test_create_jwt_auth(self) -> None:
         """Verify a valid Authorization header is added by JWTAuth"""
         auth = self.create_jwt_auth('issuer', 'issuer/key',
                                     self._private_key_pem.decode(), 'audience',
-                                    algorithm=self.algorithm)
+                                    algorithm=self.algorithm)  # type: ignore[attr-defined]
         self.assert_authorization_header_is_valid(auth)
 
     def test_create_jwt_auth_with_additional_claims(self) -> None:
@@ -66,77 +66,77 @@ def test_create_jwt_auth_with_additional_claims(self) -> None:
             'issuer',
             'issuer/key',
             self._private_key_pem.decode(),
-            algorithm=self.algorithm)
+            algorithm=self.algorithm)  # type: ignore[attr-defined]
         auth = self.auth_cls(jwt_auth_signer, 'audience',
                              additional_claims={'example': 'claim'})
         token = self.assert_authorization_header_is_valid(auth)
-        self.assertEqual(token.get('example'), 'claim')
+        self.assertEqual(token.get('example'), 'claim')  # type: ignore[attr-defined]
 
     def test_do_not_reuse_jwts(self) -> None:
         auth = self.create_jwt_auth('issuer', 'issuer/key',
                                     self._private_key_pem.decode(), 'audience',
-                                    algorithm=self.algorithm)
+                                    algorithm=self.algorithm)  # type: ignore[attr-defined]
         auth_header = self._get_auth_header(auth)
-        self.assertNotEqual(auth_header, self._get_auth_header(auth))
+        self.assertNotEqual(auth_header, self._get_auth_header(auth))  # type: ignore[attr-defined]
 
     def test_reuse_jwts(self) -> None:
         auth = self.create_jwt_auth('issuer', 'issuer/key',
                                     self._private_key_pem.decode(), 'audience',
-                                    algorithm=self.algorithm, reuse_jwts=True)
+                                    algorithm=self.algorithm, reuse_jwts=True)  # type: ignore[attr-defined]
         auth_header = self._get_auth_header(auth)
-        self.assertEqual(auth_header, self._get_auth_header(auth))
+        self.assertEqual(auth_header, self._get_auth_header(auth))  # type: ignore[attr-defined]
 
     def test_do_not_reuse_jwt_if_audience_changes(self) -> None:
         auth = self.create_jwt_auth('issuer', 'issuer/key',
                                     self._private_key_pem.decode(), 'audience',
-                                    algorithm=self.algorithm, reuse_jwts=True)
+                                    algorithm=self.algorithm, reuse_jwts=True)  # type: ignore[attr-defined]
         auth_header = self._get_auth_header(auth)
         auth._audience = 'not-' + auth._audience
-        self.assertNotEqual(auth_header, self._get_auth_header(auth))
+        self.assertNotEqual(auth_header, self._get_auth_header(auth))  # type: ignore[attr-defined]
 
     def test_do_not_reuse_jwt_if_issuer_changes(self) -> None:
         auth = self.create_jwt_auth('issuer', 'issuer/key',
                                     self._private_key_pem.decode(), 'audience',
-                                    algorithm=self.algorithm, reuse_jwts=True)
+                                    algorithm=self.algorithm, reuse_jwts=True)  # type: ignore[attr-defined]
         auth_header = self._get_auth_header(auth)
         auth._signer.issuer = 'not-' + auth._signer.issuer
-        self.assertNotEqual(auth_header, self._get_auth_header(auth))
+        self.assertNotEqual(auth_header, self._get_auth_header(auth))  # type: ignore[attr-defined]
 
     def test_do_not_reuse_jwt_if_lifetime_changes(self) -> None:
         auth = self.create_jwt_auth('issuer', 'issuer/key',
                                     self._private_key_pem.decode(), 'audience',
-                                    algorithm=self.algorithm, reuse_jwts=True)
+                                    algorithm=self.algorithm, reuse_jwts=True)  # type: ignore[attr-defined]
         auth_header = self._get_auth_header(auth)
         auth._signer.lifetime = auth._signer.lifetime - timedelta(seconds=1)
-        self.assertNotEqual(auth_header, self._get_auth_header(auth))
+        self.assertNotEqual(auth_header, self._get_auth_header(auth))  # type: ignore[attr-defined]
 
     def test_do_not_reuse_jwt_if_subject_changes(self) -> None:
         auth = self.create_jwt_auth('issuer', 'issuer/key',
                                     self._private_key_pem.decode(), 'audience',
-                                    algorithm=self.algorithm, reuse_jwts=True,
+                                    algorithm=self.algorithm, reuse_jwts=True,  # type: ignore[attr-defined]
                                     subject='subject')
         auth_header = self._get_auth_header(auth)
-        auth._signer.subject = 'not-' + auth._signer.subject
-        self.assertNotEqual(auth_header, self._get_auth_header(auth))
+        auth._signer.subject = 'not-' + auth._signer.subject  # type: ignore[operator]
+        self.assertNotEqual(auth_header, self._get_auth_header(auth))  # type: ignore[attr-defined]
 
     def test_do_not_reuse_jwt_if_additional_claims_change(self) -> None:
         auth = self.create_jwt_auth('issuer', 'issuer/key',
                                     self._private_key_pem.decode(), 'audience',
-                                    algorithm=self.algorithm, reuse_jwts=True)
+                                    algorithm=self.algorithm, reuse_jwts=True)  # type: ignore[attr-defined]
         auth_header = self._get_auth_header(auth)
         auth._additional_claims['foo'] = 'bar'
-        self.assertNotEqual(auth_header, self._get_auth_header(auth))
+        self.assertNotEqual(auth_header, self._get_auth_header(auth))  # type: ignore[attr-defined]
 
     def test_reuse_jwt_with_additional_claims(self) -> None:
         # calculating the cache key with additional claims is non-trivial
         auth = self.create_jwt_auth('issuer', 'issuer/key',
                                     self._private_key_pem.decode(), 'audience',
-                                    algorithm=self.algorithm, reuse_jwts=True)
+                                    algorithm=self.algorithm, reuse_jwts=True)  # type: ignore[attr-defined]
         auth._additional_claims['foo'] = 'bar'
         auth._additional_claims['fool'] = 'blah'
         auth._additional_claims['foot'] = 'quux'
         auth_header = self._get_auth_header(auth)
-        self.assertEqual(auth_header, self._get_auth_header(auth))
+        self.assertEqual(auth_header, self._get_auth_header(auth))  # type: ignore[attr-defined]
 
 
 class RequestsRS256Test(BaseRequestsTest,

atlassian_jwt_auth/tests/test_private_key_provider.py

@@ -25,7 +25,7 @@ class BaseDataUriPrivateKeyRetrieverTest(object):
     """ tests for the DataUriPrivateKeyRetriever class. """
 
     def setUp(self) -> None:
-        self._private_key_pem = self.get_new_private_key_in_pem_format()
+        self._private_key_pem = self.get_new_private_key_in_pem_format()  # type: ignore[attr-defined]
         self._public_key_pem = utils.get_public_key_pem_for_private_key_pem(
             self._private_key_pem)
         self._private_key_der = convert_key_pem_format_to_der_format(
@@ -41,8 +41,8 @@ def test_load_data_uri(self) -> None:
         data_uri = self.get_example_data_uri(self._private_key_der)
         provider = DataUriPrivateKeyRetriever(data_uri)
         kid, private_key_pem = provider.load('example')
-        self.assertEqual(kid.key_id, expected_kid)
-        self.assertEqual(private_key_pem,
+        self.assertEqual(kid.key_id, expected_kid)  # type: ignore[attr-defined]
+        self.assertEqual(private_key_pem,  # type: ignore[attr-defined]
                          self._private_key_pem.decode('utf-8'))
 
     def test_load_data_uri_can_be_used_with_a_signer(self) -> None:
@@ -52,7 +52,7 @@ def test_load_data_uri_can_be_used_with_a_signer(self) -> None:
         data_uri = self.get_example_data_uri(self._private_key_der)
         provider = DataUriPrivateKeyRetriever(data_uri)
         jwt_auth_signer = JWTAuthSigner(
-            'issuer', provider, algorithm=self.algorithm)
+            'issuer', provider, algorithm=self.algorithm)  # type: ignore[attr-defined]
         jwt_auth_signer.generate_jwt('aud')
 
 

atlassian_jwt_auth/tests/test_public_key_provider.py

@@ -37,7 +37,7 @@ def create_retriever(self, url) -> HTTPSPublicKeyRetriever:
         return HTTPSPublicKeyRetriever(url)
 
     def setUp(self) -> None:
-        self._private_key_pem = self.get_new_private_key_in_pem_format()
+        self._private_key_pem = self.get_new_private_key_in_pem_format()  # type: ignore[attr-defined]
         self._public_key_pem = utils.get_public_key_pem_for_private_key_pem(
             self._private_key_pem)
         self.base_url = 'https://example.com'
@@ -47,15 +47,15 @@ def test_https_public_key_retriever_does_not_support_http_url(
         """ tests that HTTPSPublicKeyRetriever does not support http://
             base urls.
         """
-        with self.assertRaises(ValueError):
+        with self.assertRaises(ValueError):  # type: ignore[attr-defined]
             self.create_retriever('http://example.com')
 
     def test_https_public_key_retriever_does_not_support_none_url(
             self) -> None:
         """ tests that HTTPSPublicKeyRetriever does not support None
             base urls.
         """
-        with self.assertRaises(ValueError):
+        with self.assertRaises(ValueError):  # type: ignore[attr-defined]
             self.create_retriever(None)
 
     def test_https_public_key_retriever_session_uses_env_proxy(self) -> None:
@@ -69,9 +69,9 @@ def test_https_public_key_retriever_session_uses_env_proxy(self) -> None:
             retriever = self.create_retriever(self.base_url)
             key_retrievers = [retriever]
             if isinstance(retriever, HTTPSMultiRepositoryPublicKeyRetriever):
-                key_retrievers = retriever._retrievers
+                key_retrievers = retriever._retrievers  # type: ignore[assignment]
             for key_retriever in key_retrievers:
-                self.assertEqual(key_retriever._proxies, expected_proxies)
+                self.assertEqual(key_retriever._proxies, expected_proxies)  # type: ignore[attr-defined]
 
     def test_https_public_key_retriever_supports_https_url(self) -> None:
         """ tests that HTTPSPublicKeyRetriever supports https://
@@ -83,9 +83,9 @@ def test_https_public_key_retriever_supports_https_url(self) -> None:
     def test_retrieve(self, mock_get_method: Mock) -> None:
         """ tests that the retrieve method works expected. """
         _setup_mock_response_for_retriever(
-            mock_get_method, self._public_key_pem)
+            mock_get_method, self._public_key_pem)  # type: ignore[arg-type]
         retriever = self.create_retriever(self.base_url)
-        self.assertEqual(
+        self.assertEqual(  # type: ignore[attr-defined]
             retriever.retrieve('example/eg'),
             self._public_key_pem)
 
@@ -99,7 +99,7 @@ def test_retrieve_with_proxy(self, mock_get_method: Mock) -> None:
         expected_proxies, proxy_dict = get_expected_and_os_proxies_dict(
             proxy_location)
         _setup_mock_response_for_retriever(
-            mock_get_method, self._public_key_pem)
+            mock_get_method, self._public_key_pem)  # type: ignore[arg-type]
         with mock.patch.dict(os.environ, proxy_dict, clear=True):
             retriever = self.create_retriever(self.base_url)
             retriever.retrieve(key_id)
@@ -122,7 +122,7 @@ def test_retrieve_with_proxy_explicitly_set(
         expected_proxies, _ = get_expected_and_os_proxies_dict(
             explicit_proxy_location)
         _setup_mock_response_for_retriever(
-            mock_get_method, self._public_key_pem)
+            mock_get_method, self._public_key_pem)  # type: ignore[arg-type]
         with mock.patch.dict(os.environ, proxy_dict, clear=True):
             retriever = self.create_retriever(self.base_url)
             retriever.retrieve(key_id, proxies=expected_proxies)
@@ -140,9 +140,9 @@ def test_retrieve_with_charset_in_content_type_h(
         """
         headers = {'content-type': 'application/x-pem-file;charset=UTF-8'}
         _setup_mock_response_for_retriever(
-            mock_get_method, self._public_key_pem, headers)
+            mock_get_method, self._public_key_pem, headers)  # type: ignore[arg-type]
         retriever = self.create_retriever(self.base_url)
-        self.assertEqual(
+        self.assertEqual(  # type: ignore[attr-defined]
             retriever.retrieve('example/eg'),
             self._public_key_pem)
 
@@ -154,9 +154,9 @@ def test_retrieve_fails_with_different_content_type(
         """
         headers = {'content-type': 'different/not-supported'}
         _setup_mock_response_for_retriever(
-            mock_get_method, self._public_key_pem, headers)
+            mock_get_method, self._public_key_pem, headers)  # type: ignore[arg-type]
         retriever = self.create_retriever(self.base_url)
-        with self.assertRaises(ValueError):
+        with self.assertRaises(ValueError):  # type: ignore[attr-defined]
             retriever.retrieve('example/eg')
 
     @mock.patch.object(requests.Session, 'get',
@@ -169,9 +169,9 @@ def test_retrieve_fails_with_forbidden_error(
         403 forbidden error.
         """
         _setup_mock_response_for_retriever(
-            mock_get_method, self._public_key_pem)
+            mock_get_method, self._public_key_pem)  # type: ignore[arg-type]
         retriever = self.create_retriever(self.base_url)
-        with self.assertRaises(ValueError):
+        with self.assertRaises(ValueError):  # type: ignore[attr-defined]
             retriever.retrieve('example/eg')
 
 
@@ -225,13 +225,13 @@ class BaseHTTPSMultiRepositoryPublicKeyRetrieverTest(
         BaseHTTPSPublicKeyRetrieverTest):
     """ tests for the HTTPSMultiRepositoryPublicKeyRetriever class. """
 
-    def create_retriever(
+    def create_retriever(  # type: ignore[override]
             self, url: str) -> HTTPSMultiRepositoryPublicKeyRetriever:
         """ returns a public key retriever created using the given url. """
         return HTTPSMultiRepositoryPublicKeyRetriever([url])
 
     def setUp(self) -> None:
-        self._private_key_pem = self.get_new_private_key_in_pem_format()
+        self._private_key_pem = self.get_new_private_key_in_pem_format()  # type: ignore[attr-defined]
         self._public_key_pem = utils.get_public_key_pem_for_private_key_pem(
             self._private_key_pem)
         self.keystore_urls = ['https://example.com', 'https://example.ly']
@@ -242,16 +242,16 @@ def test_https_multi_public_key_retriever_does_not_support_strings(
         """ tests that HTTPSMultiRepositoryPublicKeyRetriever does not
             support a string key repository url.
         """
-        with self.assertRaises(TypeError):
+        with self.assertRaises(TypeError):  # type: ignore[attr-defined]
             HTTPSMultiRepositoryPublicKeyRetriever('https://example.com')
 
     @mock.patch.object(requests.Session, 'get')
     def test_retrieve(self, mock_get_method: Mock) -> None:
         """ tests that the retrieve method works expected. """
         _setup_mock_response_for_retriever(
-            mock_get_method, self._public_key_pem)
+            mock_get_method, self._public_key_pem)  # type: ignore[arg-type]
         retriever = HTTPSMultiRepositoryPublicKeyRetriever(self.keystore_urls)
-        self.assertEqual(
+        self.assertEqual(  # type: ignore[attr-defined]
             retriever.retrieve('example/eg'),
             self._public_key_pem)
 
@@ -262,13 +262,13 @@ def test_retrieve_with_500_error(self, mock_get_method: Mock) -> None:
         """
         retriever = HTTPSMultiRepositoryPublicKeyRetriever(self.keystore_urls)
         _setup_mock_response_for_retriever(
-            mock_get_method, self._public_key_pem)
+            mock_get_method, self._public_key_pem)  # type: ignore[arg-type]
         valid_response = mock_get_method.return_value
         del mock_get_method.return_value
         server_exception = requests.exceptions.HTTPError(
             response=mock.Mock(status_code=500))
         mock_get_method.side_effect = [server_exception, valid_response]
-        self.assertEqual(
+        self.assertEqual(  # type: ignore[attr-defined]
             retriever.retrieve('example/eg'),
             self._public_key_pem)
 
@@ -280,13 +280,13 @@ def test_retrieve_with_connection_error(
         """
         retriever = HTTPSMultiRepositoryPublicKeyRetriever(self.keystore_urls)
         _setup_mock_response_for_retriever(
-            mock_get_method, self._public_key_pem)
+            mock_get_method, self._public_key_pem)  # type: ignore[arg-type]
         valid_response = mock_get_method.return_value
         del mock_get_method.return_value
         connection_exception = requests.exceptions.ConnectionError(
             response=mock.Mock(status_code=None))
         mock_get_method.side_effect = [connection_exception, valid_response]
-        self.assertEqual(
+        self.assertEqual(  # type: ignore[attr-defined]
             retriever.retrieve('example/eg'),
             self._public_key_pem)