python/knot_resolver: ruff linter fixes

Author: alesmrazek

python/knot_resolver/client/command.py

@@ -48,7 +48,7 @@ def get_socket_from_config(config: Path, optional_file: bool) -> Optional[Socket
                     f'http+unix://{quote(management["unix-socket"], safe="")}/',
                     f'Key "/management/unix-socket" in "{config}" file',
                 )
-            elif "interface" in management:
+            if "interface" in management:
                 ip = IPAddressPort(management["interface"], object_path=f"/{mkey}/interface")
                 return SocketDesc(
                     f"http://{ip.addr}:{ip.port}",

python/knot_resolver/client/commands/config.py

@@ -17,7 +17,7 @@ class Operations(Enum):
 def operation_to_method(operation: Operations) -> Literal["PUT", "GET", "DELETE"]:
     if operation == Operations.SET:
         return "PUT"
-    elif operation == Operations.DELETE:
+    if operation == Operations.DELETE:
         return "DELETE"
     return "GET"
 
@@ -91,25 +91,25 @@ def register_args_subparser(
         config_subparsers = config.add_subparsers(help="operation type")
 
         # GET operation
-        get = config_subparsers.add_parser("get", help="Get current configuration from the resolver.")
-        get.set_defaults(operation=Operations.GET, format=DataFormat.YAML)
+        get_op = config_subparsers.add_parser("get", help="Get current configuration from the resolver.")
+        get_op.set_defaults(operation=Operations.GET, format=DataFormat.YAML)
 
-        get.add_argument(
+        get_op.add_argument(
             "-p",
             "--path",
             help=path_help,
             action="store",
             type=str,
             default="",
         )
-        get.add_argument(
+        get_op.add_argument(
             "file",
             help="Optional, path to the file where to save exported configuration data. If not specified, data will be printed.",
             type=str,
             nargs="?",
         )
 
-        get_formats = get.add_mutually_exclusive_group()
+        get_formats = get_op.add_mutually_exclusive_group()
         get_formats.add_argument(
             "--json",
             help="Get configuration data in JSON format.",
@@ -126,10 +126,10 @@ def register_args_subparser(
         )
 
         # SET operation
-        set = config_subparsers.add_parser("set", help="Set new configuration for the resolver.")
-        set.set_defaults(operation=Operations.SET)
+        set_op = config_subparsers.add_parser("set", help="Set new configuration for the resolver.")
+        set_op.set_defaults(operation=Operations.SET)
 
-        set.add_argument(
+        set_op.add_argument(
             "-p",
             "--path",
             help=path_help,
@@ -138,7 +138,7 @@ def register_args_subparser(
             default="",
         )
 
-        value_or_file = set.add_mutually_exclusive_group()
+        value_or_file = set_op.add_mutually_exclusive_group()
         value_or_file.add_argument(
             "file",
             help="Optional, path to file with new configuraion.",
@@ -153,11 +153,11 @@ def register_args_subparser(
         )
 
         # DELETE operation
-        delete = config_subparsers.add_parser(
+        delete_op = config_subparsers.add_parser(
             "delete", help="Delete given configuration property or list item at the given index."
         )
-        delete.set_defaults(operation=Operations.DELETE)
-        delete.add_argument(
+        delete_op.set_defaults(operation=Operations.DELETE)
+        delete_op.add_argument(
             "-p",
             "--path",
             help=path_help,

python/knot_resolver/controller/__init__.py

@@ -86,8 +86,7 @@ async def get_controller_by_name(config: KresConfig, name: str) -> SubprocessCon
     if await controller.is_controller_available(config):
         logger.info("Selected controller '%s'", str(controller))
         return controller
-    else:
-        raise LookupError("The selected subprocess controller is not available for use on this system.")
+    raise LookupError("The selected subprocess controller is not available for use on this system.")
 
 
 # run the imports on module load

python/knot_resolver/controller/exceptions.py

@@ -3,11 +3,11 @@
 from knot_resolver import KresBaseException
 
 
-class SubprocessControllerException(KresBaseException):
+class SubprocessControllerError(KresBaseException):
     pass
 
 
-class SubprocessControllerExecException(Exception):
+class SubprocessControllerExecError(Exception):
     """
     Exception that is used to deliberately terminate system startup
     and make exec() of something else. This is used by the subprocess controller

python/knot_resolver/controller/interface.py

@@ -10,7 +10,7 @@
 from typing import Dict, Iterable, Optional, Type, TypeVar
 from weakref import WeakValueDictionary
 
-from knot_resolver.controller.exceptions import SubprocessControllerException
+from knot_resolver.controller.exceptions import SubprocessControllerError
 from knot_resolver.controller.registered_workers import register_worker, unregister_worker
 from knot_resolver.datamodel.config_schema import KresConfig
 from knot_resolver.manager.constants import kresd_config_file, policy_loader_config_file
@@ -47,8 +47,7 @@ def alloc(cls: Type[T], typ: SubprocessType) -> T:
         # find free ID closest to zero
         for i in itertools.count(start=0, step=1):
             if i not in cls._used[typ]:
-                res = cls.new(typ, i)
-                return res
+                return cls.new(typ, i)
 
         raise RuntimeError("Reached an end of an infinite loop. How?")
 
@@ -59,10 +58,9 @@ def new(cls: "Type[T]", typ: SubprocessType, n: int) -> "T":
             # typed based on subclass. I am not even sure that it's different between subclasses,
             # it's probably still the same dict. But we don't really care about it
             return cls._used[typ][n]  # type: ignore
-        else:
-            val = cls(typ, n, _i_know_what_i_am_doing=True)
-            cls._used[typ][n] = val
-            return val
+        val = cls(typ, n, _i_know_what_i_am_doing=True)
+        cls._used[typ][n] = val
+        return val
 
     def __init__(self, typ: SubprocessType, n: int, _i_know_what_i_am_doing: bool = False):
         if not _i_know_what_i_am_doing:
@@ -132,7 +130,7 @@ async def start(self, new_config: Optional[KresConfig] = None) -> None:
             if self.type is SubprocessType.KRESD:
                 register_worker(self)
                 self._registered_worker = True
-        except SubprocessControllerException as e:
+        except SubprocessControllerError as e:
             if config_file:
                 config_file.unlink()
             raise e

python/knot_resolver/controller/registered_workers.py

@@ -2,7 +2,7 @@
 import logging
 from typing import TYPE_CHECKING, Dict, List, Tuple
 
-from .exceptions import SubprocessControllerException
+from .exceptions import SubprocessControllerError
 
 if TYPE_CHECKING:
     from knot_resolver.controller.interface import KresID, Subprocess
@@ -21,7 +21,7 @@ def get_registered_workers_kresids() -> "List[KresID]":
 async def command_single_registered_worker(cmd: str) -> "Tuple[KresID, object]":
     for sub in _REGISTERED_WORKERS.values():
         return sub.id, await sub.command(cmd)
-    raise SubprocessControllerException(
+    raise SubprocessControllerError(
         "Unable to execute the command. There is no kresd worker running to execute the command."
         "Try start/restart the resolver.",
     )

python/knot_resolver/controller/supervisord/__init__.py

@@ -6,7 +6,7 @@
 
 import supervisor.xmlrpc  # type: ignore[import]
 
-from knot_resolver.controller.exceptions import SubprocessControllerException, SubprocessControllerExecException
+from knot_resolver.controller.exceptions import SubprocessControllerError, SubprocessControllerExecError
 from knot_resolver.controller.interface import (
     KresID,
     Subprocess,
@@ -30,14 +30,14 @@ async def _start_supervisord(config: KresConfig) -> None:
     logger.debug("Starting supervisord")
     res = await call(["supervisord", "--configuration", str(supervisord_config_file(config).absolute())])
     if res != 0:
-        raise SubprocessControllerException(f"Supervisord exited with exit code {res}")
+        raise SubprocessControllerError(f"Supervisord exited with exit code {res}")
 
 
 async def _exec_supervisord(config: KresConfig) -> NoReturn:
     logger.debug("Writing supervisord config")
     await write_config_file(config)
     logger.debug("Execing supervisord")
-    raise SubprocessControllerExecException(
+    raise SubprocessControllerExecError(
         [
             str(which.which("supervisord")),
             "supervisord",
@@ -53,7 +53,7 @@ async def _reload_supervisord(config: KresConfig) -> None:
         supervisord = _create_supervisord_proxy(config)
         supervisord.reloadConfig()
     except Fault as e:
-        raise SubprocessControllerException("supervisord reload failed") from e
+        raise SubprocessControllerError("supervisord reload failed") from e
 
 
 @async_in_a_thread
@@ -113,11 +113,10 @@ async def _is_supervisord_running(config: KresConfig) -> bool:
     pid = await _get_supervisord_pid(config)
     if pid is None:
         return False
-    elif not _is_process_runinng(pid):
+    if not _is_process_runinng(pid):
         supervisord_pid_file(config).unlink()
         return False
-    else:
-        return True
+    return True
 
 
 def _create_proxy(config: KresConfig) -> ServerProxy:
@@ -164,7 +163,7 @@ def _list_running_subprocesses(config: KresConfig) -> Dict[SupervisordKresID, Su
         supervisord = _create_supervisord_proxy(config)
         processes: Any = supervisord.getAllProcessInfo()
     except Fault as e:
-        raise SubprocessControllerException(f"failed to get info from all running processes: {e}") from e
+        raise SubprocessControllerError(f"failed to get info from all running processes: {e}") from e
 
     # there will be a manager process as well, but we don't want to report anything on ourselves
     processes = [pr for pr in processes if pr["name"] != "manager"]
@@ -199,7 +198,7 @@ def status(self) -> SubprocessStatus:
             supervisord = _create_supervisord_proxy(self._config)
             status = supervisord.getProcessInfo(self.name)
         except Fault as e:
-            raise SubprocessControllerException(f"failed to get status from '{self.id}' process: {e}") from e
+            raise SubprocessControllerError(f"failed to get status from '{self.id}' process: {e}") from e
         return _convert_subprocess_status(status)
 
     @async_in_a_thread
@@ -210,7 +209,7 @@ def _start(self) -> None:
             supervisord = _create_fast_proxy(self._config)
             supervisord.startProcess(self.name)
         except Fault as e:
-            raise SubprocessControllerException(f"failed to start '{self.id}'") from e
+            raise SubprocessControllerError(f"failed to start '{self.id}'") from e
 
     @async_in_a_thread
     def _stop(self) -> None:
@@ -253,8 +252,7 @@ async def get_all_running_instances(self) -> Iterable[Subprocess]:
                 for id_ in states
                 if states[id_] == SubprocessStatus.RUNNING
             ]
-        else:
-            return []
+        return []
 
     async def initialize_controller(self, config: KresConfig) -> None:
         self._controller_config = config

python/knot_resolver/controller/supervisord/config_file.py

@@ -36,21 +36,19 @@ def from_string(val: str) -> "SupervisordKresID":
         # the double name is checked because thats how we read it from supervisord
         if val in ("cache-gc", "cache-gc:cache-gc"):
             return SupervisordKresID.new(SubprocessType.GC, 0)
-        elif val in ("policy-loader", "policy-loader:policy-loader"):
+        if val in ("policy-loader", "policy-loader:policy-loader"):
             return SupervisordKresID.new(SubprocessType.POLICY_LOADER, 0)
-        else:
-            val = val.replace("kresd:kresd", "")
-            return SupervisordKresID.new(SubprocessType.KRESD, int(val))
+        val = val.replace("kresd:kresd", "")
+        return SupervisordKresID.new(SubprocessType.KRESD, int(val))
 
     def __str__(self) -> str:
         if self.subprocess_type is SubprocessType.GC:
             return "cache-gc"
-        elif self.subprocess_type is SubprocessType.POLICY_LOADER:
+        if self.subprocess_type is SubprocessType.POLICY_LOADER:
             return "policy-loader"
-        elif self.subprocess_type is SubprocessType.KRESD:
+        if self.subprocess_type is SubprocessType.KRESD:
             return f"kresd:kresd{self._id}"
-        else:
-            raise RuntimeError(f"Unexpected subprocess type {self.subprocess_type}")
+        raise RuntimeError(f"Unexpected subprocess type {self.subprocess_type}")
 
 
 def kres_cache_gc_args(config: KresConfig) -> str:

python/knot_resolver/controller/supervisord/plugin/fast_rpcinterface.py

@@ -73,7 +73,7 @@ def _update(self, text):
 
     # RPC API methods
 
-    def _getGroupAndProcess(self, name):
+    def _getGroupAndProcess(self, name):  # noqa: N802
         # get process to start from name
         group_name, process_name = split_namespec(name)
 
@@ -90,7 +90,7 @@ def _getGroupAndProcess(self, name):
 
         return group, process
 
-    def startProcess(self, name, wait=True):
+    def startProcess(self, name, wait=True):  # noqa: N802
         """Start a process
 
         @param string name Process name (or ``group:name``, or ``group:*``)
@@ -108,10 +108,10 @@ def startProcess(self, name, wait=True):
         # eventually fail
         try:
             filename, argv = process.get_execv_args()
-        except NotFound as why:
-            raise RPCError(Faults.NO_FILE, why.args[0])
+        except NotFound as e:
+            raise RPCError(Faults.NO_FILE, e.args[0]) from e
         except (BadCommand, NotExecutable, NoPermission) as why:
-            raise RPCError(Faults.NOT_EXECUTABLE, why.args[0])
+            raise RPCError(Faults.NOT_EXECUTABLE, why.args[0]) from why
 
         if process.get_state() in RUNNING_STATES:
             raise RPCError(Faults.ALREADY_STARTED, name)

python/knot_resolver/controller/supervisord/plugin/manager_integration.py

@@ -51,7 +51,7 @@ def check_for_runnning_manager(event: ProcessStateRunningEvent) -> None:
         systemd_notify(READY="1", STATUS="Ready")
 
 
-def ServerOptions_get_signal(self):
+def get_server_options_signal(self):
     sig = self.signal_receiver.get_signal()
     if sig == signal.SIGHUP and superd is not None:
         superd.options.logger.info("received SIGHUP, forwarding to the process 'manager'")
@@ -77,7 +77,7 @@ def inject(supervisord: Supervisor, **_config: Any) -> Any:  # pylint: disable=u
     subscribe(ProcessStateRunningEvent, check_for_runnning_manager)
 
     # forward SIGHUP to manager
-    ServerOptions.get_signal = ServerOptions_get_signal
+    ServerOptions.get_signal = get_server_options_signal
 
     # this method is called by supervisord when loading the plugin,
     # it should return XML-RPC object, which we don't care about

python/knot_resolver/controller/supervisord/plugin/patch_logger.py

@@ -20,7 +20,7 @@ def empty_function(*args, **kwargs):
 FORWARD_MSG_FORMAT: str = "%(name)s[%(pid)d]%(stream)s: %(data)s"
 
 
-def POutputDispatcher_log(self: POutputDispatcher, data: bytearray):
+def p_output_dispatcher_log(self: POutputDispatcher, data: bytearray):
     if data:
         # parse the input
         if not isinstance(data, bytes):
@@ -75,7 +75,7 @@ def inject(supervisord: Supervisor, **config: Any) -> Any:  # pylint: disable=us
         supervisord.options.logger.handlers = supervisord_handlers
 
         # replace output handler for subprocesses
-        POutputDispatcher._log = POutputDispatcher_log
+        POutputDispatcher._log = p_output_dispatcher_log  # noqa: SLF001
 
         # we forward stdio in all cases, even when logging to syslog. This should prevent the unforturtunate
         # case of swallowing an error message leaving the users confused. To make the forwarded lines obvious