Start reverting Union[] type-hints

Using | for type-hints isn't supported until Python 3.10;
since we require Python >= 3.9 we cannot use them,
so we'll have to revert to using Union[] instead.

To ensure that we don't miss out on more things like this
pass --py-version to pylint and --target-version to ruff.

Signed-off-by: David Weinehall <[email protected]>

Makefile

@@ -69,9 +69,15 @@ test_libs_symlink = clustermanagementtoolkit
 # Hence we we need to ignore one of those warnings.
 FLAKE8_IGNORE := F841,W503
 
+# Used by ruff to check for future and/or deprecated features
+RUFF_PYTHON_VERSION := py39
+
+# Used by pylint to check for future and/or deprecated features
+PYLINT_PYTHON_VERSION := 3.9
+
 # W0511 is TODO/XXX/FIXME; we know that these are things that we should fix eventually.
 # Hence we do not need warnings about them.
-PYLINT_IGNORE := W0511
+PYLINT_DISABLE := W0511
 
 # Warn about useless disable
 PYLINT_ENABLE := useless-suppression
@@ -220,7 +226,7 @@ ruff:
 			continue;; \
 		esac ;\
 		printf -- "File: $$file\n" ;\
-		$$cmd $$file ;\
+		$$cmd check --target-version $(RUFF_PYTHON_VERSION) $$file ;\
 	done
 
 pylint:
@@ -236,7 +242,7 @@ pylint:
 			continue;; \
 		esac ;\
 		printf -- "File: $$file\n" ;\
-		$$cmd --disable $(PYLINT_IGNORE) --enable $(PYLINT_ENABLE) $$file ;\
+		$$cmd --py-version $(PYTHON_VERSION) --disable $(PYLINT_DISABLE) --enable $(PYLINT_ENABLE) $$file ;\
 	done
 
 pylint-markdown:
@@ -251,7 +257,7 @@ pylint-markdown:
 		'cmtlog.py'|'noxfile.py') \
 			continue;; \
 		esac ;\
-		result=$$($$cmd --disable $(PYLINT_IGNORE) $$file | grep "Your code" | sed -e 's/Your code has been rated at //;s/ (previous run.*//') ;\
+		result=$$($$cmd --py-version $(PYTHON_VERSION) --disable $(PYLINT_DISABLE) --enable $(PYLINT_ENABLE) $$file | grep "Your code" | sed -e 's/Your code has been rated at //;s/ (previous run.*//') ;\
 		row="$$file | $$result\n" ;\
 		printf -- "$$row" >> $${tmpfile} ;\
 	done && \

build.py

@@ -1,6 +1,5 @@
 #! /bin/sh
 # vim: ts=4 filetype=python expandtab shiftwidth=4 softtabstop=4 syntax=python
