jt/git-flow
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: migrar comando merge para que utilice questionary

Browse Source
This commit is contained in:
jt
2026-07-14 00:28:20 -03:00
parent 320d40a211
commit 0782cfb041
2 changed files with 34 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/git_flow/command/merge.py
+31 -33
View File
@@ -1,7 +1,7 @@
import typer
from git_flow import GitFlowError
from git_flow.changelog import Changelog
from git_flow.command.base import *
import git_flow.command.base as base
from git_flow.git import Git
app = typer.Typer()
@@ -10,36 +10,36 @@ app = typer.Typer()
@app.command()
def merge():
"""Realiza el merge de la rama, o crea el PR para hacerlo si tiene un remoto configurado"""
ensure_initialized()
base.ensure_initialized()
branch = ensure_right_branch()
branch = base.ensure_right_branch()
if branch in environments:
if branch == environments[-1]:
if branch in base.environments:
if branch == base.environments[-1]:
raise GitFlowError("No se puede hacer un merge del último entorno.")
panel(
"Merge de entornos",
base.io_info(
"""Está por hacer un merge de 2 entornos que puede implicar muchos cambios en el proyecto.""",
"Merge de entornos",
)
if not confirm("¿Desea continuar?"):
if not base.io_confirm("¿Desea continuar?"):
raise GitFlowError("Ejecución abortada")
target = environments[environments.index(branch) + 1]
target = base.environments[base.environments.index(branch) + 1]
else:
target, _ = get_branch_env_and_type(branch)
target, _ = base.get_branch_env_and_type(branch)
if "flow.remote" in flowconfig:
remote = flowconfig["flow.remote"]
token = ensure_repository_token()
if "flow.remote" in base.flowconfig:
remote = base.flowconfig["flow.remote"]
token = base.ensure_repository_token()
run_remote(remote, token, branch, target)
else:
run_local(branch, target)
def run_remote(remote: str, token: str, branch: str, target: str):
ensure_clean_worktree(True)
base.ensure_clean_worktree(True)
Git("switch", target).exec(print="Cambiando a rama destino")
Git("pull").exec(print="Obteniendo ultimos cambios del remoto")
@@ -48,8 +48,8 @@ def run_remote(remote: str, token: str, branch: str, target: str):
check_merge_conflicts(target)
show_commits_to_merge(target)
if confirm(f"¿Crear PR de '{branch}' a '{target}'?"):
if branch not in environments:
if base.io_confirm(f"¿Crear PR de '{branch}' a '{target}'?"):
if branch not in base.environments:
Git("push", remote, branch, set_upstream=True).exec(
print="Subiendo rama al remoto para crear PR"
)
@@ -58,11 +58,11 @@ def run_remote(remote: str, token: str, branch: str, target: str):
def run_local(branch: str, target: str):
ensure_clean_worktree(False)
base.ensure_clean_worktree(False)
check_merge_conflicts(target)
show_commits_to_merge(target)
if confirm(f"¿Mergear rama '{branch}' a '{target}'?"):
if base.io_confirm(f"¿Mergear rama '{branch}' a '{target}'?"):
Git("switch", target).exec(print="Cambiando a rama destino")
Git("merge", branch, ff=False).exec(print="Mergeando")
@@ -87,13 +87,11 @@ def check_merge_conflicts(target: str):
Git("merge", abort=True).exec(print="Abortando merge de prueba", check=False)
if merge_conflicts:
error(f"La rama actual tiene conflictos con {target}")
info(
f"""Se recomienda mergear {target} a la rama actual, resolver los conflictos localmente, y ejecutar nuevamente este comando"""
)
base.io_error(f"La rama actual tiene conflictos con {target}.\n"
f"Se recomienda mergear {target} a la rama actual, resolver los conflictos localmente, y ejecutar nuevamente este comando", "Conflictos de merge")
raise GitFlowError("Ejecución abortada.")
else:
success("No se detectaron merge conflicts.")
base.io_success("No se detectaron merge conflicts.")
def create_pull_request(token: str, branch: str, target: str):
@@ -101,28 +99,28 @@ def create_pull_request(token: str, branch: str, target: str):
commits = Git("log", target + "..", format="%s").lines()
title = commits[0] if len(commits) == 1 else _get_pr_title_from_branch(branch)
panel("Título del PR", title)
base.panel("Título del PR", title)
if confirm("¿Desea cambiar el título del PR?", False):
title = prompt("Título del PR")
if base.io_confirm("¿Desea cambiar el título del PR?", False):
title = base.io_prompt("Título del PR", persistent=True)
message = get_remote_api(token).create_pull_request(
message = base.get_remote_api(token).create_pull_request(
branch,
target,
title,
changelog.generate_content(target, branch),
branch not in environments
branch not in base.environments
)
success(message)
base.io_success(message, "Rama creada")
if confirm("¿Desea cambiar a la rama objetivo y bajar los cambios?"):
if base.io_confirm("¿Desea cambiar a la rama objetivo y bajar los cambios?"):
Git("switch", target).exec(print="Cambiando a rama objetivo")
Git("pull").exec(print="Obteniendo cambios")
def _get_pr_title_from_branch(branch: str) -> str:
if branch in environments:
target = environments[environments.index(branch) + 1]
if branch in base.environments:
target = base.environments[base.environments.index(branch) + 1]
return f"Sincronización de entorno {branch} a {target}"
components = branch.split("/") # <type>/<desc> or release/<env>/<type>/<desc>
@@ -136,7 +134,7 @@ def _get_pr_title_from_branch(branch: str) -> str:
if first_dash >= 0 and second_dash >= 0:
maybe_ticket = branch_desc[:second_dash]
if is_valid_ticket(maybe_ticket):
if base.is_valid_ticket(maybe_ticket):
branch_ticket = maybe_ticket
branch_desc = branch_desc[second_dash+1:]