Apply pre-commit fix

From the artifact of the previous workflow run

c2cwsgiutils/acceptance/__init__.py

@@ -20,6 +20,7 @@ def retry(
         tries: number of times to try (not retry) before giving up
         delay: initial delay between retries in seconds
         backoff: backoff multiplier e.g. value of 2 will double the delay each retry
+
     """
 
     def deco_retry(f: typing.Callable[..., typing.Any]) -> typing.Callable[..., typing.Any]:

c2cwsgiutils/acceptance/image.py

@@ -89,6 +89,7 @@ def check_image(
       level: The minimum similarity level (between 0.0 and 1.0), default to 1.0
       generate_expected_image: If `True` generate the expected image instead of checking it
       use_mask: If `False` don't use the mask event if the file exists
+
     """
     assert image_to_check is not None, "Image required"
     image_file_basename = os.path.splitext(os.path.basename(expected_filename))[0]
@@ -122,7 +123,7 @@ def check_image(
         if np.issubdtype(mask.dtype, np.floating):
             mask = (mask * 255).astype("uint8")
 
-        assert ((0 < mask) & (mask < 255)).sum() == 0, "Mask should be only black and white image"
+        assert ((mask > 0) & (mask < 255)).sum() == 0, "Mask should be only black and white image"
 
         # Convert to boolean
         mask = mask == 0
@@ -139,7 +140,7 @@ def check_image(
         return
     if not os.path.isfile(expected_filename):
         skimage.io.imsave(expected_filename, image_to_check)
-        assert False, "Expected image not found: " + expected_filename
+        raise AssertionError("Expected image not found: " + expected_filename)
     expected = skimage.io.imread(expected_filename)
     assert expected is not None, "Wrong image: " + expected_filename
     expected = normalize_image(expected)
@@ -201,6 +202,7 @@ def check_screenshot(
       level: See `check_image`
       generate_expected_image: See `check_image`
       use_mask: See `check_image`
+
     """
     if headers is None:
         headers = {}

c2cwsgiutils/acceptance/print.py

@@ -21,6 +21,7 @@ def __init__(self, base_url: str, origin: str) -> None:
             base_url: The base URL to the print server (including the /print)
             app: The name of the application to use
             origin: The origin and referrer to include in the requests
+
         """
         super().__init__(base_url=base_url, origin=origin)
         self.session.headers["Referrer"] = origin

c2cwsgiutils/acceptance/utils.py

@@ -32,6 +32,7 @@ def retry_timeout(what: Callable[[], Any], timeout: float = _DEFAULT_TIMEOUT, in
         what: the function to try
         timeout: the timeout to get a success
         interval: the interval between try
+
     """
     timeout = time.perf_counter() + timeout
     while True:
@@ -46,7 +47,7 @@ def retry_timeout(what: Callable[[], Any], timeout: float = _DEFAULT_TIMEOUT, in
             error = str(e)
             _LOG.info("  Failed: %s", e)
         if time.perf_counter() > timeout:
-            assert False, "Timeout: " + error
+            raise AssertionError("Timeout: " + error)
         time.sleep(interval)
 
 

c2cwsgiutils/auth.py

@@ -209,6 +209,7 @@ def check_access(
         request: is the request object.
         repo: is the repository to check access to (<organization>/<repository>).
         access_type: is the type of access to check (admin|push|pull).
+
     """
     if not is_auth(request):
         return False

c2cwsgiutils/broadcast/redis.py

@@ -85,7 +85,8 @@ def callback(msg: Mapping[str, Any]) -> None:
                 cond.notify()
 
         answer_channel = self._get_channel(channel) + "".join(
-            random.choice(string.ascii_uppercase + string.digits) for _ in range(10)  # nosec
+            random.choice(string.ascii_uppercase + string.digits)
+            for _ in range(10)  # nosec
         )
         _LOG.debug("Subscribing for broadcast answers on %s", answer_channel)
         self._pub_sub.subscribe(**{answer_channel: callback})

c2cwsgiutils/db.py

@@ -66,6 +66,7 @@ def setup_session(
         force_slave: The method/paths that needs to use the slave
 
     Returns: The SQLAlchemy session, the R/W engine and the R/O engine
+
     """
     warnings.warn("setup_session function is deprecated; use init and request.dbsession instead")
     if slave_prefix is None:
@@ -122,6 +123,7 @@ def create_session(
         engine_config: The rest of the parameters are passed as is to the sqlalchemy.create_engine function
 
     Returns: The SQLAlchemy session
+
     """
     warnings.warn("create_session function is deprecated; use init and request.dbsession instead")
     if slave_url is None:
@@ -374,6 +376,7 @@ def init(
         force_slave: The method/paths that needs to use the slave
 
     Returns: The SQLAlchemy session
+
     """
     settings = config.get_settings()
     settings["tm.manager_hook"] = "pyramid_tm.explicit_manager"

c2cwsgiutils/debug/__init__.py

@@ -37,6 +37,8 @@ def init_daemon(config: Optional[pyramid.config.Configurator] = None) -> None:
     those requests.
     """
     if config_utils.env_or_config(config, ENV_KEY, CONFIG_KEY, type_=config_utils.config_bool):
-        from c2cwsgiutils.debug import _listeners  # pylint: disable=import-outside-toplevel
+        from c2cwsgiutils.debug import (
+            _listeners,  # pylint: disable=import-outside-toplevel
+        )
 
         _listeners.init()

c2cwsgiutils/errors.py

@@ -9,7 +9,12 @@
 import pyramid.request
 import sqlalchemy.exc
 from cornice import cors
-from pyramid.httpexceptions import HTTPError, HTTPException, HTTPRedirection, HTTPSuccessful
+from pyramid.httpexceptions import (
+    HTTPError,
+    HTTPException,
+    HTTPRedirection,
+    HTTPSuccessful,
+)
 from webob.request import DisconnectionError
 
 from c2cwsgiutils import auth, config_utils

c2cwsgiutils/health_check.py

@@ -235,6 +235,7 @@ def add_db_session_check(
             engine_type: whether to check only the RW, RO or both engines
             rw_engin: the RW engine to use (if None, use the session one)
             ro_engin: the RO engine to use (if None, use the session one)
+
         """
         if query_cb is None:
             query_cb = self._at_least_one(at_least_one_model)
@@ -265,6 +266,7 @@ def add_alembic_check(
             version_table: override the table name for the version
             rw_engin: the RW engine to use (if None, use the session one)
             ro_engin: the RO engine to use (if None, use the session one)
+
         """
         version_ = _get_alembic_version(alembic_ini_path, name)
 
@@ -321,9 +323,8 @@ def add_url_check(
             Mapping[str, str], Callable[[pyramid.request.Request], Mapping[str, str]], None
         ] = None,
         name: Optional[str] = None,
-        check_cb: Callable[
-            [pyramid.request.Request, requests.Response], Any
-        ] = lambda request, response: None,
+        check_cb: Callable[[pyramid.request.Request, requests.Response], Any] = lambda request,
+        response: None,
         timeout: float = 3,
         level: int = 1,
     ) -> None:
@@ -339,6 +340,7 @@ def add_url_check(
                          response as parameters)
             timeout: the timeout
             level: the level of the health check
+
         """
 
         def check(request: pyramid.request.Request) -> Any:
@@ -365,6 +367,7 @@ def add_redis_check(self, name: Optional[str] = None, level: int = 1) -> None:
         Arguments:
             name: the name of the check (defaults to url)
             level: the level of the health check
+
         """
 
         def check(request: pyramid.request.Request) -> Any:
@@ -413,6 +416,7 @@ def add_version_check(self, name: str = "version", level: int = 2) -> None:
         Arguments:
             name: the name of the check (defaults to "version")
             level: the level of the health check
+
         """
 
         def check(request: pyramid.request.Request) -> dict[str, Any]:
@@ -441,6 +445,7 @@ def add_custom_check(
             name: the name of the check
             check_cb: the callback to call (takes the request as parameter)
             level: the level of the health check
+
         """
         assert name
         self._checks.append((name, check_cb, level))
@@ -453,9 +458,8 @@ def _view(self, request: pyramid.request.Request) -> Mapping[str, Any]:
             "successes": {},
         }
         checks = None
-        if "checks" in request.params:
-            if request.params["checks"] != "":
-                checks = request.params["checks"].split(",")
+        if "checks" in request.params and request.params["checks"] != "":
+            checks = request.params["checks"].split(",")
         for name, check, level in self._checks:
             if level <= max_level and (checks is None or name in checks):
                 self._run_one(check, is_auth, level, name, request, results)
@@ -498,11 +502,10 @@ def _create_db_engine_check(
     ) -> tuple[str, Callable[[pyramid.request.Request], None]]:
         def check(request: pyramid.request.Request) -> None:
             del request  # unused
-            with binding as session:
-                with _PROMETHEUS_DB_SUMMARY.labels(
-                    connection=binding.name(), check="database", configuration="<default>"
-                ).time():
-                    return query_cb(session)
+            with binding as session, _PROMETHEUS_DB_SUMMARY.labels(
+                connection=binding.name(), check="database", configuration="<default>"
+            ).time():
+                return query_cb(session)
 
         return "db_engine_" + binding.name(), check
 

c2cwsgiutils/prometheus.py

@@ -159,6 +159,7 @@ def __init__(self, memory_type: str = "pss", pids: Optional[list[str]] = None):
         Arguments:
             memory_type: can be rss, pss or size
             pids: the list of pids or none
+
         """
         super().__init__()
         self.memory_type = memory_type

c2cwsgiutils/pyramid.py

@@ -31,6 +31,7 @@ def includeme(config: pyramid.config.Configurator) -> None:
 
     Arguments:
         config: The pyramid Configuration
+
     """
     logging.captureWarnings(True)
     config.include(coverage_setup.includeme)

c2cwsgiutils/scripts/genversion.py

@@ -24,7 +24,7 @@ def _get_package_version(comp: str) -> tuple[Optional[str], Optional[str]]:
     if matcher:
         return cast(tuple[str, str], matcher.groups())
     else:
-        if len(comp) > 0 and not comp[:3] == "-e ":
+        if len(comp) > 0 and comp[:3] != "-e ":
             print("Cannot parse package version: " + comp)
         return None, None
 

c2cwsgiutils/scripts/test_print.py

@@ -1,5 +1,6 @@
 #!/usr/bin/env python3
 """Test a MapfishPrint server."""
+
 import argparse
 import logging
 import pprint

c2cwsgiutils/sqlalchemylogger/handlers.py

@@ -11,7 +11,10 @@
 from sqlalchemy.orm import sessionmaker
 from sqlalchemy_utils import create_database, database_exists
 
-from c2cwsgiutils.sqlalchemylogger._filters import ContainsExpression, DoesNotContainExpression
+from c2cwsgiutils.sqlalchemylogger._filters import (
+    ContainsExpression,
+    DoesNotContainExpression,
+)
 from c2cwsgiutils.sqlalchemylogger._models import Base, create_log_class
 
 _LOG = logging.getLogger(__name__)
@@ -34,7 +37,7 @@ def __init__(
         self.engine = create_engine(sqlalchemy_url["url"])
         self.Log = create_log_class(  # pylint: disable=invalid-name
             tablename=sqlalchemy_url.get("tablename", "logs"),
-            tableargs=sqlalchemy_url.get("tableargs", None),  # type: ignore
+            tableargs=sqlalchemy_url.get("tableargs"),  # type: ignore
         )
         Base.metadata.bind = self.engine
         self.session = sessionmaker(bind=self.engine)()  # noqa

c2cwsgiutils/stats_pyramid/__init__.py

@@ -20,6 +20,7 @@ def includeme(config: pyramid.config.Configurator) -> None:
 
     Arguments:
         config: The Pyramid config
+
     """
     _pyramid_spy.init(config)
     init_db_spy()

c2cwsgiutils/stats_pyramid/_pyramid_spy.py

@@ -99,6 +99,7 @@ def init(config: pyramid.config.Configurator) -> None:  # pragma: nocover
 
     Arguments:
         config: The Pyramid config
+
     """
     config.add_subscriber(_request_callback, pyramid.events.NewRequest)
     config.add_subscriber(_before_rendered_callback, pyramid.events.BeforeRender)