Source code for

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""Tools to help CI-based builds and artifact deployment."""

import contextlib
import distutils.version
import os

import git

from .build import load_order_file
from .log import echo_info, get_logger

logger = get_logger(__name__)

[docs]def is_master(refname, tag, repodir): """Tells if we're on the master branch via ref_name or tag. This function checks if the name of the branch being built is "master". If a tag is set, then it checks if the tag is on the master branch. If so, then also returns ``True``, otherwise, ``False``. Args: refname: The value of the environment variable ``CI_COMMIT_REF_NAME`` tag: The value of the environment variable ``CI_COMMIT_TAG`` - (may be ``None``) Returns: a boolean, indicating we're building the master branch **or** that the tag being built was issued on the master branch. """ if tag is not None: repo = git.Repo(repodir) _tag = repo.tag("refs/tags/%s" % tag) return _tag.commit in repo.iter_commits(rev="master") return refname == "master"
[docs]def is_private(baseurl, package): """Tells if a given package (with a namespace) is public or private This function checks if a fully qualified package in the format ``<namespace>/<name>`` is publicly accessible. It does this by trying to access ``info/refs?service=git-upload-pack`` from the package in question. This method does **not** rely on the fact the user has access to Gitlab. .. warning:: This method only works for fully qualified package names (i.e., containing at least one forward-slash ``/``). Args: baseurl: The base URL for the gitlab service to consult package: Fully qualified (i.e., with a namespace) package name. For example: ``bob/bob.extension``. Returns: a boolean, indicating if the package is private (``True``) or not (``False``). """ from urllib.error import HTTPError from urllib.request import urlopen private = True try: r = urlopen( baseurl + "/" + package + "/info/refs?service=git-upload-pack" ) private = r.getcode() != 200 except HTTPError as e: private = e.getcode() == 401 return private
[docs]def is_stable(package, refname, tag, repodir): """Determines if the package being published is stable. This is done by checking if a tag was set for the package. If that is the case, we still cross-check the tag is on the "master" branch. If everything checks out, we return ``True``. Else, ``False``. Args: package: Package name in the format "group/name" refname: The current value of the environment ``CI_COMMIT_REF_NAME`` tag: The current value of the enviroment ``CI_COMMIT_TAG`` (may be ``None``) repodir: The directory that contains the clone of the git repository Returns: a boolean, indicating if the current build is for a stable release """ if tag is not None:'Project %s tag is "%s"', package, tag) parsed_tag = distutils.version.LooseVersion( tag[1:] ).version # remove 'v' is_prerelease = any([isinstance(k, str) for k in parsed_tag]) if is_prerelease: logger.warn( "Pre-release detected - not publishing to stable channels" ) return False if is_master(refname, tag, repodir): return True else: logger.warn("Tag %s in non-master branch will be ignored", tag) return False"No tag information available at build")"Considering this to be a pre-release build") return False
[docs]def read_packages(filename): """Return a python list of tuples (repository, branch), given a file containing one package (and branch) per line. Comments are excluded """ lines = load_order_file(filename) packages = [] for line in lines: if "," in line: # user specified a branch path, branch = [k.strip() for k in line.split(",", 1)] packages.append((path, branch)) else: packages.append((line, "master")) return packages
[docs]def select_build_file(basename, paths, branch): """Selects the file to use for a build. This method will return the name of the most adequate build-accessory file (conda_build_config.yaml, recipe_append.yaml) for a given build, in this order of priority: 1. The first file found is returned 2. We first search for a *specific* file if ``branch`` is set 3. If that file does not exist, returns the unbranded filename if that exists in one of the paths 4. If no candidates exists, returns ``None`` The candidate filename is built using ``os.path.splitext(os.path.basename(basename))[0]``. Args: basename: Name of the file to use for the search paths (list): A list of paths leading to the location of the variants file to use. Priority is given to paths that come first branch (str): Optional key to be set when searching for the variants file to use. This is typically the git-branch name of the current branch of the repo being built. Returns: str: A string containing the full, resolved path of the file to use. Returns ``None``, if no candidate is found """ import os basename, extension = os.path.splitext(os.path.basename(basename)) if branch: specific_basename = "%s-%s" % (basename, branch) for path in paths: path = os.path.realpath(path) candidate = os.path.join( path, "%s%s" % (specific_basename, extension) ) if os.path.exists(candidate): return candidate for path in paths: path = os.path.realpath(path) candidate = os.path.join(path, "%s%s" % (basename, extension)) if os.path.exists(candidate): return candidate
[docs]def select_conda_build_config(paths, branch): """Selects the default conda_build_config.yaml. See :py:func:`select_build_file` for implementation details. If no build config file is found by :py:func:`select_build_file`, then returns the default ``conda_build_config.yaml`` shipped with this package. """ from .constants import CONDA_BUILD_CONFIG as default return select_build_file(default, paths, branch) or default
[docs]def select_conda_recipe_append(paths, branch): """Selects the default recipe_append.yaml. See :py:func:`select_build_file` for implementation details. If no recipe append file is found by :py:func:`select_build_file`, then returns the default ``recipe_append.yaml`` shipped with this package. """ from .constants import CONDA_RECIPE_APPEND as default return select_build_file(default, paths, branch) or default
[docs]def select_user_condarc(paths, branch): """Selects the user condarc file to read (if any) See :py:func:`select_build_file` for implementation details. If no recipe condarc is found by :py:func:`select_build_file`, then returns ``None``. """ return select_build_file("condarc", paths, branch)
[docs]def cleanup(dry_run, username, password, includes): """Cleans-up WebDAV resources. Executes if ``dry_run==False`` only. Parameters: dry_run (bool): If set, then does not execute any action, just print what it would do instead. username (str): The user to use for interacting with the WebDAV service password (str): Password the the above user includes (re.SRE_Pattern): A regular expression that matches the names of packages that should be considered for clean-up. For example: for Bob and BATL packages, you may use ``^(bob|batl|gridtk).*`` For BEAT packages you may use ``^beat.*`` """ from .constants import SERVER, WEBDAV_PATHS from .dav import remove_old_beta_packages from .deploy import _setup_webdav_client for public in (True, False): server_info = WEBDAV_PATHS[False][public] davclient = _setup_webdav_client( SERVER, server_info["root"], username, password ) # go through all possible variants: archs = [ "linux-64", "linux-32", "linux-armv6l", "linux-armv7l", "linux-ppc64le", "linux-aarch64", "osx-64", "osx-32", "osx-arm64", "win-64", "win-32", "noarch", ] path = server_info["conda"] for arch in archs: arch_path = "/".join((path, arch)) if not (davclient.check(arch_path) and davclient.is_dir(arch_path)): # it is normal if the directory does not exist continue server_path = davclient.get_url(arch_path) echo_info("Cleaning beta packages from %s" % server_path) remove_old_beta_packages( client=davclient, path=arch_path, dry_run=dry_run, pyver=True, includes=includes, )
[docs]@contextlib.contextmanager def temporary_cwd(path): """Temporarily changes the working directory to a given path A context manager that temporarily changes the path to the given directory. The working directory changes the working directory back to the starting directory when it exits. Args: path: The directory to temporarily change to """ oldpwd = os.getcwd() os.chdir(path) try: yield finally: os.chdir(oldpwd)