dc-ops/lib/helper.py

# pylint: disable=missing-module-docstring

import logging as log
import subprocess
import sys
from pathlib import Path

import git


def do_selfupdate():
    """perform git pull on script's base directory"""
    repo_path = Path(sys.argv[0]).parent.absolute()
    repo = git.Repo(repo_path)
    try:
        pull_res = repo.remotes.origin.pull()[0]
    except git.exc.GitCommandError as e:
        log.error(str(e))
        return
    if pull_res.old_commit:
        log.info("SelfUpdate: successfully updated myself - exiting")
        sys.exit(0)
    log.info("SelfUpdate: no updates found for myself")


def run_subprocess(command: str, workdir: str) -> bool:
    """execute subprocess and print output

    Args:
        cmd (str): command
        cwd (str): working directory

    Returns:
        bool: False if subprocess fails
    """
    try:
        log.debug(
            subprocess.run(command.split(" "), cwd=workdir, text=True, check=False)
        )
    except subprocess.CalledProcessError:
        return False

    return True


def update_git_repo(repo_path: str, ignore_git_status: bool) -> bool:
    """update repo and check for new commits

    Args:
        repo_path (str): path to git repo
        ignore_git_status (bool): continue even if no changes

    Returns:
        bool: False if any step fails
    """

    # create repo instance if it exists
    try:
        repo = git.Repo(repo_path)
    except git.exc.NoSuchPathError:
        log.error("-> directory not found")
        return False

    if not ignore_git_status:
        # try to fetch latest changes
        try:
            fetch_res = repo.remotes.origin.fetch()[0]
        except git.exc.GitCommandError as e:
            log.error("-> %s ", str(e))
            return False

        # check for new commits
        if not fetch_res.old_commit:
            log.info("-> no changes - skipping")
            return False

    # pull remote changes to local branch
    try:
        log.info("-> %s", repo.git.pull())
    except git.exc.GitCommandError as e:
        log.error("-> %s", str(e))
        return False

    return True