oem.py

#!/usr/bin/python3
# Copyright (C) 2020  Canonical Ltd.

# This program is free software; you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation; version 3

# This program 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 General Public License for more
# details.

# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 59 Temple
# Place, Suite 330, Boston, MA 02111-1307 USA


import argparse
import fnmatch
import logging

from functools import lru_cache
from launchpadlib.launchpad import Launchpad
from distro_info import UbuntuDistroInfo


@lru_cache
def get_lts_distroseries(lp, log):
    udi = UbuntuDistroInfo()
    supported_lts = [r for r in udi.supported() if udi.is_lts(r)]
    ret = set()
    ubuntu = lp.distributions["ubuntu"]

    return [ubuntu.getSeries(name_or_version=lts) for lts in supported_lts]


def sync_packageset(lp, log, sources):
    log.info(f"Syncing {', '.join(sources)} into packageset")

    for ds in get_lts_distroseries(lp, log):
        packageset = lp.packagesets.getByName(
            distroseries=ds, name="canonical-oem-metapackages"
        )
        sources_in_set = set(packageset.getSourcesIncluded())
        if sources_in_set:
            log.debug("Set currently contains: %s", ", ".join(sources_in_set))
        else:
            log.debug("Set is currently empty")
        sources_to_remove = sources_in_set - sources
        sources_to_add = sources - sources_in_set
        if sources_to_remove:
            log.info(
                "Removing %s from %s/%s",
                ", ".join(sources_to_remove),
                "canonical-oem-metapackages",
                ds.name,
            )
            # XXX
            # packageset.removeSources(names=sources_to_remove)
        if sources_to_add:
            log.info(
                "Adding %s to %s/%s",
                ", ".join(sources_to_add),
                "canonical-oem-metapackages",
                ds.name,
            )
            # XXX
            # packageset.addSources(names=sources_to_add)


def get_sources(lp, log, archive):
    ret = set()

    for ds in get_lts_distroseries(lp, log):
        log.debug(
            "Looking at packages for ~%s/%s/%s/%s",
            archive.owner.name,
            archive.distribution.name,
            archive.name,
            ds.name,
        )
        sources = archive.getPublishedSources(
            status="Published", source_name="oem", exact_match=False
        )
        for source in sources:
            name = source.source_package_name
            if name in ret:
                continue
            if fnmatch.fnmatch(name, "oem-*-meta"):
                log.debug("Found: %s", name)
                ret.add(name)
            else:
                log.debug("Ignoring %s", name)

    return ret


def get_archive_packages(lp, log):
    ubuntu_archive = lp.distributions["ubuntu"].main_archive

    return get_sources(lp, log, ubuntu_archive)


def get_ppa_packages(lp, log):
    # XXX: Change to ~canonical-oem-metapackage/+archive/oem-metapackage-staging, when this is created
    ppa = lp.load("~fourdollars/+archive/ubuntu/staging-ubuntu-metapkg-ppa")

    return get_sources(lp, log, ppa)


def main():
    parser = argparse.ArgumentParser(
        description="Sync canonical-oem-metapackages packageset with "
        "~canonical-oem-metapackage-uploaders"
        "/oem-metapackage-staging PPA and Ubuntu archive"
    )
    parser.add_argument("--debug", "-d", action="store_true")
    args = parser.parse_args()

    log = logging.getLogger(__name__)
    ch = logging.StreamHandler()
    formatter = logging.Formatter("%(levelname)s: %(message)s")
    ch.setFormatter(formatter)
    log.addHandler(ch)

    if args.debug:
        log.setLevel(logging.DEBUG)

    lp = Launchpad.login_with("oem-generate", "production", "devel")

    ppa_packages = get_ppa_packages(lp, log)
    archive_packages = get_archive_packages(lp, log)

    sync_packageset(lp, log, ppa_packages | archive_packages)


if __name__ == "__main__":
    main()