def _qubes_updater_parse_progress()

in sdw_updater/Updater.py [0:0]

• 30 lines of code
• 11 McCabe index (conditional complexity)

def _qubes_updater_parse_progress(stream, result, templates, progress_callback=None):
    update_progress = {}

    for template in templates:
        result[template] = UpdateStatus.UPDATES_IN_PROGRESS

    while True:
        untrusted_line = stream.readline()
        if len(untrusted_line) == 0:
            break

        line = Util.strip_ansi_colors(untrusted_line.decode("utf-8").rstrip())
        try:
            vm, status, info = line.split()
        except ValueError:
            sdlog.warn("Line in Qubes updater's output could not be parsed")
            continue

        if status == "updating":
            if update_progress.get(vm) is None:
                sdlog.info(f"Starting update on template: '{vm}'")
                update_progress[vm] = 0
            else:
                vm_progress = int(float(info))
                update_progress[vm] = vm_progress
                if progress_callback:
                    progress_callback(sum(update_progress.values()) // len(templates))

        # First time complete (status "done") may be repeated various times
        if status == "done" and result[vm] == UpdateStatus.UPDATES_IN_PROGRESS:
            result[vm] = UpdateStatus.from_qubes_updater_name(info)
            if result[vm] == UpdateStatus.UPDATES_OK:
                sdlog.info(f"Update successful for template: '{vm}'")
                update_progress[vm] = 100
            else:
                sdlog.error(f"Update failed for template: '{vm}'")