diff --git a/.secrets.baseline b/.secrets.baseline index 018293f2fc..66eab27f70 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -3,7 +3,7 @@ "files": "build/bin/config/oscap/ssg-rhel9-ds.xml|^.secrets.baseline$", "lines": null }, - "generated_at": "2026-01-29T07:04:47Z", + "generated_at": "2026-02-12T15:32:55Z", "plugins_used": [ { "name": "AWSKeyDetector" @@ -805,6 +805,46 @@ "verified_result": null } ], + "python/test/mirror/test_mirror_advanced.py": [ + { + "hashed_secret": "206c80413b9a96c1312cc346b7d2517b84463edd", + "is_secret": false, + "is_verified": false, + "line_number": 52, + "type": "Secret Keyword", + "verified_result": null + } + ], + "python/test/mirror/test_mirror_auth.py": [ + { + "hashed_secret": "1c58bd92003bbaa0538e249fff6ee19a270dec5f", + "is_secret": false, + "is_verified": false, + "line_number": 118, + "type": "Secret Keyword", + "verified_result": null + } + ], + "python/test/mirror/test_mirror_basic.py": [ + { + "hashed_secret": "789cbe0407840b1c2041cb33452ff60f19bf58cc", + "is_secret": false, + "is_verified": false, + "line_number": 137, + "type": "Secret Keyword", + "verified_result": null + } + ], + "python/test/mirror/test_mirror_packages.py": [ + { + "hashed_secret": "206c80413b9a96c1312cc346b7d2517b84463edd", + "is_secret": false, + "is_verified": false, + "line_number": 292, + "type": "Secret Keyword", + "verified_result": null + } + ], "tekton/src/tasks/fvt/mas-fvt-assist-desktop.yml.j2": [ { "hashed_secret": "ff0c19d6cf999d585c440a2143db929839c48fb6", diff --git a/image/cli/mascli/mas b/image/cli/mascli/mas index 47b9db3375..d45337bb1c 100755 --- a/image/cli/mascli/mas +++ b/image/cli/mascli/mas @@ -181,7 +181,7 @@ case $1 in teardown_mirror_registry "$@" ;; - mirror|mirror-images|mirror-ibm-images) + mirror-images|mirror-ibm-images) echo "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!" >> $LOGFILE echo "!! mirror-ibm-images !!" >> $LOGFILE echo "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!" >> $LOGFILE @@ -242,6 +242,16 @@ case $1 in configure_ocp_for_mirror "$@" ;; + mirror) + echo "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!" >> $LOGFILE + echo "!! mirror !!" >> $LOGFILE + echo "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!" >> $LOGFILE + # Take the first parameter off (it will be "mirror") + shift + # Run the new Python-based mirror + mas-cli mirror "$@" + ;; + install) echo "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!" >> $LOGFILE echo "!! install !!" >> $LOGFILE diff --git a/python/setup.py b/python/setup.py index 561c949659..c21b674b0e 100644 --- a/python/setup.py +++ b/python/setup.py @@ -59,18 +59,19 @@ def get_version(rel_path): long_description=long_description, install_requires=[ 'mas-devops >= 5.2.0', # EPL + 'alive-progress', # MIT License 'halo', # MIT License 'prompt_toolkit', # BSD License 'openshift', # Apache Software License - 'kubernetes == 33.1.0', # Apache Software License # Version lock to be removed once https://github.com/kubernetes-client/python/issues/2460 is resolved - 'tabulate' # MIT License + 'kubernetes == 33.1.0', # Apache Software License, version lock to be removed once https://github.com/kubernetes-client/python/issues/2460 is resolved + 'tabulate', # MIT License ], extras_require={ 'dev': [ 'build', # MIT License 'flake8', # MIT License 'pytest', # MIT License - 'pyinstaller' # GPL, https://pyinstaller.org/en/stable/license.html & https://github.com/pyinstaller/pyinstaller/wiki/FAQ#license + 'pyinstaller', # GPL, https://pyinstaller.org/en/stable/license.html & https://github.com/pyinstaller/pyinstaller/wiki/FAQ#license ] }, classifiers=[ @@ -82,6 +83,6 @@ def get_version(rel_path): 'Programming Language :: Python :: 3.12', 'Topic :: Communications', 'Topic :: Internet', - 'Topic :: Software Development :: Libraries :: Python Modules' + 'Topic :: Software Development :: Libraries :: Python Modules', ] ) diff --git a/python/src/mas-cli b/python/src/mas-cli index 8d8a778158..d0f6c6f391 100644 --- a/python/src/mas-cli +++ b/python/src/mas-cli @@ -20,6 +20,7 @@ from mas.cli.aiservice.upgrade.app import AiServiceUpgradeApp from mas.cli.update.app import UpdateApp from mas.cli.upgrade.app import UpgradeApp from mas.cli.uninstall.app import UninstallApp +from mas.cli.mirror.app import MirrorApp from prompt_toolkit import HTML, print_formatted_text from urllib3.exceptions import MaxRetryError @@ -44,7 +45,7 @@ def usage(): + " - mas-cli update Apply updates and security fixes\n" # noqa: W503 + " - mas-cli upgrade Upgrade to a new MAS release\n" # noqa: W503 + " - mas-cli uninstall Remove MAS from the cluster\n" # noqa: W503 - + + " - mas-cli mirror Mirror container images \n" # noqa: W503 )) print_formatted_text(HTML("For usage information run mas-cli [action] --help\n")) @@ -71,6 +72,9 @@ if __name__ == '__main__': elif function == "upgrade": app = UpgradeApp() app.upgrade(argv[2:]) + elif function == "mirror": + app = MirrorApp() + app.mirror(argv[2:]) elif function in ["-h", "--help"]: usage() exit(0) diff --git a/python/src/mas/cli/cli.py b/python/src/mas/cli/cli.py index 33b700e869..8eb844d5fb 100644 --- a/python/src/mas/cli/cli.py +++ b/python/src/mas/cli/cli.py @@ -19,7 +19,7 @@ from subprocess import PIPE, Popen, TimeoutExpired import threading import json -from typing import List, Dict, Any, Callable, Type +from typing import List, Dict, Any, Callable, Type, NoReturn # Use of the openshift client rather than the kubernetes client allows us access to "apply" from kubernetes import config @@ -294,7 +294,7 @@ def getCompatibleVersions(self, coreChannel: str, appId: str) -> List[str]: return [] @logMethodCall - def fatalError(self, message: str, exception: Exception | None = None) -> None: + def fatalError(self, message: str, exception: Exception | None = None) -> NoReturn: if exception is not None: logger.error(message) logger.exception(exception, stack_info=True) diff --git a/python/src/mas/cli/displayMixins.py b/python/src/mas/cli/displayMixins.py index 342c0a8793..e01ba36c8c 100644 --- a/python/src/mas/cli/displayMixins.py +++ b/python/src/mas/cli/displayMixins.py @@ -54,19 +54,25 @@ def printHighlight(self, message: str | list[str]) -> None: print_formatted_text(HTML(f"{message.replace(' & ', ' & ')}")) def printWarning(self, message: str) -> None: + logger.warning(message) print_formatted_text(HTML(f"Warning: {message.replace(' & ', ' & ')}")) def printSummary(self, title: str, value: str) -> None: titleLength = len(title) message = f"{title} {'.' * (40 - titleLength)} {value}" + + logger.debug(f"Summary: {title} = {value}") print_formatted_text(HTML(f" <{SUMMARYCOLOR}>{message.replace(' & ', ' & ')}")) def printParamSummary(self, message: str, param: str) -> None: if self.getParam(param) is None: # type: ignore + logger.debug(f"Parameter Summary: {param} = undefined") self.printSummary(message, f"<{UNDEFINEDPARAMCOLOR}>Undefined") elif self.getParam(param) == "": # type: ignore + logger.debug(f"Parameter Summary: {param} = \"\"") self.printSummary(message, f"<{UNDEFINEDPARAMCOLOR}>Default") else: + logger.debug(f"Parameter Summary: {param} = {self.getParam(param)}") # type: ignore self.printSummary(message, self.getParam(param)) # type: ignore diff --git a/python/src/mas/cli/install/__init__.py b/python/src/mas/cli/install/__init__.py index 2ca23962a6..85df29608e 100644 --- a/python/src/mas/cli/install/__init__.py +++ b/python/src/mas/cli/install/__init__.py @@ -1,5 +1,5 @@ # ***************************************************************************** -# Copyright (c) 2024 IBM Corporation and other Contributors. +# Copyright (c) 2026 IBM Corporation and other Contributors. # # All rights reserved. This program and the accompanying materials # are made available under the terms of the Eclipse Public License v1.0 diff --git a/python/src/mas/cli/install/app.py b/python/src/mas/cli/install/app.py index 6b9f5a8d6c..8a804a4613 100644 --- a/python/src/mas/cli/install/app.py +++ b/python/src/mas/cli/install/app.py @@ -19,6 +19,8 @@ import calendar from openshift.dynamic.exceptions import NotFoundError +from typing import Dict, Any + from prompt_toolkit import prompt, print_formatted_text, HTML from prompt_toolkit.completion import WordCompleter @@ -84,11 +86,14 @@ def wrapper(self, *args, **kwargs): class InstallApp(BaseApp, InstallSettingsMixin, InstallSummarizerMixin, ConfigGeneratorMixin, installArgBuilderMixin): @logMethodCall def validateCatalogSource(self): - # Check supported OCP versions - ocpVersion = getClusterVersion(self.dynamicClient) - supportedReleases = self.chosenCatalog.get("ocp_compatibility", []) - if len(supportedReleases) > 0 and not isClusterVersionInRange(ocpVersion, supportedReleases): - self.fatalError(f"IBM Maximo Operator Catalog {self.getParam('mas_catalog_version')} is not compatible with OpenShift v{ocpVersion}. Compatible OpenShift releases are {supportedReleases}") + # Check supported OCP versions - but we can only do this in non-development mode because in development mode + # we do not load catalog metadata files + if not self.devMode: + assert self.chosenCatalog is not None, "validateCatalogSource() called before catalog was chosen" + ocpVersion = getClusterVersion(self.dynamicClient) + supportedReleases = self.chosenCatalog.get("ocp_compatibility", []) + if len(supportedReleases) > 0 and not isClusterVersionInRange(ocpVersion, supportedReleases): + self.fatalError(f"IBM Maximo Operator Catalog {self.getParam('mas_catalog_version')} is not compatible with OpenShift v{ocpVersion}. Compatible OpenShift releases are {supportedReleases}") # Compare with any existing installed catalog catalogsAPI = self.dynamicClient.resources.get(api_version="operators.coreos.com/v1alpha1", kind="CatalogSource") @@ -105,6 +110,7 @@ def validateCatalogSource(self): catalogId = "v8-amd64" else: self.fatalError(f"IBM Maximo Operator Catalog is already installed on this cluster. However, it is not possible to identify its version. If you wish to install a new MAS instance using the {self.getParam('mas_catalog_version')} catalog please first run 'mas update' to switch to this catalog, this will ensure the appropriate actions are performed as part of the catalog update") + assert False, "fatalError() should have exited" # Let basepyright know that fatalError() will exit if catalogId != self.getParam("mas_catalog_version"): self.fatalError(f"IBM Maximo Operator Catalog {catalogId} is already installed on this cluster, if you wish to install a new MAS instance using the {self.getParam('mas_catalog_version')} catalog please first run 'mas update' to switch to this catalog, this will ensure the appropriate actions are performed as part of the catalog update") @@ -181,6 +187,7 @@ def formatCatalog(self, name: str) -> str: @logMethodCall def processCatalogChoice(self) -> list: + assert self.chosenCatalog is not None, "processCatalogChoice() called before catalog was chosen" self.catalogDigest = self.chosenCatalog["catalog_digest"] self.catalogMongoDbVersion = self.chosenCatalog["mongo_extras_version_default"] self.catalogDb2Channel = self.chosenCatalog.get("db2_channel_default", "v110509.0") # Returns fallback "v110509.0" for old catalogs without this field @@ -499,6 +506,7 @@ def configCP4D(self): if self.getParam("mas_catalog_version") in self.catalogOptions: # Note: this will override any version provided by the user (which is intentional!) logger.debug(f"Using automatic CP4D product version: {self.getParam('cpd_product_version')}") + assert self.chosenCatalog is not None, "chosenCatalog should be set in this scenario but was not" self.setParam("cpd_product_version", self.chosenCatalog["cpd_product_version_default"]) elif self.getParam("cpd_product_version") == "": if self.noConfirm: @@ -594,7 +602,7 @@ def configCATrust(self) -> None: ]) self.yesOrNo("Trust default CAs", "mas_trust_default_cas") else: - self.setParam("mas_trust_default_cas", True) + self.setParam("mas_trust_default_cas", "true") @logMethodCall def configOperationMode(self): @@ -888,8 +896,8 @@ def configDNSAndCerts(self): self.setParam("mas_domain", "") self.setParam("mas_cluster_issuer", "") self.manualCerts = self.yesOrNo("Configure manual certificates") - self.setParam("mas_manual_cert_mgmt", self.manualCerts) - if self.getParam("mas_manual_cert_mgmt"): + self.setParam("mas_manual_cert_mgmt", str(self.manualCerts).lower()) + if self.getParam("mas_manual_cert_mgmt").lower() == "true": self.manualCertsDir = self.promptForDir("Enter the path containing the manual certificates", mustExist=True) else: self.manualCertsDir = None @@ -1124,7 +1132,7 @@ def assistSettings(self) -> None: # We still use the old name for ODF (OCS) if self.getParam("cos_type") == "odf": - self.setParam("cos_type") == "ocs" + self.setParam("cos_type", "ocs") if self.getParam("cos_type") == "ibm": self.promptForString("IBM Cloud API Key", "cos_apikey", isPassword=True) @@ -1462,7 +1470,7 @@ def nonInteractiveMode(self) -> None: self.installAIService = False self.slsLicenseFileLocal = None - self.approvals = { + self.approvals: Dict[str, Dict[str, Any]] = { "approval_core": {"id": "suite-verify"}, # After Core Platform verification has completed "approval_assist": {"id": "app-cfg-assist"}, # After Assist workspace has been configured "approval_iot": {"id": "app-cfg-iot"}, # After IoT workspace has been configured @@ -1577,7 +1585,7 @@ def nonInteractiveMode(self) -> None: self.setParam("db2_action_aiservice", "install") self.installAIService = True # Set manage - bind - AI Service params same as provided AI Service's params - self.setParam("manage_bind_aiservice_instance_id", vars(self.args).get("aiservice_instance_id")) + self.setParam("manage_bind_aiservice_instance_id", vars(self.args).get("aiservice_instance_id", "")) self.setParam("manage_bind_aiservice_tenant_id", "user") elif key == "manage_bind_aiservice_instance_id": # only set if AI Service not being installed @@ -1658,14 +1666,14 @@ def nonInteractiveMode(self) -> None: elif key == "manual_certificates": if value is not None: - self.setParam("mas_manual_cert_mgmt", True) + self.setParam("mas_manual_cert_mgmt", "true") self.manualCertsDir = value else: - self.setParam("mas_manual_cert_mgmt", False) + self.setParam("mas_manual_cert_mgmt", "false") self.manualCertsDir = None elif key == "enable_ipv6": - self.setParam("enable_ipv6", True) + self.setParam("enable_ipv6", "true") elif key == "install_minio_aiservice": if vars(self.args).get("aiservice_instance_id"): @@ -1729,6 +1737,7 @@ def nonInteractiveMode(self) -> None: # Verifiy if any of the props that needs to be in a file are given if self.getParam("mas_ws_facilities_storage_log_size") != "" or self.getParam("mas_ws_facilities_storage_userfiles_size") != "" or self.getParam("mas_ws_facilities_db_maxconnpoolsize") or self.getParam("mas_ws_facilities_dwfagents"): self.selectLocalConfigDir() + assert self.localConfigDir is not None, "localConfigDir is None" facilitiesConfigsPath = path.join(self.localConfigDir, "facilities-configs.yaml") self.generateFacilitiesCfg(destination=facilitiesConfigsPath) self.setParam("mas_ws_facilities_config_map_name", "facilities-config") @@ -1826,6 +1835,7 @@ def install(self, argv): self.isAirgap() # Configure the installOptions for the appropriate architecture + assert self.architecture is not None, "Target architecture is not set" self.catalogOptions = supportedCatalogs[self.architecture] # Basic settings before the user provides any input @@ -1962,9 +1972,9 @@ def install(self, argv): # Based on the parameters set the annotations correctly self.configAnnotations() - self.displayInstallSummary() + continueWithInstall = True if not self.noConfirm: print() self.printDescription([ @@ -1973,7 +1983,7 @@ def install(self, argv): continueWithInstall = self.yesOrNo("Proceed with these settings") # Prepare the namespace and launch the installation pipeline - if self.noConfirm or continueWithInstall: + if continueWithInstall: self.createTektonFileWithDigest() self.printH1("Launch Install") diff --git a/python/src/mas/cli/install/argBuilder.py b/python/src/mas/cli/install/argBuilder.py index c2a174d733..a31b9865d6 100644 --- a/python/src/mas/cli/install/argBuilder.py +++ b/python/src/mas/cli/install/argBuilder.py @@ -92,10 +92,10 @@ def buildCommand(self) -> str: if self.operationalMode == 2: command += f" --non-prod{newline}" - if self.getParam('mas_trust_default_cas') == "false": + if self.getParam('mas_trust_default_cas').lower() == "false": command += f" --disable-ca-trust{newline}" - if self.getParam('mas_manual_cert_mgmt') is True: + if self.getParam('mas_manual_cert_mgmt').lower() == "true": command += f" --manual-certificates \"{self.manualCertsDir}\"{newline}" if self.getParam('mas_routing_mode') != "": @@ -104,7 +104,7 @@ def buildCommand(self) -> str: if self.getParam('mas_ingress_controller_name') != "": command += f" --ingress-controller \"{self.getParam('mas_ingress_controller_name')}\"{newline}" - if self.getParam('mas_configure_ingress') is True: + if self.getParam('mas_configure_ingress').lower() == "true": command += f" --configure-ingress{newline}" if self.getParam('mas_domain') != "": @@ -125,19 +125,19 @@ def buildCommand(self) -> str: if self.getParam('mas_cluster_issuer') != "": command += f" --mas-cluster-issuer \"{self.getParam('mas_cluster_issuer')}\"{newline}" - if self.getParam('mas_enable_walkme') == "false": + if self.getParam('mas_enable_walkme').lower() == "false": command += f" --disable-walkme{newline}" - if self.getParam('mas_feature_usage') == "false": + if self.getParam('mas_feature_usage').lower() == "false": command += f" --disable-feature-usage{newline}" - if self.getParam('mas_usability_metrics') == "false": + if self.getParam('mas_usability_metrics').lower() == "false": command += f" --disable-usability-metrics{newline}" - if self.getParam('mas_deployment_progression') == "false": + if self.getParam('mas_deployment_progression').lower() == "false": command += f" --disable-deployment-progression{newline}" - if self.getParam('enable_ipv6') is True: + if self.getParam('enable_ipv6').lower() == "true": command += f" --enable-ipv6{newline}" # Storage diff --git a/python/src/mas/cli/install/settings/additionalConfigs.py b/python/src/mas/cli/install/settings/additionalConfigs.py index 330e48ddd7..52f978f0a3 100644 --- a/python/src/mas/cli/install/settings/additionalConfigs.py +++ b/python/src/mas/cli/install/settings/additionalConfigs.py @@ -1,5 +1,5 @@ # ***************************************************************************** -# Copyright (c) 2024 IBM Corporation and other Contributors. +# Copyright (c) 2024, 2026 IBM Corporation and other Contributors. # # All rights reserved. This program and the accompanying materials # are made available under the terms of the Eclipse Public License v1.0 @@ -8,7 +8,7 @@ # # ***************************************************************************** -from typing import TYPE_CHECKING, Dict, List, Any +from typing import TYPE_CHECKING, Dict, List, Any, NoReturn from os import path from base64 import b64encode from glob import glob @@ -32,7 +32,7 @@ class AdditionalConfigsMixin(): noConfirm: bool templatesDir: str slsLicenseFileLocal: str | None - manualCertsDir: str + manualCertsDir: str | None showAdvancedOptions: bool additionalConfigsSecret: Dict[str, Any] | None podTemplatesSecret: Dict[str, Any] | None @@ -46,7 +46,7 @@ def setParam(self, param: str, value: str) -> None: def getParam(self, param: str) -> str: ... - def fatalError(self, message: str, exception: Exception | None = None) -> None: + def fatalError(self, message: str, exception: Exception | None = None) -> NoReturn: ... # Methods from PrintMixin @@ -196,7 +196,7 @@ def podTemplates(self) -> None: def manualCertificates(self) -> None: - if self.getParam("mas_manual_cert_mgmt"): + if self.getParam("mas_manual_cert_mgmt").lower() == "true": certsSecret = { "apiVersion": "v1", "kind": "Secret", diff --git a/python/src/mas/cli/install/settings/db2Settings.py b/python/src/mas/cli/install/settings/db2Settings.py index db19163b3e..bcbac3232d 100644 --- a/python/src/mas/cli/install/settings/db2Settings.py +++ b/python/src/mas/cli/install/settings/db2Settings.py @@ -1,5 +1,5 @@ # ***************************************************************************** -# Copyright (c) 2024 IBM Corporation and other Contributors. +# Copyright (c) 2024, 2026 IBM Corporation and other Contributors. # # All rights reserved. This program and the accompanying materials # are made available under the terms of the Eclipse Public License v1.0 @@ -31,6 +31,8 @@ class Db2SettingsMixin(): manageAppName: str showAdvancedOptions: bool localConfigDir: str | None + catalogDb2Channel: str + chosenCatalog: Dict[str, Any] | None # Methods from BaseApp def setParam(self, param: str, value: str) -> None: @@ -80,7 +82,7 @@ def promptForListSelect( def selectLocalConfigDir(self) -> None: ... - def generateJDBCCfg(self, **kwargs: Any) -> None: + def generateJDBCCfg(self, instanceId: str, scope: str, destination: str, appId: str = "", workspaceId: str = "") -> None: ... # In silentMode, no prompts will show up for "happy path" DB2 configuration scenarios. Prompts will still show up when an input is absolutely required diff --git a/python/src/mas/cli/install/settings/kafkaSettings.py b/python/src/mas/cli/install/settings/kafkaSettings.py index cf8328aa0f..50c2541bba 100644 --- a/python/src/mas/cli/install/settings/kafkaSettings.py +++ b/python/src/mas/cli/install/settings/kafkaSettings.py @@ -1,5 +1,5 @@ # ***************************************************************************** -# Copyright (c) 2024 IBM Corporation and other Contributors. +# Copyright (c) 2024, 2026 IBM Corporation and other Contributors. # # All rights reserved. This program and the accompanying materials # are made available under the terms of the Eclipse Public License v1.0 @@ -8,7 +8,7 @@ # # ***************************************************************************** -from typing import TYPE_CHECKING, Dict, List +from typing import TYPE_CHECKING, Dict, List, NoReturn from os import path from prompt_toolkit import print_formatted_text @@ -33,7 +33,7 @@ def setParam(self, param: str, value: str) -> None: def getParam(self, param: str) -> str: ... - def fatalError(self, message: str, exception: Exception | None = None) -> None: + def fatalError(self, message: str, exception: Exception | None = None) -> NoReturn: ... # Methods from PrintMixin diff --git a/python/src/mas/cli/install/settings/manageSettings.py b/python/src/mas/cli/install/settings/manageSettings.py index f46149233f..7f006559cf 100644 --- a/python/src/mas/cli/install/settings/manageSettings.py +++ b/python/src/mas/cli/install/settings/manageSettings.py @@ -1,5 +1,5 @@ # ***************************************************************************** -# Copyright (c) 2024 IBM Corporation and other Contributors. +# Copyright (c) 2024, 2026 IBM Corporation and other Contributors. # # All rights reserved. This program and the accompanying materials # are made available under the terms of the Eclipse Public License v1.0 @@ -8,7 +8,7 @@ # # ***************************************************************************** -from typing import TYPE_CHECKING, Dict, List +from typing import TYPE_CHECKING, Dict, List, NoReturn from prompt_toolkit.completion import WordCompleter from mas.cli.validators import LanguageValidator from mas.devops.aiservice import listAiServiceTenantInstances, listAiServiceInstances @@ -35,7 +35,10 @@ class ManageSettingsMixin(): installManage: bool installAIService: bool supportedLanguages: List[str] - dynamicClient: DynamicClient + + @property + def dynamicClient(self) -> DynamicClient: + ... # Methods from BaseApp def setParam(self, param: str, value: str) -> None: @@ -47,7 +50,7 @@ def getParam(self, param: str) -> str: def isSNO(self) -> bool: ... - def fatalError(self, message: str, exception: Exception | None = None) -> None: + def fatalError(self, message: str, exception: Exception | None = None) -> NoReturn: ... # Methods from PrintMixin diff --git a/python/src/mas/cli/install/settings/mongodbSettings.py b/python/src/mas/cli/install/settings/mongodbSettings.py index e6d4528c22..aee2ffe40c 100644 --- a/python/src/mas/cli/install/settings/mongodbSettings.py +++ b/python/src/mas/cli/install/settings/mongodbSettings.py @@ -1,5 +1,5 @@ # ***************************************************************************** -# Copyright (c) 2024 IBM Corporation and other Contributors. +# Copyright (c) 2024, 2026 IBM Corporation and other Contributors. # # All rights reserved. This program and the accompanying materials # are made available under the terms of the Eclipse Public License v1.0 @@ -8,7 +8,7 @@ # # ***************************************************************************** -from typing import TYPE_CHECKING, Dict, List, Any +from typing import TYPE_CHECKING, Dict, List from os import path from prompt_toolkit import print_formatted_text @@ -59,7 +59,7 @@ def promptForString( def selectLocalConfigDir(self) -> None: ... - def generateMongoCfg(self, **kwargs: Any) -> None: + def generateMongoCfg(self, instanceId: str, destination: str) -> None: ... def configMongoDb(self) -> None: diff --git a/python/src/mas/cli/mirror/__init__.py b/python/src/mas/cli/mirror/__init__.py new file mode 100644 index 0000000000..85df29608e --- /dev/null +++ b/python/src/mas/cli/mirror/__init__.py @@ -0,0 +1,11 @@ +# ***************************************************************************** +# Copyright (c) 2026 IBM Corporation and other Contributors. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Eclipse Public License v1.0 +# which accompanies this distribution, and is available at +# http://www.eclipse.org/legal/epl-v10.html +# +# ***************************************************************************** + +from ..cli import BaseApp # noqa: F401 diff --git a/python/src/mas/cli/mirror/app.py b/python/src/mas/cli/mirror/app.py new file mode 100644 index 0000000000..50f7f0aa85 --- /dev/null +++ b/python/src/mas/cli/mirror/app.py @@ -0,0 +1,747 @@ +#!/usr/bin/env python +# ***************************************************************************** +# Copyright (c) 2026 IBM Corporation and other Contributors. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Eclipse Public License v1.0 +# which accompanies this distribution, and is available at +# http://www.eclipse.org/legal/epl-v10.html +# +# ***************************************************************************** + +import base64 +import json +import logging +import re +import selectors +import shutil +import subprocess +import yaml +import urllib.request +import urllib.error +from typing import List, Dict, Optional +from dataclasses import dataclass +from os import path, environ, makedirs + +from alive_progress import alive_bar +from prompt_toolkit import print_formatted_text, HTML + +from mas.devops.data import getCatalog + +from ..cli import BaseApp +from .argParser import mirrorArgParser +from .config import PACKAGE_CONFIGS + + +logger = logging.getLogger(__name__) + + +def logMethodCall(func): + def wrapper(self, *args, **kwargs): + logger.debug(f">>> MirrorApp.{func.__name__}") + result = func(self, *args, **kwargs) + logger.debug(f"<<< MirrorApp.{func.__name__}") + return result + return wrapper + + +@dataclass +class MirrorResult: + """Result of a mirror operation.""" + images: int + mirrored: int + + @property + def success(self) -> bool: + """ + Determine if the mirror operation was successful. + + Returns: + True if all images were mirrored successfully, False otherwise. + """ + return self.images != 0 and self.images == self.mirrored + + +def stripLogPrefix(line: str) -> str: + """ + Strip timestamp and log level prefix from command output. + + Handles format: "2026/02/02 18:12:25 [INFO] : {actual message}" + Removes everything up to and including the first ": " after a log level. + + Args: + line: The log line to process + + Returns: + The line with prefix stripped, or original line if no match + """ + # Check if line starts with a timestamp pattern (with or without ANSI codes) + # If it does, find the first ": " after a log level and remove everything before it + if re.match(r'^.*?\d{4}/\d{2}/\d{2}\s+\d{2}:\d{2}:\d{2}', line): + # Find position of ": " after the log level + # Split on first occurrence of ": " that comes after a bracket + parts = line.split(': ', 1) + if len(parts) == 2 and '[' in parts[0]: + return parts[1] + + return line + + +def countImagesInConfig(configPath: str) -> int: + """ + Parse YAML config file and count images in mirror.additionalImages. + + Args: + configPath: Path to the YAML configuration file + + Returns: + Number of images to be mirrored, or 0 if parsing fails + """ + try: + with open(configPath, 'r') as f: + config = yaml.safe_load(f) + + additionalImages = config.get('mirror', {}).get('additionalImages', []) + imageCount = len(additionalImages) + logger.debug(f"Found {imageCount} images in {configPath}") + return imageCount + except FileNotFoundError: + logger.error(f"Config file not found: {configPath}") + return 0 + except yaml.YAMLError as e: + logger.error(f"Failed to parse YAML config {configPath}: {e}") + return 0 + except Exception as e: + logger.error(f"Unexpected error reading config {configPath}: {e}") + return 0 + + +def getISC(configPath: str) -> str: + """ + Get the Image Set Config file, downloading from GitHub if it doesn't exist locally. + + The config file is expected to be in ~/.ibm-mas/{configPath}. + If the file doesn't exist, it will be downloaded from: + https://github.com/ibm-mas/image-set-configs/blob/master/{configPath} + + Args: + configPath: Relative path to the config file (e.g., "catalogs/v9-260129-amd64.yaml") + + Returns: + Full path to the local config file + + Raises: + FileNotFoundError: If the file doesn't exist and cannot be downloaded + """ + # Get home directory + homeDir = environ.get('HOME') or environ.get('USERPROFILE') or '' + if not homeDir: + raise FileNotFoundError("Could not determine home directory") + + # Construct full local path with .ibm-mas prefix + localPath = path.join(homeDir, '.ibm-mas', 'image-set-configs', configPath) + + # If file exists, return it + if path.exists(localPath): + logger.info(f"Using existing config file: {localPath}") + return localPath + + # File doesn't exist, try to download it + logger.info(f"Config file not found locally: {localPath}") + + # Construct GitHub raw content URL + # Convert blob URL to raw content URL + githubUrl = f"https://raw.githubusercontent.com/ibm-mas/image-set-configs/master/{configPath}" + + logger.info(f"Attempting to download from: {githubUrl}") + + try: + # Create directory if it doesn't exist + localDir = path.dirname(localPath) + makedirs(localDir, exist_ok=True) + + # Download the file + with urllib.request.urlopen(githubUrl) as response: + content = response.read() + + # Write to local file + with open(localPath, 'wb') as f: + f.write(content) + + logger.info(f"Successfully downloaded config file to: {localPath}") + return localPath + + except urllib.error.HTTPError as e: + logger.error(f"Failed to download config file from GitHub: HTTP {e.code} - {e.reason}") + raise FileNotFoundError(f"Config file not found locally and could not be downloaded from GitHub: {configPath}") from e + except urllib.error.URLError as e: + logger.error(f"Failed to download config file from GitHub: {e.reason}") + raise FileNotFoundError(f"Config file not found locally and could not be downloaded from GitHub: {configPath}") from e + except Exception as e: + logger.error(f"Unexpected error downloading config file: {e}") + raise FileNotFoundError(f"Config file not found locally and could not be downloaded from GitHub: {configPath}") from e + + +def _processStreams(process: subprocess.Popen, resultData: Dict, progressBar=None) -> None: + """ + Process stdout and stderr streams from a subprocess using selectors. + + Uses non-blocking I/O to efficiently read from both streams without threading. + Filters output and captures result information. + + Args: + process: The subprocess.Popen object with stdout and stderr pipes + resultData: Dictionary to store captured result information + progressBar: Optional alive-progress bar instance to update on image copy success + """ + # Ensure streams are available + if process.stdout is None or process.stderr is None: + return + + # Compile filter patterns into a single case-insensitive regex for performance + filterPatterns = [ + "Hello, welcome to oc-mirror", + "setting up the environment for you...", + "using digest to pull, but tag only for mirroring" + ] + # Escape special regex characters and join with OR operator + filterRegex = re.compile('|'.join(re.escape(pattern) for pattern in filterPatterns), re.IGNORECASE) + + # Set up selector for non-blocking I/O + sel = selectors.DefaultSelector() + sel.register(process.stdout, selectors.EVENT_READ, data='stdout') + sel.register(process.stderr, selectors.EVENT_READ, data='stderr') + + # Track which streams are still open (store file objects, not selectors) + streamsOpen = {process.stdout.fileno(), process.stderr.fileno()} + + while streamsOpen: + # Wait for data to be available on any stream + events = sel.select(timeout=0.1) + + for key, _ in events: + streamType = key.data + + # Get the actual file object from the key + if streamType == 'stdout': + stream = process.stdout + else: + stream = process.stderr + + if stream is None: + continue + + line = stream.readline() + + if not line: + # Stream closed + streamsOpen.discard(stream.fileno()) + sel.unregister(stream) + continue + + lineStripped = line.rstrip() + + # Capture result information BEFORE stripping prefix + resultMatch = re.search(r'(\d+)\s+/\s+(\d+)\s+additional images mirrored successfully', lineStripped) + if resultMatch: + resultData['mirrored'] = int(resultMatch.group(1)) + resultData['images'] = int(resultMatch.group(2)) + logger.debug(f"Captured result: {resultData['mirrored']}/{resultData['images']}") + + # Detect "Success copying" and update progress bar + successMatch = re.search(r'Success copying .+ ➡️', lineStripped) + if successMatch and progressBar is not None: + progressBar() # Increment progress bar + logger.debug("Progress bar incremented") + + # Strip duplicate timestamp/level prefix from command output + cleanLine = stripLogPrefix(lineStripped) + + # Skip lines matching the filter regex (case-insensitive) + if not filterRegex.search(lineStripped): + # Log to appropriate level based on stream + if streamType == 'stdout': + logger.debug(cleanLine) + else: + logger.error(cleanLine) + + sel.close() + + +def runCommand(cmd: List[str], progressBar=None) -> tuple[int, Dict]: + """ + Execute a command and stream output/errors in real-time. + + Args: + cmd: List of command arguments to execute + progressBar: Optional alive-progress bar instance to update on image copy success + + Returns: + Tuple of (exitCode, resultData) where resultData contains captured information + """ + logger.info(f"Executing: {' '.join(cmd)}") + + # Dictionary to capture result data from output + resultData = {} + + try: + with subprocess.Popen( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + bufsize=1 # Line buffered for real-time output + ) as process: + # Process streams using selectors for efficient non-blocking I/O + _processStreams(process, resultData, progressBar) + + # Wait for process to complete + returnCode = process.wait() + + if returnCode != 0: + logger.error(f"Command failed with exit code {returnCode}") + + return returnCode, resultData + + except Exception as e: + logger.error(f"Error executing command: {e}") + return 1, {} + + +def _executeMirror(configPath: str, displayName: str, workspacePath: str, mode: str, + targetRegistry: str = "", ocMirrorPath: str = "oc-mirror", + authFilePath: Optional[str] = None, rootDir: str = "", + destTlsVerify: bool = True, imageTimeout: str = "20m") -> MirrorResult: + """ + Execute the mirror operation for a given configuration. + + This is a common function used by both mirrorPackage and mirrorCatalog. + + Args: + configPath: Path to the YAML configuration file + displayName: Display name for progress bar (e.g., "ibm-mas v9.0.5 (amd64)" or "catalog v9-260129-amd64") + workspacePath: Workspace path for the mirror operation (e.g., "package/arch/version" or "catalog/version") + mode: Mirror mode ("m2m", "m2d", or "d2m") + targetRegistry: Target registry for m2m and d2m modes + ocMirrorPath: Path to oc-mirror binary (default: "oc-mirror") + authFilePath: Path to authentication file (default: ~/.ibm-mas/auth.json) + rootDir: Root directory for mirror operations (workspace for m2m, disk storage for m2d/d2m) + destTlsVerify: Verify TLS certificates for destination registry (default: True) + imageTimeout: Timeout for image operations (default: "20m") + + Returns: + MirrorResult object with images, mirrored, and success status. + Returns images=0, mirrored=0, success=False if operation failed or results couldn't be parsed. + """ + logger.info(f"Using configuration: {configPath}") + + # Set default auth file path if not provided + if authFilePath is None: + homeDir = environ.get('HOME') or environ.get('USERPROFILE') or '' + authFilePath = path.join(homeDir, '.ibm-mas', 'auth.json') + + # Count images in config file + totalImages = countImagesInConfig(configPath) + if totalImages == 0: + logger.error(f"No images found in config or failed to parse: {configPath}") + print(f"❌ {displayName} - No images found in config") + return MirrorResult(images=0, mirrored=0) + + logger.info(f"Found {totalImages} images to mirror") + + # Build TLS verify flag + tlsVerifyFlag = f"--dest-tls-verify={'true' if destTlsVerify else 'false'}" + + if mode == "m2m": + cmd = [ + ocMirrorPath, "--v2", "--config", configPath, "--authfile", authFilePath, + "--workspace", f"file://{rootDir}/{workspacePath}", + tlsVerifyFlag, "--image-timeout", imageTimeout, + f"docker://{targetRegistry}" + ] + elif mode == "m2d": + cmd = [ + ocMirrorPath, "--v2", "--config", configPath, "--authfile", authFilePath, + "--image-timeout", imageTimeout, + f"file://{rootDir}/{workspacePath}", + ] + elif mode == "d2m": + cmd = [ + ocMirrorPath, "--v2", "--config", configPath, "--authfile", authFilePath, + "--from", f"file://{rootDir}/{workspacePath}", + tlsVerifyFlag, "--image-timeout", imageTimeout, + f"docker://{targetRegistry}" + ] + else: + logger.error(f"Unsupported mirror mode: {mode}") + print(f"❌ {displayName} - Unsupported mirror mode: {mode}") + return MirrorResult(images=0, mirrored=0) + + # Execute command with progress bar + # Use fixed-width title (50 chars) for alignment, with in-progress icon + barTitleBase = displayName.ljust(50) + barTitle = f"{barTitleBase} ⏳" + with alive_bar(totalImages, title=barTitle, length=20, enrich_print=False) as bar: + exitCode, resultData = runCommand(cmd, progressBar=bar) + + # Update bar title with status icon after completion + if exitCode != 0: + bar.title = f"{barTitleBase} ❌" + logger.error(f"Mirror operation failed with exit code {exitCode}") + return MirrorResult(images=0, mirrored=0) + + # Create result object from captured data + if 'images' in resultData and 'mirrored' in resultData: + result = MirrorResult( + images=resultData['images'], + mirrored=resultData['mirrored'] + ) + logger.info(f"Mirror operation completed: {result.mirrored}/{result.images} images mirrored (success={result.success})") + + if result.success: + bar.title = f"{barTitleBase} ✅" + else: + bar.title = f"{barTitleBase} ⚠️" + + return result + else: + bar.title = f"{barTitleBase} ⚠️" + logger.warning("Mirror operation completed but could not parse result statistics") + return MirrorResult(images=0, mirrored=0) + + +def mirrorPackage(package: str, version: str, arch: str, mode: str, + targetRegistry: str = "", flag: bool = True, + ocMirrorPath: str = "oc-mirror", authFilePath: Optional[str] = None, + rootDir: str = "", destTlsVerify: bool = True, + imageTimeout: str = "20m") -> MirrorResult: + """ + Mirror a package and return the result. + + Args: + package: Package name (e.g., "ibm-mas") + version: Package version (e.g., "9.0.5") + arch: Architecture (e.g., "amd64") + mode: Mirror mode ("m2m", "m2d", or "d2m") + targetRegistry: Target registry for m2m and d2m modes + flag: Whether to actually perform the mirror operation + ocMirrorPath: Path to oc-mirror binary (default: "oc-mirror") + authFilePath: Path to authentication file (default: ~/.ibm-mas/auth.json) + rootDir: Root directory for mirror operations (workspace for m2m, disk storage for m2d/d2m) + destTlsVerify: Verify TLS certificates for destination registry (default: True) + imageTimeout: Timeout for image operations (default: "20m") + + Returns: + MirrorResult object with images, mirrored, and success status. + Returns images=0, mirrored=0, success=False if operation failed or results couldn't be parsed. + """ + # Extract major.minor version (first two components) + versionParts = version.split('.') + + # Validate version format + if len(versionParts) < 2: + logger.error(f"Invalid version format: '{version}'. Expected format: 'major.minor.patch' (e.g., '9.0.5')") + return MirrorResult(images=0, mirrored=0) + + majorMinor = f"{versionParts[0]}.{versionParts[1]}" + + if not flag: + logger.info(f"Skipping {package} version {version} for {arch} architecture") + # Add empty progress bar to align with other status messages + emptyBar = "|" + " " * 20 + "|" + print(f"{package} v{version} ({arch})".ljust(50) + f" ⏭️ {emptyBar} Mirroring disabled by user") + return MirrorResult(images=0, mirrored=0) + + logger.info(f"Mirroring {package} version {version} for {arch} architecture") + + # Get or download the config file + relativeConfigPath = f"packages/{package}/{majorMinor}/{arch}/{package}-{version}-{arch}.yaml" + try: + configPath = getISC(relativeConfigPath) + except FileNotFoundError as e: + logger.error(f"Failed to get config file: {e}") + print(f"❌ {package} v{version} ({arch}) - Config file not found") + return MirrorResult(images=0, mirrored=0) + + displayName = f"{package} v{version} ({arch})" + workspacePath = f"{package}/{arch}/{version}" + + return _executeMirror(configPath, displayName, workspacePath, mode, targetRegistry, + ocMirrorPath, authFilePath, rootDir, destTlsVerify, imageTimeout) + + +def mirrorCatalog(version: str, mode: str, targetRegistry: str = "", + ocMirrorPath: str = "oc-mirror", authFilePath: Optional[str] = None, + rootDir: str = "", destTlsVerify: bool = True, + imageTimeout: str = "20m") -> MirrorResult: + """ + Mirror a catalog and return the result. + + Args: + version: Catalog version (e.g., "v9-260129-amd64") + mode: Mirror mode ("m2m", "m2d", or "d2m") + targetRegistry: Target registry for m2m and d2m modes + ocMirrorPath: Path to oc-mirror binary (default: "oc-mirror") + authFilePath: Path to authentication file (default: ~/.ibm-mas/auth.json) + rootDir: Root directory for mirror operations (workspace for m2m, disk storage for m2d/d2m) + destTlsVerify: Verify TLS certificates for destination registry (default: True) + imageTimeout: Timeout for image operations (default: "20m") + + Returns: + MirrorResult object with images, mirrored, and success status. + Returns images=0, mirrored=0, success=False if operation failed or results couldn't be parsed. + """ + logger.info(f"Mirroring catalog {version}") + + # Get or download the config file + relativeConfigPath = f"catalogs/{version}.yaml" + try: + configPath = getISC(relativeConfigPath) + except FileNotFoundError as e: + logger.error(f"Failed to get config file: {e}") + print(f"❌ catalog {version} - Config file not found") + return MirrorResult(images=0, mirrored=0) + + displayName = f"catalog {version}" + workspacePath = f"catalog/{version}" + + return _executeMirror(configPath, displayName, workspacePath, mode, targetRegistry, + ocMirrorPath, authFilePath, rootDir, destTlsVerify, imageTimeout) + + +def validateEnvironmentVariables(mode: str, targetRegistry: str) -> None: + """ + Validate that required environment variables are set based on the mirror mode. + + Args: + mode: Mirror mode ("m2m", "m2d", or "d2m") + targetRegistry: Target registry for m2m and d2m modes + + Raises: + ValueError: If required environment variables are not set + """ + missingVars = [] + + # Check for target registry credentials (m2m or d2m) + if mode in ["m2m", "d2m"]: + if not environ.get('REGISTRY_USERNAME'): + missingVars.append('REGISTRY_USERNAME') + if not environ.get('REGISTRY_PASSWORD'): + missingVars.append('REGISTRY_PASSWORD') + + # Check for IBM Entitlement Key (m2m or m2d) + if mode in ["m2m", "m2d"]: + if not environ.get('IBM_ENTITLEMENT_KEY'): + missingVars.append('IBM_ENTITLEMENT_KEY') + + if missingVars: + raise ValueError(f"Missing required environment variables: {', '.join(missingVars)}") + + +def generateAuthFile(mode: str, targetRegistry: str) -> str: + """ + Generate an authentication file from environment variables. + + Args: + mode: Mirror mode ("m2m", "m2d", or "d2m") + targetRegistry: Target registry for m2m and d2m modes + + Returns: + Path to the generated auth file + + Raises: + ValueError: If required environment variables are not set + """ + # Validate environment variables first + validateEnvironmentVariables(mode, targetRegistry) + + # Get home directory + homeDir = environ.get('HOME') or environ.get('USERPROFILE') or '' + if not homeDir: + raise ValueError("Could not determine home directory") + + # Create auth file path + authFilePath = path.join(homeDir, '.ibm-mas', 'auth.json') + authDir = path.dirname(authFilePath) + + # Create directory if it doesn't exist + makedirs(authDir, exist_ok=True) + + # Build auth configuration + authConfig = {} + + # Add target registry credentials (m2m or d2m) + if mode in ["m2m", "d2m"]: + registryUsername = environ.get('REGISTRY_USERNAME', '') + registryPassword = environ.get('REGISTRY_PASSWORD', '') + authString = f"{registryUsername}:{registryPassword}" + authBase64 = base64.b64encode(authString.encode()).decode() + authConfig[targetRegistry] = { + "auth": authBase64 + } + + # Add IBM Entitlement Key (m2m or m2d) + if mode in ["m2m", "m2d"]: + ibmEntitlementKey = environ.get('IBM_ENTITLEMENT_KEY', '') + authString = f"cp:{ibmEntitlementKey}" + authBase64 = base64.b64encode(authString.encode()).decode() + authConfig["cp.icr.io/cp"] = { + "auth": authBase64 + } + + auths = { + "auths": authConfig + } + + # Write auth file + with open(authFilePath, 'w') as f: + json.dump(auths, f, indent=2) + + logger.info(f"Generated auth file: {authFilePath}") + return authFilePath + + +class MirrorApp(BaseApp): + + @logMethodCall + def interactiveMode(self, simplified: bool, advanced: bool) -> None: + # Interactive mode + self._interactiveMode = True + + @logMethodCall + def nonInteractiveMode(self) -> None: + self._interactiveMode = False + + @logMethodCall + def mirror(self, argv): + """ + Main mirror function that orchestrates the mirroring of catalogs and packages. + + Args: + argv: Command line arguments + """ + args = mirrorArgParser.parse_args(args=argv) + + catalogVersion = args.catalog + release = args.release + mode = args.mode + targetRegistry = args.target_registry or "" + authFile = args.authfile + rootDir = args.dir + destTlsVerify = args.dest_tls_verify + imageTimeout = args.image_timeout + mirrorAll = args.all + + # Validate that oc-mirror is available on PATH + if not shutil.which("oc-mirror"): + logger.error("oc-mirror executable not found on PATH") + self.fatalError("oc-mirror executable not found on PATH. Please install oc-mirror and ensure it is available in your PATH.") + return + + # Validate that --target-registry is provided for m2m and d2m modes + if mode in ["m2m", "d2m"] and not targetRegistry: + logger.error(f"--target-registry is required when mode is '{mode}'") + self.fatalError(f"--target-registry is required when mode is '{mode}'") + return + + # Handle authfile parameter + if authFile: + # Validate that the file exists + if not path.exists(authFile): + logger.error(f"Auth file does not exist: {authFile}") + self.fatalError(f"Auth file does not exist: {authFile}") + return + logger.info(f"Using provided auth file: {authFile}") + authFilePath = authFile + else: + # Generate auth file from environment variables + try: + authFilePath = generateAuthFile(mode, targetRegistry) + except ValueError as e: + logger.error(f"Failed to generate auth file: {e}") + self.fatalError(f"Failed to generate auth file: {e}") + return + + catalog = getCatalog(catalogVersion) + if catalog is None: + self.fatalError(f"Catalog {catalogVersion} not found") + else: + arch = catalogVersion.split("-")[-1] + + self.printH2("Mirror Configuration") + self.printSummary("Catalog", catalogVersion) + self.printSummary("Architecture", arch) + self.printSummary("Release", release) + self.printSummary("Mode", mode) + self.printSummary("Authentication File", authFilePath) + + self.printH2("Mirror Target") + if mode == "m2d": + self.printSummary("Destination", rootDir) + else: + self.printSummary("Destination", targetRegistry) + self.printSummary("Verify Registry Certificate", destTlsVerify) + self.printSummary("Mirror Image Timeout", imageTimeout) + + self.printH2("IBM Maximo Operator Catalog") + mirrorCatalog( + version=catalogVersion, + mode=mode, + targetRegistry=targetRegistry, + authFilePath=authFilePath, + rootDir=rootDir, + destTlsVerify=destTlsVerify, + imageTimeout=imageTimeout + ) + + # Mirror each package with common parameters using shared configuration + currentGroup = None + for group, argName, packageName, catalogKey in PACKAGE_CONFIGS: + # Print section header when group changes + if group != currentGroup: + self.printH2(group) + currentGroup = group + + # Get version from catalog - handle both direct keys and release-specific keys + perReleaseVersions = [ + "aiservice_version", + "mas_core_version", + "mas_assist_version", + "mas_iot_version", + "mas_facilities_version", + "mas_manage_version", + "mas_monitor_version", + "mas_predict_version", + "mas_optimizer_version", + "mas_visualinspection_version" + ] + if catalogKey in perReleaseVersions: + version = catalog[catalogKey][release] + else: + version = catalog[catalogKey] + + # Remove any +buildnum properties from the version in the metadata file + try: + version = version.split("+")[0] + except AttributeError: + # This likely means we have the perReleaseVersions configuration incorrect + logger.exception(f"Failed to parse version for {packageName} ({catalogKey}) from catalog: {catalogVersion}") + raise + + # Get the flag value from args, or use mirrorAll if --all is set + flag = mirrorAll or getattr(args, argName.replace("-", "_")) + + mirrorPackage( + package=packageName, + version=version, + arch=arch, + mode=mode, + targetRegistry=targetRegistry, + flag=flag, + authFilePath=authFilePath, + rootDir=rootDir, + destTlsVerify=destTlsVerify, + imageTimeout=imageTimeout + ) + + print_formatted_text(HTML("\n✅ Mirror operation completed")) diff --git a/python/src/mas/cli/mirror/argParser.py b/python/src/mas/cli/mirror/argParser.py new file mode 100644 index 0000000000..32ff4ff34e --- /dev/null +++ b/python/src/mas/cli/mirror/argParser.py @@ -0,0 +1,169 @@ +# ***************************************************************************** +# Copyright (c) 2026 IBM Corporation and other Contributors. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Eclipse Public License v1.0 +# which accompanies this distribution, and is available at +# http://www.eclipse.org/legal/epl-v10.html +# +# ***************************************************************************** + +from collections import defaultdict +import argparse +import re +from itertools import groupby + +from .config import PACKAGE_CONFIGS +from .. import __version__ as packageVersion +from ..cli import getHelpFormatter + + +def validate_timeout(value): + """ + Validate that the timeout string is in a valid format. + + Valid formats: "1h20m10s", "1h", "20m", "10s", "1h20m", etc. + + Args: + value: The timeout string to validate + + Returns: + The validated timeout string + + Raises: + argparse.ArgumentTypeError: If the format is invalid + """ + # Pattern matches combinations of hours (h), minutes (m), and seconds (s) + # Must have at least one unit and units must be in order (h, m, s) + pattern = r'^(\d+h)?(\d+m)?(\d+s)?$' + + if not re.match(pattern, value): + raise argparse.ArgumentTypeError( + f"Invalid timeout format: '{value}'. " + "Expected format: combinations of hours (h), minutes (m), and seconds (s), " + "e.g., '1h20m10s', '1h', '20m', '10s'" + ) + + # Ensure at least one unit is present + if not any(unit in value for unit in ['h', 'm', 's']): + raise argparse.ArgumentTypeError( + f"Invalid timeout format: '{value}'. " + "Must include at least one time unit (h, m, or s)" + ) + + return value + + +mirrorArgParser = argparse.ArgumentParser( + prog="mas mirror", + description="\n".join([ + f"IBM Maximo Application Suite Admin CLI v{packageVersion}", + "Mirror IBM Maximo content to a private container registry.", + ]), + epilog="Refer to the online documentation for more information: https://ibm-mas.github.io/cli/", + formatter_class=getHelpFormatter(), + add_help=False +) + +mainGroup = mirrorArgParser.add_argument_group("Primary Configuration") +mainGroup.add_argument( + "--catalog", + required=True, + help="Catalog version (e.g., v9-240625-amd64, v9-260129-amd64)" +) +mainGroup.add_argument( + "--release", + required=True, + help="MAS release version", + choices=["8.10.x", "8.11.x", "9.0.x", "9.1.x"] +) +mainGroup.add_argument( + "--mode", + required=True, + help="Mirror mode", + choices=["m2m", "m2d", "d2m"] +) +mainGroup.add_argument( + "--target-registry", + required=False, + type=str, + help="Target registry for m2m and d2m modes (e.g., registry.example.com/namespace)" +) +mainGroup.add_argument( + "--dir", + required=True, + type=str, + help="Root directory for mirror operations (workspace for m2m, disk storage for m2d/d2m)" +) +mainGroup.add_argument( + "--authfile", + required=False, + type=str, + help="Path to authentication file (must exist). If not provided, will be generated from environment variables (REGISTRY_USERNAME, REGISTRY_PASSWORD, and IBM_ENTITLEMENT_KEY)." +) + +# Add package-specific arguments dynamically, organized by group +# First, deduplicate by argName and aggregate package names + +# Group configs by (groupName, argName) to deduplicate and aggregate packages +arg_map = defaultdict(list) +for group, argName, packageName, versionKey in PACKAGE_CONFIGS: + arg_map[(group, argName)].append(packageName) + +# Now create arguments with deduplicated argNames and aggregated package lists +for groupName, groupItems in groupby(PACKAGE_CONFIGS, key=lambda x: x[0]): + argGroup = mirrorArgParser.add_argument_group(groupName) + # Track which argNames we've already added to this group + added_args = set() + + for group, argName, packageName, _ in groupItems: + if argName not in added_args: + added_args.add(argName) + # Get all package names for this argName + packages = arg_map[(group, argName)] + + # Create help text based on number of packages + if len(packages) == 1: + help_text = f"Mirror images for the {packages[0]} package" + else: + package_list = ", ".join(packages) + help_text = f"Mirror images for packages: {package_list}" + + argGroup.add_argument( + f"--{argName}", + required=False, + help=help_text, + action="store_true" + ) + +advancedGroup = mirrorArgParser.add_argument_group("Advanced Configuration") +advancedGroup.add_argument( + "--all", + required=False, + action="store_true", + help="Mirror all packages for the chosen release" +) +advancedGroup.add_argument( + "--dest-tls-verify", + required=False, + type=lambda x: x.lower() == 'true', + default=True, + help="Verify TLS certificates for destination registry (default: true)" +) +advancedGroup.add_argument( + "--image-timeout", + required=False, + type=validate_timeout, + default="20m", + help="Timeout for image operations (e.g., '1h20m10s', '1h', '20m', default: '20m')" +) + +otherArgGroup = mirrorArgParser.add_argument_group( + "More" +) +otherArgGroup.add_argument( + "-h", "--help", + action="help", + default=False, + help="Show this help message and exit" +) diff --git a/python/src/mas/cli/mirror/config.py b/python/src/mas/cli/mirror/config.py new file mode 100644 index 0000000000..cdde7374b0 --- /dev/null +++ b/python/src/mas/cli/mirror/config.py @@ -0,0 +1,50 @@ +# ***************************************************************************** +# Copyright (c) 2026 IBM Corporation and other Contributors. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Eclipse Public License v1.0 +# which accompanies this distribution, and is available at +# http://www.eclipse.org/legal/epl-v10.html +# +# ***************************************************************************** + +PACKAGE_CONFIGS = [ + ("Required Dependencies", "sls", "ibm-sls", "sls_version"), + ("Required Dependencies", "tsm", "ibm-truststore-mgr", "tsm_version"), + + ("Optional Dependencies", "amlen", "amlen", "amlen_extras_version"), + + ("Optional Dependencies", "aiservice", "ibm-aiservice", "aiservice_version"), + ("Optional Dependencies", "data-dictionary", "ibm-data-dictionary", "dd_version"), + + ("Optional Dependencies", "db2u-s11", "ibm-db2uoperator-s11", "db2u_version"), + ("Optional Dependencies", "db2u-s12", "ibm-db2uoperator-s12", "db2u_version"), + + ("Optional Dependencies", "mongodb-ce", "mongodb-ce", "mongo_extras_version_default"), + + ("Maximo Application Suite", "core", "ibm-mas", "mas_core_version"), + ("Maximo Application Suite", "assist", "ibm-mas-assist", "mas_assist_version"), + ("Maximo Application Suite", "iot", "ibm-mas-iot", "mas_iot_version"), + ("Maximo Application Suite", "facilities", "ibm-mas-facilities", "mas_facilities_version"), + ("Maximo Application Suite", "manage", "ibm-mas-manage", "mas_manage_version"), + ("Maximo Application Suite", "manage-icd", "ibm-mas-manage-icd", "mas_manage_version"), + ("Maximo Application Suite", "monitor", "ibm-mas-monitor", "mas_monitor_version"), + ("Maximo Application Suite", "predict", "ibm-mas-predict", "mas_predict_version"), + ("Maximo Application Suite", "optimizer", "ibm-mas-optimizer", "mas_optimizer_version"), + ("Maximo Application Suite", "visualinspection", "ibm-mas-visualinspection", "mas_visualinspection_version"), + + ("Cloud Pak for Data - Platform", "cp4d-platform", "ibm-cp-common-services", "common_svcs_version"), + ("Cloud Pak for Data - Platform", "cp4d-platform", "ibm-zen", "ibm_zen_version"), + ("Cloud Pak for Data - Platform", "cp4d-platform", "ibm-cp-datacore", "cp4d_platform_version"), + ("Cloud Pak for Data - Platform", "cp4d-platform", "ibm-licensing", "ibm_licensing_version"), + ("Cloud Pak for Data - Platform", "cp4d-platform", "ibm-ccs", "ccs_build"), + ("Cloud Pak for Data - Platform", "cp4d-platform", "ibm-cloud-native-postgresql", "postgress_version"), + ("Cloud Pak for Data - Platform", "cp4d-platform", "ibm-datarefinery", "datarefinery_version"), + ("Cloud Pak for Data - Platform", "cp4d-platform", "ibm-elasticsearch-operator", "elasticsearch_version"), + ("Cloud Pak for Data - Platform", "cp4d-platform", "ibm-opensearch-operator", "opensearch_version"), + ("Cloud Pak for Data - WSL", "cp4d-wsl", "ibm-wsl", "wsl_version"), + ("Cloud Pak for Data - WSL", "cp4d-wsl", "ibm-wsl-runtimes", "wsl_runtimes_version"), + ("Cloud Pak for Data - WML", "cp4d-wml", "ibm-wml-cpd", "wml_version"), + ("Cloud Pak for Data - Spark", "cp4d-spark", "ibm-analyticsengine", "spark_version"), + ("Cloud Pak for Data - Cognos", "cp4d-cognos", "ibm-cognos-analytics-prod", "cognos_version"), +] diff --git a/python/test/mirror/README.md b/python/test/mirror/README.md new file mode 100644 index 0000000000..a722f11a89 --- /dev/null +++ b/python/test/mirror/README.md @@ -0,0 +1,296 @@ +# Mirror Command Unit Tests + +This directory contains unit tests for the `mas mirror` command, following the patterns established by the install test helper. + +## Overview + +The mirror test framework provides a structured way to test the mirror command functionality without requiring actual container registries, oc-mirror binary execution, or network access. All external dependencies are mocked. + +## Test File Organization + +The test suite is organized into modular files by category for better maintainability: + +### Test Files (23 tests total) + +1. **test_mirror_basic.py** (5 tests) + - Basic mirror operations across different modes (m2d, m2m, d2m) + - Custom auth file usage + - Config file download from GitHub + +2. **test_mirror_errors.py** (7 tests) + - Missing oc-mirror binary + - Invalid catalog version + - Missing credentials (IBM entitlement key, registry credentials) + - Command failures and partial failures + - Timeout handling + +3. **test_mirror_auth.py** (3 tests) + - Authentication file generation for m2d mode + - Authentication file generation for m2m mode + - Authentication file generation for d2m mode + +4. **test_mirror_config.py** (3 tests) + - Successful config download from GitHub + - Config download failure handling + - Invalid YAML config handling + +5. **test_mirror_packages.py** (3 tests) + - Mirroring all available packages + - Selective package mirroring + - DB2 package special handling (db2u-s11, db2u-s12) + +6. **test_mirror_advanced.py** (2 tests) + - TLS verification disabled + - Custom image timeout + +## Architecture + +### Test Helper Components + +1. **MirrorTestConfig** (`test/utils/mirror_test_helper.py`) + - Dataclass that defines test scenario configuration + - Specifies mirror mode, catalog version, packages, and expected behavior + - Automatically builds command line arguments from configuration + +2. **MirrorTestHelper** (`test/utils/mirror_test_helper.py`) + - Main test execution class + - Sets up all necessary mocks (subprocess, file I/O, network) + - Includes watchdog thread to detect hanging tests + - Validates command construction and execution + +3. **run_mirror_test()** (`test/utils/mirror_test_helper.py`) + - Convenience function to run a test with minimal code + - Takes tmpdir and MirrorTestConfig as parameters + +## Running Tests + +Run all mirror tests: +```bash +pytest test/mirror/ +``` + +Run specific test file: +```bash +pytest test/mirror/test_mirror_basic.py +pytest test/mirror/test_mirror_errors.py +pytest test/mirror/test_mirror_auth.py +pytest test/mirror/test_mirror_config.py +pytest test/mirror/test_mirror_packages.py +pytest test/mirror/test_mirror_advanced.py +``` + +Run specific test: +```bash +pytest test/mirror/test_mirror_basic.py::test_mirror_m2d_catalog_only +``` + +## Test Scenarios + +### test_mirror_m2d_catalog_only +**Purpose**: Test basic mirror-to-disk mode with catalog only + +**Configuration**: +- Mode: m2d (mirror to disk) +- Catalog: v9-260129-amd64 +- Release: 9.1.x +- Packages: None (catalog only) + +**What it tests**: +- Basic m2d mode execution +- Catalog mirroring without packages +- Success result parsing +- Command argument construction + +### test_mirror_m2m_with_packages +**Purpose**: Test mirror-to-mirror mode with packages + +**Configuration**: +- Mode: m2m (mirror to mirror) +- Catalog: v9-260129-amd64 +- Release: 9.1.x +- Packages: SLS, Core +- Target registry: registry.example.com/mas + +**What it tests**: +- m2m mode with target registry +- Multiple package mirroring +- Authentication file generation +- Registry credentials handling + +### test_mirror_d2m_resume +**Purpose**: Test disk-to-mirror mode for resuming + +**Configuration**: +- Mode: d2m (disk to mirror) +- Catalog: v9-260129-amd64 +- Release: 9.1.x +- Target registry: registry.example.com/mas + +**What it tests**: +- d2m mode execution +- Resuming from disk storage +- Registry-only authentication (no IBM entitlement needed) + +### test_mirror_with_custom_authfile +**Purpose**: Test using a pre-existing authentication file + +**Configuration**: +- Mode: m2d +- Custom authfile path provided + +**What it tests**: +- Custom auth file usage +- Skipping auth file generation +- Auth file validation + +### test_mirror_with_config_download +**Purpose**: Test config file download from GitHub + +**Configuration**: +- Mode: m2d +- Config file doesn't exist locally + +**What it tests**: +- GitHub config file download +- Network mock handling +- Fallback to remote config + +## Writing New Tests + +### Basic Pattern + +```python +def test_my_scenario(tmpdir): + """Test description.""" + + # Create test configuration + config = MirrorTestConfig( + mode='m2d', + catalog_version='v9-260129-amd64', + release='9.1.x', + root_dir=str(tmpdir), + packages={'sls': True}, # Enable packages as needed + mock_oc_mirror_output=[ + '2026/02/09 17:00:15 [INFO] : 10 / 10 additional images mirrored successfully', + ], + mock_image_count=10, + expect_success=True, + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-key', + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + ) + + # Run the test + run_mirror_test(tmpdir, config) +``` + +### Configuration Options + +#### Required Parameters +- `mode`: Mirror mode ('m2m', 'm2d', or 'd2m') +- `catalog_version`: Catalog version string (e.g., 'v9-260129-amd64') +- `release`: MAS release version (e.g., '9.1.x') + +#### Optional Parameters +- `target_registry`: Target registry for m2m/d2m modes +- `root_dir`: Root directory for mirror operations (default: '/tmp/mirror') +- `packages`: Dict of package flags (e.g., `{'sls': True, 'core': True}`) +- `mock_oc_mirror_output`: List of output lines to simulate +- `mock_image_count`: Number of images in config (default: 10) +- `expect_success`: Whether operation should succeed (default: True) +- `timeout_seconds`: Test timeout in seconds (default: 30) +- `argv`: Custom command line arguments (auto-generated if not provided) +- `authfile`: Path to custom auth file +- `dest_tls_verify`: TLS verification flag (default: True) +- `image_timeout`: Image operation timeout (default: '20m') +- `env_vars`: Environment variables dict +- `mock_catalog_data`: Custom catalog data dict +- `config_exists_locally`: Whether config file exists locally (default: True) + +## Mock Behavior + +### Subprocess (oc-mirror) +- Mocked via `subprocess.Popen` +- Returns configured output lines +- Simulates success/failure based on `expect_success` +- Validates command arguments + +### File System +- `path.exists()`: Returns True for local configs or custom authfile +- `makedirs()`: No-op, doesn't create actual directories +- `open()`: Returns mock file with YAML content + +### Network +- `urllib.request.urlopen()`: Returns mock YAML content for GitHub downloads +- Only used when `config_exists_locally=False` + +### Catalog Data +- `getCatalog()`: Returns mock catalog with version information +- Can be customized via `mock_catalog_data` parameter + +## Running Tests + +### Run all mirror tests +```bash +cd ../cli/python +pytest test/mirror/ +``` + +### Run specific test +```bash +pytest test/mirror/test_mirror.py::test_mirror_m2d_catalog_only +``` + +### Run with verbose output +```bash +pytest test/mirror/ -v +``` + +### Run with debug logging +```bash +pytest test/mirror/ -v --log-cli-level=DEBUG +``` + +## Troubleshooting + +### Test Timeout +If a test times out, check: +- `timeout_seconds` is sufficient for the test +- Mocks are properly configured +- No actual subprocess execution is occurring + +### Import Errors +Ensure the test directory is in the Python path: +```python +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) +``` + +### Mock Not Working +Verify the mock patch path matches the actual import in the code: +- Use `mas.cli.mirror.app.subprocess.Popen` not just `subprocess.Popen` +- Check the module where the function is used, not where it's defined + +## Comparison with Install Test Helper + +| Feature | Install Helper | Mirror Helper | +|---------|---------------|---------------| +| **User Interaction** | Yes (prompt tracking) | No (non-interactive) | +| **External Deps** | Kubernetes API | oc-mirror, file system | +| **Async Operations** | Pipeline launch | Subprocess execution | +| **Watchdog** | Yes | Yes | +| **Mock Complexity** | High (K8s resources) | Medium (subprocess, I/O) | + +## Future Enhancements + +Potential areas for expansion: +- Error scenario tests (network failures, invalid configs) +- Performance tests (large image counts) +- Partial failure scenarios (some images fail) +- Resume/retry logic testing +- Multi-architecture testing +- Custom timeout validation + +## Made with Bob \ No newline at end of file diff --git a/python/test/mirror/__init__.py b/python/test/mirror/__init__.py new file mode 100644 index 0000000000..666b115068 --- /dev/null +++ b/python/test/mirror/__init__.py @@ -0,0 +1,11 @@ +# ***************************************************************************** +# Copyright (c) 2026 IBM Corporation and other Contributors. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Eclipse Public License v1.0 +# which accompanies this distribution, and is available at +# http://www.eclipse.org/legal/epl-v10.html +# +# ***************************************************************************** + +# Made with Bob diff --git a/python/test/mirror/test_mirror_advanced.py b/python/test/mirror/test_mirror_advanced.py new file mode 100644 index 0000000000..ff7f261008 --- /dev/null +++ b/python/test/mirror/test_mirror_advanced.py @@ -0,0 +1,92 @@ +#!/usr/bin/env python +# ***************************************************************************** +# Copyright (c) 2026 IBM Corporation and other Contributors. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Eclipse Public License v1.0 +# which accompanies this distribution, and is available at +# http://www.eclipse.org/legal/epl-v10.html +# +# ***************************************************************************** + +""" +Advanced configuration tests for mirror command. + +Tests advanced mirror options including TLS verification settings and +custom timeout configurations. +""" + +from utils import MirrorTestConfig, run_mirror_test +import sys +import os + +# Add test directory to path for utils import +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) + + +def test_mirror_with_tls_verify_disabled(tmpdir): + """ + Test mirror with TLS verification disabled. + + This scenario tests: + - dest-tls-verify=false flag + - Should pass flag to oc-mirror command + """ + config = MirrorTestConfig( + mode='m2m', + catalog_version='v9-260129-amd64', + release='9.1.x', + target_registry='registry.example.com/mas', + root_dir=str(tmpdir), + dest_tls_verify=False, # Disable TLS verification + packages={}, + mock_oc_mirror_output=[ + '2026/02/09 17:00:15 [INFO] : 10 / 10 additional images mirrored successfully', + ], + mock_image_count=10, + expect_success=True, + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key', + 'REGISTRY_USERNAME': 'testuser', + 'REGISTRY_PASSWORD': 'testpass', + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + ) + + run_mirror_test(tmpdir, config) + + +def test_mirror_with_custom_image_timeout(tmpdir): + """ + Test mirror with custom image timeout. + + This scenario tests: + - Custom --image-timeout flag + - Should pass timeout to oc-mirror command + """ + config = MirrorTestConfig( + mode='m2d', + catalog_version='v9-260129-amd64', + release='9.1.x', + root_dir=str(tmpdir), + image_timeout='30m', # Custom timeout + packages={}, + mock_oc_mirror_output=[ + '2026/02/09 17:00:15 [INFO] : 10 / 10 additional images mirrored successfully', + ], + mock_image_count=10, + expect_success=True, + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key', + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + ) + + run_mirror_test(tmpdir, config) + + +# Made with Bob diff --git a/python/test/mirror/test_mirror_auth.py b/python/test/mirror/test_mirror_auth.py new file mode 100644 index 0000000000..89878e769a --- /dev/null +++ b/python/test/mirror/test_mirror_auth.py @@ -0,0 +1,127 @@ +#!/usr/bin/env python +# ***************************************************************************** +# Copyright (c) 2026 IBM Corporation and other Contributors. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Eclipse Public License v1.0 +# which accompanies this distribution, and is available at +# http://www.eclipse.org/legal/epl-v10.html +# +# ***************************************************************************** + +""" +Authentication tests for mirror command. + +Tests authentication file generation for different mirror modes (m2d, m2m, d2m) +with various credential combinations. +""" + +from utils import MirrorTestConfig, run_mirror_test +import sys +import os + +# Add test directory to path for utils import +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) + + +def test_mirror_auth_file_generation_m2d(tmpdir): + """ + Test auth file generation for m2d mode. + + This scenario tests: + - Auth file is generated with IBM entitlement key + - Mode: m2d + - No registry credentials needed + """ + config = MirrorTestConfig( + mode='m2d', + catalog_version='v9-260129-amd64', + release='9.1.x', + root_dir=str(tmpdir), + packages={}, + mock_oc_mirror_output=[ + '2026/02/09 17:00:15 [INFO] : 10 / 10 additional images mirrored successfully', + ], + mock_image_count=10, + expect_success=True, + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key-12345', + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + ) + + run_mirror_test(tmpdir, config) + + # Verify auth file was created (mocked) + tmpdir.join('.mas', 'mirror', 'auth.json') + # In real scenario, would verify file contents + + +def test_mirror_auth_file_generation_m2m(tmpdir): + """ + Test auth file generation for m2m mode. + + This scenario tests: + - Auth file is generated with both IBM entitlement and registry credentials + - Mode: m2m + """ + config = MirrorTestConfig( + mode='m2m', + catalog_version='v9-260129-amd64', + release='9.1.x', + target_registry='registry.example.com/mas', + root_dir=str(tmpdir), + packages={}, + mock_oc_mirror_output=[ + '2026/02/09 17:00:15 [INFO] : 10 / 10 additional images mirrored successfully', + ], + mock_image_count=10, + expect_success=True, + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key', + 'REGISTRY_USERNAME': 'testuser', + 'REGISTRY_PASSWORD': 'testpass123', + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + ) + + run_mirror_test(tmpdir, config) + + +def test_mirror_auth_file_generation_d2m(tmpdir): + """ + Test auth file generation for d2m mode. + + This scenario tests: + - Auth file is generated with only registry credentials + - Mode: d2m (no IBM entitlement needed) + """ + config = MirrorTestConfig( + mode='d2m', + catalog_version='v9-260129-amd64', + release='9.1.x', + target_registry='registry.example.com/mas', + root_dir=str(tmpdir), + packages={}, + mock_oc_mirror_output=[ + '2026/02/09 17:00:15 [INFO] : 10 / 10 additional images mirrored successfully', + ], + mock_image_count=10, + expect_success=True, + timeout_seconds=30, + env_vars={ + 'REGISTRY_USERNAME': 'testuser', + 'REGISTRY_PASSWORD': 'testpass123', + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + ) + + run_mirror_test(tmpdir, config) + + +# Made with Bob diff --git a/python/test/mirror/test_mirror_basic.py b/python/test/mirror/test_mirror_basic.py new file mode 100644 index 0000000000..ab736a7703 --- /dev/null +++ b/python/test/mirror/test_mirror_basic.py @@ -0,0 +1,213 @@ +#!/usr/bin/env python +# ***************************************************************************** +# Copyright (c) 2026 IBM Corporation and other Contributors. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Eclipse Public License v1.0 +# which accompanies this distribution, and is available at +# http://www.eclipse.org/legal/epl-v10.html +# +# ***************************************************************************** + +""" +Basic mirror operation tests. + +Tests the fundamental mirror operations across different modes (m2d, m2m, d2m). +""" + +from utils import MirrorTestConfig, run_mirror_test +import sys +import os + +# Add test directory to path for utils import +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) + + +def test_mirror_m2d_catalog_only(tmpdir): + """ + Test mirror-to-disk (m2d) mode with catalog only. + + This is the simplest mirror scenario: + - Mode: m2d (mirror to disk) + - Only mirrors the operator catalog + - No additional packages + - Uses local config file (no download) + - Simulates successful mirroring of 10 images + """ + config = MirrorTestConfig( + mode='m2d', + catalog_version='v9-260129-amd64', + release='9.1.x', + root_dir=str(tmpdir), + packages={}, # No packages, catalog only + mock_oc_mirror_output=[ + '2026/02/09 17:00:00 [INFO] : Hello, welcome to oc-mirror', + '2026/02/09 17:00:01 [INFO] : setting up the environment for you...', + '2026/02/09 17:00:05 [INFO] : Success copying image1 ➡️', + '2026/02/09 17:00:06 [INFO] : Success copying image2 ➡️', + '2026/02/09 17:00:07 [INFO] : Success copying image3 ➡️', + '2026/02/09 17:00:08 [INFO] : Success copying image4 ➡️', + '2026/02/09 17:00:09 [INFO] : Success copying image5 ➡️', + '2026/02/09 17:00:10 [INFO] : Success copying image6 ➡️', + '2026/02/09 17:00:11 [INFO] : Success copying image7 ➡️', + '2026/02/09 17:00:12 [INFO] : Success copying image8 ➡️', + '2026/02/09 17:00:13 [INFO] : Success copying image9 ➡️', + '2026/02/09 17:00:14 [INFO] : Success copying image10 ➡️', + '2026/02/09 17:00:15 [INFO] : 10 / 10 additional images mirrored successfully', + ], + mock_image_count=10, + expect_success=True, + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key', + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + ) + + run_mirror_test(tmpdir, config) + + +def test_mirror_m2m_with_packages(tmpdir): + """ + Test mirror-to-mirror (m2m) mode with catalog and packages. + + This scenario tests: + - Mode: m2m (mirror to mirror) + - Mirrors catalog + SLS + Core packages + - Requires target registry + - Requires authentication (registry + IBM entitlement) + - Simulates successful mirroring + """ + config = MirrorTestConfig( + mode='m2m', + catalog_version='v9-260129-amd64', + release='9.1.x', + target_registry='registry.example.com/mas', + root_dir=str(tmpdir), + packages={ + 'sls': True, + 'core': True, + }, + mock_oc_mirror_output=[ + '2026/02/09 17:00:00 [INFO] : Hello, welcome to oc-mirror', + '2026/02/09 17:00:15 [INFO] : 10 / 10 additional images mirrored successfully', + ], + mock_image_count=10, + expect_success=True, + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key', + 'REGISTRY_USERNAME': 'test-user', + 'REGISTRY_PASSWORD': 'test-password', + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + ) + + run_mirror_test(tmpdir, config) + + +def test_mirror_d2m_resume(tmpdir): + """ + Test disk-to-mirror (d2m) mode for resuming from disk. + + This scenario tests: + - Mode: d2m (disk to mirror) + - Resumes mirroring from previously saved disk content + - Requires target registry + - Requires registry authentication only (no IBM entitlement needed) + """ + config = MirrorTestConfig( + mode='d2m', + catalog_version='v9-260129-amd64', + release='9.1.x', + target_registry='registry.example.com/mas', + root_dir=str(tmpdir), + packages={}, # Catalog only + mock_oc_mirror_output=[ + '2026/02/09 17:00:00 [INFO] : Hello, welcome to oc-mirror', + '2026/02/09 17:00:15 [INFO] : 10 / 10 additional images mirrored successfully', + ], + mock_image_count=10, + expect_success=True, + timeout_seconds=30, + env_vars={ + 'REGISTRY_USERNAME': 'test-user', + 'REGISTRY_PASSWORD': 'test-password', + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + ) + + run_mirror_test(tmpdir, config) + + +def test_mirror_with_custom_authfile(tmpdir): + """ + Test mirror with custom authentication file. + + This scenario tests: + - Using a pre-existing auth file instead of generating one + - Mode: m2d + - Catalog only + """ + # Create a mock auth file + auth_file_path = str(tmpdir.join('custom-auth.json')) + tmpdir.join('custom-auth.json').write('{"auths": {}}') + + config = MirrorTestConfig( + mode='m2d', + catalog_version='v9-260129-amd64', + release='9.1.x', + root_dir=str(tmpdir), + authfile=auth_file_path, + packages={}, + mock_oc_mirror_output=[ + '2026/02/09 17:00:15 [INFO] : 10 / 10 additional images mirrored successfully', + ], + mock_image_count=10, + expect_success=True, + timeout_seconds=30, + env_vars={ + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + ) + + run_mirror_test(tmpdir, config) + + +def test_mirror_with_config_download(tmpdir): + """ + Test mirror with config file download from GitHub. + + This scenario tests: + - Config file doesn't exist locally + - Should download from GitHub + - Mode: m2d + - Catalog only + """ + config = MirrorTestConfig( + mode='m2d', + catalog_version='v9-260129-amd64', + release='9.1.x', + root_dir=str(tmpdir), + packages={}, + mock_oc_mirror_output=[ + '2026/02/09 17:00:15 [INFO] : 10 / 10 additional images mirrored successfully', + ], + mock_image_count=10, + expect_success=True, + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key', + 'HOME': str(tmpdir), + }, + config_exists_locally=False, # Force download + ) + + run_mirror_test(tmpdir, config) + + +# Made with Bob diff --git a/python/test/mirror/test_mirror_config.py b/python/test/mirror/test_mirror_config.py new file mode 100644 index 0000000000..f1dcc50f1a --- /dev/null +++ b/python/test/mirror/test_mirror_config.py @@ -0,0 +1,137 @@ +#!/usr/bin/env python +# ***************************************************************************** +# Copyright (c) 2026 IBM Corporation and other Contributors. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Eclipse Public License v1.0 +# which accompanies this distribution, and is available at +# http://www.eclipse.org/legal/epl-v10.html +# +# ***************************************************************************** + +""" +Configuration handling tests for mirror command. + +Tests config file download from GitHub, local file handling, and error scenarios +with invalid or missing configuration files. +""" + +from utils import MirrorTestConfig, run_mirror_test +import sys +import os +import urllib.error +from unittest import mock + +# Add test directory to path for utils import +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) + + +def test_mirror_config_download_success(tmpdir): + """ + Test successful config file download from GitHub. + + This scenario tests: + - Config file doesn't exist locally + - Successfully downloads from GitHub + - Proceeds with mirror operation + """ + config = MirrorTestConfig( + mode='m2d', + catalog_version='v9-260129-amd64', + release='9.1.x', + root_dir=str(tmpdir), + packages={}, + mock_oc_mirror_output=[ + '2026/02/09 17:00:15 [INFO] : 10 / 10 additional images mirrored successfully', + ], + mock_image_count=10, + expect_success=True, + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key', + 'HOME': str(tmpdir), + }, + config_exists_locally=False, # Force download + ) + + run_mirror_test(tmpdir, config) + + +def test_mirror_config_download_failure(tmpdir): + """ + Test handling of config file download failure. + + This scenario tests: + - Config file doesn't exist locally + - GitHub download fails (network error, 404, etc.) + - Should fail gracefully + """ + config = MirrorTestConfig( + mode='m2d', + catalog_version='v9-260129-amd64', + release='9.1.x', + root_dir=str(tmpdir), + packages={}, + mock_oc_mirror_output=[], + mock_image_count=0, + expect_success=False, + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key', + 'HOME': str(tmpdir), + }, + config_exists_locally=False, + ) + + # Mock urllib to raise exception + with mock.patch('urllib.request.urlopen') as mock_urlopen: + mock_urlopen.side_effect = urllib.error.URLError('Network error') + + try: + run_mirror_test(tmpdir, config) + assert False, "Expected exception but test completed" + except (SystemExit, Exception): + # Expected to fail + pass + + +def test_mirror_invalid_yaml_config(tmpdir): + """ + Test handling of invalid YAML in config file. + + This scenario tests: + - Config file exists but contains invalid YAML + - Should fail during config parsing + """ + # Create invalid YAML file + config_dir = tmpdir.join('.mas', 'mirror', 'configs') + config_dir.ensure(dir=True) + config_file = config_dir.join('v9-260129-amd64.yaml') + config_file.write('invalid: yaml: content: [unclosed') + + config = MirrorTestConfig( + mode='m2d', + catalog_version='v9-260129-amd64', + release='9.1.x', + root_dir=str(tmpdir), + packages={}, + mock_oc_mirror_output=[], + mock_image_count=0, + expect_success=False, + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key', + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + ) + + try: + run_mirror_test(tmpdir, config) + assert False, "Expected exception but test completed" + except (SystemExit, Exception): + # Expected to fail + pass + + +# Made with Bob diff --git a/python/test/mirror/test_mirror_errors.py b/python/test/mirror/test_mirror_errors.py new file mode 100644 index 0000000000..353855aac2 --- /dev/null +++ b/python/test/mirror/test_mirror_errors.py @@ -0,0 +1,300 @@ +#!/usr/bin/env python +# ***************************************************************************** +# Copyright (c) 2026 IBM Corporation and other Contributors. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Eclipse Public License v1.0 +# which accompanies this distribution, and is available at +# http://www.eclipse.org/legal/epl-v10.html +# +# ***************************************************************************** + +""" +Error handling tests for mirror command. + +Tests various error scenarios including missing dependencies, invalid inputs, +command failures, and timeout handling. +""" + +from utils import MirrorTestConfig, run_mirror_test +import sys +import os +import time +from unittest import mock +from unittest.mock import MagicMock + +# Add test directory to path for utils import +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) + + +def test_mirror_missing_oc_mirror(tmpdir): + """ + Test error handling when oc-mirror binary is not found. + + This scenario tests: + - Missing oc-mirror binary + - Should fail gracefully with appropriate error message + - App logs error but doesn't exit (continues to show summary) + """ + config = MirrorTestConfig( + mode='m2d', + catalog_version='v9-260129-amd64', + release='9.1.x', + root_dir=str(tmpdir), + packages={}, + mock_oc_mirror_output=[], + mock_image_count=5, # Need images for oc-mirror to be called + expect_success=False, + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key', + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + ) + + # Override which() to return None for oc-mirror + with mock.patch('mas.cli.cli.which') as mock_which: + def which_side_effect(cmd): + if cmd == 'oc-mirror': + return None + elif cmd == 'kubectl': + return '/usr/bin/kubectl' + return None + mock_which.side_effect = which_side_effect + + # The app will log an error but complete (not exit) + # This is the expected behavior - it shows the error and summary + run_mirror_test(tmpdir, config) + # Test passes if no exception is raised + + +def test_mirror_invalid_catalog_version(tmpdir): + """ + Test error handling with invalid catalog version. + + This scenario tests: + - Invalid catalog version format + - Should fail during catalog lookup + """ + config = MirrorTestConfig( + mode='m2d', + catalog_version='invalid-version', + release='9.1.x', + root_dir=str(tmpdir), + packages={}, + mock_oc_mirror_output=[], + mock_image_count=0, + expect_success=False, + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key', + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + ) + + try: + run_mirror_test(tmpdir, config) + assert False, "Expected exception but test completed" + except (SystemExit, Exception): + # Expected to fail + pass + + +def test_mirror_missing_entitlement_key_m2d(tmpdir): + """ + Test error handling when IBM_ENTITLEMENT_KEY is missing for m2d mode. + + This scenario tests: + - Missing IBM_ENTITLEMENT_KEY environment variable + - Mode: m2d (requires entitlement key) + - Should fail during auth file generation + """ + config = MirrorTestConfig( + mode='m2d', + catalog_version='v9-260129-amd64', + release='9.1.x', + root_dir=str(tmpdir), + packages={}, + mock_oc_mirror_output=[], + mock_image_count=0, + expect_success=False, + timeout_seconds=30, + env_vars={ + 'HOME': str(tmpdir), + # IBM_ENTITLEMENT_KEY intentionally missing + }, + config_exists_locally=True, + ) + + try: + run_mirror_test(tmpdir, config) + assert False, "Expected SystemExit but test completed" + except SystemExit as e: + # Expected to fail + assert e.code != 0 + + +def test_mirror_missing_registry_credentials_m2m(tmpdir): + """ + Test error handling when registry credentials are missing for m2m mode. + + This scenario tests: + - Missing REGISTRY_USERNAME/PASSWORD for m2m mode + - Should fail during auth file generation + """ + config = MirrorTestConfig( + mode='m2m', + catalog_version='v9-260129-amd64', + release='9.1.x', + target_registry='registry.example.com/mas', + root_dir=str(tmpdir), + packages={}, + mock_oc_mirror_output=[], + mock_image_count=0, + expect_success=False, + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key', + 'HOME': str(tmpdir), + # REGISTRY_USERNAME/PASSWORD intentionally missing + }, + config_exists_locally=True, + ) + + try: + run_mirror_test(tmpdir, config) + assert False, "Expected SystemExit but test completed" + except SystemExit as e: + # Expected to fail + assert e.code != 0 + + +def test_mirror_oc_mirror_command_failure(tmpdir): + """ + Test error handling when oc-mirror command fails. + + This scenario tests: + - oc-mirror command returns non-zero exit code + - Should detect failure and report it + """ + config = MirrorTestConfig( + mode='m2d', + catalog_version='v9-260129-amd64', + release='9.1.x', + root_dir=str(tmpdir), + packages={}, + mock_oc_mirror_output=[ + '2026/02/09 17:00:00 [ERROR] : Failed to connect to registry', + '2026/02/09 17:00:01 [ERROR] : Mirror operation failed', + ], + mock_image_count=5, # Need images for oc-mirror to be called + expect_success=False, + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key', + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + ) + + # Mock subprocess to return failure + with mock.patch('subprocess.Popen') as mock_popen: + mock_process = MagicMock() + mock_process.returncode = 1 + mock_process.stdout.readline.side_effect = [ + line.encode() + b'\n' for line in config.mock_oc_mirror_output + ] + [b''] + mock_process.poll.return_value = 1 + mock_popen.return_value = mock_process + + try: + run_mirror_test(tmpdir, config) + # Test should complete but report failure + except SystemExit: + # May exit on failure + pass + + +def test_mirror_partial_failure(tmpdir): + """ + Test handling of partial mirror failure. + + This scenario tests: + - Some images mirror successfully, others fail + - Should report partial success + """ + config = MirrorTestConfig( + mode='m2d', + catalog_version='v9-260129-amd64', + release='9.1.x', + root_dir=str(tmpdir), + packages={}, + mock_oc_mirror_output=[ + '2026/02/09 17:00:00 [INFO] : Hello, welcome to oc-mirror', + '2026/02/09 17:00:05 [INFO] : Success copying image1 ➡️', + '2026/02/09 17:00:06 [INFO] : Success copying image2 ➡️', + '2026/02/09 17:00:07 [ERROR] : Failed copying image3', + '2026/02/09 17:00:08 [INFO] : Success copying image4 ➡️', + '2026/02/09 17:00:09 [ERROR] : Failed copying image5', + '2026/02/09 17:00:15 [INFO] : 3 / 5 additional images mirrored successfully', + ], + mock_image_count=5, + expect_success=False, # Partial failure + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key', + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + ) + + run_mirror_test(tmpdir, config) + + +def test_mirror_timeout(tmpdir): + """ + Test timeout handling when mirror operation hangs. + + This scenario tests: + - Mirror operation exceeds timeout + - Watchdog thread should detect and terminate + """ + config = MirrorTestConfig( + mode='m2d', + catalog_version='v9-260129-amd64', + release='9.1.x', + root_dir=str(tmpdir), + packages={}, + mock_oc_mirror_output=[ + '2026/02/09 17:00:00 [INFO] : Hello, welcome to oc-mirror', + # No completion message - simulates hang + ], + mock_image_count=10, + expect_success=False, + timeout_seconds=2, # Short timeout for test + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key', + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + ) + + # Mock subprocess to simulate hanging + with mock.patch('subprocess.Popen') as mock_popen: + mock_process = MagicMock() + mock_process.returncode = None + mock_process.stdout.readline.side_effect = lambda: time.sleep(5) or b'' + mock_process.poll.return_value = None + mock_popen.return_value = mock_process + + try: + run_mirror_test(tmpdir, config) + # Should timeout + except (SystemExit, Exception): + # Expected to fail/timeout + pass + + +# Made with Bob diff --git a/python/test/mirror/test_mirror_packages.py b/python/test/mirror/test_mirror_packages.py new file mode 100644 index 0000000000..1c4c101f91 --- /dev/null +++ b/python/test/mirror/test_mirror_packages.py @@ -0,0 +1,310 @@ +#!/usr/bin/env python +# ***************************************************************************** +# Copyright (c) 2026 IBM Corporation and other Contributors. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Eclipse Public License v1.0 +# which accompanies this distribution, and is available at +# http://www.eclipse.org/legal/epl-v10.html +# +# ***************************************************************************** + +""" +Package selection tests for mirror command. + +Tests mirroring with different package combinations including all packages, +selective packages, and special handling for DB2. +""" + +from utils import MirrorTestConfig, run_mirror_test +import sys +import os + +# Add test directory to path for utils import +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) + + +def test_mirror_all_packages(tmpdir): + """ + Test mirroring with all available packages. + + This scenario tests: + - All packages enabled (sls, core, assist, iot, manage, monitor, optimizer, predict, visualinspection) + - Mode: m2m + - Should generate config with all package images + """ + config = MirrorTestConfig( + mode='m2m', + catalog_version='v9-260129-amd64', + release='9.1.x', + target_registry='registry.example.com/mas', + root_dir=str(tmpdir), + packages={ + 'sls': True, + 'core': True, + 'assist': True, + 'iot': True, + 'manage': True, + 'monitor': True, + 'optimizer': True, + 'predict': True, + 'visualinspection': True, + }, + mock_oc_mirror_output=[ + '2026/02/09 17:00:00 [INFO] : Hello, welcome to oc-mirror', + '2026/02/09 17:00:15 [INFO] : 50 / 50 additional images mirrored successfully', + ], + mock_image_count=50, + expect_success=True, + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key', + 'REGISTRY_USERNAME': 'testuser', + 'REGISTRY_PASSWORD': 'testpass', + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + ) + + run_mirror_test(tmpdir, config) + + +def test_mirror_selective_packages(tmpdir): + """ + Test mirroring with selective packages. + + This scenario tests: + - Only specific packages enabled (sls, manage, monitor) + - Mode: m2m + - Should generate config with only selected package images + """ + config = MirrorTestConfig( + mode='m2m', + catalog_version='v9-260129-amd64', + release='9.1.x', + target_registry='registry.example.com/mas', + root_dir=str(tmpdir), + packages={ + 'sls': True, + 'manage': True, + 'monitor': True, + }, + mock_oc_mirror_output=[ + '2026/02/09 17:00:00 [INFO] : Hello, welcome to oc-mirror', + '2026/02/09 17:00:15 [INFO] : 20 / 20 additional images mirrored successfully', + ], + mock_image_count=20, + expect_success=True, + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key', + 'REGISTRY_USERNAME': 'testuser', + 'REGISTRY_PASSWORD': 'testpass', + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + ) + + run_mirror_test(tmpdir, config) + + +def test_mirror_db2_package_special_handling(tmpdir): + """ + Test DB2 package special version handling. + + This scenario tests: + - DB2 packages (db2u-s11 and db2u-s12) use different version format + - Should handle DB2 version correctly + """ + config = MirrorTestConfig( + mode='m2m', + catalog_version='v9-260129-amd64', + release='9.1.x', + target_registry='registry.example.com/mas', + root_dir=str(tmpdir), + packages={ + 'db2u-s11': True, + 'db2u-s12': True, + }, + mock_oc_mirror_output=[ + '2026/02/09 17:00:00 [INFO] : Hello, welcome to oc-mirror', + '2026/02/09 17:00:15 [INFO] : 5 / 5 additional images mirrored successfully', + ], + mock_image_count=5, + expect_success=True, + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key', + 'REGISTRY_USERNAME': 'testuser', + 'REGISTRY_PASSWORD': 'testpass', + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + ) + + run_mirror_test(tmpdir, config) + + +def test_mirror_cp4d_platform_packages(tmpdir): + """ + Test mirroring CP4D Platform packages with multiple package names. + + This scenario tests: + - CP4D Platform package (cp4d-platform) which includes multiple packages: + ibm-cp-common-services, ibm-zen, ibm-cp-datacore, ibm-licensing, ibm-ccs, + ibm-cloud-native-postgresql, ibm-datarefinery, ibm-elasticsearch-operator, + ibm-opensearch-operator + - Mode: m2m + - Should generate config with all CP4D platform package images + """ + config = MirrorTestConfig( + mode='m2m', + catalog_version='v9-260129-amd64', + release='9.1.x', + target_registry='registry.example.com/mas', + root_dir=str(tmpdir), + packages={ + 'cp4d-platform': True, + }, + mock_oc_mirror_output=[ + '2026/02/09 17:00:00 [INFO] : Hello, welcome to oc-mirror', + '2026/02/09 17:00:15 [INFO] : 30 / 30 additional images mirrored successfully', + ], + mock_image_count=30, + expect_success=True, + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key', + 'REGISTRY_USERNAME': 'testuser', + 'REGISTRY_PASSWORD': 'testpass', + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + ) + + run_mirror_test(tmpdir, config) + + +def test_mirror_cp4d_wsl_packages(tmpdir): + """ + Test mirroring CP4D WSL packages with multiple package names. + + This scenario tests: + - CP4D WSL package (cp4d-wsl) which includes: ibm-wsl, ibm-wsl-runtimes + - Mode: m2m + - Should generate config with all CP4D WSL package images + """ + config = MirrorTestConfig( + mode='m2m', + catalog_version='v9-260129-amd64', + release='9.1.x', + target_registry='registry.example.com/mas', + root_dir=str(tmpdir), + packages={ + 'cp4d-wsl': True, + }, + mock_oc_mirror_output=[ + '2026/02/09 17:00:00 [INFO] : Hello, welcome to oc-mirror', + '2026/02/09 17:00:15 [INFO] : 10 / 10 additional images mirrored successfully', + ], + mock_image_count=10, + expect_success=True, + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key', + 'REGISTRY_USERNAME': 'testuser', + 'REGISTRY_PASSWORD': 'testpass', + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + ) + + run_mirror_test(tmpdir, config) + + +def test_mirror_all_cp4d_packages(tmpdir): + """ + Test mirroring all CP4D packages together. + + This scenario tests: + - All CP4D packages: cp4d-platform, cp4d-wsl, cp4d-wml, cp4d-spark, cp4d-cognos + - Mode: m2m + - Should generate config with all CP4D package images + """ + config = MirrorTestConfig( + mode='m2m', + catalog_version='v9-260129-amd64', + release='9.1.x', + target_registry='registry.example.com/mas', + root_dir=str(tmpdir), + packages={ + 'cp4d-platform': True, + 'cp4d-wsl': True, + 'cp4d-wml': True, + 'cp4d-spark': True, + 'cp4d-cognos': True, + }, + mock_oc_mirror_output=[ + '2026/02/09 17:00:00 [INFO] : Hello, welcome to oc-mirror', + '2026/02/09 17:00:15 [INFO] : 60 / 60 additional images mirrored successfully', + ], + mock_image_count=60, + expect_success=True, + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key', + 'REGISTRY_USERNAME': 'testuser', + 'REGISTRY_PASSWORD': 'testpass', + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + ) + + run_mirror_test(tmpdir, config) + + +def test_mirror_with_all_flag(tmpdir): + """ + Test mirroring with --all flag to enable all packages. + + This scenario tests: + - Using --all flag instead of individual package flags + - Should mirror all available packages + - Mode: m2m + """ + config = MirrorTestConfig( + mode='m2m', + catalog_version='v9-260129-amd64', + release='9.1.x', + target_registry='registry.example.com/mas', + root_dir=str(tmpdir), + packages={}, # No individual packages specified + mock_oc_mirror_output=[ + '2026/02/09 17:00:00 [INFO] : Hello, welcome to oc-mirror', + '2026/02/09 17:00:15 [INFO] : 100 / 100 additional images mirrored successfully', + ], + mock_image_count=100, + expect_success=True, + timeout_seconds=30, + env_vars={ + 'IBM_ENTITLEMENT_KEY': 'test-entitlement-key', + 'REGISTRY_USERNAME': 'testuser', + 'REGISTRY_PASSWORD': 'testpass', + 'HOME': str(tmpdir), + }, + config_exists_locally=True, + # Override argv to use --all flag + argv=[ + '--catalog', 'v9-260129-amd64', + '--release', '9.1.x', + '--mode', 'm2m', + '--dir', str(tmpdir), + '--target-registry', 'registry.example.com/mas', + '--all', + ] + ) + + run_mirror_test(tmpdir, config) + + +# Made with Bob diff --git a/python/test/utils/__init__.py b/python/test/utils/__init__.py index 8b519fa770..f303cad85d 100644 --- a/python/test/utils/__init__.py +++ b/python/test/utils/__init__.py @@ -11,7 +11,17 @@ from .prompt_tracker import PromptTracker, create_prompt_handler from .install_test_helper import InstallTestConfig, InstallTestHelper, run_install_test +from .mirror_test_helper import MirrorTestConfig, MirrorTestHelper, run_mirror_test -__all__ = ['PromptTracker', 'create_prompt_handler', 'InstallTestConfig', 'InstallTestHelper', 'run_install_test'] +__all__ = [ + 'PromptTracker', + 'create_prompt_handler', + 'InstallTestConfig', + 'InstallTestHelper', + 'run_install_test', + 'MirrorTestConfig', + 'MirrorTestHelper', + 'run_mirror_test' +] # Made with Bob diff --git a/python/test/utils/mirror_test_helper.py b/python/test/utils/mirror_test_helper.py new file mode 100644 index 0000000000..13e819e47d --- /dev/null +++ b/python/test/utils/mirror_test_helper.py @@ -0,0 +1,345 @@ +#!/usr/bin/env python +# ***************************************************************************** +# Copyright (c) 2026 IBM Corporation and other Contributors. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Eclipse Public License v1.0 +# which accompanies this distribution, and is available at +# http://www.eclipse.org/legal/epl-v10.html +# +# ***************************************************************************** + +import time +import threading +import yaml +from typing import Dict, Optional, List +from unittest import mock +from unittest.mock import MagicMock +from dataclasses import dataclass, field +from mas.cli.mirror.app import MirrorApp + + +@dataclass +class MirrorTestConfig: + """Configuration for a mirror test scenario.""" + + mode: str # m2m, m2d, or d2m + catalog_version: str + release: str + target_registry: str = "" + root_dir: str = "/tmp/mirror" + packages: Dict[str, bool] = field(default_factory=dict) + mock_oc_mirror_output: List[str] = field(default_factory=list) + mock_image_count: int = 10 + expect_success: bool = True + timeout_seconds: int = 30 + argv: Optional[list] = None + authfile: Optional[str] = None + dest_tls_verify: bool = True + image_timeout: str = "20m" + # Environment variables for auth file generation + env_vars: Dict[str, str] = field(default_factory=dict) + # Mock catalog data + mock_catalog_data: Optional[Dict] = None + # Whether config file should exist locally (vs download from GitHub) + config_exists_locally: bool = True + + def __post_init__(self): + """Set default argv if not provided.""" + if self.argv is None: + self.argv = self._build_default_argv() + + def _build_default_argv(self) -> list: + """Build default command line arguments from config.""" + args = [ + '--catalog', self.catalog_version, + '--release', self.release, + '--mode', self.mode, + '--dir', self.root_dir, + ] + + if self.target_registry: + args.extend(['--target-registry', self.target_registry]) + + if self.authfile: + args.extend(['--authfile', self.authfile]) + + if not self.dest_tls_verify: + args.extend(['--dest-tls-verify', 'false']) + + if self.image_timeout != "20m": + args.extend(['--image-timeout', self.image_timeout]) + + # Add package flags + for package, enabled in self.packages.items(): + if enabled: + args.append(f'--{package}') + + return args + + +class MirrorTestHelper: + """Helper class to run mirror tests with minimal code duplication.""" + + def __init__(self, tmpdir, config: MirrorTestConfig): + """ + Initialize the test helper. + + Args: + tmpdir: pytest tmpdir fixture + config: Test configuration + """ + self.tmpdir = tmpdir + self.config = config + self.test_failed = {'failed': False, 'message': ''} + self.last_activity_time = {'time': time.time()} + self.watchdog_thread = None + self.oc_mirror_call_count = 0 + + def start_watchdog(self): + """Start watchdog thread to detect hanging tests.""" + def watchdog(): + while not self.test_failed['failed']: + time.sleep(1) + elapsed = time.time() - self.last_activity_time['time'] + if elapsed > self.config.timeout_seconds: + self.test_failed['failed'] = True + self.test_failed['message'] = f"Test hung: No activity for {self.config.timeout_seconds}s" + break + + self.watchdog_thread = threading.Thread(target=watchdog, daemon=True) + self.watchdog_thread.start() + + def stop_watchdog(self): + """Stop the watchdog thread.""" + self.test_failed['failed'] = True + + def update_activity(self): + """Update last activity time to prevent watchdog timeout.""" + self.last_activity_time['time'] = time.time() + + def create_mock_subprocess(self): + """ + Create a mock subprocess.Popen that simulates oc-mirror execution. + + Returns: + Mock Popen object configured to return test output + """ + mock_process = MagicMock() + + # Create mock stdout and stderr + mock_stdout = MagicMock() + mock_stderr = MagicMock() + + # Configure readline to return mock output lines + if self.config.mock_oc_mirror_output: + # Add lines one by one, then empty string to signal EOF + stdout_lines = [line + '\n' for line in self.config.mock_oc_mirror_output] + [''] + mock_stdout.readline.side_effect = stdout_lines + mock_stdout.fileno.return_value = 1 + else: + # Default success output + default_output = [ + f"{self.config.mock_image_count} / {self.config.mock_image_count} additional images mirrored successfully\n", + '' + ] + mock_stdout.readline.side_effect = default_output + mock_stdout.fileno.return_value = 1 + + # Empty stderr + mock_stderr.readline.side_effect = [''] + mock_stderr.fileno.return_value = 2 + + mock_process.stdout = mock_stdout + mock_process.stderr = mock_stderr + mock_process.wait.return_value = 0 if self.config.expect_success else 1 + + return mock_process + + def setup_mocks(self): + """Setup all mock objects and return context managers.""" + # Mock prompt_toolkit's print_formatted_text to avoid Windows console issues + mock_print = mock.patch('prompt_toolkit.shortcuts.utils.print_formatted_text') + mock_print.start() + + # Create mock catalog data + if self.config.mock_catalog_data is None: + # Default catalog data structure + self.config.mock_catalog_data = { + 'sls_version': '3.10.0', + 'tsm_version': '1.5.0', + 'mas_core_version': {'9.1.x': '9.1.0'}, + 'mas_assist_version': {'9.1.x': '9.1.0'}, + 'mas_iot_version': {'9.1.x': '9.1.0'}, + 'mas_manage_version': {'9.1.x': '9.1.0'}, + 'mas_monitor_version': {'9.1.x': '9.1.0'}, + 'mas_predict_version': {'9.1.x': '9.1.0'}, + 'mas_optimizer_version': {'9.1.x': '9.1.0'}, + 'mas_visualinspection_version': {'9.1.x': '9.1.0'}, + 'mas_facilities_version': {'9.1.x': '9.1.0'}, + 'db2u_version': '11.5.9.0+123', + 'amlen_extras_version': '1.0.0', + 'aiservice_version': {'9.1.x': '1.0.0'}, + 'dd_version': '1.0.0', + 'mongo_extras_version_default': '6.0.0', + # CP4D Platform version keys + 'common_svcs_version': '4.13.0', + 'ibm_zen_version': '6.2.0+20250530.152516.232', + 'cp4d_platform_version': '5.2.0+20250709.170324', + 'ibm_licensing_version': '4.2.17', + 'ccs_build': '11.0.0+20250605.130237.468', + 'postgress_version': '5.16.0+20250827.110911.2626', + 'datarefinery_version': '11.0.0+20250513.203727.232', + 'elasticsearch_version': '1.1.2667', + 'opensearch_version': '1.1.2494', + # CP4D WSL version keys + 'wsl_version': '11.0.0+20250521.202913.73', + 'wsl_runtimes_version': '11.0.0+20250515.090949.21', + # CP4D WML version key + 'wml_version': '11.0.0+20250530.193146.282', + # CP4D Spark version key + 'spark_version': '11.0.0+20250604.163055.2097', + # CP4D Cognos version key + 'cognos_version': '28.0.0+20250515.175459.10054', + } + + # Mock YAML config file content + mock_yaml_content = { + 'mirror': { + 'additionalImages': [ + {'name': f'image{i}'} for i in range(self.config.mock_image_count) + ] + } + } + + return mock_yaml_content + + def run_mirror_test(self): + """ + Run the mirror test with all mocks configured. + + Raises: + TimeoutError: If test times out + AssertionError: If validation fails + """ + self.start_watchdog() + + mock_yaml_content = self.setup_mocks() + + # Create a custom open mock that only affects YAML config files + original_open = open + + def selective_open(file, mode='r', *args, **kwargs): + # Only mock YAML config files, let everything else (including log files) use real open + if isinstance(file, str) and ('.yaml' in file or 'auth.json' in file): + if mode == 'r' or 'r' in mode: + # Return mock file with YAML content for reading + from io import StringIO + return StringIO(yaml.dump(mock_yaml_content)) + else: + # For writing (auth.json), return a mock that accepts writes + mock_file = MagicMock() + mock_file.__enter__ = MagicMock(return_value=mock_file) + mock_file.__exit__ = MagicMock(return_value=False) + return mock_file + # Use original open for everything else (log files, etc.) + return original_open(file, mode, *args, **kwargs) + + with ( + # Mock kubectl check in BaseApp.__init__ (which is imported from shutil) + mock.patch('mas.cli.cli.which') as mock_kubectl_which, + # Mock oc-mirror availability + mock.patch('mas.cli.mirror.app.shutil.which') as mock_which, + mock.patch('mas.cli.mirror.app.subprocess.Popen') as mock_popen, + mock.patch('mas.cli.mirror.app.getCatalog') as mock_get_catalog, + mock.patch('builtins.open', side_effect=selective_open) as mock_file, # noqa: F841 + mock.patch('mas.cli.mirror.app.path.exists') as mock_path_exists, + mock.patch('mas.cli.mirror.app.makedirs') as mock_makedirs, # noqa: F841 + mock.patch('mas.cli.mirror.app.urllib.request.urlopen') as mock_urlopen, + mock.patch('mas.cli.mirror.app.environ', self.config.env_vars) as mock_environ, # noqa: F841 + ): + # Configure kubectl mock (for BaseApp.__init__) + mock_kubectl_which.return_value = '/usr/local/bin/kubectl' + # Configure oc-mirror mock + mock_which.return_value = '/usr/local/bin/oc-mirror' + mock_get_catalog.return_value = self.config.mock_catalog_data + + # Configure path.exists based on config + if self.config.authfile: + # If authfile is provided, it should exist + mock_path_exists.side_effect = lambda p: self.config.authfile in p or self.config.config_exists_locally + else: + # Config files exist locally, auth file will be generated + mock_path_exists.return_value = self.config.config_exists_locally + + # Configure subprocess mock + def popen_side_effect(*args, **kwargs): + self.update_activity() + self.oc_mirror_call_count += 1 + return self.create_mock_subprocess() + + mock_popen.side_effect = popen_side_effect + + # Configure urllib for GitHub downloads (if needed) + if not self.config.config_exists_locally: + mock_response = MagicMock() + mock_response.read.return_value = yaml.dump(mock_yaml_content).encode() + mock_urlopen.return_value.__enter__.return_value = mock_response + + try: + # Run the mirror command + app = MirrorApp() + app.mirror(argv=self.config.argv) + + # Update activity after completion + self.update_activity() + + finally: + self.stop_watchdog() + + # Check if test timed out + if self.test_failed['message']: + raise TimeoutError(self.test_failed['message']) + + # Verify oc-mirror was called + assert self.oc_mirror_call_count > 0, "oc-mirror command was not executed" + + # Verify oc-mirror was called with correct arguments + if mock_popen.called: + call_args = mock_popen.call_args[0][0] # Get the command list + assert 'oc-mirror' in call_args[0] or call_args[0].endswith('oc-mirror'), \ + f"Expected oc-mirror in command, got: {call_args[0]}" + assert '--v2' in call_args, "Expected --v2 flag" + assert '--config' in call_args, "Expected --config flag" + assert self.config.mode in ['m2m', 'm2d', 'd2m'], f"Invalid mode: {self.config.mode}" + + # Verify mode-specific arguments + if self.config.mode == 'm2m': + assert f"docker://{self.config.target_registry}" in call_args, \ + "Expected target registry in m2m mode" + elif self.config.mode == 'm2d': + assert any('file://' in arg for arg in call_args), \ + "Expected file:// destination in m2d mode" + elif self.config.mode == 'd2m': + assert '--from' in call_args, "Expected --from flag in d2m mode" + assert f"docker://{self.config.target_registry}" in call_args, \ + "Expected target registry in d2m mode" + + +def run_mirror_test(tmpdir, config: MirrorTestConfig): + """ + Convenience function to run a mirror test. + + Args: + tmpdir: pytest tmpdir fixture + config: Test configuration + + Raises: + TimeoutError: If test times out + AssertionError: If validation fails + """ + helper = MirrorTestHelper(tmpdir, config) + helper.run_mirror_test() + + +# Made with Bob