Merge pull request #108 from pvarki/pytest

Pytest updates

Author: rambo

poetry.lock

@@ -4,8 +4,7 @@
 import aiohttp
 
 
-from .base import anon_session, get_result_cert, CFSSLError, ocsprest_base
-from ..rmsettings import RMSettings
+from .base import anon_session, get_result_cert, CFSSLError, ocsprest_base, default_timeout
 
 LOGGER = logging.getLogger(__name__)
 
@@ -23,7 +22,7 @@ async def anon_sign_csr(csr: str, bundle: bool = True) -> str:
         url = f"{ocsprest_base()}/api/v1/csr/sign"
         payload = {"certificate_request": csr, "profile": "client", "bundle": bundle}
         try:
-            async with session.post(url, json=payload, timeout=RMSettings.singleton().cfssl_timeout) as response:
+            async with session.post(url, json=payload, timeout=default_timeout()) as response:
                 resp = await get_result_cert(response)
                 return resp
         except aiohttp.ClientError as exc:

src/rasenmaeher_api/cfssl/anoncsr.py

@@ -9,7 +9,11 @@
 from ..rmsettings import RMSettings
 
 LOGGER = logging.getLogger(__name__)
-DEFAULT_TIMEOUT = RMSettings.singleton().cfssl_timeout
+
+
+def default_timeout() -> aiohttp.ClientTimeout:
+    """Return configured timeout wrapped in the new aiohttp way"""
+    return aiohttp.ClientTimeout(total=RMSettings.singleton().cfssl_timeout)
 
 
 class CFSSLError(RuntimeError):

src/rasenmaeher_api/cfssl/base.py

@@ -9,7 +9,7 @@
 import cryptography.x509
 from libadvian.tasks import TaskMaster
 
-from .base import base_url, get_result_cert, CFSSLError, get_result, NoResult, ocsprest_base, DBLocked
+from .base import base_url, get_result_cert, CFSSLError, get_result, NoResult, ocsprest_base, DBLocked, default_timeout
 from .mtls import mtls_session
 from ..rmsettings import RMSettings
 
@@ -29,7 +29,7 @@ async def post_ocsprest(
     async with (await mtls_session()) as session:
         try:
             LOGGER.debug("POSTing to {}, payload={}".format(url, send_payload))
-            async with session.post(url, data=send_payload, timeout=timeout) as response:
+            async with session.post(url, data=send_payload, timeout=aiohttp.ClientTimeout(total=timeout)) as response:
                 resp_payload = await response.json()
                 LOGGER.debug("resp_payload={}".format(resp_payload))
                 if not resp_payload["success"]:
@@ -59,7 +59,7 @@ async def sign_csr(csr: str, bundle: bool = True) -> str:
         payload = {"certificate_request": csr, "profile": "client", "bundle": bundle}
         try:
             LOGGER.debug("Calling {}".format(url))
-            async with session.post(url, json=payload, timeout=RMSettings.singleton().cfssl_timeout) as response:
+            async with session.post(url, json=payload, timeout=default_timeout()) as response:
                 resp = await get_result_cert(response)
                 TaskMaster.singleton().create_task(refresh_ocsp())
                 return resp
@@ -80,7 +80,7 @@ async def sign_ocsp(cert: str, status: str = "good") -> Any:
         url = f"{base_url()}/api/v1/cfssl/ocspsign"
         payload = {"certificate": cert, "status": status}
         try:
-            async with session.post(url, json=payload, timeout=RMSettings.singleton().cfssl_timeout) as response:
+            async with session.post(url, json=payload, timeout=default_timeout()) as response:
                 return await get_result(response)
         except aiohttp.ClientError as exc:
             raise CFSSLError(str(exc)) from exc
@@ -137,7 +137,7 @@ async def revoke_serial(serialno: str, authority_key_id: str, reason: ReasonType
             "reason": str(reason.value).replace("_", ""),
         }
         try:
-            async with session.post(url, json=payload, timeout=RMSettings.singleton().cfssl_timeout) as response:
+            async with session.post(url, json=payload, timeout=default_timeout()) as response:
                 try:
                     await get_result(response)
                 except NoResult:
@@ -179,7 +179,7 @@ async def certadd_pem(pem: Union[str, Path], status: str = "good") -> Any:
         }
         try:
             LOGGER.debug("POSTing {} to {}".format(payload, url))
-            async with session.post(url, json=payload, timeout=RMSettings.singleton().cfssl_timeout) as response:
+            async with session.post(url, json=payload, timeout=default_timeout()) as response:
                 return await get_result(response)
         except aiohttp.ClientError as exc:
             raise CFSSLError(str(exc)) from exc

src/rasenmaeher_api/cfssl/private.py

@@ -5,9 +5,17 @@
 
 import aiohttp
 
-from .base import base_url, anon_session, get_result, get_result_cert, CFSSLError, get_result_bundle, ocsprest_base
+from .base import (
+    base_url,
+    anon_session,
+    get_result,
+    get_result_cert,
+    CFSSLError,
+    get_result_bundle,
+    ocsprest_base,
+    default_timeout,
+)
 from .private import refresh_ocsp
-from ..rmsettings import RMSettings
 
 
 LOGGER = logging.getLogger(__name__)
@@ -25,7 +33,7 @@ async def get_ca() -> str:
         payload: Dict[str, Any] = {}
         # PONDER: Why does this need to be a POST ??
         try:
-            async with session.post(url, json=payload, timeout=RMSettings.singleton().cfssl_timeout) as response:
+            async with session.post(url, json=payload, timeout=default_timeout()) as response:
                 return await get_result_cert(response)
         except aiohttp.ClientError as exc:
             raise CFSSLError(str(exc)) from exc
@@ -54,9 +62,7 @@ async def get_crl() -> bytes:
     async with (await anon_session()) as session:
         url = f"{base_url()}/api/v1/cfssl/crl"
         try:
-            async with session.get(
-                url, params={"expiry": CRL_LIFETIME}, timeout=RMSettings.singleton().cfssl_timeout
-            ) as response:
+            async with session.get(url, params={"expiry": CRL_LIFETIME}, timeout=default_timeout()) as response:
                 crl_b64 = await get_result(response)
                 data = base64.b64decode(crl_b64)
                 return data
@@ -76,7 +82,7 @@ async def get_bundle(cert: str) -> str:
         url = f"{base_url()}/api/v1/cfssl/bundle"
         payload: Dict[str, Any] = {"certificate": cert, "flavor": "optimal"}
         try:
-            async with session.post(url, json=payload, timeout=RMSettings.singleton().cfssl_timeout) as response:
+            async with session.post(url, json=payload, timeout=default_timeout()) as response:
                 return await get_result_bundle(response)
         except aiohttp.ClientError as exc:
             raise CFSSLError(str(exc)) from exc

src/rasenmaeher_api/cfssl/public.py

@@ -10,7 +10,9 @@
 from starlette.datastructures import Secret
 
 LOGGER = logging.getLogger(__name__)
-config = Config()  # not supporting .env files anymore because https://github.com/encode/starlette/discussions/2446
+config = cast(
+    Callable[..., Any], Config()
+)  # not supporting .env files anymore because https://github.com/encode/starlette/discussions/2446
 
 # FIXME: this should probably be in some common library of ours