158 lines
5.1 KiB
Python
158 lines
5.1 KiB
Python
import typer
|
|
|
|
import git_flow.command.base as base
|
|
from git_flow import GitFlowError
|
|
from git_flow.changelog import Changelog
|
|
from git_flow.git import Git
|
|
|
|
app = typer.Typer()
|
|
|
|
|
|
@app.command()
|
|
def merge():
|
|
"""Realiza el merge de la rama, o crea el PR para hacerlo si tiene un remoto configurado"""
|
|
base.ensure_initialized()
|
|
|
|
branch = base.ensure_right_branch()
|
|
|
|
if branch in base.environments:
|
|
if branch == base.environments[-1]:
|
|
raise GitFlowError("No se puede hacer un merge del último entorno.")
|
|
|
|
base.io_info(
|
|
"""Está por hacer un merge de 2 entornos que puede implicar muchos cambios en el proyecto.""",
|
|
"Merge de entornos",
|
|
)
|
|
|
|
if not base.io_confirm("¿Desea continuar?"):
|
|
raise GitFlowError("Ejecución abortada")
|
|
|
|
target = base.environments[base.environments.index(branch) + 1]
|
|
else:
|
|
target, _ = base.get_branch_env_and_type(branch)
|
|
|
|
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):
|
|
base.ensure_clean_worktree(True)
|
|
|
|
Git("switch", target).exec(print="Cambiando a rama destino")
|
|
Git("pull").exec(print="Obteniendo ultimos cambios del remoto")
|
|
Git("switch", "-").exec(print="Volviendo a rama a mergear")
|
|
|
|
check_merge_conflicts(target)
|
|
show_commits_to_merge(target)
|
|
|
|
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"
|
|
)
|
|
|
|
create_pull_request(token, branch, target)
|
|
|
|
|
|
def run_local(branch: str, target: str):
|
|
base.ensure_clean_worktree(False)
|
|
check_merge_conflicts(target)
|
|
show_commits_to_merge(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")
|
|
|
|
|
|
def show_commits_to_merge(target):
|
|
commits = Git("log", target + "..", format="%s").lines()
|
|
|
|
if not commits:
|
|
raise GitFlowError("No hay cambios a mergear.")
|
|
|
|
print("Cambios a mergear:")
|
|
|
|
for commit in commits:
|
|
print("- " + commit)
|
|
|
|
|
|
def check_merge_conflicts(target: str):
|
|
merge_conflicts = Git("merge", target, ff=False, commit=False).code(
|
|
print="Realizando merge de prueba para verificar conflictos"
|
|
)
|
|
|
|
Git("merge", abort=True).exec(print="Abortando merge de prueba", check=False)
|
|
|
|
if merge_conflicts:
|
|
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,\n"
|
|
"y ejecutar nuevamente este comando",
|
|
title="Conflictos de merge",
|
|
)
|
|
raise GitFlowError("Ejecución abortada.")
|
|
else:
|
|
base.io_success("No se detectaron merge conflicts.")
|
|
|
|
|
|
def create_pull_request(token: str, branch: str, target: str):
|
|
changelog = Changelog()
|
|
commits = Git("log", target + "..", format="%s").lines()
|
|
title = commits[0] if len(commits) == 1 else _get_pr_title_from_branch(branch)
|
|
|
|
base.panel("Título del PR", title)
|
|
|
|
if base.io_confirm("¿Desea cambiar el título del PR?", False):
|
|
title = base.io_prompt("Título del PR", persistent=True)
|
|
|
|
message = base.get_remote_api(token).create_pull_request(
|
|
branch,
|
|
target,
|
|
title,
|
|
changelog.generate_content(target, branch),
|
|
branch not in base.environments,
|
|
)
|
|
|
|
base.io_success(message, "Rama creada")
|
|
|
|
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 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>
|
|
branch_type = components[-2]
|
|
branch_desc = components[-1]
|
|
|
|
# detect if branch description has ticket as prefix: <desc> = ABC-123-my-branch-name
|
|
branch_ticket = None
|
|
first_dash = branch_desc.find("-")
|
|
second_dash = branch_desc.find("-", first_dash + 1)
|
|
|
|
if first_dash >= 0 and second_dash >= 0:
|
|
maybe_ticket = branch_desc[:second_dash]
|
|
if base.is_valid_ticket(maybe_ticket):
|
|
branch_ticket = maybe_ticket
|
|
branch_desc = branch_desc[second_dash + 1 :]
|
|
|
|
return " ".join(
|
|
filter(
|
|
None,
|
|
[
|
|
f"[Pasaje a {components[1]}]" if len(components) == 4 else None,
|
|
branch_type + ":",
|
|
branch_ticket,
|
|
branch_desc.replace("-", " "),
|
|
],
|
|
)
|
|
)
|