diff --git a/convert2rhel/logger.py b/convert2rhel/logger.py index fbbcaeb3a..12d009fae 100644 --- a/convert2rhel/logger.py +++ b/convert2rhel/logger.py @@ -257,17 +257,6 @@ def colorize(message, color="OKGREEN"): return "".join((getattr(bcolors, color), message, bcolors.ENDC)) -class ConversionStage: - stages = {"prepare": "Prepare"} - current = None # type: str|None - - @classmethod - def set_stage(cls, stage): # type: (str) -> None - if stage == None or stage in cls.stages: - cls.current = cls.stages[stage] - raise NotImplementedError("The stage {} is not implemented in the ConversionStage class".format(stage)) - - class CustomFormatter(logging.Formatter): """ Custom formatter to handle different logging formats based on logging level. diff --git a/convert2rhel/main.py b/convert2rhel/main.py index cfe7ee708..6165482ef 100644 --- a/convert2rhel/main.py +++ b/convert2rhel/main.py @@ -24,6 +24,7 @@ from convert2rhel import logger as logger_module from convert2rhel import pkghandler, pkgmanager, subscription, systeminfo, toolopts, utils from convert2rhel.actions import level_for_raw_action_data, report +from convert2rhel.utils.phase import ConversionPhases loggerinst = logger_module.root_logger.getChild(__name__) @@ -40,25 +41,16 @@ class _InhibitorsFound(Exception): pass -class ConversionPhase: - POST_CLI = 1 - # PONR means Point Of No Return - PRE_PONR_CHANGES = 2 - # Phase to exit the Analyze SubCommand early - ANALYZE_EXIT = 3 - POST_PONR_CHANGES = 4 - - _REPORT_MAPPING = { - ConversionPhase.ANALYZE_EXIT: ( + ConversionPhases.ANALYZE_EXIT.name: ( report.CONVERT2RHEL_PRE_CONVERSION_JSON_RESULTS, report.CONVERT2RHEL_PRE_CONVERSION_TXT_RESULTS, ), - ConversionPhase.PRE_PONR_CHANGES: ( + ConversionPhases.PRE_PONR_CHANGES.name: ( report.CONVERT2RHEL_PRE_CONVERSION_JSON_RESULTS, report.CONVERT2RHEL_PRE_CONVERSION_TXT_RESULTS, ), - ConversionPhase.POST_PONR_CHANGES: ( + ConversionPhases.POST_PONR_CHANGES.name: ( report.CONVERT2RHEL_POST_CONVERSION_JSON_RESULTS, report.CONVERT2RHEL_POST_CONVERSION_TXT_RESULTS, ), @@ -129,14 +121,14 @@ def main_locked(): pre_conversion_results = None post_conversion_results = None - process_phase = ConversionPhase.POST_CLI + ConversionPhases.set_current(ConversionPhases.POST_CLI) # since we now have root, we can add the FileLogging # and also archive previous logs initialize_file_logging("convert2rhel.log", logger_module.LOG_DIR) try: - logger_module.ConversionStage.set_stage("prepare") + ConversionPhases.set_current(ConversionPhases.PREPARE) perform_boilerplate() gather_system_info() @@ -146,11 +138,11 @@ def main_locked(): # we don't fail in case rollback is triggered during # actions.run_pre_actions() (either from a bug or from the user hitting # Ctrl-C) - process_phase = ConversionPhase.PRE_PONR_CHANGES + ConversionPhases.set_current(ConversionPhases.PRE_PONR_CHANGES) pre_conversion_results = actions.run_pre_actions() if toolopts.tool_opts.activity == "analysis": - process_phase = ConversionPhase.ANALYZE_EXIT + ConversionPhases.set_current(ConversionPhases.ANALYZE_EXIT) raise _AnalyzeExit() _raise_for_skipped_failures(pre_conversion_results) @@ -172,7 +164,7 @@ def main_locked(): loggerinst.warning("********************************************************") utils.ask_to_continue() - process_phase = ConversionPhase.POST_PONR_CHANGES + ConversionPhases.set_current(ConversionPhases.POST_PONR_CHANGES) post_conversion_results = actions.run_post_actions() _raise_for_skipped_failures(post_conversion_results) @@ -202,25 +194,25 @@ def main_locked(): return ConversionExitCodes.SUCCESSFUL except _InhibitorsFound as err: loggerinst.critical_no_exit(str(err)) - results = _pick_conversion_results(process_phase, pre_conversion_results, post_conversion_results) - _handle_main_exceptions(process_phase, results) + results = _pick_conversion_results(pre_conversion_results, post_conversion_results) + _handle_main_exceptions(results) return _handle_inhibitors_found_exception() except exceptions.CriticalError as err: loggerinst.critical_no_exit(err.diagnosis) - results = _pick_conversion_results(process_phase, pre_conversion_results, post_conversion_results) - return _handle_main_exceptions(process_phase, results) + results = _pick_conversion_results(pre_conversion_results, post_conversion_results) + return _handle_main_exceptions(results) except (Exception, SystemExit, KeyboardInterrupt) as err: - results = _pick_conversion_results(process_phase, pre_conversion_results, post_conversion_results) - return _handle_main_exceptions(process_phase, results) + results = _pick_conversion_results(pre_conversion_results, post_conversion_results) + return _handle_main_exceptions(results) finally: if not backup.backup_control.rollback_failed: # Write the assessment to a file as json data so that other tools can # parse and act upon it. - results = _pick_conversion_results(process_phase, pre_conversion_results, post_conversion_results) - - if results and process_phase in _REPORT_MAPPING: - json_report, txt_report = _REPORT_MAPPING[process_phase] + results = _pick_conversion_results(pre_conversion_results, post_conversion_results) + current_phase = ConversionPhases.current_phase + if results and current_phase and current_phase.name in _REPORT_MAPPING: + json_report, txt_report = _REPORT_MAPPING[current_phase.name] report.summary_as_json(results, json_report) report.summary_as_txt(results, txt_report) @@ -248,29 +240,29 @@ def _raise_for_skipped_failures(results): # TODO(r0x0d): Better function name -def _pick_conversion_results(process_phase, pre_conversion, post_conversion): +def _pick_conversion_results(pre_conversion, post_conversion): """Utilitary function to define which action results to use Maybe not be necessary (or even correct), but it is the best approximation idea for now. """ - if process_phase == ConversionPhase.POST_PONR_CHANGES: + if ConversionPhases.current_phase == ConversionPhases.POST_PONR_CHANGES: return post_conversion return pre_conversion -def _handle_main_exceptions(process_phase, results=None): +def _handle_main_exceptions(results=None): """Common steps to handle graceful exit due to several different Exception types.""" breadcrumbs.breadcrumbs.finish_collection() no_changes_msg = "No changes were made to the system." utils.log_traceback(toolopts.tool_opts.debug) - if process_phase == ConversionPhase.POST_CLI: + if ConversionPhases.is_current(ConversionPhases.POST_CLI): loggerinst.info(no_changes_msg) return ConversionExitCodes.FAILURE - elif process_phase == ConversionPhase.PRE_PONR_CHANGES: + elif ConversionPhases.is_current(ConversionPhases.PRE_PONR_CHANGES): # Update RHSM custom facts only when this returns False. Otherwise, # sub-man get uninstalled and the data is removed from the RHSM server. if not subscription.should_subscribe(): @@ -281,7 +273,7 @@ def _handle_main_exceptions(process_phase, results=None): pre_conversion_results=results, include_all_reports=True, ) - elif process_phase == ConversionPhase.POST_PONR_CHANGES: + elif ConversionPhases.is_current(ConversionPhases.POST_PONR_CHANGES): # After the process of subscription is done and the mass update of # packages is started convert2rhel will not be able to guarantee a # system rollback without user intervention. If a proper rollback diff --git a/convert2rhel/utils/phase.py b/convert2rhel/utils/phase.py new file mode 100644 index 000000000..966eba0df --- /dev/null +++ b/convert2rhel/utils/phase.py @@ -0,0 +1,87 @@ +# -*- coding: utf-8 -*- +# +# Copyright(C) 2024 Red Hat, Inc. +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + + +class ConversionPhase: + """ + Conversion phase to hold name and logging-friendly name. + """ + + def __init__(self, name, log_name=None): # type: (str, str|None) -> None + self.name = name + self.log_name = log_name + + def __str__(self): + return self.log_name if self.log_name else self.name + + def __repr__(self): + return self.name + + def __hash__(self): + return hash(self.name) + + def __eq__(self, value): + return self.__hash__() == value.__hash__() + + +class ConversionPhases: + """During conversion we will be in different states depending on + where we are in the execution. This class establishes the different phases + that we have as well as what the current phase is set to. + """ + + POST_CLI = ConversionPhase(name="POST_CLI") + PREPARE = ConversionPhase(name="PREPARE", log_name="Prepare") + # PONR means Point Of No Return + PRE_PONR_CHANGES = ConversionPhase(name="PRE_PONR_CHANGES", log_name="Analyze") + # Phase to exit the Analyze SubCommand early + ANALYZE_EXIT = ConversionPhase(name="ANALYZE_EXIT") + POST_PONR_CHANGES = ConversionPhase(name="POST_PONR_CHANGES", log_name="Convert") + ROLLBACK = (ConversionPhase(name="ROLLBACK", log_name="Rollback"),) + + current_phase = None # type: ConversionPhase|None + + @classmethod + def get(cls, key): # type: (str) -> ConversionPhase + return next((phase for phase in cls.__dict__ if isinstance(phase, ConversionPhase) and phase.name == key)) + + @classmethod + def has(cls, key): # type: (str) -> bool + try: + cls.get(key) + return True + except StopIteration: + return False + + @classmethod + def set_current(cls, phase): # type: (str|ConversionPhase|None) -> None + if phase is None: + cls.current_phase = None + elif isinstance(phase, str) and cls.has(phase): + cls.current_phase = cls.get(phase) + elif isinstance(phase, ConversionPhase) and phase.name in cls.__dict__: + cls.current_phase = phase + else: + raise NotImplementedError("The {} phase is not implemented in the {} class".format(phase, cls.__name__)) + + @classmethod + def is_current(cls, phase): # type: (str|ConversionPhase) -> bool + if isinstance(phase, str): + return cls.current_phase == cls.get(phase) + elif isinstance(phase, ConversionPhase) and phase.name in cls.__dict__: + return cls.current_phase == phase + raise TypeError("Unexpected type, wanted str or {}".format(ConversionPhase.__name__))