scripts: add type hints

Signed-off-by: Gaëtan Lehmann <gaetan.lehmann@vates.tech>

Author: glehmann

scripts/install_xcpng.py

@@ -1,5 +1,6 @@
-#!/usr/bin/env python3
+from __future__ import annotations
 
+#!/usr/bin/env python3
 import argparse
 import atexit
 import logging
@@ -25,7 +26,8 @@
 
 logging.basicConfig(format='[%(levelname)s] %(message)s', level=logging.INFO)
 
-def generate_answerfile(directory, installer, hostname_or_ip, target_hostname, action, hdd, netinstall_gpg_check):
+def generate_answerfile(directory: str, installer: str, hostname_or_ip: str, target_hostname: str | None, action: str,
+                        hdd: str, netinstall_gpg_check: str) -> None:
     password = host_data(hostname_or_ip)['password']
     cmd = ['openssl', 'passwd', '-6', password]
     res = subprocess.run(cmd, stdout=subprocess.PIPE)
@@ -69,33 +71,33 @@ def generate_answerfile(directory, installer, hostname_or_ip, target_hostname, a
         else:
             raise Exception(f"Unknown action: `{action}`")
 
-def is_ip_active(ip):
+def is_ip_active(ip: str) -> bool:
     return not os.system(f"ping -c 3 -W 10 {ip} > /dev/null 2>&1")
 
-def is_ssh_up(ip):
+def is_ssh_up(ip: str) -> bool:
     try:
         ssh(ip, 'true', options=['-o', 'ConnectTimeout 10'])
         return True
     except SSHCommandFailed:
         # probably not up yet
         return False
 
-def get_new_host_ip(mac_address):
+def get_new_host_ip(mac_address: str) -> str | None:
     candidate_ips = pxe.arp_addresses_for(mac_address)
     logging.debug("Candidate IPs: " + ", ".join(candidate_ips))
     for ip in candidate_ips:
         if is_ip_active(ip) and is_ssh_up(ip):
             return ip
     return None
 
-def is_new_host_ready(ip_address):
+def is_new_host_ready(ip_address: str) -> bool:
     try:
         output = ssh(ip_address, 'xe host-list enabled=true --minimal')
         return is_uuid(output)
     except Exception:
         return False
 
-def check_mac_address(host: Host, mac_address):
+def check_mac_address(host: Host, mac_address: str) -> None:
     bridge = host.inventory['MANAGEMENT_INTERFACE']
     host_mac_address = host.ssh(f'cat /sys/class/net/{bridge}/address')
     if mac_address != host_mac_address:
@@ -104,15 +106,15 @@ def check_mac_address(host: Host, mac_address):
             f"Expected: `{mac_address}`"
         )
 
-def url_checker(url):
+def url_checker(url: str) -> None:
     try:
         response = requests.get(url)
         if not response:
             raise Exception(f"{url}: URL is not reachable, status_code: {response.status_code}")
     except requests.exceptions.RequestException as e:
         raise SystemExit(f"{url}: URL is not reachable\nErr: {e}")
 
