Use mypy for static type checking

# Conflicts:
#	.pre-commit-config.yaml
#	docker/mongodb-kubernetes-tests/kubetester/certs.py

# Conflicts:
#	.pre-commit-config.yaml

# Conflicts:
#	docker/mongodb-kubernetes-tests/kubetester/certs.py

Author: lsierant

.pre-commit-config.yaml

@@ -129,6 +129,16 @@ repos:
             ^scripts/funcs/[^/]+$
           )
 
+  - repo: local
+    hooks:
+      - id: ty
+        name: ty
+        entry: bash -c 'cd docker/mongodb-kubernetes-tests && uvx ty check tests/'
+        language: system
+        pass_filenames: false
+        files: docker/mongodb-kubernetes-tests/tests/.*\.py$
+        stages: [pre-commit]
+
   - repo: https://github.com/golangci/golangci-lint
     rev: v2.0.2
     hooks:

docker/mongodb-kubernetes-tests/kubeobject/customobject.py

@@ -2,7 +2,7 @@
 
 import copy
 from datetime import datetime, timedelta
-from typing import Dict, Optional
+from typing import Any, Dict, Optional, Self
 
 import yaml
 from kubernetes import client
@@ -43,6 +43,7 @@ def __init__(
                 version=version,
                 api_client=api_client,
             )
+            assert crd is not None, "Could not find CRD matching the given parameters"
             self.kind = crd.spec.names.kind
             self.plural = crd.spec.names.plural
             self.group = crd.spec.group
