diff --git a/src/git_flow/command/base.py b/src/git_flow/command/base.py index 0fb4f71..056e805 100644 --- a/src/git_flow/command/base.py +++ b/src/git_flow/command/base.py @@ -1,6 +1,5 @@ from argparse import Namespace, ArgumentParser from abc import ABC, abstractmethod -import os from os.path import isfile from git_flow import ( @@ -45,6 +44,11 @@ class Command(ABC): 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) @@ -145,7 +149,7 @@ class Command(ABC): raise GitFlowError("Ejecución cancelada.") def ensure_repository_token(self): - if not os.path.isfile(REPOSITORY_TOKEN_FILENAME): + if not isfile(REPOSITORY_TOKEN_FILENAME): raise GitFlowError( "No existe un token para el repositorio en " + REPOSITORY_TOKEN_FILENAME ) @@ -153,6 +157,23 @@ class Command(ABC): 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(" ") diff --git a/src/git_flow/command/merge.py b/src/git_flow/command/merge.py index b3faafb..cf89a2b 100644 --- a/src/git_flow/command/merge.py +++ b/src/git_flow/command/merge.py @@ -27,7 +27,7 @@ class MergeCommand(Command): self.run_local(base, branch, target) def run_remote(self, remote: str, token: str, base: str, branch: str, target: str): - self.check_pending_changes(True) + self.ensure_clean_worktree(True) Git("switch", target).exec(print="Cambiando a rama destino") Git("pull").exec(print="Obteniendo ultimos cambios del remoto") @@ -44,7 +44,7 @@ class MergeCommand(Command): self.create_pull_request(token, base, branch, target) def run_local(self, base: str, branch: str, target: str): - self.check_pending_changes(False) + self.ensure_clean_worktree(False) self.check_merge_conflicts(target) self.show_commits_to_merge(base) @@ -52,23 +52,6 @@ class MergeCommand(Command): Git("switch", target).exec(print="Cambiando a rama destino") Git("merge", branch, ff=False).exec(print="Mergeando") - def check_pending_changes(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 show_commits_to_merge(self, base): commits = Git("log", base + "..", format="%s").lines() diff --git a/src/git_flow/command/release.py b/src/git_flow/command/release.py new file mode 100644 index 0000000..1eb70a4 --- /dev/null +++ b/src/git_flow/command/release.py @@ -0,0 +1,67 @@ +from argparse import Namespace +from git_flow import COMMIT_TYPES, GitFlowError +from git_flow.command.base import Command +from git_flow.git import Git + + +class ReleaseCommand(Command): + def name(self) -> str: + return "release" + + def description(self) -> str: + return """Realiza el merge de la rama, o crea el PR para hacerlo si tiene un remoto configurado""" + + def run(self, args: Namespace = Namespace()): + self.ensure_initialized() + + branch = self.ensure_right_branch() + envs = self.flowconfig["flow.branches"].split(",") + (env, _) = self.get_branch_env_and_type(branch) + + if env == envs[-1]: + raise GitFlowError( + "No se puede hacer release de una rama en el ultimo entorno." + ) + + has_remote = "flow.remote" in self.flowconfig + + self.ensure_clean_worktree(has_remote) + + next_env = envs[envs.index(env) + 1] + next_branch = ( + branch.replace(f"/{env}/", f"/{next_env}/") + if branch.startswith("release/") + else f"release/{next_env}/{branch}" + ) + + if has_remote: + Git("pull").exec(print="Sincronizando cambios de la rama") + Git("switch", next_env).exec(print="Cambiando al siguiente entorno") + Git("pull").exec(print="Sincronizando cambios del entorno") + Git("switch", "-").exec(print="Volviendo a la rama original") + + base = Git.get_first_fork_point(branch, env) + commits = int(Git("rev-list", base + "..", count=True).firstline()) + + if commits > 1 and self.confirm( + "Desea reemplazar los {commits} commits de la rama por uno solo?" + ): + Git("switch", next_branch, base, create=True).exec( + print="Creando rama release en base" + ) + Git("merge", branch, squash=True).exec( + print="Squasheando commits en uno solo" + ) + + commit_type = self.choice("Tipo de commit", COMMIT_TYPES[:-1]) + commit_message = self.prompt( + "Mensaje (máximo recomendado: 100 caracteres)", persistent=True + ) + message = commit_type + ": " + commit_message + Git("commit", m=message).exec(print="Creando commit único") + else: + Git("switch", next_branch, create=True).exec(print="Creando rama release") + + Git("rebase", base, next_branch, onto=next_env).exec( + print="Moviendo cambios hacia el siguiente ambiente" + ) diff --git a/src/git_flow/main.py b/src/git_flow/main.py index 23d0a38..1099e09 100755 --- a/src/git_flow/main.py +++ b/src/git_flow/main.py @@ -1,27 +1,22 @@ #!/usr/bin/env python3 -import os -from os.path import isfile -import sys -import datetime -import requests import locale from argparse import Namespace, ArgumentParser from git_flow import GitFlowError -from git_flow.changelog import Changelog from git_flow.command.base import Command from git_flow.command.commit import CommitCommand from git_flow.command.init import InitCommand from git_flow.command.merge import MergeCommand from git_flow.command.new import NewCommand +from git_flow.command.release import ReleaseCommand from git_flow.command.tag import TagCommand -from git_flow.git import BUMP_VERSION, FLOWCONFIG_FILE, Git from git_flow.io import * -REPOSITORY_TOKEN_PATH=".repository-token" -LOCALE = 'es_AR.UTF-8' +REPOSITORY_TOKEN_PATH = ".repository-token" +LOCALE = "es_AR.UTF-8" + class GitFlowCommand(Command): commands: tuple[Command, ...] @@ -33,14 +28,17 @@ class GitFlowCommand(Command): return "git-flow" def description(self) -> str: - return """Git flow es un workflow automatizado siguiendo conventional commits, semver, y - deploys aislados a distintos entornos.""" + return """Git flow es una herramienta para automatizar un workflow siguiendo conventional commits, conventional branch y semver.""" def setup_parser(self, parser: ArgumentParser) -> ArgumentParser: - subparsers = parser.add_subparsers(title="comandos", dest="command", required=True) + subparsers = parser.add_subparsers( + title="comandos", dest="command", required=True + ) for command in self.commands: - subparser = subparsers.add_parser(command.name(), description=command.description()) + subparser = subparsers.add_parser( + command.name(), description=command.description() + ) command.setup_parser(subparser) return parser @@ -55,7 +53,7 @@ class GitFlowCommand(Command): for command in self.commands: if command.name() == args.command: - command.flowconfig = Git.get_config(FLOWCONFIG_FILE) if isfile(FLOWCONFIG_FILE) else {} + command.init() return command.run(args) @@ -71,6 +69,7 @@ def main(): CommitCommand(), MergeCommand(), TagCommand(), + ReleaseCommand(), ) try: command.run() @@ -81,440 +80,6 @@ def main(): return - args = sys.argv - - if len(args) < 2: - print_error("missing command") - return - - command = args[1] - - try: - if command == "tag": - tag_command() - elif command == "release": - release_command() - elif command == "help": - help_command() - else: - help_command(command) - except Exception as e: - print_error(f"Ocurrio un error al ejecutar el comando: {e}") - - - -BRANCH_TYPES = [ - "feature", - "refactor", - "bugfix", - "hotfix", - "perf", - "docs", - "typo", - "testing", - "breaking", -] - -BRANCH_TYPES_INCREMENT = { - "feature": "minor", - "refactor": "minor", - "bugfix": "patch", - "hotfix": "patch", - "perf": "minor", - "docs": "patch", - "typo": "patch", - "testing": "patch", - "breaking": "major", -} - -COMMIT_TYPES = [ - "feature", - "bugfix", - "hotfix", - "refactor", - "perf", - "docs", - "typo", - "testing", - "ignore", - "wip" -] - -DEFAULT_REMOTE = "origin" - - -def parse_url(url: str) -> list[str]: - if "@" in url: - # git@github.com:username/repository.git - start = url.find("@") + 1 - end = url.find(":") - service = url[start : end] - start = end + 1 - end = url.find(".git") if ".git" in url else len(url) - repository = url[start : end] - elif "://" in url: - # https://github.com/username/repository.git - start = url.index("://") + 3 - end = url.index("/", start) - service = url[start : end] - start = end + 1 - end = url.find(".git") if ".git" in url else len(url) - repository = url[start : end] - else: - raise RuntimeError("Invalid url: " + url) - - # Esto permite utilizar un Host ficticio si tenemos 2 cuentas de bitbucket (personal y trabajo) - if service.endswith(".bitbucket.org"): - service = "bitbucket.org" - - return [service, repository] - - -def get_branch_env_and_type(branch: str) -> tuple[str, str]: - components = branch.split("/") - target_branches = Git.flow_config("flow.branches").firstline().split(" ") - - if len(components) == 2: - if components[0] not in BRANCH_TYPES: - raise ValueError(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 ValueError( - "Branch release inválido, debe tener el siguiente formato: " - "release///, pero es: " + branch - ) - - return (components[1], components[2]) - else: - raise ValueError("Branch inválido: " + branch) - - -def tag_command(): - ensure_initialized() - ci = len(sys.argv) == 3 # ./git-flow tag - token = sys.argv[2] if ci else None - - if token is None: - with open(REPOSITORY_TOKEN_PATH, "r") as file: - token = file.readline().strip() - - output = Git("log", first_parent=True, merges=True, max_count=1, format="%H,%s").lines() - - if not output: - print_error("No hay merges a taggear") - return - - output = output[0] - pos = output.index(',') - commit = output[:pos] - message = output[pos+1:] - merged_branch = None - - output = Git("describe", commit, tags=True, exact_match=True).firstline(check=False) - - if output: - print_error("El ultimo merge ya tiene tag: " + output) - return - - if message.startswith("Merged in "): - merged_branch = message.removeprefix("Merged in ") - merged_branch = merged_branch[:merged_branch.index(' ')] - elif message.startswith("Merge branch '"): - merged_branch = message.removeprefix("Merge branch '") - merged_branch = merged_branch[:merged_branch.index("'")] - else: - print_error("No se pudo obtener el nombre de la rama mergeada") - return - - remote = Git.flow_config("flow.remote").firstline() - branch = Git.get_current_branch() - last_version = Git.get_current_tag() or "v1.0.0" - from_major = 1 - until_dash = last_version.index("-") if "-" in last_version else None - [major, minor, patch] = map(int, last_version[from_major:until_dash].split(".")) - - (_, merged_branch_type) = get_branch_env_and_type(merged_branch) - increment = BRANCH_TYPES_INCREMENT[merged_branch_type] - - if increment == "major": - major = major + 1 - minor = 0 - patch = 0 - elif increment == "minor": - minor = minor + 1 - patch = 0 - elif increment == "patch": - patch = patch + 1 - - last_branch = Git.flow_config("flow.branches").firstline().split(" ")[-1] - suffix = "-" + branch if branch != last_branch else "" - new_version = f"v{major}.{minor}.{patch}" + suffix - - if ci: - url = Git("remote", "get-url", remote).firstline() - - if url: - [service, repository] = parse_url(url) - create_tag(service, repository, token, new_version, commit) - - merged_commit = Git("show", "HEAD^2", patch=False, format="%H").firstline() - if update_changelog(merged_commit, branch): - Git("add", Changelog.FILENAME).exec() - Git("commit", message=BUMP_VERSION).exec() - Git("push").exec() - else: - Git("tag", new_version, commit).exec() - - -def release_command(): - ensure_initialized() - - branch = ensure_right_branch() - env_branches = Git.flow_config("flow.branches").firstline().split(" ") - (env, _) = get_branch_env_and_type(branch) - - if env == env_branches[-1]: - raise RuntimeError("No se puede hacer release de una rama en el ultimo entorno.") - - base = Git.get_first_fork_point(branch, env) - next_env = env_branches[env_branches.index(env) + 1] - next_branch = branch.replace(f"/{env}/", f"/{next_env}/") if branch.startswith("release/") else f"release/{next_env}/{branch}" - - Git("switch", next_branch, create=True).exec() - commits = int(Git("rev-list", base + ".." + next_branch, count=True).firstline()) - - if commits > 1 and confirm(f"Desea reemplazar los {commits} commits de la rama por uno solo?"): - Git("reset", base, soft=True).exec() - commit_type = choose("Tipo de commit", COMMIT_TYPES[:-1]) # Solo permitir commits NO wip - Git("commit", m=commit_type + ": " + get_commit_message()).exec() - - Git("rebase", base, next_branch, onto=next_env).exec() - - -def help_command(name: str|None = None): - if name is not None: - print_error(f"Comando inválido: {name}") - - script = sys.argv[0] - - print(f"""USO: - {script} - -COMANDOS: - init inicializa el repositorio por única vez para utilizar git-flow - new crea una nueva rama para realizar cambios - commit crea un commit siguiendo el formato de Conventional Commit - merge actualiza el changelog y crea un nuevo PR para mergear la rama - tag crea un nuevo tag para el ultimo merge - release crea una nueva rama release para pasar la rama actual al siguiente entorno - help muestra este texto de ayuda""", file=sys.stderr) - - -def ensure_initialized(): - initialized = Git.flow_config("flow.initialized").firstline() - - if initialized == "": - raise RuntimeError("El repositorio no fue inicializado, antes de continuar ejecute `git flow init`.") - elif initialized != "true": - raise RuntimeError(f"El valor de `flow.initialized` ({initialized}) es inválido.") - - -def get_commit_message(): - message = None - - while message is None: - message = input("Mensaje (recomendado: 100 caracteres): ").strip() - - if not message: - print_error("El mensaje no puede ser vacío.") - message = None - - return message - - -def ensure_right_branch(): - branch = Git.get_current_branch() - - if branch != "HEAD": - if confirm(f"Se encuentra sobre la rama '{branch}'. ¿Es correcto?"): - return branch - - raise RuntimeError("Abortando operación") - - -def get_base_branch(branch: str): - if branch.startswith("release/"): - start = len("release/") - end = branch.index("/", start) - return branch[start:end] - else: - return Git.flow_config("flow.branches").firstline().split(" ")[0] - - -def get_unique_branch_name(branch_type): - print_info(f"Ingrese las palabras clave que describan el cambio de tipo '{branch_type}'.") - branch = None - branches = Git.get_branches() - - while branch is None: - keywords = input("Palabras clave (por ej. 'edit worker form'): ") - - if len(keywords) == 0: - print_error("Debe ingresar al menos una palabra clave.") - else: - keywords = list(filter(lambda k: len(k) > 0, keywords.split(" "))) - branch = branch_type + "/" + "-".join(keywords) - - if branch in branches: - print_error( - f"La rama '{branch}' ya existe, cambie las palabras claves para generar otra" - ) - branch = None - - return branch - - -def update_changelog(branch: str, target: str): - last_commit = Git("show", "HEAD", no_patch=True, format="%s").firstline() - - if last_commit == BUMP_VERSION: - return False - else: - content = generate_changelog_entry(branch, target) - - if os.path.isfile(Changelog.FILENAME): - with open(Changelog.FILENAME, "r") as changelog: - content = content + changelog.read() - - with open(Changelog.FILENAME, "w") as changelog: - changelog.write(content) - - - return True - - -def get_changelog_header_lines(branch: str): - name = Git("config", "user.name").firstline() - email = Git("config", "user.email").firstline() - - now = datetime.datetime.now() - weekday = now.strftime("%A").capitalize() - month = now.strftime("%B").capitalize() - timestamp = now.strftime(f"{weekday}, %e de {month} de %Y, %H:%M") - - return [ - f"## {timestamp}", - "", - f"- Autor: [{name}](mailto:{email})", - f"- Rama: `{branch}`", - "", - "### Commits", - "" - ] - -def get_changelog_content_lines(branch: str, target: str): - lines = [] - base = Git.get_first_fork_point(branch, target) - commits = Git("log", base + ".." + branch, first_parent=True, format="%h", merges=False).lines() - - for commit in commits: - message = Git("show", commit, patch=False, format="%s").firstline() - email = Git("show", commit, patch=False, format="%ae").firstline() - username = email[:email.index('@')] - - if message == BUMP_VERSION: - continue - - index = message.index(":") - commit_type = message[:index] - commit_message = message[index+1:] - - lines.append(f"- **{commit_type}**: {commit_message} [[{username}](mailto:{email})] ({commit})") - - return lines - -def get_changelog_footer_lines(): - return [ - "", - "---", - "", - "" - ] - -def generate_changelog_entry(branch: str, target: str): - entry_lines = get_changelog_header_lines(branch) - entry_lines += get_changelog_content_lines(branch, target) - entry_lines += get_changelog_footer_lines() - - return "\n".join(entry_lines) - - -def create_pull_request(service: str, repository: str, source: str, destination: str): - endpoint = get_api_endpoint(service, repository, "/pullrequests") - title = source.replace("/", ": ").replace("-", " ") - description = "\n".join(get_changelog_content_lines(source, destination)) - headers = get_headers() - - json = { - "title": title, - "description": description, - "source": {"branch": {"name": source}}, - "destination": {"branch": {"name": destination}}, - "close_source_branch": True - } - - request = requests.post(endpoint, headers=headers, json=json) - - if request.ok: - json = request.json() - print_success("PR creado exitosamente: " + json['links']['html']['href']) - else: - print_error("Ocurrió un error al crear el PR: " + request.text) - - -def create_tag(service: str, repository: str, token: str, tag: str, commit: str): - endpoint = get_api_endpoint(service, repository, "/refs/tags") - headers = get_headers(token) - json = { - "name": tag, - "target": { - "hash": commit - } - } - - response = requests.post(endpoint, headers=headers, json=json) - - if response.ok: - print("Tag creado exitosamente.") - else: - print_error("Ocurrió un error al crear el tag: " + response.text) - - -def get_headers(token: str | None = None): - if not token: - with open(REPOSITORY_TOKEN_PATH) as file: - token = file.readline().strip() - - return { - "Accept": "application/json", - "Content-Type": "application/json", - "Authorization": "Bearer " + token, - } - - -def get_api_endpoint(service: str, repository: str, path: str): - if service == "bitbucket.org": - return "https://api.bitbucket.org/2.0/repositories/" + repository + path - - return "" if __name__ == "__main__": main()