Fix new Prospector issues

.pre-commit-config.yaml

@@ -65,7 +65,7 @@ repos:
         args:
           - --line-length=110
   - repo: https://github.com/PyCQA/prospector
-    rev: v1.15.0
+    rev: v1.15.1
     hooks:
       - id: prospector
         args:
@@ -75,15 +75,15 @@ repos:
           - --profile=app/.prospector.yaml
         additional_dependencies:
           - prospector-profile-duplicated==1.10.4 # pypi
-          - prospector-profile-utils==1.21.4 # pypi
+          - prospector-profile-utils==1.21.6 # pypi
       - id: prospector
         args:
           - --die-on-tool-error
           - --output-format=pylint
           - --profile=utils:tests
           - --profile=utils:pre-commit
         additional_dependencies:
-          - prospector-profile-utils==1.21.4 # pypi
+          - prospector-profile-utils==1.21.6 # pypi
   - repo: https://github.com/sbrunner/jsonschema-validator
     rev: 1.0.0
     hooks:

app/shared_config_manager/security.py

@@ -27,13 +27,14 @@ class User:
     def __init__(
         self,
         auth_type: str,
-        login: str | None,
-        name: str | None,
-        url: str | None,
-        is_auth: bool,
-        token: str | None,
-        request: pyramid.request.Request,
+        login: str | None = None,
+        name: str | None = None,
+        url: str | None = None,
+        is_auth: bool = False,
+        token: str | None = None,
+        request: pyramid.request.Request | None = None,
     ) -> None:
+        assert request is not None
         self.auth_type = auth_type
         self.login = login
         self.name = name
@@ -82,7 +83,7 @@ def identity(self, request: pyramid.request.Request) -> User:
                     our_signature,
                     request.headers["X-Hub-Signature-256"].split("=", 1)[1],
                 ):
