def start_assessment()

in src/core/src/core_logic/PatchAssessor.py [0:0]
52 lines of code
10 McCabe index (conditional complexity)

    def start_assessment(self):
        """ Start a patch assessment """
        self.status_handler.set_current_operation(Constants.ASSESSMENT)
        self.raise_if_telemetry_unsupported()
        self.raise_if_min_python_version_not_met()

        if self.execution_config.exec_auto_assess_only and not self.should_auto_assessment_run():
            self.composite_logger.log("\nSkipping automatic patch assessment... [ShouldAutoAssessmentRun=False]\n")
            self.lifecycle_manager.lifecycle_status_check()
            return True

        self.composite_logger.log("\nStarting patch assessment... [MachineId: " + self.env_layer.platform.vm_name() + "][ActivityId: " + self.execution_config.activity_id + "][StartTime: " + self.execution_config.start_time + "]")
        self.write_assessment_state()   # success / failure does not matter, only that an attempt started

        self.stopwatch.start()
        self.status_handler.set_assessment_substatus_json(status=Constants.STATUS_TRANSITIONING)
        retry_count = 0

        for i in range(0, Constants.MAX_ASSESSMENT_RETRY_COUNT):
            try:
                self.composite_logger.log("\n\nGetting available patches...")
                self.package_manager.refresh_repo()
                self.status_handler.reset_assessment_data()

                if self.lifecycle_manager is not None:
                    self.lifecycle_manager.lifecycle_status_check()     # may terminate the code abruptly, as designed

                # All updates
                retry_count = retry_count + 1
                
                # All updates
                packages, package_versions = self.package_manager.get_all_updates()
                self.telemetry_writer.write_event("Full assessment: " + str(packages), Constants.TelemetryEventLevel.Verbose)
                self.status_handler.set_package_assessment_status(packages, package_versions)
                if self.lifecycle_manager is not None:
                    self.lifecycle_manager.lifecycle_status_check()     # may terminate the code abruptly, as designed
                sec_packages, sec_package_versions = self.package_manager.get_security_updates()

                # Tag security updates
                self.telemetry_writer.write_event("Security assessment: " + str(sec_packages), Constants.TelemetryEventLevel.Verbose)
                self.status_handler.set_package_assessment_status(sec_packages, sec_package_versions, Constants.PackageClassification.SECURITY)

                # Set the security-esm packages in status.
                self.package_manager.set_security_esm_package_status(Constants.ASSESSMENT, packages=[])

                # ensure reboot status is set
                reboot_pending = self.package_manager.is_reboot_pending()
                self.status_handler.set_reboot_pending(reboot_pending)

                self.status_handler.set_assessment_substatus_json(status=Constants.STATUS_SUCCESS)
                break   # avoid retries for success

            except Exception as error:
                if i < Constants.MAX_ASSESSMENT_RETRY_COUNT - 1:
                    error_msg = 'Retriable error retrieving available patches: ' + repr(error)
                    self.composite_logger.log_warning(error_msg)
                    self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.DEFAULT_ERROR)
                    time.sleep(2*(i + 1))
                else:
                    error_msg = 'Error retrieving available patches: ' + repr(error)
                    self.composite_logger.log_error(error_msg)
                    self.write_assessment_perf_logs(retry_count, Constants.TaskStatus.FAILED, error_msg)
                    self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.DEFAULT_ERROR)
                    if Constants.ERROR_ADDED_TO_STATUS not in repr(error):
                        error.args = (error.args, "[{0}]".format(Constants.ERROR_ADDED_TO_STATUS))
                    self.status_handler.set_assessment_substatus_json(status=Constants.STATUS_ERROR)
                    raise

        self.write_assessment_perf_logs(retry_count, Constants.TaskStatus.SUCCEEDED, "")
        self.composite_logger.log("\nPatch assessment completed.\n")
        return True