-def main():
+def main() -> None:
     parser = argparse.ArgumentParser()
     parser.add_argument(
         "host",

scripts/xcpng-fs-diff.py

@@ -47,6 +47,9 @@
 import tempfile
 from enum import StrEnum, auto
 from fnmatch import fnmatch
+from subprocess import CompletedProcess
+
+from typing import Any, cast
 
 class DataType(StrEnum):
     FILE = auto()
@@ -55,10 +58,10 @@ class DataType(StrEnum):
     BROKEN_SYMLINK = auto()
     PACKAGE = auto()
 
-def ignore_file(filename, ignored_files):
+def ignore_file(filename: str, ignored_files: list[str]) -> bool:
     return any(fnmatch(filename, i) for i in ignored_files)
 
-def ssh_cmd(host, cmd):
+def ssh_cmd(host: str, cmd: str) -> str:
     args = ["ssh", f"root@{host}", cmd]
 
     cmdres = subprocess.run(args, capture_output=True, text=True)
@@ -67,10 +70,10 @@ def ssh_cmd(host, cmd):
 
     return cmdres.stdout
 
-def ssh_get_files(host, file_type, folders):
+def ssh_get_files(host: str, file_type: DataType, folders: list[str]) -> dict[str, str] | None:
     md5sum = False
     readlink = False
-    folders = " ".join(folders)
+    folders_concatenated = " ".join(folders)
 
     match file_type:
         case DataType.FILE:
@@ -89,7 +92,7 @@ def ssh_get_files(host, file_type, folders):
             print("Unknown file type: ", file=sys.stderr)
             return None
 
-    find_cmd = f"find {folders} {find_type}"
+    find_cmd = f"find {folders_concatenated} {find_type}"
     if readlink:
         find_cmd += " -exec readlink -n {} \\; -exec echo -n '  ' \\; -print"
     elif md5sum:
@@ -99,14 +102,14 @@ def ssh_get_files(host, file_type, folders):
 
     rawres = ssh_cmd(host, find_cmd)
 
-    res = dict()
+    res: dict[str, str] = dict()
     for line in rawres.splitlines():
         entry = line.split(' ', 1)
         res[entry[1].strip()] = entry[0].strip()
 
     return res
 
-def ssh_get_packages(host):
+def ssh_get_packages(host: str) -> dict[str, str]:
     packages = dict()
 
     res = ssh_cmd(host, "rpm -qa --queryformat '%{NAME} %{VERSION}\n'")
@@ -116,8 +119,8 @@ def ssh_get_packages(host):
 
     return packages
 
-def get_data(host, folders):
-    ref_data = dict()
+def get_data(host: str, folders: list[str]) -> dict[DataType, dict[str, str] | None]:
+    ref_data: dict[DataType, dict[str, str] | None] = dict()
 
     try:
         ref_data[DataType.FILE] = ssh_get_files(host, DataType.FILE, folders)
@@ -131,7 +134,7 @@ def get_data(host, folders):
 
     return ref_data
 
-def sftp_get(host, remote_file, local_file):
+def sftp_get(host: str, remote_file: str, local_file: str) -> CompletedProcess[bytes]:
     opts = '-o "StrictHostKeyChecking no" -o "LogLevel ERROR" -o "UserKnownHostsFile /dev/null"'
 
     args = f"sftp {opts} -b - root@{host}"
@@ -150,7 +153,7 @@ def sftp_get(host, remote_file, local_file):
 
     return res
 
-def remote_diff(host_ref, host_test, filename):
+def remote_diff(host_ref: str, host_test: str, filename: str) -> None:
     file_ref = None
     file_test = None
     try:
@@ -189,7 +192,7 @@ def remote_diff(host_ref, host_test, filename):
         if file_test is not None and os.path.exists(file_test):
             os.remove(file_test)
 
-def print_results(results, show_diff, show_ignored):
+def print_results(results: dict[str, Any], show_diff: bool, show_ignored: bool) -> None:
     # Print what differs
     for dtype in DataType:
         if dtype == DataType.PACKAGE:
@@ -213,11 +216,13 @@ def print_results(results, show_diff, show_ignored):
         for f in results['ignored_files']:
             print(f"{f}")
 
-def compare_data(ref, test, ignored_file_patterns):
+def compare_data(
+    ref: dict[str, Any], test: dict[str, Any], ignored_file_patterns: list[str]
+) -> tuple[dict[str, Any], int]:
     ref_data = ref['data']
     test_data = test['data']
     err = 0
-    results = {
+    results: dict[str | DataType, Any] = {
         'host': {
             'ref': ref['host'],
             'test': test['host']
@@ -282,27 +287,27 @@ def compare_data(ref, test, ignored_file_patterns):
     return results, err
 
 # Load a previously saved json file containing reference data
-def load_reference_files(filename):
+def load_reference_files(filename: str) -> dict[DataType, dict[str, str] | None]:
     try:
         with open(filename, 'r') as fd:
-            return json.load(fd)
+            return cast(dict[DataType, dict[str, str] | None], json.load(fd))
     except Exception as e:
         print(f"Error: {e}", file=sys.stderr)
         exit(-1)
 
 # Save files from a reference host in json format
-def save_reference_data(files, filename):
+def save_reference_data(files: dict[DataType, dict[str, str] | None], filename: str) -> None:
     try:
         with open(filename, 'w') as fd:
             json.dump(files, fd, indent=4)
     except Exception as e:
         print(f"Error: {e}", file=sys.stderr)
         exit(-1)
 
-def main():
-    ref_data = None
-    folders = ["/boot", "/etc", "/opt", "/usr"]
-    ignored_file_patterns = [
+def main() -> int:
+    ref_data: dict[DataType, dict[str, str] | None] | None = None
+    folders: list[str] = ["/boot", "/etc", "/opt", "/usr"]
+    ignored_file_patterns: list[str] = [
         '/boot/initrd-*',
         '/boot/efi/*',
         '/boot/grub/*',
@@ -361,13 +366,13 @@ def main():
     ]
 
     parser = argparse.ArgumentParser(description='Spot filesystem differences between 2 XCP-ng hosts')
-    parser.add_argument('--reference-host', '-r', dest='ref_host',
+    parser.add_argument('--reference-host', '-r', dest='ref_host', type=str,
                         help='The XCP-ng host used as reference')
-    parser.add_argument('--test-host', '-t', dest='test_host',
+    parser.add_argument('--test-host', '-t', dest='test_host', type=str,
                         help='The XCP-ng host to be tested after install or upgrade')
-    parser.add_argument('--save-reference', '-s', dest='save_ref',
+    parser.add_argument('--save-reference', '-s', dest='save_ref', type=str,
                         help='Save filesystem information of the reference host to a file')
-    parser.add_argument('--load-reference', '-l', dest='load_ref',
+    parser.add_argument('--load-reference', '-l', dest='load_ref', type=str,
                         help='Load reference filesystem information from a file')
     parser.add_argument('--show-diff', '-d', action='store_true', dest='show_diff',
                         help='Show diff of text files that differ. A reference host must be supplied with -r')
@@ -412,8 +417,8 @@ def main():
         print(f"Get test host data from {args.test_host}")
     test_data = get_data(args.test_host, args.folders)
 
-    ref = dict([('data', ref_data), ('host', args.ref_host)])
-    test = dict([('data', test_data), ('host', args.test_host)])
+    ref: dict[str, Any] = dict([('data', ref_data), ('host', args.ref_host)])
+    test: dict[str, Any] = dict([('data', test_data), ('host', args.test_host)])
 
     results, err = compare_data(ref, test, args.ignored_file_patterns)
 

scripts/xva_bridge.py

@@ -11,23 +11,25 @@
 import libarchive  # type: ignore
 import libarchive.ffi  # type: ignore
 
+from typing import Generator
+
 class XvaHeaderMember:
     def __init__(self, member: minidom.Element):
         self.member = member
 
-    def get_name(self):
+    def get_name(self) -> str | None:
         for child in self.member.childNodes:
             if child.nodeType == minidom.Node.ELEMENT_NODE and child.tagName == "name" and child.firstChild:
                 return child.firstChild.nodeValue
         return None
 
-    def get_value(self):
+    def get_value(self) -> str | None:
         for child in self.member.childNodes:
             if child.nodeType == minidom.Node.ELEMENT_NODE and child.tagName == "value" and child.firstChild:
                 return child.firstChild.nodeValue
         return None
 
-    def set_value(self, value: str):
+    def set_value(self, value: str) -> None:
         for child in self.member.childNodes:
             if child.nodeType == minidom.Node.ELEMENT_NODE and child.tagName == "value" and child.firstChild:
                 child.firstChild.nodeValue = value  # type: ignore
@@ -38,18 +40,20 @@ class XvaHeader:
     def __init__(self, header_bytes: bytes):
         self.xml = minidom.parseString(header_bytes.decode())
 
-    def members(self):
+    def members(self) -> Generator[XvaHeaderMember, None, None]:
         for member in self.xml.getElementsByTagName("member"):
             if member.nodeType == minidom.Node.ELEMENT_NODE:
                 yield XvaHeaderMember(member)
 
-    def get_bridge(self):
+    def get_bridge(self) -> str:
         for member in self.members():
             if member.get_name() == "bridge":
-                return member.get_value()
+                v = member.get_value()
+                assert v is not None
+                return v
         raise ValueError("Could not find bridge value in XVA header")
 
-    def set_bridge(self, bridge: str):
+    def set_bridge(self, bridge: str) -> None:
         for member in self.members():
             if member.get_name() == "bridge":
                 member.set_value(bridge)