list_updatable_packages

#!/usr/bin/env python3

import json
import os
import subprocess
import sys
from packaging.version import Version
from itertools import chain
from pathlib import Path
from typing import Dict, Generator, Iterable, Mapping, Optional, Tuple

import requests
import tempfile

SUPPORTED_TEMPLATES = ('foreman_plugin', 'hammer_plugin', 'smart_proxy_plugin')
EXTRA_SUPPORTED_GEMS = ('foreman_maintain',)
EXCLUDED_GEMS = ()
SESSION = requests.Session()
DEBUG = (len(sys.argv) == 2 and sys.argv[1] == 'debug')

class Spec:
    def __init__(self, path: Path):
        self.path = path
        spec = path.read_text()
        self.lines = spec.splitlines()
        self._globals: Optional[Mapping[str, str]] = None

    @property
    def directory(self) -> str:
        return self.path.parent.as_posix()

    @property
    def package_name(self) -> str:
        return self.path.name[:-len('.spec')]

    @property
    def template(self) -> Optional[str]:
        for line in self.lines:
            if line.startswith('# template:'):
                _, _, template = line.split(None, 2)
                return template

        return None

    @property
    def globals(self) -> Mapping[str, str]:
        """
        Perform very naive spec parsing to extract the globals
        """
        if self._globals is None:
            self._globals = {}
            for line in self.lines:
                if line.startswith('%global'):
                    definition, name, value = line.split(None, 2)
                    if definition == '%global':
                        self._globals[name] = value

        return self._globals

    @property
    def is_nightly(self) -> bool:
        return 'prereleasesource' in self.globals

    @property
    def is_excluded(self) -> bool:
        return self.gem_name in EXCLUDED_GEMS

    @property
    def is_updateable(self) -> bool:
        return not self.is_nightly and not self.is_excluded

    @property
    def is_supported(self) -> bool:
        return self.template in SUPPORTED_TEMPLATES or self.gem_name in EXTRA_SUPPORTED_GEMS

    @property
    def gem_name(self) -> Optional[str]:
        return self.globals.get('gem_name')

    @property
    def npm_name(self) -> Optional[str]:
        return self.globals.get('npm_name')

    @property
    def npm_strategy(self) -> Optional[str]:
        if any('registry.npmjs.org.tgz' in file for file in os.listdir(self.path.parent)):
            return 'bundle'
        else:
            return 'single'

    @property
    def version(self) -> str:
        cmd = ['rpmspec', '--srpm', '--query', '--queryformat=%{version}', self.path.as_posix()]
        return subprocess.check_output(cmd, stderr=subprocess.DEVNULL, universal_newlines=True)

    def __eq__(self, other):
        return self.path == other.path

    def __hash__(self):
        return hash(self.path)


def find_gemfile_lock_specs(ci_job: str, subdir: str, project: str) -> Generator[Tuple[Spec, str, str], None, None]:
    """
    Find all specs based on what's in the bundler lockfile. This Gemfile.lock is retrieved from
    the last successful build in CI.
    """
    url = f'https://ci.theforeman.org/job/{ci_job}/lastSuccessfulBuild/artifact/Gemfile.lock'
    base_path = Path('packages') / subdir

    for gem in parse_gemfile_lock(fetch_gemfile_lock(url)):
        package_name = f'rubygem-{gem["name"]}'
        path = base_path / package_name / f'{package_name}.spec'
        if path.is_file():
            spec = Spec(path)
            if spec.is_updateable:
                yield spec, gem['version'], project


def find_package_lock_specs(ci_job: str, subdir: str, project: str) -> Generator[Tuple[Spec, str, str], None, None]:
    """
    Find all specs based on what's in the NPM package-lock.json. This package-lock.json is retrieved from
    the last successful build in CI.
    """
    url = f'https://ci.theforeman.org/job/{ci_job}/lastSuccessfulBuild/artifact/package-lock.json'
    base_path = Path('packages') / subdir

    for npm_module in parse_package_lock(fetch_package_lock(url)):
        npm_name = npm_module['name'].replace('@', '').replace('/', '-')
        package_name = f'nodejs-{npm_name}'
        path = base_path / package_name / f'{package_name}.spec'
        if path.is_file():
            spec = Spec(path)
            if spec.is_updateable:
                yield spec, npm_module['version'], project


def find_packaged_specs() -> Generator[Tuple[Spec, None, None], None, None]:
    if DEBUG: print('Finding supported leaf specs')
    for path in Path('packages').glob('*/*/*.spec'):
        spec = Spec(path)
        if spec.is_supported and spec.is_updateable:
            yield spec, None, None