-                    user = User("github_webhook", None, None, None, True, None, request)
+                    user = User("github_webhook", is_auth=True, request=request)
                 else:
                     _LOG.warning("Invalid GitHub signature")
                     _LOG.debug(
@@ -101,7 +102,7 @@ def identity(self, request: pyramid.request.Request) -> User:
 
             elif "X-Scm-Secret" in request.headers and "SCM_SECRET" in os.environ:
                 if request.headers["X-Scm-Secret"] == os.environ["SCM_SECRET"]:
-                    user = User("scm_internal", None, None, None, True, None, request)
+                    user = User("scm_internal", is_auth=True, request=request)
                 else:
                     _LOG.warning("Invalid SCM secret")
 
@@ -120,7 +121,7 @@ def identity(self, request: pyramid.request.Request) -> User:
 
             request.user = user
 
-        return request.user  # type: ignore
+        return request.user  # type: ignore[misc]
 
     def authenticated_userid(self, request: pyramid.request.Request) -> str | None:
         """Return a string ID for the user."""

app/shared_config_manager/services.py

@@ -25,24 +25,27 @@
 __BRANCH_NAME_SANITIZER = re.compile(r"[^0-9a-zA-z-_]")
 
 
-@_refresh_service.get()  # type: ignore
+@_refresh_service.get()  # type: ignore[misc]
 def _refresh_view(request: pyramid.request.Request) -> dict[str, Any]:
     id_ = request.matchdict["id"]
     source, _ = registry.get_source_check_auth(id_=id_, request=request)
     if source is None:
-        raise HTTPNotFound(f"Unknown id {id_}")
+        message = f"Unknown id {id_}"
+        raise HTTPNotFound(message)
     return _refresh(request)
 
 
-@_refresh_service.post()  # type: ignore
+@_refresh_service.post()  # type: ignore[misc]
 def _refresh_webhook(request: pyramid.request.Request) -> dict[str, Any]:
     id_ = request.matchdict["id"]
     source, _ = registry.get_source_check_auth(id_=id_, request=request)
     if source is None:
-        raise HTTPNotFound(f"Unknown id {id_}")
+        message = f"Unknown id {id_}"
+        raise HTTPNotFound(message)
 
     if source.get_type() != "git":
-        raise HTTPServerError(f"Non GIT source {id_} cannot be refreshed by a webhook")
+        message = f"Non GIT source {id_} cannot be refreshed by a webhook"
+        raise HTTPServerError(message)
 
     source_git = cast(git.GitSource, source)
 
@@ -52,7 +55,8 @@ def _refresh_webhook(request: pyramid.request.Request) -> dict[str, Any]:
 
     ref = request.json.get("ref")
     if ref is None:
-        raise HTTPServerError(f"Webhook for {id_} is missing the ref")
+        message = "Webhook for {id_} is missing the ref"
+        raise HTTPServerError(message)
     if ref != "refs/heads/" + source_git.get_branch():
         _LOG.info(
             "Ignoring webhook notif for non-matching branch %s on %s",
@@ -70,10 +74,11 @@ def _refresh(request: pyramid.request.Request) -> dict[str, Any]:
     return {"status": 200}
 
 
-@_refresh_all_service.get()  # type: ignore
+@_refresh_all_service.get()  # type: ignore[misc]
 def _refresh_all(request: pyramid.request.Request) -> dict[str, Any]:
     if not registry.MASTER_SOURCE:
-        raise HTTPServerError("Master source not initialized")
+        message = "Master source not initialized"
+        raise HTTPServerError(message)
     registry.MASTER_SOURCE.validate_auth(request)
     nb_refresh = 0
     for id_ in registry.get_sources():
@@ -82,10 +87,11 @@ def _refresh_all(request: pyramid.request.Request) -> dict[str, Any]:
     return {"status": 200, "nb_refresh": nb_refresh}
 
 
-@_refresh_all_service.post()  # type: ignore
+@_refresh_all_service.post()  # type: ignore[misc]
 def _refresh_all_webhook(request: pyramid.request.Request) -> dict[str, Any]:
     if not registry.MASTER_SOURCE:
-        raise HTTPServerError("Master source not initialized")
+        message = "Master source not initialized"
+        raise HTTPServerError(message)
     registry.MASTER_SOURCE.validate_auth(request=request)
 
     if request.headers.get("X-GitHub-Event") != "push":
@@ -94,7 +100,8 @@ def _refresh_all_webhook(request: pyramid.request.Request) -> dict[str, Any]:
 
     ref = request.json.get("ref")
     if ref is None:
-        raise HTTPServerError("Webhook is missing the ref")
+        message = "Webhook is missing the ref"
+        raise HTTPServerError(message)
 
     nb_refresh = 0
     for id_, source in registry.get_sources().items():
@@ -117,7 +124,7 @@ def _refresh_all_webhook(request: pyramid.request.Request) -> dict[str, Any]:
     return {"status": 200, "nb_refresh": nb_refresh}
 
 
-@_status_service.get()  # type: ignore
+@_status_service.get()  # type: ignore[misc]
 def _stats(request: pyramid.request.Request) -> dict[str, Any]:
     if not registry.MASTER_SOURCE:
         return {"slaves": {}}
@@ -128,12 +135,13 @@ def _stats(request: pyramid.request.Request) -> dict[str, Any]:
     return {"slaves": slaves}
 
 
-@_source_stats_service.get()  # type: ignore
+@_source_stats_service.get()  # type: ignore[misc]
 def _source_stats(request: pyramid.request.Request) -> dict[str, Any]:
     id_ = request.matchdict["id"]
     source, _ = registry.get_source_check_auth(id_=id_, request=request)
     if source is None:
-        raise HTTPNotFound(f"Unknown id {id_}")
+        message = f"Unknown id {id_}"
+        raise HTTPNotFound(message)
     slaves: list[SourceStatus] | None = slave_status.get_source_status(id_=id_)
     assert slaves is not None
     statuses: list[SourceStatus] = []
@@ -154,20 +162,23 @@ def _cleanup_slave_status(status: BroadcastObject) -> BroadcastObject:
     return result
 
 
-@_tarball_service.get()  # type: ignore
+@_tarball_service.get()  # type: ignore[m]
 def _tarball(request: pyramid.request.Request) -> pyramid.response.Response:
     id_ = request.matchdict["id"]
     source, filtered = registry.get_source_check_auth(id_=id_, request=request)
     if source is None:
-        raise HTTPNotFound(f"Unknown id {id_}")
+        message = f"Unknown id {id_}"
+        raise HTTPNotFound(message)
     if not source.is_loaded():
-        raise HTTPNotFound("Not loaded yet")
+        message = "Not loaded yet"
+        raise HTTPNotFound(message)
     assert not filtered
     path = source.get_path()
 
-    if not os.path.isdir(path):
+    if not path.is_dir():
         _LOG.error("The path %s does not exists or is not a path, for the source %s.", path, source.get_id())
-        raise HTTPNotFound("Not loaded yet: path didn't exists")
+        message = "Not loaded yet: path didn't exists"
+        raise HTTPNotFound(message)
 
     response: pyramid.response.Response = request.response
 
@@ -190,9 +201,10 @@ def _tarball(request: pyramid.request.Request) -> pyramid.response.Response:
 
 def _proc_iter(proc: subprocess.Popen[bytes]) -> Iterable[bytes | Any]:
     while True:
-        block = proc.stdout.read(4096)  # type: ignore
+        block = proc.stdout.read(4096)  # type: ignore[m]
         if not block:
             break
         yield block
     if proc.wait() != 0:
-        raise HTTPServerError("Error building the tarball")
+        message = "Error building the tarball"
+        raise HTTPServerError(message)

app/shared_config_manager/sources/base.py

@@ -1,10 +1,10 @@
 import copy
 import logging
 import os
-import pathlib
 import shutil
 import subprocess  # nosec
 import time
+from pathlib import Path
 from typing import Any, cast
 
 import pyramid.request
@@ -19,8 +19,8 @@
 from shared_config_manager.sources import mode
 
 _LOG = logging.getLogger(__name__)
-_TARGET = os.environ.get("TARGET", "/config")
-_MASTER_TARGET = os.environ.get("MASTER_TARGET", "/master_config")
+_TARGET = Path(os.environ.get("TARGET", "/config"))
+_MASTER_TARGET = Path(os.environ.get("MASTER_TARGET", "/master_config"))
 _RETRY_NUMBER = int(os.environ.get("SCM_RETRY_NUMBER", 3))
 _RETRY_DELAY = int(os.environ.get("SCM_RETRY_DELAY", 1))
 
@@ -99,7 +99,7 @@ def _eval_templates(self) -> None:
         # all the files that are created by template engines (see the --delete rsync flag in
         # BaseSource._copy).
         root_dir = self.get_path()
-        files = [os.path.relpath(str(p), root_dir) for p in pathlib.Path(root_dir).glob("**/*")]
+        files = [os.path.relpath(str(p), root_dir) for p in root_dir.glob("**/*")]
 
         for engine in self._template_engines:
             with _TEMPLATE_SUMMARY.labels(self.get_id(), engine.get_type()).time():
@@ -134,9 +134,9 @@ def _do_fetch(self) -> None:
                 _LOG.info("Doing a fetch of %s", self.get_id())
                 response = requests.get(url, headers={"X-Scm-Secret": os.environ["SCM_SECRET"]}, stream=True)
                 response.raise_for_status()
-                if os.path.exists(path):
+                if path.exists():
                     shutil.rmtree(path)
-                os.makedirs(path, exist_ok=True)
+                path.mkdir(parent=True, exist_ok=True)
                 with subprocess.Popen(  # nosec
                     [
                         "tar",
@@ -150,11 +150,10 @@ def _do_fetch(self) -> None:
                     cwd=path,
                     stdin=subprocess.PIPE,
                 ) as tar:
-                    shutil.copyfileobj(response.raw, tar.stdin)  # type: ignore
-                    tar.stdin.close()  # type: ignore
+                    shutil.copyfileobj(response.raw, tar.stdin)  # type: ignore[m]
+                    tar.stdin.close()  # type: ignore[m]
                     assert tar.wait() == 0
-                return
-            except Exception as exception:  # pylint: disable=broad-exception-caught
+            except Exception as exception:  # pylint: disable=broad-exception-caught # noqa: PERF203
                 _DO_FETCH_ERROR_COUNTER.labels(self.get_id()).inc()
                 retry_message = f" (will retry in {_RETRY_DELAY}s)" if i else " (failed)"
                 _LOG.warning(
@@ -167,9 +166,11 @@ def _do_fetch(self) -> None:
                     time.sleep(_RETRY_DELAY)
                 else:
                     raise
+            else:
+                return
 
     def _copy(self, source: str, excludes: list[str] | None = None) -> None:
-        os.makedirs(self.get_path(), exist_ok=True)
+        self.get_path().mkdir(parents=True, exist_ok=True)
         cmd = [
             "rsync",
             "--recursive",
@@ -184,31 +185,32 @@ def _copy(self, source: str, excludes: list[str] | None = None) -> None:
             cmd += ["--exclude=" + exclude for exclude in excludes]
         if "excludes" in self._config:
             cmd += ["--exclude=" + exclude for exclude in self._config["excludes"]]
-        cmd += [source + "/", self.get_path()]
+        cmd += [source + "/", str(self.get_path())]
         with _COPY_SUMMARY.labels(self.get_id()).time():
             self._exec(*cmd)
 
     def delete_target_dir(self) -> None:
         dest = self.get_path()
         _LOG.info("Deleting target dir %s", dest)
-        if os.path.isdir(dest):
+        if dest.is_dir():
             shutil.rmtree(dest)
 
-    def get_path(self) -> str:
+    def get_path(self) -> Path:
         if "target_dir" in self._config:
             target_dir = self._config["target_dir"]
             if target_dir.startswith("/"):
-                return target_dir
-            return _MASTER_TARGET if self._is_master else os.path.join(_TARGET, target_dir)
-        return _MASTER_TARGET if self._is_master else os.path.join(_TARGET, self.get_id())
+                return Path(target_dir)
+            return _MASTER_TARGET if self._is_master else _TARGET / target_dir
+        return _MASTER_TARGET if self._is_master else _TARGET / self.get_id()
 
     def get_id(self) -> str:
         return self._id
 
     def validate_auth(self, request: pyramid.request.Request) -> None:
         permission = request.has_permission("all", self.get_config())
         if not isinstance(permission, Allowed):
-            raise HTTPForbidden("Not allowed to access this source")
+            message = "Not allowed to access this source"
+            raise HTTPForbidden(message)
 
     def is_master(self) -> bool:
         return self._is_master
@@ -255,10 +257,11 @@ def _exec(*args: Any, **kwargs: Any) -> str:
             )
             if output:
                 _LOG.debug(output)
-            return output
         except subprocess.CalledProcessError as exception:
             _LOG.error(exception.output.decode("utf-8").strip())
             raise
+        else:
+            return output
 
     def is_loaded(self) -> bool:
         return self._is_loaded