175 lines
4.6 KiB
Python
175 lines
4.6 KiB
Python
"""Utility functions.
|
|
|
|
While mostly a vague concept and sometimes a bad idea, the utilities module is
|
|
somewhere where reusable system related, data processing and other general
|
|
functions can live. This codebase seems to have a few of those, so it seems
|
|
like a necessary module. Let us be critical of our requirements when adding
|
|
things here.
|
|
"""
|
|
|
|
from os import chdir
|
|
from os.path import exists
|
|
from pathlib import Path
|
|
from subprocess import call, check_output
|
|
from sys import exit as sys_exit
|
|
|
|
from pexpect import spawn
|
|
from psutil import process_iter
|
|
from PyInquirer import prompt
|
|
|
|
from autonomic.config import CONFIG_DIR, CONFIG_YAML, INFRA_DIR
|
|
from autonomic.logger import log
|
|
from autonomic.yaml import yaml
|
|
|
|
|
|
def run(
|
|
cmd, cwd=None, interactive=False, pexpect=False, pexpected=None, **kwargs
|
|
):
|
|
"""Run a system command.
|
|
|
|
Please note, all **kwargs will be passed into the check_output command so
|
|
that system call customisation can happen. Please name your keyword
|
|
arguments (like `cwd`) if you intend that they are used for other logic.
|
|
"""
|
|
try:
|
|
if cwd:
|
|
chdir(cwd)
|
|
log.info("Changed directory to {}".format(cwd))
|
|
|
|
log.info("Running {}".format(" ".join(cmd)))
|
|
|
|
if interactive:
|
|
return call(cmd, **kwargs)
|
|
|
|
if pexpect:
|
|
child = spawn(" ".join(cmd))
|
|
for expected, response in pexpected.items():
|
|
try:
|
|
child.expect(expected, timeout=5)
|
|
child.sendline(response)
|
|
except Exception as exception:
|
|
msg = "Pexpecting failed, saw {}".format(str(exception))
|
|
exit(msg)
|
|
return child.read().decode("utf-8")
|
|
|
|
output = check_output(cmd, **kwargs)
|
|
return output.decode("utf-8")
|
|
except Exception as exception:
|
|
msg = "{} failed! Saw {}".format(" ".join(cmd), str(exception))
|
|
exit(msg)
|
|
|
|
|
|
def exit(msg, code=1):
|
|
"""Exit and log appropriate level."""
|
|
if code != 0:
|
|
log.critical(msg)
|
|
else:
|
|
log.info(msg)
|
|
|
|
sys_exit(code)
|
|
|
|
|
|
def question_ask(name, message, choices):
|
|
"""Ask a question."""
|
|
question = [
|
|
{
|
|
"type": "list",
|
|
"name": name,
|
|
"message": message,
|
|
"choices": choices,
|
|
"filter": lambda answer: answer.lower(),
|
|
}
|
|
]
|
|
|
|
return answer(question, name)
|
|
|
|
|
|
def pass_ask(message):
|
|
"""Ask for a password."""
|
|
question = [{"type": "password", "message": message, "name": "password"}]
|
|
return answer(question, "password")
|
|
|
|
|
|
def input_ask(message):
|
|
"""Ask for input."""
|
|
question = [{"type": "input", "message": message, "name": "input"}]
|
|
return answer(question, "input")
|
|
|
|
|
|
def answer(question, key):
|
|
"""Retrieve an answer from a prompt"""
|
|
result = prompt(question)
|
|
|
|
if not result:
|
|
exit("Prompt cancelled")
|
|
|
|
try:
|
|
return result[key]
|
|
except KeyError:
|
|
exit("{} key is missing".format(key))
|
|
|
|
|
|
def yaml_load(target, text=False):
|
|
"""Load a YAML file or text."""
|
|
if text is True:
|
|
try:
|
|
return yaml.load(target)
|
|
except Exception as exception:
|
|
log.error(str(exception))
|
|
|
|
try:
|
|
with open(target, "r") as handle:
|
|
return yaml.load(handle.read())
|
|
except Exception as exception:
|
|
log.error(str(exception))
|
|
|
|
|
|
def yaml_dump(fpath, data):
|
|
"""Dump a YAML file."""
|
|
try:
|
|
with open(fpath, "w") as handle:
|
|
yaml.dump(data, handle)
|
|
except Exception as exception:
|
|
log.error(str(exception))
|
|
|
|
|
|
def is_proc(name):
|
|
"""Determine if a process is running or not."""
|
|
for process in process_iter():
|
|
try:
|
|
if name.lower() in process.name().lower():
|
|
return True
|
|
except Exception:
|
|
pass
|
|
return False
|
|
|
|
|
|
def git_status(fpath):
|
|
"""Check if Git reports changes to be committed."""
|
|
cmd = ["git", "status", "--porcelain"]
|
|
output = run(cmd, cwd=fpath)
|
|
|
|
if output:
|
|
msg = "warning: git reports uncommitted changes in {}".format(fpath)
|
|
log.warning(msg)
|
|
log.warning(output)
|
|
else:
|
|
msg = "No unstaged changes found in {}".format(INFRA_DIR)
|
|
log.info(msg)
|
|
|
|
|
|
def ensure_config_dir():
|
|
"""Ensure configuration directory is in place."""
|
|
if not exists(CONFIG_DIR):
|
|
msg = "{} is missing, did you run 'autonomic init'?".format(CONFIG_YAML)
|
|
exit(msg)
|
|
|
|
|
|
def ensure_deploy_d_dir():
|
|
"""Ensure deploy.d directory is in place."""
|
|
deploy_d_dir = (Path(".") / "deploy.d").absolute()
|
|
|
|
if not exists(deploy_d_dir):
|
|
msg = "No deploy.d folder found, are you in the right place?"
|
|
exit(msg)
|