fedora-coreos-config/versionary

#!/usr/bin/python3 -u

# This file originally lived in
# https://github.com/coreos/fedora-coreos-releng-automation. See that repo for
# archeological git research.

'''
    Implements the Fedora CoreOS versioning scheme as per:
        https://github.com/coreos/fedora-coreos-tracker/issues/81
        https://github.com/coreos/fedora-coreos-tracker/issues/211
'''

import argparse
import dotenv
import json
import os
import platform
import re
import subprocess
import sys
import time
import yaml

from datetime import datetime

# streams which don't use lockfiles
UNLOCKED_STREAMS = [
    'bodhi-updates-testing',
    'bodhi-updates',
]

# https://github.com/coreos/fedora-coreos-tracker/issues/211#issuecomment-543547587
STREAM_TO_NUM = {
    'next': 1,
    'testing': 2,
    'stable': 3,
    'next-devel': 10,
    'testing-devel': 20,
    'rawhide': 91,
    'branched': 92,
    'bodhi-updates-testing': 93,
    'bodhi-updates': 94,
}


def main():
    args = parse_args()
    if args.workdir is not None:
        os.chdir(args.workdir)
    assert os.path.isdir('builds'), 'Missing builds/ dir'

    config = dotenv.dotenv_values('src/config/build-args.conf')
    x, y, z = (get_x(config), get_y(config), get_z(config, args.dev))
    n = get_next_iteration(x, y, z)
    new_version = f'{x}.{y}.{z}.{n}'

    # sanity check the new version by trying to re-parse it
    assert parse_version(new_version) is not None
    print(new_version)


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument('--workdir', help="path to cosa workdir")
    parser.add_argument('--dev', action='store_true', help="generate a developer version")
    return parser.parse_args()


def get_x(config):
    """
        X is the base release version on which we're based.
    """
    base_version = config['BASE_VERSION']
    eprint(f"x: {base_version} (from config)")
    return int(base_version)


def get_y(config):
    """
        Y is the base snapshot date in YYYYMMDD format of Fedora. We derive
        this using the timestamp in the base lockfile.
    """

    stream = config['STREAM']

    # XXX: should sanity check that the lockfiles for all the basearches have
    # matching timestamps
    exts = ['json', 'yaml']
    basearch = platform.machine()
    for ext in exts:
        try:
            with open(f"src/config/manifest-lock.{basearch}.{ext}") as f:
                lockfile = yaml.safe_load(f)
                generated = lockfile.get('metadata', {}).get('generated')
                if not generated:
                    raise Exception("Missing 'metadata.generated' key "
                                    f"from {lockfile}")
                dt = datetime.strptime(generated, '%Y-%m-%dT%H:%M:%SZ')
                assert stream not in UNLOCKED_STREAMS
                msg_src = "from lockfile"
                break
        except FileNotFoundError:
            continue
    else:
        # must be an unlocked stream
        assert stream in UNLOCKED_STREAMS
        msg_src = "unlocked stream"
        dt = datetime.now()

    ymd = dt.strftime('%Y%m%d')
    eprint(f"y: {ymd} ({msg_src})")
    return int(ymd)


def get_z(config, dev):
    """
        Z is the stream indicator.
    """
    if dev:
        eprint("z: dev (overridden)")
        return 'dev'
    stream = config['STREAM']
    assert stream in STREAM_TO_NUM, f"Unknown stream: {stream}"
    mapped = STREAM_TO_NUM[stream]
    eprint(f"z: {mapped} (mapped from stream {stream})")
    return mapped


def get_next_iteration(x, y, z):
    try:
        with open('builds/builds.json') as f:
            builds = json.load(f)
    except FileNotFoundError:
        builds = {'builds': []}

    if len(builds['builds']) == 0:
        eprint("n: 0 (no previous builds)")
        return 0

    last_buildid = builds['builds'][0]['id']
    last_version = parse_version(last_buildid)
    if not last_version:
        eprint(f"n: 0 (previous version {last_buildid} does not match scheme)")
        return 0

    if (x, y, z) != last_version[:3]:
        eprint(f"n: 0 (previous version {last_buildid} x.y.z does not match)")
        return 0

    n = last_version[3] + 1
    eprint(f"n: {n} (incremented from previous version {last_buildid})")
    return n


def parse_version(version):
    m = re.match(r'^([0-9]{2})\.([0-9]{8})\.([0-9]+|dev)\.([0-9]+)$', version)
    if m is None:
        return None
    # sanity-check date
    try:
        time.strptime(m.group(2), '%Y%m%d')
    except ValueError:
        return None
    return tuple(map(lambda x: 'dev' if x == 'dev' else int(x), m.groups()))


def eprint(*args):
    print(*args, file=sys.stderr)


if __name__ == "__main__":
    sys.exit(main())