def parse_gemfile_lock(gemfile_lock: str) -> Iterable[Mapping[str, str]]:
    with tempfile.TemporaryDirectory() as tmpdirname:
        with open(f'{tmpdirname}/Gemfile.lock', 'w') as fp:
            fp.write(gemfile_lock)

        with open(f'{tmpdirname}/Gemfile', 'w') as fp:
            pass

        if DEBUG: print('Parsing Gemfile.lock')
        script = Path(__file__).absolute().parent / 'parse-gemfile-lock'
        return json.loads(subprocess.check_output([script], universal_newlines=True, cwd=tmpdirname))


def parse_package_lock(package_lock: str) -> Iterable[Mapping[str, str]]:
    with tempfile.TemporaryDirectory() as tmpdirname:
        with open(f'{tmpdirname}/package-lock.json', 'w') as fp:
            fp.write(package_lock)

        if DEBUG: print('Parsing package-lock.json')
        script = Path(__file__).absolute().parent / 'parse-package-lock'
        return json.loads(subprocess.check_output([script], universal_newlines=True, cwd=tmpdirname))


def fetch_gemfile_lock(url) -> str:
    if DEBUG: print(f'Fetching Gemfile.lock from {url}')
    response = SESSION.get(url)
    response.raise_for_status()
    return response.text


def fetch_package_lock(url) -> str:
    if DEBUG: print(f'Fetching package-lock.json from {url}')
    response = SESSION.get(url)
    response.raise_for_status()
    return response.text


def latest_version(spec: Spec) -> str:
    if spec.gem_name:
        url = f'https://rubygems.org/api/v1/versions/{spec.gem_name}/latest.json'
        response = SESSION.get(url)
        response.raise_for_status()
        return response.json()['version']
    raise ValueError('Unable to determine latest version', spec)


def merge_specs(specs: Iterable[Tuple[Spec, Optional[str], Optional[str]]]) -> Mapping[Spec, Dict[Optional[str], Optional[str]]]:
    """
    Merge specs

    This can process the same spec file multiple times. If a spec is already listed but without a
    version, it will take the more specific constraint. The last entry is used if a constraint was
    already given while a warning is printed.
    """
    result: Dict[Spec, Dict[Optional[str], Optional[str]]] = {}
    for spec, new_version, project in specs:
        existing_spec = result.get(spec)
        if existing_spec and existing_spec['new_version'] != new_version:
            # TODO: GH annotation?
            print(f'Found conflict for {spec.package_name}: {result[spec]["new_version"]} ({result[spec]["project"]}) != {new_version} ({project})', file=sys.stderr)
            result[spec] = {'new_version': 'CONFLICT', 'project': project}
        else:
            result[spec] = {'new_version': new_version, 'project': project}
    return result


def build_matrix(specs: Mapping[Spec, Dict[Optional[str], Optional[str]]]) -> Generator[Mapping[str, Optional[str]], None, None]:
    for spec, metadata in specs.items():
        new_version = metadata['new_version']
        try:
            current_version = spec.version
        except subprocess.CalledProcessError as e:  # pylint: disable=invalid-name
            print('Failed to determine version for', spec.path.as_posix(), str(e), file=sys.stderr)
        else:
            if not new_version:
                new_version = latest_version(spec)

            if new_version != 'CONFLICT' and Version(current_version) > Version(new_version):
                print(f'Found downgrade for {spec.package_name}: {current_version} > {new_version}', file=sys.stderr)
                new_version = 'DOWNGRADE'

            if current_version != new_version and spec.is_updateable:
                entry = {
                    'directory': spec.directory,
                    'package_name': spec.package_name,
                    'gem_name': spec.gem_name,
                    'npm_name': spec.npm_name,
                    'npm_strategy': spec.npm_strategy,
                    'current_version': current_version,
                    'new_version': new_version,
                }
                yield entry


def main() -> None:
    specs = chain(
        find_packaged_specs(),
        find_gemfile_lock_specs('foreman-develop-source-release', 'foreman', 'foreman'),
        find_gemfile_lock_specs('smart-proxy-develop-source-release', 'foreman', 'smart-proxy'),
        find_gemfile_lock_specs('katello-master-source-release', 'katello', 'katello'),
        find_gemfile_lock_specs('hammer-cli-master-source-release', 'foreman', 'hammer-cli'),
        find_gemfile_lock_specs('hammer-cli-foreman-master-source-release', 'foreman', 'hammer-cli-foreman'),
        find_package_lock_specs('foreman-develop-source-release', 'foreman', 'foreman'),
        find_package_lock_specs('katello-master-source-release', 'katello', 'katello'),
    )

    matrix = list(build_matrix(merge_specs(specs)))

    if 'GITHUB_ACTION' in os.environ:
        directories = [entry['directory'] for entry in matrix]
        with open(os.environ['GITHUB_OUTPUT'], 'a') as github_output:
            print(f'directories={json.dumps(directories)}', file=github_output)
            print(f'matrix={json.dumps(matrix)}', file=github_output)
    for entry in matrix:
        print(entry['directory'], entry['new_version'])


if __name__ == '__main__':
    main()