-# pylint: disable-next=anomalous-backslash-in-string
 ''''eval version=$( ls /usr/bin/python3.* | \
     grep '.*[0-9]$' | sort -nr -k2 -t. | head -n1 ) && \
     version=${version##/usr/bin/python3.} && [ ${version} ] && \
@@ -14,7 +13,6 @@
 import os
 from pathlib import Path, PosixPath
 import sys
-import yaml
 
 try:  # pragma: no cover
     from jinja2 import Environment, FileSystemLoader, select_autoescape
@@ -34,16 +32,16 @@ def main() -> None:
 
     # This program should be called with the path to the directory to process .j2 files in
     # as well as a path to the directory that holds the variables to use in substitutions.
-    if not (2 < len(sys.argv) < 5):
-        sys.exit(f"Usage: build.py TEMPLATE_DIRECTORY VARIABLE_DIRECTORY [OUTPUT_DIRECTORY]")
+    if not 2 < len(sys.argv) < 5:
+        sys.exit("Usage: build.py TEMPLATE_DIRECTORY VARIABLE_DIRECTORY [OUTPUT_DIRECTORY]")
     template_path = sys.argv[1]
     variable_path = sys.argv[2]
     if len(sys.argv) > 3:
         output_path = sys.argv[3]
     else:
         output_path = template_path
     if template_path == variable_path:
-        sys.exit(f"TEMPLATE_DIRECTORY cannot be same as VARIABLE_DIRECTORY; aborting.")
+        sys.exit("TEMPLATE_DIRECTORY cannot be same as VARIABLE_DIRECTORY; aborting.")
     template_entry = Path(template_path)
     variable_entry = Path(variable_path)
     output_entry = Path(output_path)
@@ -60,7 +58,6 @@ def main() -> None:
     for filepath in variable_entry.iterdir():
         if not filepath.is_file() or not filepath.name.endswith(".var"):
             continue
-        string: str = ""
         with open(filepath, "r", encoding="utf-8", errors="replace") as f:
             tmp = f.read()
         context[filepath.name.removesuffix(".var")] = tmp[:-1]

clustermanagementtoolkit/commandparser.py

@@ -37,7 +37,7 @@
 # pylint: disable-next=unused-argument
 def __version(options: list[tuple[str, str]], args: list[str]) -> int:
     """
-    Display version information
+    Display version information.
 
         Parameters:
             options ([(str, str)]): Unused
@@ -61,7 +61,7 @@ def __version(options: list[tuple[str, str]], args: list[str]) -> int:
 # pylint: disable-next=too-many-locals,too-many-branches
 def __sub_usage(command: str) -> int:
     """
-    Display usage information for a single command
+    Display usage information for a single command.
 
         Parameters:
             command (str): The command to show help for
@@ -152,7 +152,7 @@ def __sub_usage(command: str) -> int:
 # pylint: disable-next=unused-argument,too-many-locals,too-many-branches,too-many-statements
 def __usage(options: list[tuple[str, str]], args: list[str]) -> int:
     """
-    Display usage information
+    Display usage information.
 
         Parameters:
             options ([(str, str)]): Options to use when executing this action
@@ -421,7 +421,7 @@ def __usage(options: list[tuple[str, str]], args: list[str]) -> int:
 
 def __command_usage(options: list[tuple[str, str]], args: list[str]) -> int:
     """
-    Display usage information for a single command
+    Display usage information for a single command.
 
         Parameters:
             options ([(str, str)]): Unused
@@ -546,7 +546,7 @@ def parse_commandline(__programname: str, __programversion: str,
                                                                  list[tuple[str, str]],
                                                                  list[str]]:
     """
-    Parse the command line
+    Parse the command line.
 
         Parameters:
             __programname (str): The name of the program

clustermanagementtoolkit/listgetters.py

@@ -2291,10 +2291,10 @@ def listgetter_path(obj: dict, **kwargs: Any) -> tuple[dict | list[dict], int]:
                 if isinstance(item, dict):
                     if enumeration == "standard":
                         deep_set(item, DictPath("_extra_data#enumeration"), i,
-                                 create_path = True)
+                                 create_path=True)
                     elif enumeration == "reverse":
                         deep_set(item, DictPath("_extra_data#enumeration"), len(tmp2) - i,
-                                 create_path = True)
+                                 create_path=True)
                     vlist.append(item)
         else:
             vlist = tmp2

cmt-install

@@ -22,7 +22,7 @@ import re
 import subprocess  # nosec
 from subprocess import PIPE, STDOUT  # nosec
 import sys
-from typing import Any
+from typing import Any, Union
 
 from clustermanagementtoolkit.cmttypes import deep_get, DictPath, FilePath, SecurityPolicy
 
@@ -292,7 +292,7 @@ def copy_files(files: list[tuple[FilePath, FilePath, int]], verbose: bool = Fals
 
 
 # pylint: disable-next=unused-argument
-def install_software_suse(packages: dict | set, verbose: bool = False) -> None:
+def install_software_suse(packages: Union[dict, set], verbose: bool = False) -> None:
     """
     Batch install SUSE packages
 
@@ -363,7 +363,7 @@ def install_software_suse(packages: dict | set, verbose: bool = False) -> None:
 
 
 # pylint: disable-next=unused-argument,too-many-locals
-def install_software_fedora(packages: dict | set, verbose: bool = False) -> None:
+def install_software_fedora(packages: Union[dict, set], verbose: bool = False) -> None:
     """
     Batch install Red Hat (Fedora) packages
 

cmtadm

@@ -30,7 +30,7 @@ import socket
 from subprocess import CalledProcessError  # nosec
 import sys
 import tempfile
-from typing import Any, cast, Optional
+from typing import Any, cast, Optional, Union
 from collections.abc import Callable
 
 try:
@@ -2958,7 +2958,7 @@ def __task_run_postpurge_playbooks(installation_info: dict, verbose: bool = Fals
 
 
 def __validate_task_index(tasks: list[tuple[list[ANSIThemeStr], Callable[..., None]]],
-                          phase: str | int) -> int:
+                          phase: Union[str, int]) -> int:
     """
     Helper that validates the installer task index.
 
@@ -3330,7 +3330,7 @@ def prepare_installation(options: list[tuple[str, Any]], args: list[str]) -> Non
     os_distro = identify_distro()
     requested_version = installation_info[cluster_name]["requested_version"]
     state = installation_info[cluster_name]["state"]
-    phase: str | int = installation_info[cluster_name]["phase"]
+    phase: Union[str, int] = installation_info[cluster_name]["phase"]
     phase_skiplist = set(installation_info[cluster_name]["phase_skiplist"])
     verbose = False
     hide_next_step = False
@@ -5330,7 +5330,7 @@ def setup_control_planes(options: list[tuple[str, Any]], args: list[str]) -> Non
 
     requested_version = installation_info[cluster_name]["requested_version"]
     state = installation_info[cluster_name]["state"]
-    phase: str | int = installation_info[cluster_name]["phase"]
+    phase: Union[str, int] = installation_info[cluster_name]["phase"]
     phase_skiplist = set(installation_info[cluster_name]["phase_skiplist"])
     cni = installation_info[cluster_name]["cni"]
     pod_network_cidr = installation_info[cluster_name]["pod_network_cidr"]
@@ -5837,7 +5837,7 @@ def teardown_control_plane(options: list[tuple[str, str]], args: list[str]) -> N
     cluster_name = deep_get(installation_info, DictPath("installation_target"))
     k8s_distro = deep_get(installation_info, DictPath(f"{cluster_name}#distro"))
     state = deep_get(installation_info, DictPath(f"{cluster_name}#state"))
-    phase: str | int = deep_get(installation_info, DictPath(f"{cluster_name}#phase"))
+    phase: Union[str, int] = deep_get(installation_info, DictPath(f"{cluster_name}#phase"))
     phase_skiplist = set(deep_get(installation_info, DictPath(f"{cluster_name}#phase_skiplist")))
     verbose = False
 
@@ -5939,7 +5939,7 @@ def purge_control_plane(options: list[tuple[str, str]], args: list[str]) -> None
     installation_info = get_installation_info()
     cluster_name = installation_info["installation_target"]
     state = installation_info[cluster_name]["state"]
-    phase: str | int = installation_info[cluster_name]["phase"]
+    phase: Union[str, int] = installation_info[cluster_name]["phase"]
     phase_skiplist = set(installation_info[cluster_name]["phase_skiplist"])
     verbose = False
 
@@ -6075,7 +6075,7 @@ def upgrade_control_plane(options: list[tuple[str, str]], args: list[str]) -> No
     version = deep_get(installation_info, DictPath(f"{cluster_name}#version"))
     requested_version = deep_get(installation_info, DictPath(f"{cluster_name}#requested_version"))
     state = deep_get(installation_info, DictPath(f"{cluster_name}#state"))
-    phase: str | int = deep_get(installation_info, DictPath(f"{cluster_name}#phase"))
+    phase: Union[str, int] = deep_get(installation_info, DictPath(f"{cluster_name}#phase"))
     phase_skiplist = set(deep_get(installation_info, DictPath(f"{cluster_name}#phase_skiplist")))
     verbose = False
     ignore_feature_gates = False

cmtinv

@@ -22,7 +22,7 @@ import os
 from pathlib import Path
 import re
 import sys
-from typing import Any, cast, Optional
+from typing import Any, cast, Optional, Union
 try:
     import yaml
 except ModuleNotFoundError:  # pragma: no cover
@@ -93,7 +93,7 @@ def set_host_vars(options: list[tuple[str, str]], args: list[str]) -> int:
         Returns:
             (int): 0
     """
-    hostvars: list[tuple[str, str | int]] = []
+    hostvars: list[tuple[str, Union[str, int]]] = []
     keyvals: list[str] = args[0].split(",")
     hosts: list[str] = args[1].split(",")
 

cmu

@@ -47,7 +47,7 @@ import socket
 import subprocess  # nosec
 from subprocess import PIPE, STDOUT  # nosec
 import sys
-from typing import Any, cast, Optional, Type
+from typing import Any, cast, Optional, Type, Union
 from collections.abc import Callable, Sequence
 try:
     import yaml
@@ -621,7 +621,7 @@ def __process_string(value: str, replace_quotes: str) -> str:
     return value
 
 
-def __process_sum_numerical(value: Sequence[int | float]) -> float | int:
+def __process_sum_numerical(value: Sequence[Union[int, float]]) -> Union[float, int]:
     return sum(value)
 
 
@@ -646,8 +646,8 @@ def __process_sum_mem_usage(values: list[str]) -> str:
     return normalise_mem_bytes_to_str(mem_usage_sum)
 
 
-def __process_timestamp(value: Sequence[int | str] | str,
-                        action: str, formatter: str) -> datetime | int:
+def __process_timestamp(value: Union[Sequence[Union[int, str]], str],
+                        action: str, formatter: str) -> Union[datetime, int]:
     new_value: Any = None
 
     if action in ("earliest", "latest") and isinstance(value, (list, tuple)):
@@ -682,7 +682,7 @@ def __process_timestamp(value: Sequence[int | str] | str,
 
 # pylint: disable-next=too-many-locals,too-many-branches,too-many-statements
 def process_value(value: Any, vtype: str, **kwargs: Any) -> \
-        int | float | str | list[str] | list[tuple[str]] | datetime | None:
+        Union[int, float, str, list[str], list[tuple[str]], datetime, None]:
     """
     Reformat values; returns data in a format suitable for further processing.
 
@@ -713,7 +713,7 @@ def process_value(value: Any, vtype: str, **kwargs: Any) -> \
                                 DictPath(f"Global#kind_format_{field_index}"),
                                 DictPath("Global#kind_format")], "mixed")
 
-    new_value: int | float | str | list[str] | datetime | None = None
+    new_value: Union[int, float, str, list[str], datetime, None] = None
 
     if vtype == "str":
         new_value = __process_string(value, replace_quotes)
@@ -745,7 +745,7 @@ def process_value(value: Any, vtype: str, **kwargs: Any) -> \
         if value is None:
             new_value = "0"
         else:
-            new_value = str(len(cast(str | Sequence, value)))
+            new_value = str(len(cast(Union[str, Sequence], value)))
     elif vtype == "unix_timestamp":
         new_value = datetime.fromtimestamp(value)
     elif vtype == "timestamp":
@@ -952,7 +952,7 @@ def when_filter(when_path: dict, item: dict, key: Optional[str] = None, value: A
 
 
 # pylint: disable-next=too-many-locals,too-many-branches,too-many-statements
-def transform_list(vlist: list | dict, transform: dict) -> list[Any]:
+def transform_list(vlist: Union[list, dict], transform: dict) -> list[Any]:
     """
     Given data as a list or dict, modify the data according the rules
     in a transformation ruleset.
@@ -2415,7 +2415,7 @@ def generate_list_header(uip: UIProps, field_dict: dict, is_taggable: bool = Fal
     uip.tabstops = tabstops
 
 
-# pylint: disable-next=too-many-arguments,too-many-locals
+# pylint: disable-next=too-many-arguments,too-many-locals,too-many-positional-arguments
 def generate_list_row(uip: UIProps, data: Type, field_list: dict,
                       ypos: int, is_selected: bool, is_taggable: bool = False,
                       is_tagged: bool = False, is_deleted: bool = False) -> None:
@@ -3137,7 +3137,7 @@ def genericlistloop(stdscr: curses.window, **kwargs: Any) -> Retval:
 
             all_ns = "<All>"
             if selected_namespace == "":
-                preselection: str | set = all_ns
+                preselection: Union[str, set] = all_ns
             else:
                 preselection = selected_namespace
             namespace_list = [{
@@ -5139,7 +5139,7 @@ def export_data(stdscr: curses.window, **kwargs: Any) -> Retval:
 
 
 # pylint: disable-next=too-many-branches,too-many-statements
-def generate_helptext(view: str | tuple[str, str], viewtype: str,
+def generate_helptext(view: Union[str, tuple[str, str]], viewtype: str,
                       additional_helptexts: list[tuple[str, str]],
                       shortcuts: dict[str, dict[str, Any]]) -> list[dict]:
     """
@@ -5981,7 +5981,7 @@ def genericinfoloop(stdscr: curses.window, **kwargs: Any) -> Retval:
                                                   DictPath("listpad#on_activation#formatter_path"))
                         on_activation_args["formatter"] = formatter
                         on_activation_args["formatter_path"] = formatter_path
-                    retval: Retval | None = None
+                    retval: Union[Retval, None] = None
                     if view == ("ConfigMap", "") and callable(uip.activatedfun):
                         retval = uip.activatedfun(uip.stdscr,
                                                   obj=deep_get(obj, DictPath(f"data#{match}"), ""),
@@ -6433,7 +6433,7 @@ def eventdispatch(stdscr: curses.window, **kwargs: Any) -> Retval:
     return resourceinfodispatch(stdscr, obj=ref, kind=kind)
 
 
-# pylint: disable-next=too-many-arguments
+# pylint: disable-next=too-many-arguments,too-many-positional-arguments
 def log_add_line(timestamps: list[datetime], facilities: list[str],
                  severities: list[LogLevel], messages: list[list[ThemeRef | ThemeStr] | str],
                  timestamp: Optional[datetime], facility: str,
@@ -7700,7 +7700,7 @@ def do_command(stdscr: curses.window, **kwargs: Any) -> None:
     stdscr.refresh()
 
 
-# pylint: disable-next=too-many-arguments,too-many-locals,too-many-branches
+# noqa: E501 pylint: disable-next=too-many-arguments,too-many-locals,too-many-branches,too-many-positional-arguments
 def executecommand(stdscr: curses.window,
                    obj: dict, container: str, msg: list[ANSIThemeStr],
                    command: list[str], waitforkeypress: bool) -> None:
@@ -7892,7 +7892,7 @@ def resourceinfodispatch(stdscr: curses.window, **kwargs: Any) -> Retval:
             (Retval): Retval.NOMATCH if no match was found or a container had invalid status,
                       otherwise Retval from the involved object whenever that returns.
     """
-    kind: str | tuple[str, str] = deep_get(kwargs, DictPath("kind"), ("", ""))
+    kind: Union[str, tuple[str, str]] = deep_get(kwargs, DictPath("kind"), ("", ""))
     obj: dict = deep_get(kwargs, DictPath("obj"))
     info = deep_get(kwargs, DictPath("info"))
 
@@ -10315,6 +10315,9 @@ def map_key(view_file: str, shortcut: str, activatedfun: Optional[Callable],
             num = int(key[1:]) + 12
             shortcut_key = deep_get(key_mappings, DictPath(key_))
             help_key = f"[Shift] + [{key.upper()}] / [{key[0].upper()}{num}]"
+        else:
+            sys.exit(f"View-file {view_file} is invalid: "
+                     f"the modifier {modifier}; cannot be combined with {shortcut_key}; aborting")
     elif key in ("f13", "f14", "f15", "f16", "f17", "f18",
                  "f19", "f20", "f21", "f22", "f23", "f24") and not modifier:
         num = int(key[1:]) - 12
@@ -11278,7 +11281,7 @@ def list_views(options: list[tuple[str, str]], args: list[str]) -> None:
                                  "of the field names.", "default")])
 
 
-def checkforview(arg: str | tuple[str, str]) -> Optional[str]:
+def checkforview(arg: Union[str, tuple[str, str]]) -> Optional[str]:
     """
     Check whether a view exists.