diff --git a/tests/test_email.py b/tests/test_email.py
index 157bd6a5fa31bdc01f7979ca87eb3dd468d63b28..381bd2e5b436f3685ca099761e71bd468cc74d83 100755
--- a/tests/test_email.py
+++ b/tests/test_email.py
@@ -1,151 +1,191 @@
#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-'''
+
+"""
Criticality: High
Check that emails can be sent.
-'''
+"""
+
import os
import subprocess
import sys
import random
import time
-import imp
+# install spf lib if not present
try:
import spf
except ImportError:
- subprocess.check_call(['apt-get', '-qq', '-y', 'install', 'python3-spf'])
+ subprocess.check_call(["apt-get", "-qq", "-y", "install", "python3-spf"])
import spf
-if subprocess.call(['which', 'netstat']) != 0:
- subprocess.check_call(['apt-get', '-qq', '-y', 'install', 'net-tools'])
-YELLOW = '\033[93m'
-GREEN = '\033[92m'
-RED = '\033[91m'
-DEF = '\033[0m'
+# install netstat if not present
+if subprocess.call(["which", "netstat"]) != 0:
+ subprocess.check_call(["apt-get", "-qq", "-y", "install", "net-tools"])
-os.chdir(os.path.dirname(__file__))
-if not os.path.isfile('../utils.py'):
- print('The envsetup configuration was not found.')
- sys.exit(1)
-else:
- es_utils = imp.load_source('es_utils', '../utils.py')
- conf = es_utils.load_conf()
+sys.path.append("..")
+# pylint: disable=wrong-import-position
+from envsetup import utils as u # noqa: E402
-def print_color(txt, col):
- print('%s%s%s' % (col, txt, DEF))
+def check_listen() -> int:
+ """Check that Postfix is listening on 127.0.0.1:25.
-def print_yellow(txt):
- print_color(txt, YELLOW)
+ :return: Exit return code
+ :rtype: int
+ """
+ print("Checking that postfix is listening locally:")
-def print_red(txt):
- print_color(txt, RED)
+ # get listening state from netstat
+ status, out = subprocess.getstatusoutput("netstat -pant | grep master | grep ':25'")
+ if status != 0 or "127.0.0.1:25" not in out:
+ u.error("Postfix is not listening on 127.0.0.1:25")
+ return 1
-def print_green(txt):
- print_color(txt, GREEN)
-
+ u.success("Postfix is listening on 127.0.0.1:25")
-def check_listening_port():
- # check that postfix listens the port 25 correctly
- status, out = subprocess.getstatusoutput('netstat -pant | grep master | grep ":25"')
- if status != 0:
- print_red('The port 25 is not listened by postfix "master" process.')
- return 1
- print_green('Postfix "master" process is listening port 25 correctly.')
- if '127.0.0.1:25' not in out:
- print_red('Postfix "master" process is not listening address 127.0.0.1, please check postfix configuration.')
- print('Postfix should listen address 127.0.0.1 to be sure that this server cannot be used as an SMTP relay by external services.')
- return 1
- print_green('Postfix "master" process is listening address 127.0.0.1 correctly.')
return 0
-def check_relay():
- print('Checking if SMTP relay conforms to conf.')
- status, out = subprocess.getstatusoutput('grep relayhost /etc/postfix/main.cf')
+def check_relay(conf: dict) -> int:
+ """Check that Postfix is not an open relay.
+
+ :param conf: EnvSetup configuration settings
+ :type conf: dict
+ :return: Exit return code
+ :rtype: int
+ """
+
+ print("Checking that SMTP relay conforms to conf:")
+
+ # get relayhost value from Postfix config
+ status, out = subprocess.getstatusoutput("grep relayhost /etc/postfix/main.cf")
+
if status == 0:
- configured_relay = out[len('relayhost'):].strip(' \t=').replace('[', '').replace(']', '')
+ configured_relay = (
+ out[len("relayhost") :].strip(" \t=").replace("[", "").replace("]", "")
+ )
else:
- configured_relay = ''
+ configured_relay = ""
+
if not configured_relay:
- # no relay configured, check relayless situations
- ip = conf.get('NETWORK_IP_NAT')
- if not ip:
- ip = conf.get('NETWORK_IP')
- if not ip:
- print_yellow('Cannot determine public IP address')
+ # check public ip address
+ ip_addr = conf.get("NETWORK_IP_NAT")
+ if not ip_addr:
+ ip_addr = conf.get("NETWORK_IP")
+ if not ip_addr:
+ u.warning("Cannot determine public IP address")
return 3
- with open('/etc/mailname', 'r') as f:
- d = f.read().strip()
- if d not in ('ubicast.tv', 'ubicast.eu'):
- print_yellow('/etc/mailname does not contain ubicast.eu or ubicast.tv, mails will probably not be received on ubicast mailing lists')
+ # check domain origin
+ with open("/etc/mailname", "r") as mailname:
+ data = mailname.read().strip()
+ if data not in ("ubicast.tv", "ubicast.eu"):
+ u.warning("/etc/mailname does not contain ubicast.eu or ubicast.tv")
return 3
# check spf
- result, explain = spf.check2(i=ip, s='support@ubicast.eu', h='')
- if result != 'pass':
- print_red('ip %s is not in ubicast.eu SPF and emails sent from support@ubicast.eu may be treated as spam' % ip)
+ result, _ = spf.check2(i=ip_addr, s="support@ubicast.eu", h="")
+ if result != "pass":
+ u.error("SPF record for {} in ubicast.eu is missing".format(ip_addr))
return 3
- conf_relay = conf.get('EMAIL_SMTP_SERVER', '').replace('[', '').replace(']', '')
+
+ # get relayhost value from envsetup config
+ conf_relay = conf.get("EMAIL_SMTP_SERVER", "").replace("[", "").replace("]", "")
+
if conf_relay != configured_relay:
- print_red('Configured STMP relay (%s) does not match the expected value (%s).' % (configured_relay, conf_relay))
+ u.error("STMP relay must be {}".format(conf_relay))
return 3
- else:
- print_green('STMP relay is properly set.')
- return 0
+ u.success("STMP relay is properly set")
+ return 0
+
+
+def check_send_test_email() -> int:
+ """Check that Postfix can send email.
+
+ :return: Exit return code
+ :rtype: int
+ """
+
+ print("Checking Postfix can send email:")
-def send_test_email():
- email = 'noreply+%s-%s@ubicast.eu' % (time.time(), random.randint(0, 1000))
- print('Sending test email to "%s".' % email)
- cmd = 'echo "This is a test email" | mail -s "Test email from `cat /etc/hostname`" %s' % email
+ # send email
+ email = "noreply+{}-{}@ubicast.eu".format(time.time(), random.randint(0, 1000))
+ u.info("Sending test email to '{}'".format(email))
+ cmd = "echo 'test email' | mail -s 'test email' {}".format(email)
subprocess.getstatusoutput(cmd)
+
+ # init vars
timeout = 120
waited = 1
delay = 2
- print('Timeout to find email sending log is set to %s seconds.' % timeout)
- out = ''
- if os.path.isfile('/var/log/mail.log'):
- cmd = 'grep "%s" /var/log/mail.log' % email
+ out = ""
+
+ # find logs
+ if os.path.isfile("/var/log/mail.log"):
+ cmd = "grep '{}' /var/log/mail.log".format(email)
else:
- print('/var/log/mail.log not found, trying journalctl')
- cmd = 'journalctl -u postfix | grep %s' % email
+ u.info("/var/log/mail.log not found, trying journalctl")
+ cmd = "journalctl -u postfix | grep {}".format(email)
+
+ # logs polling
time.sleep(1)
while True:
status, out = subprocess.getstatusoutput(cmd)
+ # found
if status == 0:
- out = out.strip().split('\n')[-1]
- if 'status=deferred' not in out:
- print('Log entry found after %s seconds.' % waited)
+ out = out.strip().split("\n")[-1]
+ if "status=deferred" not in out:
+ u.info("Log entry found after {} seconds".format(waited))
break
+ # timeout
if waited >= timeout:
- print_red('Failed to send email.')
- print('No log entry found after %s seconds using command: %s' % (waited, cmd))
+ u.error("Failed to send email.")
+ u.info(
+ "No log entry found after {} seconds using command: {}".format(
+ waited, cmd
+ )
+ )
return 1
- print('No log entry found after %s seconds, waiting %s more seconds...' % (waited, delay))
+ # not found yet
+ u.info(
+ "No log entry found after {} seconds, waiting {} more seconds...".format(
+ waited, delay
+ )
+ )
+ # wait before next iteration
time.sleep(delay)
waited += delay
delay *= 2
- if 'bounced' not in out or 'The email account that you tried to reach does not exist.' in out:
- print_green('Email sent.')
- return 0
- else:
- print_red('Failed to send email.')
- print('Sending log line:\n%s' % out)
+
+ # check output for errors
+ if "bounced" in out or "you tried to reach does not exist" in out:
+ u.error("Failed to send email")
+ u.info("Sending log line:\n{}".format(out))
return 1
+ u.success("Email sent.")
+ return 0
+
+
+def main():
+ """Run all checks and exits with corresponding exit code."""
+
+ if not os.path.exists("/etc/postfix"):
+ u.error("Postfix dir does not exists, please install postfix")
+ sys.exit(1)
+
+ conf = u.load_conf()
+
+ rcode = check_listen()
+ if rcode == 0:
+ rcode = check_relay(conf)
+ if check_send_test_email() == 1:
+ rcode = 1
-if not os.path.exists('/etc/postfix'):
- print_red('Postfix dir does not exists, please install postfix.')
- sys.exit(1)
+ sys.exit(rcode)
-rc = check_listening_port()
-if rc == 0:
- rc = check_relay()
- if send_test_email() == 1:
- rc = 1
-sys.exit(rc)
+if __name__ == "__main__":
+ main()
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index 8206300a0fea9aff209d434c4b35eceab5b3f43e..5767eca67362cfc76b85e1cc2f8e2615a111f462 100755
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -4,8 +4,6 @@ Criticality: High
This test check the current state of the PostgreSQL database cluster.
"""
-import imp
-import os
import re
import socket
import subprocess
@@ -13,45 +11,15 @@ import sys
import time
import uuid
+sys.path.append("..")
+
try:
import psycopg2
except ImportError:
sys.exit(2)
-GREEN = "\033[92m"
-YELLOW = "\033[93m"
-RED = "\033[91m"
-DEF = "\033[0m"
-
-
-def success(message: str):
- """Print formatted success message.
-
- :param message: Message to print
- :type message: str
- """
-
- print(" {}✔{} {}".format(GREEN, DEF, message))
-
-
-def warning(message: str):
- """Print formatted warning message.
-
- :param message: Message to print
- :type message: str
- """
-
- print(" {}âš {} {}".format(YELLOW, DEF, message))
-
-
-def error(message: str):
- """Print formatted error message.
-
- :param message: Message to print
- :type message: str
- """
-
- print(" {}✖{} {}".format(RED, DEF, message))
+# pylint: disable=wrong-import-position
+from envsetup import utils as u # noqa: E402
def is_ha(port: int) -> bool:
@@ -254,7 +222,6 @@ def check_replication(primary: dict, standby: dict) -> tuple:
rand = uuid.uuid4().hex
write_query = "CREATE TABLE es_test_{} (id serial PRIMARY KEY);".format(rand)
read_query = "SELECT * FROM es_test_{};".format(rand)
- del_query = "DROP TABLE es_test_{};".format(rand)
# write
try:
@@ -281,7 +248,7 @@ def check_replication(primary: dict, standby: dict) -> tuple:
# delete
try:
- primary_psql.execute(del_query)
+ primary_psql.execute("DROP TABLE es_test_{};".format(rand))
except psycopg2.Error:
pass
@@ -314,15 +281,15 @@ def check_ha(db_conn: dict, errors: int = 0) -> int:
# check haproxy
print("Checking local HAProxy frontends:")
if not check_listen(db_conn["host"], 54321):
- error("HAProxy pgsql-primary frontend is not listening")
+ u.error("HAProxy pgsql-primary frontend is not listening")
errors += 1
else:
- success("HAProxy pgsql-primary frontend is listening")
+ u.success("HAProxy pgsql-primary frontend is listening")
if not check_listen(db_conn["host"], 54322):
- error("HAProxy pgsql-standby frontend is not listening")
+ u.error("HAProxy pgsql-standby frontend is not listening")
errors += 1
else:
- success("HAProxy pgsql-standby frontend is listening")
+ u.success("HAProxy pgsql-standby frontend is listening")
# check remotes
print("Checking remote PostgreSQL nodes:")
@@ -330,19 +297,19 @@ def check_ha(db_conn: dict, errors: int = 0) -> int:
node_host = nodes[node]["host"]
node_port = nodes[node]["port"]
if not check_listen(node_host, node_port):
- error("Cannot bind {}:{}".format(node_host, node_port))
+ u.error("Cannot bind {}:{}".format(node_host, node_port))
errors += 1
else:
- success("Can bind {}:{}".format(node_host, node_port))
+ u.success("Can bind {}:{}".format(node_host, node_port))
# check fenced
print("Checking cluster state:")
fenced, node = check_fenced(nodes)
if fenced:
- error("Node `{}` is fenced".format(node))
+ u.error("Node `{}` is fenced".format(node))
errors += 1
else:
- success("No fenced node found")
+ u.success("No fenced node found")
# check replication
print("Checking replication state:")
@@ -352,10 +319,10 @@ def check_ha(db_conn: dict, errors: int = 0) -> int:
standby["port"] = 54322
status, info = check_replication(primary, standby)
if not status:
- error("Cannot replicate between primary/standby ({})".format(info))
+ u.error("Cannot replicate between primary/standby ({})".format(info))
errors += 1
else:
- success("Can replicate between primary/standby ({})".format(info))
+ u.success("Can replicate between primary/standby ({})".format(info))
return errors
@@ -371,27 +338,27 @@ def check_local(db_conn: dict, errors: int = 0) -> int:
:rtype: int
"""
- db_host = db_conn["host"]
- db_port = db_conn["port"]
- db_user = db_conn["user"]
+ host = db_conn["host"]
+ port = db_conn["port"]
+ user = db_conn["user"]
# check listen
print("Checking local PostgreSQL node:")
- if not check_listen(db_host, db_port):
- error("Cannot connect to {}:{}".format(db_host, db_port))
+ if not check_listen(host, port):
+ u.error("Cannot connect to {}:{}".format(host, port))
errors += 1
else:
- success("Can connect to {}:{}".format(db_host, db_port))
+ u.success("Can connect to {}:{}".format(host, port))
# check read
print("Checking read operation:")
read_query = "SELECT 1;"
status, info = check_psql(db_conn, read_query)
if not status:
- error("Cannot read from {}@{}:{} ({})".format(db_user, db_host, db_port, info))
+ u.error("Cannot read from {}@{}:{} ({})".format(user, host, port, info))
errors += 1
else:
- success("Can read from {}@{}:{}".format(db_user, db_host, db_port))
+ u.success("Can read from {}@{}:{}".format(user, host, port))
# check write
print("Checking write operation:")
@@ -399,10 +366,10 @@ def check_local(db_conn: dict, errors: int = 0) -> int:
write_query = "CREATE TABLE es_test_{} (id serial PRIMARY KEY);".format(rand)
status, info = check_psql(db_conn, write_query)
if not status:
- error("Cannot write on {}@{}:{} ({})".format(db_user, db_host, db_port, info))
+ u.error("Cannot write on {}@{}:{} ({})".format(user, host, port, info))
errors += 1
else:
- success("Can write on {}@{}:{}".format(db_user, db_host, db_port))
+ u.success("Can write on {}@{}:{}".format(user, host, port))
# remove test table
check_psql(db_conn, "DROP TABLE es_test_{};".format(rand))
@@ -412,20 +379,8 @@ def check_local(db_conn: dict, errors: int = 0) -> int:
def main():
"""Run all checks and exits with corresponding exit code."""
- # envsetup utils path
- cwd = os.path.dirname(__file__)
- utils = os.path.join(cwd, "..", "utils.py")
-
- # check envsetup utils presence
- if not os.path.isfile(utils):
- error("{} not found.".format(utils))
- sys.exit(1)
-
- # load envsetup utils
- es_utils = imp.load_source("es_utils", utils)
-
# load configuration
- conf = es_utils.load_conf()
+ conf = u.load_conf()
# get database configuration
db_host = conf.get("DB_HOST") if conf.get("DB_HOST") else "127.0.0.1"
diff --git a/utils.py b/utils.py
index 95c64d92c3c4affa13329c346a66f84cb41dae65..0629672530e14943b551b3284fa9f1f23bed5f75 100644
--- a/utils.py
+++ b/utils.py
@@ -1,54 +1,154 @@
#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
+
+"""EnvSetup utilities."""
+
from collections import OrderedDict
import os
import subprocess
import sys
+from typing import Any
+
+RED = "\033[91m"
+GREEN = "\033[92m"
+YELLOW = "\033[93m"
+BLUE = "\033[94m"
+DEF = "\033[0m"
+DEFAULT_CONF_PATH = "global-conf.sh"
+AUTO_CONF_PATH = "auto-generated-conf.sh"
+CONF_PATH = "conf.sh"
-DEFAULT_CONF_PATH = 'global-conf.sh'
-AUTO_CONF_PATH = 'auto-generated-conf.sh'
-CONF_PATH = 'conf.sh'
-CONF = dict()
+def log(text: str, error: bool = False):
+ """Output log message to stout or stderr.
+
+ :param text: Message to log
+ :type text: str
+ :param error: Wether it should output to stderr or not, defaults to False
+ :param error: bool, optional
+ """
-def log(text, error=False):
fo = sys.stderr if error else sys.stdout
print(text, file=fo)
fo.flush()
-def get_dir(file_path):
+def info(message: str):
+ """Print formatted info message.
+
+ :param message: Message to print
+ :type message: str
+ """
+
+ print(" {}🛈{} {}".format(GREEN, DEF, message))
+
+
+def success(message: str):
+ """Print formatted success message.
+
+ :param message: Message to print
+ :type message: str
+ """
+
+ print(" {}✔{} {}".format(GREEN, DEF, message))
+
+
+def warning(message: str):
+ """Print formatted warning message.
+
+ :param message: Message to print
+ :type message: str
+ """
+
+ print(" {}âš {} {}".format(YELLOW, DEF, message))
+
+
+def error(message: str):
+ """Print formatted error message.
+
+ :param message: Message to print
+ :type message: str
+ """
+
+ print(" {}✖{} {}".format(RED, DEF, message))
+
+
+def get_dir(file_path: str) -> str:
+ """Get the absolute directory path for the given file.
+
+ :param file_path: File path
+ :type file_path: str
+ :return: Absolute directory path
+ :rtype: str
+ """
+
return os.path.dirname(os.path.abspath(os.path.expanduser(file_path)))
-def exec_cmd(cmd, log_output=True, get_output=True):
+def exec_cmd(cmd: Any, log_output: bool = True, get_output: bool = True) -> tuple:
+ """Execute the given command.
+
+ :param cmd: Command to run
+ :type cmd: Any
+ :param log_output: Wether to log output or not, defaults to True
+ :param log_output: bool, optional
+ :param get_output: Wether to return output or not, defaults to True
+ :param get_output: bool, optional
+ :return: Return code and output
+ :rtype: tuple
+ """
+
shell = not isinstance(cmd, (tuple, list))
stdout = subprocess.PIPE if get_output or not log_output else sys.stdout
stderr = subprocess.PIPE if get_output or not log_output else sys.stderr
- p = subprocess.Popen(cmd, stdin=sys.stdin, stdout=stdout, stderr=stderr, shell=shell)
+
+ # execute
+ p = subprocess.Popen(
+ cmd, stdin=sys.stdin, stdout=stdout, stderr=stderr, shell=shell
+ )
out, err = p.communicate()
+
+ # send to the correct output
if get_output:
- out = out.decode('utf-8').strip() if out else ''
+ out = out.decode("utf-8").strip() if out else ""
if err:
if out:
- out += '\n'
- out += err.decode('utf-8').strip()
+ out += "\n"
+ out += err.decode("utf-8").strip()
out = out.strip()
if log_output:
log(out)
elif log_output:
sys.stdout.flush()
sys.stderr.flush()
+
return p.returncode, out
-def check_cmd(cmd, log_output=False):
- code, out = exec_cmd(cmd, log_output, False)
+def check_cmd(cmd: Any, log_output: bool = False) -> int:
+ """Get the return code of the given command.
+
+ :param cmd: Command to execute
+ :type cmd: Any
+ :param log_output: Wether to log output or not, defaults to False
+ :param log_output: bool, optional
+ :return: Return code
+ :rtype: int
+ """
+
+ code, _ = exec_cmd(cmd, log_output, False)
+
return code
-def load_conf():
+def load_conf() -> dict:
+ """Load EnvSetup configuration settings.
+
+ :return: Configuration settings
+ :rtype: dict
+ """
+
+ conf = {}
base_dir = get_dir(__file__)
files = (
(os.path.join(base_dir, DEFAULT_CONF_PATH), True),
@@ -60,94 +160,119 @@ def load_conf():
for path, is_default in files:
if not os.path.exists(path):
if is_default:
- log('The configuration file for EnvSetup script does not exist.\nPath of configuration file: %s' % path, error=True)
+ log(
+ "The configuration file '{}' does not exist.".format(path),
+ error=True,
+ )
return dict()
continue
# Load conf
- with open(path, 'r') as fo:
+ with open(path, "r") as fo:
content = fo.read()
# Parse conf
- for line in content.split('\n'):
+ for line in content.split("\n"):
line = line.strip()
- if line and not line.startswith('#') and '=' in line:
- name, *val = line.split('=')
- name = name.strip(' \t\'\"')
- val = ('='.join(val)).strip(' \t\'\"')
- CONF[name] = val
+ if line and not line.startswith("#") and "=" in line:
+ name, *val = line.split("=")
+ name = name.strip(" \t'\"")
+ val = ("=".join(val)).strip(" \t'\"")
+ conf[name] = val
if is_default:
override[name] = False
else:
only_default = False
override[name] = True
- CONF['_override'] = override
+ conf["_override"] = override
# Check a value to know if the config file has been changed
if only_default:
- log('\033[93mWarning:\033[0m')
- log('The configuration is using only default values.')
- log('Perhaps you forget to change the configuration.')
- log('Path of configuration file: %s' % os.path.join(base_dir, CONF_PATH))
- log('Perhaps you want to quit this script to change the configuration?\n')
- return CONF
+ log("\033[93mWarning:\033[0m")
+ log("The configuration is using only default values.")
+ log("Perhaps you forget to change the configuration.")
+ log("Path of configuration file: %s" % os.path.join(base_dir, CONF_PATH))
+ log("Perhaps you want to quit this script to change the configuration?\n")
+ return conf
+
+
+def get_conf(name: str, default: str = None) -> str:
+ """Get the given configuration parameter.
+
+ :param name: Parameter name
+ :type name: str
+ :param default: Default parameter value, defaults to None
+ :param default: str, optional
+ :return: Parameter value
+ :rtype: str
+ """
+
+ conf = load_conf()
+
+ return conf.get(name, default)
-def get_conf(name, default=None):
- if not CONF:
- load_conf()
- return CONF.get(name, default)
+def run_commands(cmds: list):
+ """Run a serie of successive commands.
+ :param cmds: List of commands
+ :type cmds: list
+ :raises Exception: Houston we have a problem
+ """
-def run_commands(cmds):
- # run a serie of successive commands
try:
# Execute commands
for cmd in cmds:
if not isinstance(cmd, dict):
cmd = dict(line=cmd)
- if cmd.get('cond'):
- cond = cmd['cond']
- negate = cmd.get('cond_neg')
- skip = cmd.get('cond_skip')
+ if cmd.get("cond"):
+ cond = cmd["cond"]
+ negate = cmd.get("cond_neg")
+ skip = cmd.get("cond_skip")
code = check_cmd(cond)
- success = code != 0 if negate else code == 0
- if not success:
- msg = 'Condition for command "%s" not fullfilled.' % cmd['line']
+ valid = code != 0 if negate else code == 0
+ if not valid:
+ msg = 'Condition for command "%s" not fullfilled.' % cmd["line"]
if skip:
- log('%s Command skipped.' % msg)
+ log("%s Command skipped." % msg)
continue
raise Exception(msg)
- if cmd['line'] == 'write':
- if not cmd.get('target'):
- raise Exception('No target file to write in.')
- if cmd.get('backup') and os.path.exists(cmd['target']) and not os.path.exists(cmd['target'] + '.back'):
- os.rename(cmd['target'], cmd['target'] + '.back')
- log('A backup file has been created for:\n%s' % cmd['target'])
+ if cmd["line"] == "write":
+ if not cmd.get("target"):
+ raise Exception("No target file to write in.")
+ if (
+ cmd.get("backup")
+ and os.path.exists(cmd["target"])
+ and not os.path.exists(cmd["target"] + ".back")
+ ):
+ os.rename(cmd["target"], cmd["target"] + ".back")
+ log("A backup file has been created for:\n%s" % cmd["target"])
# Load content from template if any
- content = cmd.get('content', '')
- if cmd.get('template'):
- if not os.path.exists(cmd['template']):
- raise Exception('Template file does not exist: %s.' % cmd['template'])
- with open(cmd['template'], 'r') as fd:
+ content = cmd.get("content", "")
+ if cmd.get("template"):
+ if not os.path.exists(cmd["template"]):
+ raise Exception(
+ "Template file does not exist: %s." % cmd["template"]
+ )
+ with open(cmd["template"], "r") as fd:
content = fd.read()
- if cmd.get('params'):
- for k, v in cmd['params']:
+ if cmd.get("params"):
+ for k, v in cmd["params"]:
content = content.replace(k, v)
# Write target file
- with open(cmd['target'], 'w+') as fd:
+ with open(cmd["target"], "w+") as fd:
fd.write(content)
- log('File %s written' % cmd['target'])
- elif cmd['line'] == 'backup':
- if not cmd.get('target'):
- raise Exception('No target file to backup.')
- if not os.path.exists(cmd['target'] + '.back'):
- os.rename(cmd['target'], cmd['target'] + '.back')
- log('A backup file has been created for:\n%s' % cmd['target'])
+ log("File %s written" % cmd["target"])
+ elif cmd["line"] == "backup":
+ if not cmd.get("target"):
+ raise Exception("No target file to backup.")
+ if not os.path.exists(cmd["target"] + ".back"):
+ os.rename(cmd["target"], cmd["target"] + ".back")
+ log("A backup file has been created for:\n%s" % cmd["target"])
else:
- log('A backup file already exist for:\n%s' % cmd['target'])
+ log("A backup file already exist for:\n%s" % cmd["target"])
else:
- log('>>> ' + cmd['line'])
- code = check_cmd(cmd['line'], log_output=True)
+ log(">>> " + cmd["line"])
+ code = check_cmd(cmd["line"], log_output=True)
if code != 0:
- raise Exception('Command exited with code %s.' % code)
+ raise Exception("Command exited with code %s." % code)
except Exception as e:
- log('Command failed:\n%s' % e)
+ log("Command failed:\n%s" % e)
raise