in src/core/src/service_interfaces/StatusHandler.py [0:0]
def load_status_file_components(self, initial_load=False):
""" Loads currently persisted status data into memory.
:param initial_load: If no status file exists AND initial_load is true, a default initial status file is created.
:return: None
"""
# Initializing records safely
self.__installation_substatus_json = None
self.__installation_summary_json = None
self.__installation_packages = []
self.__installation_errors = []
self.__installation_packages_map = collections.OrderedDict()
self.__installation_substatus_msg_copy = None
self.__installation_patches_copy = []
self.__installation_patches_removed = []
self.__assessment_substatus_json = None
self.__assessment_summary_json = None
self.__assessment_packages = []
self.__assessment_errors = []
self.__assessment_packages_map = collections.OrderedDict()
self.__assessment_substatus_msg_copy = None
self.__assessment_patches_copy = []
self.__assessment_patches_removed = []
self.__metadata_for_healthstore_substatus_json = None
self.__metadata_for_healthstore_summary_json = None
self.__configure_patching_substatus_json = None
self.__configure_patching_summary_json = None
self.__configure_patching_errors = []
self.__configure_patching_auto_assessment_errors = []
self.__truncation_timestamp = datetime.datetime(1971, 1, 1, 0, 0, 0)
self.__truncated_status_file_json_dumps = None
self.__force_truncation_on = False
self.composite_logger.log_debug("Loading status file components [InitialLoad={0}].".format(str(initial_load)))
# Retain 10 complete status files, and remove older files
self.__removed_older_complete_status_files(self.execution_config.status_folder)
# Verify the status file exists - if not, reset status file
if not os.path.exists(self.complete_status_file_path) and initial_load:
self.composite_logger.log_warning("Status file not found at initial load. Resetting status file to defaults.")
self.__reset_status_file()
return
# Load status data and sanity check structure - raise exception if data loss risk is detected on corrupt data
complete_status_file_data = self.__load_complete_status_file_data(self.complete_status_file_path)
if 'status' not in complete_status_file_data or 'substatus' not in complete_status_file_data['status']:
self.composite_logger.log_error("Malformed status file. Resetting status file for safety.")
self.__reset_status_file()
return
# Load portions of data that need to be built on for next write - raise exception if corrupt data is encountered
# todo: refactor
self.__high_level_status_message = complete_status_file_data['status']['formattedMessage']['message']
for i in range(0, len(complete_status_file_data['status']['substatus'])):
name = complete_status_file_data['status']['substatus'][i]['name']
if name == Constants.PATCH_INSTALLATION_SUMMARY: # if it exists, it must be to spec, or an exception will get thrown
if self.execution_config.exec_auto_assess_only:
self.__installation_substatus_json = complete_status_file_data['status']['substatus'][i]
else:
self.__installation_summary_json = self.__get_substatus_message(complete_status_file_data, i)
# Reload patches into installation ordered map for fast look up
self.__installation_packages_map = collections.OrderedDict((package["patchId"], package) for package in self.__installation_summary_json['patches'])
self.__installation_packages = list(self.__installation_packages_map.values())
self.__maintenance_window_exceeded = bool(self.__installation_summary_json['maintenanceWindowExceeded'])
self.__installation_reboot_status = self.__installation_summary_json['rebootStatus']
errors = self.__installation_summary_json['errors']
if errors is not None and errors['details'] is not None:
self.__installation_errors = errors['details']
self.__installation_total_error_count = self.__get_total_error_count_from_prev_status(errors['message'])
if name == Constants.PATCH_ASSESSMENT_SUMMARY: # if it exists, it must be to spec, or an exception will get thrown
self.__assessment_summary_json = self.__get_substatus_message(complete_status_file_data, i)
# Reload patches into assessment ordered map for fast look up
self.__assessment_packages_map = collections.OrderedDict((package["patchId"], package) for package in self.__assessment_summary_json['patches'])
self.__assessment_packages = list(self.__assessment_packages_map.values())
errors = self.__assessment_summary_json['errors']
if errors is not None and errors['details'] is not None:
self.__assessment_errors = errors['details']
self.__assessment_total_error_count = self.__get_total_error_count_from_prev_status(errors['message'])
if name == Constants.PATCH_METADATA_FOR_HEALTHSTORE: # if it exists, it must be to spec, or an exception will get thrown
if self.execution_config.exec_auto_assess_only:
self.__metadata_for_healthstore_substatus_json = complete_status_file_data['status']['substatus'][i]
else:
self.__metadata_for_healthstore_summary_json = self.__get_substatus_message(complete_status_file_data, i)
if name == Constants.CONFIGURE_PATCHING_SUMMARY: # if it exists, it must be to spec, or an exception will get thrown
if self.execution_config.exec_auto_assess_only:
self.__configure_patching_substatus_json = complete_status_file_data['status']['substatus'][i]
else:
self.__configure_patching_summary_json = self.__get_substatus_message(complete_status_file_data, i)
errors = self.__configure_patching_summary_json['errors']
if errors is not None and errors['details'] is not None:
self.__configure_patching_errors = errors['details']
self.__configure_patching_top_level_error_count = self.__get_total_error_count_from_prev_status(errors['message'])