centos-stream-tag-fixer.py

#!/usr/bin/python

import click
import koji
import koji_cli.lib
import rpm


TAG='dist-c8-stream'

class Build:
    __slots__ = ['name', 'epoch', 'version', 'release', 'nvr']

    def __init__(self, build_info):
        self.name = build_info['name']
        self.epoch = str(build_info['epoch']) if build_info['epoch'] else build_info['epoch']
        self.version = build_info['version']
        self.release = build_info['release']
        self.nvr = build_info['nvr']

    def __repr__(self):
        if self.epoch:
            return f'<{self.__class__.__name__}: "{self.nvr} (epoch {self.epoch})">'
        else:
            return f'<{self.__class__.__name__}: "{self.nvr}">'

    def __str__(self):
        return f'{self.nvr}'

    def __eq__(self, other):
        return rpm.labelCompare(
            (self.epoch, self.version, self.release),
            (other.epoch, other.version, other.release)
        ) == 0

    def __lt__(self, other):
        return rpm.labelCompare(
            (self.epoch, self.version, self.release),
            (other.epoch, other.version, other.release)
        ) == -1

    def __gt__(self, other):
        return rpm.labelCompare(
            (self.epoch, self.version, self.release),
            (other.epoch, other.version, other.release)
        ) == 1


@click.command()
@click.option('-p', '--profile', default='centos', help='koji profile')
@click.argument('package')
def main(profile, package):
    # Set up koji session.
    profile_module = koji.get_profile_module(profile)
    session = profile_module.ClientSession(profile_module.config.server)
    try:
        koji_cli.lib.activate_session(session, profile_module.config)
    except KeyboardInterrupt:
        raise click.ClickException('Aborting koji session activation')
    except profile_module.ServerOffline:
        raise click.ClickException('failed to activate koji session')

    # get builds in tag
    try:
        builds = [Build(build_info) for build_info in session.listTagged(TAG, package=package)]
    except profile_module.ServerOffline:
        raise click.ClickException('failed to get builds')

    if not builds:
        raise click.ClickException(f'no builds found for {package}')

    builds.sort()

    # determine latest build
    try:
        [latest_build_info] = session.getLatestBuilds(TAG, package=package)
    except profile_module.ServerOffline:
        raise click.ClickException('failed to get latest build')

    latest_build = Build(latest_build_info)

    # output builds, with the latest one highlighted
    for build in builds:
        if build == latest_build:
            click.secho(build.nvr, fg='cyan')
        else:
            click.echo(build.nvr)

    if builds[-1] != latest_build:
        if click.confirm('fix?'):
            try:
                session.untagBuild(TAG, builds[-1].nvr)
            except profile_module.ServerOffline:
                raise click.ClickException(f'failed to untag {builds[-1]}')

            try:
                session.tagBuild(TAG, builds[-1].nvr)
            except profile_module.ServerOffline:
                raise click.ClickException(f'failed to tag {builds[-1]}')


if __name__ == '__main__':
    main()