Install CRDs in operator upgrade tests (#127)

# Summary

#63 rebased on
`master` branch.

## Proof of Work

<!-- Enter your proof that it works here.-->

## Checklist
- [ ] Have you linked a jira ticket and/or is the ticket in the title?
- [ ] Have you checked whether your jira ticket required DOCSP changes?
- [ ] Have you checked for release_note changes?

## Reminder (Please remove this when merging)
- Please try to Approve or Reject Changes the PR, keep PRs in review as
short as possible
- Our Short Guide for PRs:
[Link](https://docs.google.com/document/d/1T93KUtdvONq43vfTfUt8l92uo4e4SEEvFbIEKOxGr44/edit?tab=t.0)
- Remember the following Communication Standards - use comment prefixes
for clarity:
  * **blocking**: Must be addressed before approval.
  * **follow-up**: Can be addressed in a later PR or ticket.
  * **q**: Clarifying question.
  * **nit**: Non-blocking suggestions.
  * **note**: Side-note, non-actionable. Example: Praise 
  * --> no prefix is considered a question

---------

Co-authored-by: Łukasz Sierant <lukasz.sierant@mongodb.com>

Author: mircea-cosbuc

docker/mongodb-kubernetes-tests/Dockerfile

@@ -43,6 +43,10 @@ RUN mkdir -p /tmp/mongodb-tools && \
     cp /tmp/mongodb-tools/*/bin/* /usr/local/bin/ && \
     rm -rf /tmp/mongodb-tools /tmp/mongodb-tools.tgz
 
+RUN curl -LO "https://storage.googleapis.com/kubernetes-release/release/$(curl -s https://storage.googleapis.com/kubernetes-release/release/stable.txt)/bin/linux/amd64/kubectl" \
+    && chmod +x ./kubectl \
+    && mv ./kubectl /usr/local/bin/kubectl
+
 COPY --from=builder /venv /venv
 
 ENV PATH="/venv/bin:${PATH}"

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

@@ -1,3 +1,4 @@
+import glob
 import logging
 import os
 import re
@@ -120,8 +121,16 @@ def helm_repo_add(repo_name: str, url: str):
 
 def process_run_and_check(args, **kwargs):
     try:
+        logger.debug(f"subprocess.run: {args}")
         completed_process = subprocess.run(args, **kwargs)
-        completed_process.check_returncode()
+        # always print process output
+        if completed_process.stdout is not None:
+            stdout = completed_process.stdout.decode("utf-8")
+            logger.debug(f"stdout: {stdout}")
+        if completed_process.stderr is not None:
+            stderr = completed_process.stderr.decode("utf-8")
+            logger.debug(f"stderr: {stderr}")
+            completed_process.check_returncode()
     except subprocess.CalledProcessError as exc:
         if exc.stdout is not None:
             stdout = exc.stdout.decode("utf-8")
@@ -141,10 +150,17 @@ def helm_upgrade(
     helm_options: Optional[List[str]] = None,
     helm_override_path: Optional[bool] = False,
     custom_operator_version: Optional[str] = None,
+    apply_crds_first: bool = False,
 ):
     if not helm_chart_path:
         logger.warning("Helm chart path is empty, defaulting to 'helm_chart'")
         helm_chart_path = "helm_chart"
+
+    chart_dir = helm_chart_path if helm_override_path else _helm_chart_dir(helm_chart_path)
+
+    if apply_crds_first:
+        apply_crds_from_chart(chart_dir)
+
     command_args = _create_helm_args(helm_args, helm_options)
     args = [
         "helm",
@@ -154,19 +170,30 @@ def helm_upgrade(
         *command_args,
         name,
     ]
+
     if custom_operator_version:
         args.append(f"--version={custom_operator_version}")
-    if helm_override_path:
-        args.append(helm_chart_path)
-    else:
-        args.append(_helm_chart_dir(helm_chart_path))
+
+    args.append(chart_dir)
 
     command = " ".join(args)
-    logger.debug("Running helm upgrade command:")
-    logger.debug(command)
     process_run_and_check(command, check=True, capture_output=True, shell=True)
 
 
+def apply_crds_from_chart(chart_dir: str):
+    crd_files = glob.glob(os.path.join(chart_dir, "crds", "*.yaml"))
+
+    if not crd_files:
+        raise Exception(f"No CRD files found in chart directory: {chart_dir}")
+
+    logger.info(f"Found {len(crd_files)} CRD files to apply:")
+
+    for crd_file in crd_files:
+        logger.info(f"Applying CRD from file: {crd_file}")
+        args = ["kubectl", "apply", "-f", crd_file]
+        process_run_and_check(" ".join(args), check=True, capture_output=True, shell=True)
+
+
 def helm_uninstall(name):
     args = ("helm", "uninstall", name)
     logger.info(args)

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

@@ -95,7 +95,9 @@ def install(self, custom_operator_version: Optional[str] = None) -> Operator:
 
         return self
 
-    def upgrade(self, multi_cluster: bool = False, custom_operator_version: Optional[str] = None) -> Operator:
+    def upgrade(
+        self, multi_cluster: bool = False, custom_operator_version: Optional[str] = None, apply_crds_first: bool = False
+    ) -> Operator:
         """Upgrades the Operator in Kubernetes cluster using 'helm upgrade', waits until it's running"""
         helm_upgrade(
             self.name,
@@ -104,6 +106,7 @@ def upgrade(self, multi_cluster: bool = False, custom_operator_version: Optional
             helm_chart_path=self.helm_chart_path,
             helm_options=self.helm_options,
             custom_operator_version=custom_operator_version,
+            apply_crds_first=apply_crds_first,
         )
         self._wait_for_operator_ready()
         self._wait_operator_webhook_is_ready(multi_cluster=multi_cluster)

docker/mongodb-kubernetes-tests/tests/conftest.py

@@ -107,7 +107,7 @@ def get_version_id():
 
 
 @fixture(scope="module")
-def operator_installation_config(namespace: str) -> Dict[str, str]:
+def operator_installation_config(namespace: str) -> dict[str, str]:
     return get_operator_installation_config(namespace)
 
 
@@ -126,7 +126,7 @@ def get_operator_installation_config(namespace):
 
 
 @fixture(scope="module")
-def monitored_appdb_operator_installation_config(operator_installation_config: Dict[str, str]) -> Dict[str, str]:
+def monitored_appdb_operator_installation_config(operator_installation_config: dict[str, str]) -> dict[str, str]:
     """Returns the ConfigMap containing configuration data for the Operator to be created
     and for the AppDB to be monitored.
     Created in the single_e2e.sh"""
@@ -135,7 +135,7 @@ def monitored_appdb_operator_installation_config(operator_installation_config: D
     return config
 
 
-def get_multi_cluster_operator_installation_config(namespace: str) -> Dict[str, str]:
+def get_multi_cluster_operator_installation_config(namespace: str) -> dict[str, str]:
     """Returns the ConfigMap containing configuration data for the Operator to be created.
     Created in the single_e2e.sh"""
     config = KubernetesTester.read_configmap(
@@ -150,7 +150,7 @@ def get_multi_cluster_operator_installation_config(namespace: str) -> Dict[str,
 @fixture(scope="module")
 def multi_cluster_operator_installation_config(
     central_cluster_client: kubernetes.client.ApiClient, namespace: str
-) -> Dict[str, str]:
+) -> dict[str, str]:
     return get_multi_cluster_operator_installation_config(namespace)
 
 
@@ -159,15 +159,15 @@ def multi_cluster_monitored_appdb_operator_installation_config(
     central_cluster_client: kubernetes.client.ApiClient,
     namespace: str,
     multi_cluster_operator_installation_config: dict[str, str],
-) -> Dict[str, str]:
+) -> dict[str, str]:
     multi_cluster_operator_installation_config["customEnvVars"] = f"OPS_MANAGER_MONITOR_APPDB=true"
     return multi_cluster_operator_installation_config
 
 
 @fixture(scope="module")
 def operator_clusterwide(
     namespace: str,
-    operator_installation_config: Dict[str, str],
+    operator_installation_config: dict[str, str],
 ) -> Operator:
     return get_operator_clusterwide(namespace, operator_installation_config)
 
@@ -181,7 +181,7 @@ def get_operator_clusterwide(namespace, operator_installation_config):
 @fixture(scope="module")
 def operator_vault_secret_backend(
     namespace: str,
-    monitored_appdb_operator_installation_config: Dict[str, str],
+    monitored_appdb_operator_installation_config: dict[str, str],
 ) -> Operator:
     helm_args = monitored_appdb_operator_installation_config.copy()
     helm_args["operator.vaultSecretBackend.enabled"] = "true"
@@ -191,7 +191,7 @@ def operator_vault_secret_backend(
 @fixture(scope="module")
 def operator_vault_secret_backend_tls(
     namespace: str,
-    monitored_appdb_operator_installation_config: Dict[str, str],
+    monitored_appdb_operator_installation_config: dict[str, str],
 ) -> Operator:
     helm_args = monitored_appdb_operator_installation_config.copy()
     helm_args["operator.vaultSecretBackend.enabled"] = "true"
@@ -200,7 +200,7 @@ def operator_vault_secret_backend_tls(
 
 
 @fixture(scope="module")
-def operator_installation_config_quick_recovery(operator_installation_config: Dict[str, str]) -> Dict[str, str]:
+def operator_installation_config_quick_recovery(operator_installation_config: dict[str, str]) -> dict[str, str]:
     """
     This functions appends automatic recovery settings for CLOUDP-189433. In order to make the test runnable in
     reasonable time, we override the Recovery back off to 120 seconds. This gives enough time for the initial
@@ -480,9 +480,9 @@ def get_custom_om_version():
 @fixture(scope="module")
 def default_operator(
     namespace: str,
-    operator_installation_config: Dict[str, str],
+    operator_installation_config: dict[str, str],
     central_cluster_name: str,
-    multi_cluster_operator_installation_config: Dict[str, str],
+    multi_cluster_operator_installation_config: dict[str, str],
     central_cluster_client: client.ApiClient,
     member_cluster_clients: List[MultiClusterClient],
     member_cluster_names: List[str],
@@ -500,24 +500,23 @@ def default_operator(
 
 
 def get_default_operator(
-    namespace: str,
-    operator_installation_config: Dict[str, str],
+    namespace: str, operator_installation_config: dict[str, str], apply_crds_first: bool = False
 ) -> Operator:
     """Installs/upgrades a default Operator used by any test not interested in some custom Operator setting.
     TODO we use the helm template | kubectl apply -f process so far as Helm install/upgrade needs more refactoring in
     the shared environment"""
     operator = Operator(
         namespace=namespace,
         helm_args=operator_installation_config,
-    ).upgrade()
+    ).upgrade(apply_crds_first=apply_crds_first)
 
     return operator
 
 
 @fixture(scope="module")
 def operator_with_monitored_appdb(
     namespace: str,
-    monitored_appdb_operator_installation_config: Dict[str, str],
+    monitored_appdb_operator_installation_config: dict[str, str],
 ) -> Operator:
     """Installs/upgrades a default Operator used by any test that needs the AppDB monitoring enabled."""
     return Operator(
@@ -628,7 +627,7 @@ def member_cluster_clients() -> List[MultiClusterClient]:
 def multi_cluster_operator(
     namespace: str,
     central_cluster_name: str,
-    multi_cluster_operator_installation_config: Dict[str, str],
+    multi_cluster_operator_installation_config: dict[str, str],
     central_cluster_client: client.ApiClient,
     member_cluster_clients: List[MultiClusterClient],
     member_cluster_names: List[str],
@@ -646,10 +645,11 @@ def multi_cluster_operator(
 def get_multi_cluster_operator(
     namespace: str,
     central_cluster_name: str,
-    multi_cluster_operator_installation_config: Dict[str, str],
+    multi_cluster_operator_installation_config: dict[str, str],
     central_cluster_client: client.ApiClient,
     member_cluster_clients: List[MultiClusterClient],
     member_cluster_names: List[str],
+    apply_crds_first: bool = False,
 ) -> Operator:
     os.environ["HELM_KUBECONTEXT"] = central_cluster_name
 
@@ -667,14 +667,15 @@ def get_multi_cluster_operator(
             "operator.createOperatorServiceAccount": "false",
         },
         central_cluster_name,
+        apply_crds_first=apply_crds_first,
     )
 
 
 @fixture(scope="module")
 def multi_cluster_operator_with_monitored_appdb(
     namespace: str,
     central_cluster_name: str,
-    multi_cluster_monitored_appdb_operator_installation_config: Dict[str, str],
+    multi_cluster_monitored_appdb_operator_installation_config: dict[str, str],
     central_cluster_client: client.ApiClient,
     member_cluster_clients: List[MultiClusterClient],
     member_cluster_names: List[str],
@@ -703,7 +704,7 @@ def multi_cluster_operator_with_monitored_appdb(
 def multi_cluster_operator_manual_remediation(
     namespace: str,
     central_cluster_name: str,
-    multi_cluster_operator_installation_config: Dict[str, str],
+    multi_cluster_operator_installation_config: dict[str, str],
     central_cluster_client: client.ApiClient,
     member_cluster_clients: List[MultiClusterClient],
     member_cluster_names: List[str],
@@ -754,7 +755,7 @@ def get_multi_cluster_operator_clustermode(namespace: str) -> Operator:
 def multi_cluster_operator_clustermode(
     namespace: str,
     central_cluster_name: str,
-    multi_cluster_operator_installation_config: Dict[str, str],
+    multi_cluster_operator_installation_config: dict[str, str],
     central_cluster_client: client.ApiClient,
     member_cluster_clients: List[MultiClusterClient],
     member_cluster_names: List[str],
@@ -767,7 +768,7 @@ def multi_cluster_operator_clustermode(
 def install_multi_cluster_operator_set_members_fn(
     namespace: str,
     central_cluster_name: str,
-    multi_cluster_operator_installation_config: Dict[str, str],
+    multi_cluster_operator_installation_config: dict[str, str],
     central_cluster_client: client.ApiClient,
     member_cluster_clients: List[MultiClusterClient],
 ) -> Callable[[List[str]], Operator]:
@@ -793,14 +794,15 @@ def _fn(member_cluster_names: List[str]) -> Operator:
 
 def _install_multi_cluster_operator(
     namespace: str,
-    multi_cluster_operator_installation_config: Dict[str, str],
+    multi_cluster_operator_installation_config: dict[str, str],
     central_cluster_client: client.ApiClient,
     member_cluster_clients: List[MultiClusterClient],
-    helm_opts: Dict[str, str],
+    helm_opts: dict[str, str],
     central_cluster_name: str,
     operator_name: Optional[str] = MULTI_CLUSTER_OPERATOR_NAME,
     helm_chart_path: Optional[str] = LOCAL_HELM_CHART_DIR,
     custom_operator_version: Optional[str] = None,
+    apply_crds_first: bool = False,
 ) -> Operator:
     multi_cluster_operator_installation_config.update(helm_opts)
 
@@ -822,7 +824,7 @@ def _install_multi_cluster_operator(
         helm_args=multi_cluster_operator_installation_config,
         api_client=central_cluster_client,
         helm_chart_path=helm_chart_path,
-    ).upgrade(multi_cluster=True, custom_operator_version=custom_operator_version)
+    ).upgrade(multi_cluster=True, custom_operator_version=custom_operator_version, apply_crds_first=apply_crds_first)
 
     # If we're running locally, then immediately after installing the deployment, we scale it to zero.
     # This way operator in POD is not interfering with locally running one.
@@ -840,7 +842,7 @@ def _install_multi_cluster_operator(
 def official_operator(
     namespace: str,
     managed_security_context: str,
-    operator_installation_config: Dict[str, str],
+    operator_installation_config: dict[str, str],
     central_cluster_name: str,
     central_cluster_client: client.ApiClient,
     member_cluster_clients: List[MultiClusterClient],
@@ -919,7 +921,7 @@ def install_legacy_deployment_state_meko(
 def install_official_operator(
     namespace: str,
     managed_security_context: str,
-    operator_installation_config: Dict[str, str],
+    operator_installation_config: dict[str, str],
     central_cluster_name: Optional[str],
     central_cluster_client: Optional[client.ApiClient],
     member_cluster_clients: Optional[List[MultiClusterClient]],
@@ -1033,7 +1035,7 @@ def log_deployment_and_images(deployments):
 
 # Extract container images and deployments names from the nested dict returned by kubetester
 # Handles any missing key gracefully
-def extract_container_images_and_deployments(deployments) -> (Dict[str, str], List[str]):
+def extract_container_images_and_deployments(deployments) -> (dict[str, str], List[str]):
     deployment_images = {}
     deployment_names = []
     deployments = deployments.to_dict()
@@ -1318,7 +1320,7 @@ def get_api_servers_from_kubeconfig_secret(
     return get_api_servers_from_pod_kubeconfig(kubeconfig_secret["kubeconfig"], cluster_clients)
 
 
-def get_api_servers_from_test_pod_kubeconfig(namespace: str, member_cluster_names: List[str]) -> Dict[str, str]:
+def get_api_servers_from_test_pod_kubeconfig(namespace: str, member_cluster_names: List[str]) -> dict[str, str]:
     test_pod_cluster = get_test_pod_cluster_name()
     cluster_clients = get_clients_for_clusters(member_cluster_names)