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

from argparse import Namespace, ArgumentParser
from abc import ABC, abstractmethod
from os.path import isfile
import rich
from rich.prompt import Prompt, Confirm

from git_flow import (
    BRANCH_TYPES,
    COLOR_BLUE,
    COLOR_GREEN,
    COLOR_RED,
    COLOR_RESET,
    COLOR_YELLOW,
    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

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):
        rich.print(f":white_check_mark: [green]{msg}[/green]")

    def warning(self, msg: str):
        rich.print(f":warning: [yellow]{msg}[/yellow]")

    def error(self, msg: str):
        rich.print(f":x: [red]{msg}[/red]")

    def info(self, msg: str):
        rich.print(f":information: [blue]{msg}[/blue]")

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

        while onetime or persistent:
            result = Prompt.ask(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):
        return Confirm.ask(question, default=default)

    def choice(self, prompt: str, options: list[str]):
        return Prompt.ask(prompt, choices = options)

    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 ""