git-flow/src/git_flow/command/base.py

from argparse import Namespace, ArgumentParser
from abc import ABC, abstractmethod
from os.path import isfile

from git_flow import (
    BRANCH_TYPES,
    REPOSITORY_TOKEN_FILENAME,
    GitFlowError,
)
from git_flow.git import FLOWCONFIG_FILE, Git
from git_flow.remote.base import RemoteAPI
from git_flow.remote.bitbucket import BitbucketRemoteAPI
from git_flow.remote.github import GithubRemoteAPI

COLOR_BLACK = "\033[30m"
COLOR_RED = "\033[31m"
COLOR_GREEN = "\033[32m"
COLOR_YELLOW = "\033[33m"
COLOR_BLUE = "\033[34m"
COLOR_5 = "\033[35m"
COLOR_6 = "\033[36m"
COLOR_7 = "\033[37m"
COLOR_BLACK_BOLD = "\033[1;30m"
COLOR_RESET = "\033[0m"

TYPE_SUCCESS = 0
TYPE_WARNING = 1
TYPE_ERROR = 2
TYPE_INFO = 3


class Command(ABC):
    flowconfig: dict[str, str]

    @abstractmethod
    def name(self) -> str:
        pass

    @abstractmethod
    def description(self) -> str:
        pass

    @abstractmethod
    def run(self, args: Namespace = Namespace()):
        pass

    def init(self):
        self.flowconfig = (
            Git.get_config(FLOWCONFIG_FILE) if isfile(FLOWCONFIG_FILE) else {}
        )

    def success(self, msg: str):
        self._print(TYPE_SUCCESS, msg)

    def warning(self, msg: str):
        self._print(TYPE_WARNING, msg)

    def error(self, msg: str):
        self._print(TYPE_ERROR, msg)

    def info(self, msg: str):
        self._print(TYPE_INFO, msg)

    def prompt(
        self,
        prompt: str,
        default: str | None = None,
        persistent: bool = False,
        strip: bool = True,
    ) -> str:
        onetime = not persistent

        while onetime or persistent:
            if default:
                prompt += f" [{default}]"

            prompt += ": "
            result = input(prompt)
            result = result.strip() if strip else result

            if result:
                return result
            elif default is not None:
                return default
            elif persistent:
                self.error("Debe ingresar un valor no vacío.")
            else:
                return result

    def confirm(self, question: str, default: bool = True):
        suffix = " [Y/n]: " if default else " [y/N]: "
        answer = input(question + suffix)

        return default if len(answer) == 0 else answer.startswith("y")

    def choice(self, prompt: str, options: list[str]):
        print(prompt)

        for i, option in enumerate(options):
            print(f"\t{i+1}. {option}")

        selection = None

        while selection is None:
            answer = input(f"Seleccione una opción [1-{len(options)}] o escribala: ")

            if answer.isdigit():
                answer = int(answer)

                if 1 <= answer and answer <= len(options):
                    selection = options[answer - 1]

                    if not self.confirm(
                        f"Seleccionó la opción {answer} ({selection}), ¿es correcto?"
                    ):
                        selection = None
                else:
                    self.error(f"La opción {answer} está fuera del rango permitido.")
            elif answer in options:
                selection = answer
            else:
                self.error(f"La opción '{answer}' es inválida.")

        return selection

    def setup_parser(self, parser: ArgumentParser) -> ArgumentParser:
        return parser

    def ensure_initialized(self):
        initialized = self.flowconfig["flow.initialized"] if self.flowconfig else None

        if not initialized:
            raise GitFlowError(
                "El repositorio no fue inicializado, debe ejecutar el comando `init`."
            )
        elif initialized != "true":
            raise GitFlowError(
                f"El valor de `flow.initialized` ({initialized}) es inválido."
            )

    def ensure_right_branch(self):
        branch = Git.get_current_branch()

        if branch == "HEAD":
            raise GitFlowError("No se encuentra parado sobre una rama.")
        elif self.confirm(f"Se encuentra sobre la rama '{branch}'. ¿Es correcto?"):
            return branch
        else:
            raise GitFlowError("Ejecución cancelada.")

    def ensure_repository_token(self):
        if not isfile(REPOSITORY_TOKEN_FILENAME):
            raise GitFlowError(
                "No existe un token para el repositorio en " + REPOSITORY_TOKEN_FILENAME
            )

        with open(REPOSITORY_TOKEN_FILENAME) as f:
            return f.readline().strip()

    def ensure_clean_worktree(self, has_remote: bool):
        status = Git.status()

        if not status:
            return

        not_empty = any(map(lambda s: s.index != " " or s.worktree != " ", status))

        if not_empty:
            if not has_remote:
                self.warning("Existen cambios en tu entorno de trabajo sin commitear.")
            else:
                raise GitFlowError("No se puede continuar con cambios pendientes.")

            if not self.confirm("¿Desea continuar?", False):
                raise GitFlowError("Ejecución abortada")

    def get_branch_env_and_type(self, branch: str) -> tuple[str, str]:
        components = branch.split("/")
        target_branches = self.flowconfig["flow.branches"].split(" ")

        if len(components) == 2:
            if components[0] not in BRANCH_TYPES:
                raise GitFlowError(
                    f"Branch inválida, el tipo '{components[0]}' no es válido."
                )

            return (target_branches[0], components[0])
        elif len(components) == 4:
            target_branches = target_branches[1:]

            if (
                components[0] != "release"
                or components[1] not in target_branches
                or components[2] not in BRANCH_TYPES
            ):
                raise GitFlowError(
                    "Branch release inválido, debe tener el siguiente formato: "
                    "release/<env>/<type>/<name>, pero es: " + branch
                )

            return (components[1], components[2])
        else:
            raise GitFlowError("Branch inválido: " + branch)

    def get_remote_api(self, token: str) -> RemoteAPI:
        if "flow.remote" not in self.flowconfig:
            raise GitFlowError("El repositorio no tiene configurado un remoto.")

        [host, repository] = RemoteAPI.parse(self.flowconfig["flow.remote"])

        if host == "bitbucket.org":
            return BitbucketRemoteAPI(repository, token)
        elif host == "github.com":
            return GithubRemoteAPI(repository, token)
        else:
            raise GitFlowError("El host del repositorio remoto es inválido")

    def _print(self, type: int, msg: str):
        print(self._get_color(type) + self._get_tag(type) + " " + msg + COLOR_RESET)

    def _get_color(self, type: int) -> str:
        if type == TYPE_SUCCESS:
            return COLOR_GREEN
        elif type == TYPE_WARNING:
            return COLOR_YELLOW
        elif type == TYPE_ERROR:
            return COLOR_RED
        elif type == TYPE_INFO:
            return COLOR_BLUE
        return ""

    def _get_tag(self, type: int) -> str:
        if type == TYPE_SUCCESS:
            return "[success]"
        elif type == TYPE_WARNING:
            return "[warning]"
        elif type == TYPE_ERROR:
            return "[error]"
        elif type == TYPE_INFO:
            return "[info]"

        return ""