Skip to content

Commit

Permalink
Change format to use new stages style
Browse files Browse the repository at this point in the history
  • Loading branch information
Venefilyn committed Sep 12, 2024
1 parent bc0067d commit 034d54a
Show file tree
Hide file tree
Showing 3 changed files with 109 additions and 44 deletions.
11 changes: 0 additions & 11 deletions convert2rhel/logger.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
58 changes: 25 additions & 33 deletions convert2rhel/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__)
Expand All @@ -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,
),
Expand Down Expand Up @@ -129,14 +121,14 @@ def main_locked():

pre_conversion_results = None
post_conversion_results = None
process_phase = ConversionPhase.POST_CLI
ConversionPhases.set_current("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("POST_CLI")
perform_boilerplate()

gather_system_info()
Expand All @@ -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("PRE_PONR_CHANGES")
pre_conversion_results = actions.run_pre_actions()

if toolopts.tool_opts.activity == "analysis":
process_phase = ConversionPhase.ANALYZE_EXIT
ConversionPhases.set_current("ANALYZE_EXIT")
raise _AnalyzeExit()

_raise_for_skipped_failures(pre_conversion_results)
Expand All @@ -172,7 +164,7 @@ def main_locked():
loggerinst.warning("********************************************************")
utils.ask_to_continue()

process_phase = ConversionPhase.POST_PONR_CHANGES
ConversionPhases.set_current("POST_PONR_CHANGES")
post_conversion_results = actions.run_post_actions()

_raise_for_skipped_failures(post_conversion_results)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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.get("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():
Expand All @@ -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
Expand Down
84 changes: 84 additions & 0 deletions convert2rhel/utils/phase.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,84 @@
# -*- 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 <https://www.gnu.org/licenses/>.


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):

Check warning

Code scanning / CodeQL

Inconsistent equality and hashing Warning

Class
ConversionPhase
implements __hash__ but does not define __eq__.
return hash(self.name)


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__))

0 comments on commit 034d54a

Please sign in to comment.