@@ -71,7 +72,7 @@ def __init__(
         self.auto_reload_period = timedelta(seconds=2)
 
         # Last time this object was updated
-        self.last_update: datetime = None
+        self.last_update: Optional[datetime] = None
 
         # Sets the API used for this particular type of object
         self.api = client.CustomObjectsApi(api_client=api_client)
@@ -85,7 +86,7 @@ def __init__(
                 "status": {},
             }
 
-    def load(self) -> CustomObject:
+    def load(self) -> Self:
         """Loads this object from the API."""
 
         obj = self.api.get_namespaced_custom_object(self.group, self.version, self.namespace, self.plural, self.name)
@@ -202,7 +203,7 @@ def from_yaml(cls, yaml_file, name=None, namespace=None, cluster_scoped=False):
 
     @classmethod
     def define(
-        cls: CustomObject,
+        cls,
         name: str,
         kind: Optional[str] = None,
         plural: Optional[str] = None,
@@ -284,7 +285,7 @@ def get_crd_names(
     group: Optional[str] = None,
     version: Optional[str] = None,
     api_client: Optional[client.ApiClient] = None,
-) -> Optional[Dict]:
+) -> Optional[Any]:
     """Gets the CRD entry that matches all the parameters passed."""
     api = client.ApiextensionsV1Api(api_client=api_client)
 
@@ -313,6 +314,8 @@ def get_crd_names(
         if found:
             return crd
 
+    return None
+
 
 def create_or_update(resource: CustomObject) -> CustomObject:
     """

docker/mongodb-kubernetes-tests/kubetester/__init__.py

@@ -75,7 +75,7 @@ def delete_service_account(namespace: str, name: str) -> str:
 
 def get_service(
     namespace: str, name: str, api_client: Optional[kubernetes.client.ApiClient] = None
-) -> client.V1ServiceSpec:
+) -> Optional[client.V1Service]:
     """Gets a service with `name` in `namespace.
     :return None if the service does not exist
     """
@@ -148,6 +148,13 @@ def create_or_update_service(
     service: Optional[client.V1Service] = None,
 ) -> str:
     print("Logging inside create_or_update_service")
+    if service_name is None and service is not None:
+        if isinstance(service, dict):
+            service_name = service.get("metadata", {}).get("name")
+        elif hasattr(service, "metadata") and service.metadata is not None:
+            service_name = service.metadata.name
+    if service_name is None:
+        raise ValueError("service_name must not be None")
     try:
         create_service(namespace, service_name, cluster_ip=cluster_ip, ports=ports, selector=selector, service=service)
     except kubernetes.client.ApiException as e:
@@ -247,8 +254,8 @@ def delete_pod(namespace: str, name: str, api_client: Optional[kubernetes.client
 
 def create_or_update_namespace(
     namespace: str,
-    labels: dict = None,
-    annotations: dict = None,
+    labels: Optional[dict] = None,
+    annotations: Optional[dict] = None,
     api_client: Optional[kubernetes.client.ApiClient] = None,
 ):
     namespace_resource = client.V1Namespace(
@@ -368,7 +375,7 @@ def get_pod_when_ready(
     """
     cnt = 0
 
-    while True and cnt < default_retry:
+    while default_retry is not None and cnt < default_retry:
         print(f"get_pod_when_ready: namespace={namespace}, label_selector={label_selector}")
 
         if cnt > 0:
@@ -429,7 +436,7 @@ def is_pod_ready(
     return None
 
 
-def get_default_storage_class() -> str:
+def get_default_storage_class() -> Optional[str]:
     default_class_annotations = (
         "storageclass.kubernetes.io/is-default-class",  # storage.k8s.io/v1
         "storageclass.beta.kubernetes.io/is-default-class",  # storage.k8s.io/v1beta1
@@ -440,6 +447,7 @@ def get_default_storage_class() -> str:
             sc.metadata.annotations.get(a) == "true" for a in default_class_annotations
         ):
             return sc.metadata.name
+    return None
 
 
 def decode_secret(data: Dict[str, str]) -> Dict[str, str]:

docker/mongodb-kubernetes-tests/kubetester/automation_config_tester.py

@@ -104,6 +104,7 @@ def assert_oidc_providers_size(self, expected_size: int):
         assert len(self.automation_config["oidcProviderConfigs"]) == expected_size
 
     def assert_oidc_configuration(self, oidc_config: Optional[Dict] = None):
+        assert oidc_config is not None, "oidc_config must not be None"
         actual_configs = self.automation_config["oidcProviderConfigs"]
         assert len(actual_configs) == len(
             oidc_config

docker/mongodb-kubernetes-tests/kubetester/certs.py

@@ -81,14 +81,14 @@ class Issuer(IssuerType, WaitForConditions):
     Reason = "KeyPairVerified"
 
 
-class ClusterIssuer(ClusterIssuerType, WaitForConditions):
+class ClusterIssuer(ClusterIssuerType, WaitForConditions):  # type: ignore[valid-type, misc]
     Reason = "KeyPairVerified"
 
 
 def generate_cert(
     namespace: str,
-    pod: str,
-    dns: str,
+    pod: str | list[str],
+    dns: str | list[str],
     issuer: str,
     spec: Optional[Dict] = None,
     additional_domains: Optional[List[str]] = None,
@@ -113,15 +113,18 @@ def generate_cert(
     cert = Certificate(namespace=namespace, name=secret_name)
 
     if multi_cluster_mode:
-        dns_names = dns_list
+        dns_names = dns_list if dns_list is not None else []
     else:
-        dns_names = [dns]
+        dns_names = dns if isinstance(dns, list) else [dns]
 
     if not multi_cluster_mode:
-        dns_names.append(pod)
+        if isinstance(pod, list):
+            dns_names.extend(pod)
+        else:
+            dns_names.append(pod)
 
     if additional_domains is not None:
-        dns_names += additional_domains
+        dns_names = dns_names + additional_domains
 
     issuerRef = {"name": issuer, "kind": "Issuer"}
     if clusterwide:
@@ -176,8 +179,8 @@ def create_tls_certs(
     namespace: str,
     resource_name: str,
     replicas: int = 3,
-    replicas_cluster_distribution: Optional[List[int]] = None,
-    service_name: str = None,
+    replicas_cluster_distribution: Optional[List[Optional[int]]] = None,
+    service_name: Optional[str] = None,
     spec: Optional[Dict] = None,
     secret_name: Optional[str] = None,
     additional_domains: Optional[List[str]] = None,
@@ -308,8 +311,8 @@ def create_mongodb_tls_certs(
     resource_name: str,
     bundle_secret_name: str,
     replicas: int = 3,
-    replicas_cluster_distribution: Optional[List[int]] = None,
-    service_name: str = None,
+    replicas_cluster_distribution: Optional[List[Optional[int]]] = None,
+    service_name: Optional[str] = None,
     spec: Optional[Dict] = None,
     additional_domains: Optional[List[str]] = None,
     secret_backend: Optional[str] = None,
@@ -342,7 +345,7 @@ def create_mongodb_tls_certs(
 def multi_cluster_service_fqdns(
     resource_name: str,
     namespace: str,
-    external_domain: str,
+    external_domain: Optional[str],
     cluster_index: int,
     replicas: int,
 ) -> List[str]:
@@ -374,11 +377,14 @@ def create_x509_mongodb_tls_certs(
     resource_name: str,
     bundle_secret_name: str,
     replicas: int = 3,
-    replicas_cluster_distribution: Optional[List[int]] = None,
-    service_name: str = None,
+    replicas_cluster_distribution: Optional[List[Optional[int]]] = None,
+    service_name: Optional[str] = None,
+    spec: Optional[Dict] = None,
     additional_domains: Optional[List[str]] = None,
     secret_backend: Optional[str] = None,
     vault_subpath: Optional[str] = None,
+    process_hostnames: Optional[List[str]] = None,
+    clusterwide: bool = False,
 ) -> str:
     spec = get_mongodb_x509_subject(namespace)
 
@@ -394,6 +400,8 @@ def create_x509_mongodb_tls_certs(
         additional_domains=additional_domains,
         secret_backend=secret_backend,
         vault_subpath=vault_subpath,
+        process_hostnames=process_hostnames,
+        clusterwide=clusterwide,
     )
 
 
@@ -484,9 +492,9 @@ def create_sharded_cluster_certs(
     additional_domains: Optional[List[str]] = None,
     secret_prefix: Optional[str] = None,
     secret_backend: Optional[str] = None,
-    shard_distribution: Optional[List[int]] = None,
-    mongos_distribution: Optional[List[int]] = None,
-    config_srv_distribution: Optional[List[int]] = None,
+    shard_distribution: Optional[List[Optional[int]]] = None,
+    mongos_distribution: Optional[List[Optional[int]]] = None,
+    config_srv_distribution: Optional[List[Optional[int]]] = None,
 ):
     cert_generation_func = create_mongodb_tls_certs
     if x509_certs:

docker/mongodb-kubernetes-tests/kubetester/certs_mongodb_multi.py

@@ -128,7 +128,7 @@ def create_multi_cluster_tls_certs(
     secret_name: str,
     central_cluster_client: kubernetes.client.ApiClient,
     member_clients: List[MultiClusterClient],
-    mongodb_multi: Optional[CustomObject] = None,
+    mongodb_multi: Optional[MongoDBMulti] = None,
     namespace: Optional[str] = None,
     secret_backend: Optional[str] = None,
     additional_domains: Optional[List[str]] = None,
@@ -137,6 +137,7 @@ def create_multi_cluster_tls_certs(
     spec: Optional[dict] = None,
 ) -> str:
     if service_fqdns is None:
+        assert mongodb_multi is not None, "mongodb_multi is required when service_fqdns is not provided"
         service_fqdns = [f"{mongodb_multi.name}-svc.{mongodb_multi.namespace}.svc.cluster.local"]
 
         for client in member_clients:
@@ -145,6 +146,7 @@ def create_multi_cluster_tls_certs(
                 external_domain = cluster_spec["externalAccess"]["externalDomain"]
             except KeyError:
                 external_domain = None
+            assert client.cluster_index is not None
             service_fqdns.extend(
                 multi_cluster_service_fqdns(
                     mongodb_multi.name,
@@ -156,6 +158,7 @@ def create_multi_cluster_tls_certs(
             )
 
     if namespace is None:
+        assert mongodb_multi is not None, "mongodb_multi is required when namespace is not provided"
         namespace = mongodb_multi.namespace
 
     generate_cert(

docker/mongodb-kubernetes-tests/kubetester/crypto.py

@@ -49,7 +49,7 @@ def generate_csr(namespace: str, host: str, servicename: str):
     )
 
 
-def get_pem_certificate(name: str) -> Optional[str]:
+def get_pem_certificate(name: str) -> Optional[bytes]:
     body = client.CertificatesV1Api().read_certificate_signing_request_status(name)
     if body.status.certificate is None:
         return None

docker/mongodb-kubernetes-tests/kubetester/helm.py

@@ -34,6 +34,7 @@ def helm_template(
         command_args.append("--show-only")
         command_args.append(templates)
 
+    helm_chart_path = helm_chart_path or "helm_chart"
     args = ("helm", "template", *command_args, helm_chart_path)
     logger.info(" ".join(args))
 
@@ -53,6 +54,7 @@ def helm_install(
     custom_operator_version: Optional[str] = None,
 ):
     command_args = _create_helm_args(helm_args, helm_options)
+    helm_chart_path = helm_chart_path or "helm_chart"
     args = [
         "helm",
         "upgrade",
@@ -176,7 +178,8 @@ def helm_registry_login_to_ecr(helm_registry: str, region: str):
 
         # Close the stdout stream of aws_proc in the parent process
         # to prevent resource leakage (only needed if you plan to do more processing)
-        aws_proc.stdout.close()
+        if aws_proc.stdout:
+            aws_proc.stdout.close()
 
         # Wait for the Helm command (helm_proc) to finish and capture its output
         helm_stdout, helm_stderr = helm_proc.communicate()
@@ -297,17 +300,19 @@ def _create_helm_args(helm_args: Dict[str, str], helm_options: Optional[List[str
 def helm_chart_path_and_version(helm_chart_path: str, operator_version: str) -> tuple[str, str]:
     # if helm_chart_path is not passed, use the default value from DEFAULT_HELM_CHART_PATH
     if not helm_chart_path:
-        helm_chart_path = os.getenv(DEFAULT_HELM_CHART_PATH_ENV_VAR_NAME)
+        helm_chart_path = os.getenv(DEFAULT_HELM_CHART_PATH_ENV_VAR_NAME, "")
 
     if helm_chart_path.startswith("oci://"):
         # If operator_version is not passed, we want to install the current version.
         if not operator_version:
-            operator_version = os.environ.get(OCI_HELM_VERSION_ENV_VAR_NAME)
+            operator_version = os.environ.get(OCI_HELM_VERSION_ENV_VAR_NAME, "")
 
-        registry = os.environ.get(OCI_HELM_REGISTRY_ENV_VAR_NAME)
+        registry = os.environ.get(OCI_HELM_REGISTRY_ENV_VAR_NAME, "")
         # If ECR we need to login first to the OCI container registry
         if registry == OCI_HELM_REGISTRY_ECR:
             region = os.environ.get(OCI_HELM_REGION_ENV_VAR_NAME)
+            if not region:
+                raise ValueError(f"{OCI_HELM_REGION_ENV_VAR_NAME} must be set when using ECR registry")
             try:
                 helm_registry_login_to_ecr(registry, region)
             except Exception as e:

docker/mongodb-kubernetes-tests/kubetester/http.py

@@ -30,7 +30,7 @@ def get_retriable_https_session(*, tls_verify: bool) -> requests.Session:
     return get_retriable_session("https", tls_verify)
 
 
-def https_endpoint_is_reachable(url: str, auth: Tuple[str], *, tls_verify: bool) -> bool:
+def https_endpoint_is_reachable(url: str, auth: Tuple[str, str], *, tls_verify: bool) -> bool:
     """
     Checks that `url` is reachable, using `auth` basic credentials.
     """

docker/mongodb-kubernetes-tests/kubetester/kmip.py

@@ -124,7 +124,7 @@ def _create_tls_certs_kmip(
         resource_name: str,
         bundle_secret_name: str,
         replicas: int = 3,
-        service_name: str = None,
+        service_name: Optional[str] = None,
         spec: Optional[Dict] = None,
     ) -> str:
         ca = read_configmap(namespace, self.ca_configmap)
@@ -135,7 +135,7 @@ def _create_tls_certs_kmip(
             replicas=replicas,
             service_name=service_name,
             spec=spec,
-            additional_domains=[service_name],
+            additional_domains=[service_name] if service_name else None,
         )
         secret = read_secret(namespace, cert_secret_name)
         create_or_update_secret(