56 lines
No EOL
2 KiB
Python
56 lines
No EOL
2 KiB
Python
#+------------------------------------------------------------------+
|
|
#| Imports |
|
|
#+------------------------------------------------------------------+
|
|
from Src.Ops.Def import *
|
|
|
|
#+------------------------------------------------------------------+
|
|
#| Funciones |
|
|
#+------------------------------------------------------------------+
|
|
# Hooks dict ya te pide que el dict que se le pasases sea como tal ese..
|
|
# data["hooks"] y se siga la esuturca correcta
|
|
def execute_hooks(hooks_dict: dict, hook_type: str) -> bool:
|
|
"""Ejecuta los hooks especificados con timeout"""
|
|
|
|
# En caso no exista dicha propiedad
|
|
if not hooks_dict or hook_type not in hooks_dict:
|
|
return True
|
|
|
|
# Comandos
|
|
commands : dict = hooks_dict[hook_type]
|
|
if not commands:
|
|
return True
|
|
|
|
click.echo(f"Ejecutando {hook_type}...")
|
|
|
|
all_success : bool = True
|
|
for item in commands:
|
|
cmd : str = item["command"]
|
|
permitir_fallo : bool = item.get("permitir_fallo", False)
|
|
timeout_sec : int = item.get("timeout_ms", 60000) / 1000 # Convertir a segundos
|
|
|
|
# click.echo(f"{cmd} (timeout: {timeout_sec}s)")
|
|
try:
|
|
subprocess.run(
|
|
cmd,
|
|
shell=True,
|
|
check=True,
|
|
timeout=timeout_sec
|
|
)
|
|
# click.echo(f"Ok")
|
|
|
|
except subprocess.TimeoutExpired:
|
|
msg : str= f"Timeout después de {timeout_sec}s"
|
|
if permitir_fallo:
|
|
click.echo(f"{msg} (permitido)")
|
|
else:
|
|
click.echo(f"{msg} (error)", err=True)
|
|
all_success = False
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
if permitir_fallo:
|
|
click.echo(f"Error: {e} (permitido)")
|
|
else:
|
|
click.echo(f"Error: {e} (FATAL)", err=True)
|
|
all_success = False
|
|
|
|
return all_success |