Merge pull request #95 from IdentityPython/httpc_params_loader

feat: httpc_params_loader (httpc default timeout refactor)

Author: peppelinux

pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+markers =
+    network: mark a test as a network.

src/cryptojwt/key_bundle.py

@@ -29,13 +29,13 @@
 from .jwk.rsa import RSAKey
 from .jwk.rsa import new_rsa_key
 from .utils import as_unicode
+from .utils import httpc_params_loader
 
 __author__ = "Roland Hedberg"
 
 KEYLOADERR = "Failed to load %s key from '%s' (%s)"
 REMOTE_FAILED = "Remote key update from '{}' failed, HTTP status {}"
 MALFORMED = "Remote key update from {} failed, malformed JWKS."
-DEFAULT_HTTPC_TIMEOUT = 10
 
 LOGGER = logging.getLogger(__name__)
 
@@ -254,9 +254,7 @@ def __init__(
         else:
             self.httpc = requests.request
 
-        self.httpc_params = httpc_params or {}
-        if "timeout" not in self.httpc_params:
-            self.httpc_params["timeout"] = DEFAULT_HTTPC_TIMEOUT
+        self.httpc_params = httpc_params_loader(httpc_params)
 
         if keys:
             self.source = None

src/cryptojwt/key_issuer.py

@@ -8,11 +8,11 @@
 
 from .jwe.utils import alg2keytype as jwe_alg2keytype
 from .jws.utils import alg2keytype as jws_alg2keytype
-from .key_bundle import DEFAULT_HTTPC_TIMEOUT
 from .key_bundle import KeyBundle
 from .key_bundle import build_key_bundle
 from .key_bundle import key_diff
 from .key_bundle import update_key_bundle
+from .utils import httpc_params_loader
 from .utils import importer
 from .utils import qualified_name
 
@@ -58,9 +58,7 @@ def __init__(
 
         self.ca_certs = ca_certs
         self.httpc = httpc or request
-        self.httpc_params = httpc_params or {}
-        if "timeout" not in self.httpc_params:
-            self.httpc_params["timeout"] = DEFAULT_HTTPC_TIMEOUT
+        self.httpc_params = httpc_params_loader(httpc_params)
         self.keybundle_cls = keybundle_cls
         self.name = name
         self.remove_after = remove_after

src/cryptojwt/key_jar.py

@@ -8,12 +8,12 @@
 from .exception import IssuerNotFound
 from .jwe.jwe import alg2keytype as jwe_alg2keytype
 from .jws.utils import alg2keytype as jws_alg2keytype
-from .key_bundle import DEFAULT_HTTPC_TIMEOUT
 from .key_bundle import KeyBundle
 from .key_issuer import KeyIssuer
 from .key_issuer import build_keyissuer
 from .key_issuer import init_key_issuer
 from .utils import deprecated_alias
+from .utils import httpc_params_loader
 from .utils import importer
 from .utils import qualified_name
 
@@ -51,9 +51,7 @@ def __init__(
         self.keybundle_cls = keybundle_cls
         self.remove_after = remove_after
         self.httpc = httpc or request
-        self.httpc_params = httpc_params or {}
-        if "timeout" not in self.httpc_params:
-            self.httpc_params["timeout"] = DEFAULT_HTTPC_TIMEOUT
+        self.httpc_params = httpc_params_loader(httpc_params)
 
         # Now part of httpc_params
         # self.verify_ssl = verify_ssl

src/cryptojwt/utils.py

@@ -10,6 +10,8 @@
 
 from cryptojwt.exception import BadSyntax
 
+DEFAULT_HTTPC_TIMEOUT = 10
+
 # ---------------------------------------------------------------------------
 # Helper functions
 
@@ -255,3 +257,10 @@ def rename_kwargs(func_name, kwargs, aliases):
                 raise TypeError("{} received both {} and {}".format(func_name, alias, new))
             warnings.warn("{} is deprecated; use {}".format(alias, new), DeprecationWarning)
             kwargs[new] = kwargs.pop(alias)
+
+
+def httpc_params_loader(httpc_params):
+    httpc_params = httpc_params or {}
+    if "timeout" not in httpc_params:
+        httpc_params["timeout"] = DEFAULT_HTTPC_TIMEOUT
+    return httpc_params

tests/test_04_key_jar.py

@@ -1058,7 +1058,7 @@ def test_similar():
 
     kj = KeyJar()
     kb = KeyBundle(JWK2)
-    kj.add_kb(issuer=ISSUER, kb=kb)
+    kj.add_kb(issuer_id=ISSUER, kb=kb)
 
     keys1 = kj.get_issuer_keys(ISSUER)
     keys2 = kj[ISSUER].all_keys()

tests/test_50_argument_alias.py

@@ -65,33 +65,35 @@ def setup(self):
         self.bob_keyjar["Bob"] = self.bob_keyjar[""]
 
         # To Alice's keyjar add Bob's public keys
-        self.alice_keyjar.import_jwks(self.bob_keyjar.export_jwks(issuer="Bob"), "Bob")
+        self.alice_keyjar.import_jwks(self.bob_keyjar.export_jwks(issuer_id="Bob"), "Bob")
 
         # To Bob's keyjar add Alice's public keys
-        self.bob_keyjar.import_jwks(self.alice_keyjar.export_jwks(issuer="Alice"), "Alice")
+        self.bob_keyjar.import_jwks(self.alice_keyjar.export_jwks(issuer_id="Alice"), "Alice")
 
         _jws = JWS('{"aud": "Bob", "iss": "Alice"}', alg="RS256")
-        sig_key = self.alice_keyjar.get_signing_key("rsa", owner="Alice")[0]
+        sig_key = self.alice_keyjar.get_signing_key("rsa", issuer_id="Alice")[0]
         self.sjwt_a = _jws.sign_compact([sig_key])
 
         _jws = JWS('{"aud": "Alice", "iss": "Bob"}', alg="RS256")
-        sig_key = self.bob_keyjar.get_signing_key("rsa", owner="Bob")[0]
+        sig_key = self.bob_keyjar.get_signing_key("rsa", issuer_id="Bob")[0]
         self.sjwt_b = _jws.sign_compact([sig_key])
 
     def test_no_kid_multiple_keys_no_kid_issuer(self):
-        a_kids = [k.kid for k in self.alice_keyjar.get_verify_key(owner="Alice", key_type="RSA")]
+        a_kids = [
+            k.kid for k in self.alice_keyjar.get_verify_key(issuer_id="Alice", key_type="RSA")
+        ]
         no_kid_issuer = {"Alice": a_kids}
         _jwt = factory(self.sjwt_a)
         _jwt.jwt.headers["kid"] = ""
         keys = self.bob_keyjar.get_jwt_verify_keys(_jwt.jwt, no_kid_issuer=no_kid_issuer)
         assert len(keys) == 3
 
     def test_aud(self):
-        self.alice_keyjar.import_jwks(JWK1, issuer="D")
-        self.bob_keyjar.import_jwks(JWK1, issuer="D")
+        self.alice_keyjar.import_jwks(JWK1, issuer_id="D")
+        self.bob_keyjar.import_jwks(JWK1, issuer_id="D")
 
         _jws = JWS('{"iss": "D", "aud": "A"}', alg="HS256")
-        sig_key = self.alice_keyjar.get_signing_key("oct", owner="D")[0]
+        sig_key = self.alice_keyjar.get_signing_key("oct", issuer_id="D")[0]
         _sjwt = _jws.sign_compact([sig_key])
 
         no_kid_issuer = {"D": []}
@@ -121,7 +123,9 @@ def test_init_key_jar_dump_private():
             os.unlink(_file)
 
     # New set of keys, JWKSs with keys and public written to file
-    _keyjar = init_key_jar(private_path=PRIVATE_FILE, key_defs=KEYSPEC, owner="https://example.com")
+    _keyjar = init_key_jar(
+        private_path=PRIVATE_FILE, key_defs=KEYSPEC, issuer_id="https://example.com"
+    )
     assert list(_keyjar.owners()) == ["https://example.com"]
 
     # JWKS will be read from disc, not created new
@@ -138,7 +142,7 @@ def test_init_key_jar_update():
     _keyjar_1 = init_key_jar(
         private_path=PRIVATE_FILE,
         key_defs=KEYSPEC,
-        owner="https://example.com",
+        issuer_id="https://example.com",
         public_path=PUBLIC_FILE,
         read_only=False,
     )