image-refresh

#!/usr/bin/env python3

# This file is part of Cockpit.
#
# Copyright (C) 2016-2024 Red Hat, Inc.
#
# Cockpit is free software; you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation; either version 2.1 of the License, or
# (at your option) any later version.
#
# Cockpit 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
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Cockpit; If not, see <http://www.gnu.org/licenses/>.

import json
import os
import shlex
import subprocess
import sys
from typing import Any

import task
from lib import testmap
from lib.constants import BOTS_DIR, SCRIPTS_DIR

sys.dont_write_bytecode = True


def run(*cmd: str, check: bool = True, **kwargs: Any) -> int:
    print('\n+', shlex.join(cmd), file=sys.stderr)
    result = subprocess.run(cmd, **kwargs, check=check)
    return result.returncode


def with_logs(n: int, log_url: str | None) -> int | tuple[int, str]:
    if log_url:
        return n, log_url
    return n


def image_refresh(image: str, **kwargs: Any) -> int | tuple[int, str]:
    try:
        log_url = os.environ.get('COCKPIT_CI_LOG_URL')

        dry_run: bool = kwargs['dry']
        triggers = testmap.tests_for_image(image)

        # Cleanup any extraneous disk usage elsewhere
        run('./vm-reset')

        # download the current image, for comparing them; that may not exist yet for newly introduced images
        if run('./image-download', image, check=False) == 0:
            old_image = os.path.realpath(f'{BOTS_DIR}/images/{image}')
        else:
            old_image = None

        # create the new image
        run('./image-create', '--verbose', image,
            env={**os.environ, 'VIRT_BUILDER_NO_CACHE': "yes"})

        # upload the new image
        if dry_run:
            task.would('./image-upload', '--prune-s3', image)
        else:
            run('./image-upload', '--prune-s3', image)

        # compare it to the previous one (on hosts we can ssh to)
        if old_image and os.path.exists(f'{SCRIPTS_DIR}/{image}.setup'):
            run('./image-diff', old_image, image)

        # create branch and push it
        branch = task.branch(image, f"images: Update {image} image", pathspec="images", **kwargs)

        # trigger tests if it is not a pull request
        if branch and "pull" not in kwargs:
            pull = task.pull(branch, labels=['bot', 'no-test'], run_tests=False, **kwargs)

            if log_url:
                # Create a synthetic status for the log URL
                log_status = {
                    'state': 'success',
                    'context': f'image-refresh/{image}',
                    'description': 'Forwarded status',
                    'target_url': log_url
                }
                if dry_run:
                    task.would('add status', json.dumps(log_status, indent=4))
                else:
                    task.api.post(f"statuses/{pull['head']['sha']}", log_status)

            # Trigger this pull request
            if dry_run:
                task.would('trigger tests:', json.dumps(triggers, indent=4))
            else:
                head = pull["head"]["sha"]
                for trigger in triggers:
                    task.api.post(f"statuses/{head}", {
                        "state": "pending",
                        "context": trigger,
                        "description": task.github.NOT_TESTED_DIRECT
                    })

    except subprocess.CalledProcessError as exc:
        return with_logs(exc.returncode, log_url)
    else:
        return with_logs(0, log_url)


if __name__ == '__main__':
    task.main(function=image_refresh, title="Refresh image")