badblocks/docker-easy-haproxy
1
0
Fork 0
Code Activity

Merge pull request #58

Browse source 
Log Refactor
This commit is contained in:
Joao M 2024-11-19 10:18:40 -06:00 • committed by GitHub
commit 64bc49b0be
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 166 additions and 175 deletions
Download patch file Download diff file Expand all files Collapse all files

133
src/functions/__init__.py
View file

@ -1,14 +1,16 @@
import os
import shlex
import subprocess
import sys
import time
import logging
from datetime import datetime
from multiprocessing import Process
from typing import Final
import requests
from OpenSSL import crypto
class ContainerEnv:
@staticmethod
def read():
@ -82,7 +84,7 @@ class ContainerEnv:
env_vars["certbot"]["eab_hmac_key"] = os.environ['EASYHAPROXY_CERTBOT_EAB_HMAC_KEY'] = resp["eab_hmac_key"]
else:
del os.environ["EASYHAPROXY_CERTBOT_EMAIL"]
Functions.log(Functions.CERTBOT_LOG, Functions.ERROR, "Could not obtain ZeroSSL credentials " + resp["error"]["type"])
loggerCertbot.error("Could not obtain ZeroSSL credentials " + resp["error"]["type"])
os.environ['EASYHAPROXY_CERTBOT_SERVER'] = env_vars["certbot"]["server"]
@ -90,34 +92,37 @@ class ContainerEnv:
class Functions:
HAPROXY_LOG = "HAPROXY"
EASYHAPROXY_LOG = "EASYHAPROXY"
CERTBOT_LOG = "CERTBOT"
INIT_LOG = "INIT"
HAPROXY_LOG: Final[str] = "HAPROXY"
EASYHAPROXY_LOG: Final[str] = "EASYHAPROXY"
CERTBOT_LOG: Final[str] = "CERTBOT"
INIT_LOG: Final[str] = "INIT"
TRACE = "TRACE"
DEBUG = "DEBUG"
INFO = "INFO"
WARN = "WARN"
ERROR = "ERROR"
FATAL = "FATAL"
debug_log = None
TRACE: Final[str] = "TRACE"
DEBUG: Final[str] = "DEBUG"
INFO: Final[str] = "INFO"
WARN: Final[str] = "WARN"
ERROR: Final[str] = "ERROR"
FATAL: Final[str] = "FATAL"
@staticmethod
def skip_log(source, log_level_str):
level = os.getenv("%s_LOG_LEVEL" % (source.upper()), "").upper()
def setup_log(source):
level = os.getenv("%s_LOG_LEVEL" % (source.name.upper()), "").upper()
level_importance = {
Functions.TRACE: 0,
Functions.DEBUG: 1,
Functions.INFO: 2,
Functions.WARN: 3,
Functions.ERROR: 4,
Functions.FATAL: 5
Functions.TRACE: logging.DEBUG,
Functions.DEBUG: logging.DEBUG,
Functions.INFO: logging.INFO,
Functions.WARN: logging.WARNING,
Functions.ERROR: logging.ERROR,
Functions.FATAL: logging.FATAL
}
level_required = 1 if level not in level_importance else level_importance[level]
level_asked = 1 if log_level_str.upper() not in level_importance else level_importance[log_level_str.upper()]
return level_asked < level_required
selected_level = level_importance[level] if level in level_importance else logging.INFO
log_source_handler = logging.StreamHandler(sys.stdout)
log_source_formatter = logging.Formatter('%(name)s [%(asctime)s] %(levelname)s - %(message)s')
log_source_handler.setFormatter(log_source_formatter)
source.setLevel(selected_level)
source.addHandler(log_source_handler)
return selected_level
@staticmethod
def load(filename):
@ -130,24 +135,7 @@ class Functions:
file.write(contents)
@staticmethod
def log(source, level, message):
if message is None or message == "":
return
if Functions.skip_log(source, level):
return
if not isinstance(message, (list, tuple)):
message = [message]
for line in message:
log = "[%s] %s [%s]: %s" % (source, datetime.now().strftime("%x %X"), level, line.rstrip())
print(log)
if Functions.debug_log is not None:
Functions.debug_log.append(log)
@staticmethod
def run_bash(source, command, log_output=True, return_result=True):
def run_bash(log_source, command, log_output=True, return_result=True):
if not isinstance(command, (list, tuple)):
command = shlex.split(command)
@ -161,22 +149,24 @@ class Functions:
while True:
line = process.stdout.readline().rstrip()
error_line = process.stderr.readline().rstrip()
output.append(line) if return_result else None
Functions.log(source, Functions.INFO, line) if log_output else None
Functions.log(source, Functions.WARN, process.stderr.readline())
log_source.info(line) if log_output and len(line) > 0 else None
log_source.warning(error_line) if len(error_line) > 0 else None
return_code = process.poll()
if return_code is not None:
lines = []
error_line = process.stderr.readline().rstrip()
for line in process.stdout.readlines():
output.append(line.rstrip()) if return_result else None
lines.append(line.rstrip())
Functions.log(source, Functions.INFO, lines) if log_output else None
Functions.log(source, Functions.WARN, process.stderr.readlines())
log_source.info(lines) if log_output and len(lines) > 0 else None
log_source.warning(error_line) if len(error_line) > 0 else None
break
return [return_code, output]
except Exception as e:
Functions.log(source, Functions.ERROR, "%s" % e)
log_source.error("%s" % e)
return [-99, e]
@ -189,6 +179,9 @@ class Consts:
class DaemonizeHAProxy:
HAPROXY_START: Final[str] = "start"
HAPROXY_RELOAD: Final[str] = "reload"
def __init__(self, custom_config_folder = None):
self.process = None
self.thread = None
@ -207,17 +200,16 @@ class DaemonizeHAProxy:
def get_haproxy_command(self, action, pid_file="/run/haproxy.pid"):
custom_config_files = ""
if len(list(self.get_custom_config_files().keys())) != 0:
custom_config_files = "-f %s" % (self.custom_config_folder)
custom_config_files = "-f %s" % self.custom_config_folder
if action == "start":
if action == DaemonizeHAProxy.HAPROXY_START:
return "/usr/sbin/haproxy -W -f /etc/haproxy/haproxy.cfg %s -p %s -S /var/run/haproxy.sock" % (custom_config_files, pid_file)
else:
return_code, output = Functions().run_bash(Functions.HAPROXY_LOG, "cat %s" % pid_file, log_output=False)
return_code, output = Functions().run_bash(loggerHaproxy, "cat %s" % pid_file, log_output=False)
pid = "".join(output)
return "/usr/sbin/haproxy -W -f /etc/haproxy/haproxy.cfg %s -p %s -x /var/run/haproxy.sock -sf %s" % (custom_config_files, pid_file, pid)
def __prepare(self, command):
source = Functions.HAPROXY_LOG
if not isinstance(command, (list, tuple)):
command = shlex.split(command)
@ -230,20 +222,19 @@ class DaemonizeHAProxy:
universal_newlines=True)
except Exception as e:
Functions.log(source, Functions.ERROR, "%s" % e)
loggerHaproxy.error("%s" % e)
def __start(self):
source = Functions.HAPROXY_LOG
try:
with self.process.stdout:
for line in iter(self.process.stdout.readline, b''):
Functions.log(source, Functions.INFO, line)
loggerHaproxy.info(line)
return_code = self.process.wait()
Functions.log(source, Functions.DEBUG, "Return code %s" % return_code)
loggerHaproxy.debug("Return code %s" % return_code)
except Exception as e:
Functions.log(source, Functions.ERROR, "%s" % e)
loggerHaproxy.error("%s" % e)
def is_alive(self):
return self.thread.is_alive()
@ -330,14 +321,13 @@ class Certbot:
elif host in self.freeze_issue:
freeze_count = self.freeze_issue.pop(host, 0)
if freeze_count > 0:
Functions.log(Functions.CERTBOT_LOG, Functions.DEBUG,
"Waiting freezing period (%d) for %s due previous errors" % (freeze_count, host))
loggerCertbot.debug("Waiting freezing period (%d) for %s due previous errors" % (freeze_count, host))
self.freeze_issue[host] = freeze_count-1
elif cert_status == "not_found" or cert_status == "expired":
Functions.log(Functions.CERTBOT_LOG, Functions.DEBUG, "[%s] Request new certificate for %s" % (cert_status, host))
loggerCertbot.debug("[%s] Request new certificate for %s" % (cert_status, host))
request_certs.append(host_arg)
elif cert_status == "expiring":
Functions.log(Functions.CERTBOT_LOG, Functions.DEBUG, "[%s] Renew certificate for %s" % (cert_status, host))
loggerCertbot.debug("[%s] Renew certificate for %s" % (cert_status, host))
renew_certs.append(host_arg)
certbot_certonly = ('/usr/bin/certbot certonly {acme_server}'
@ -368,11 +358,11 @@ class Certbot:
return_code_issue = 0
return_code_renew = 0
if len(request_certs) > 0:
return_code_issue, output = Functions.run_bash(Functions.CERTBOT_LOG, certbot_certonly, return_result=False)
return_code_issue, output = Functions.run_bash(loggerCertbot, certbot_certonly, return_result=False)
ret_reload = True
if len(renew_certs) > 0:
return_code_renew, output = Functions.run_bash(Functions.CERTBOT_LOG, "/usr/bin/certbot renew", return_result=False)
return_code_renew, output = Functions.run_bash(loggerCertbot, "/usr/bin/certbot renew", return_result=False)
ret_reload = True
if ret_reload:
@ -385,7 +375,7 @@ class Certbot:
return ret_reload
except Exception as e:
Functions.log(Functions.CERTBOT_LOG, Functions.ERROR, "%s" % e)
loggerCertbot.error("%s" % e)
return False
@staticmethod
@ -420,7 +410,7 @@ class Certbot:
elif (expiration_after - current_time) // (24 * 3600) <= 15:
return "expiring"
except Exception as e:
Functions.log(Functions.CERTBOT_LOG, Functions.ERROR, "Certificate %s error %s" % (host, e))
loggerCertbot.error("Certificate %s error %s" % (host, e))
return "error"
return "ok"
@ -432,4 +422,15 @@ class Certbot:
cert_status = self.get_certificate_status(host)
if cert_status != "ok":
self.freeze_issue[host] = self.retry_count
Functions.log(Functions.CERTBOT_LOG, Functions.DEBUG, "Freeze issuing ssl for %s due failure. The certificate is %s" % (host, cert_status))
loggerCertbot.debug("Freeze issuing ssl for %s due failure. The certificate is %s" % (host, cert_status))
# ####################################################################################################################
# Setup Global Log
loggerInit = logging.getLogger(Functions.INIT_LOG)
loggerHaproxy = logging.getLogger(Functions.HAPROXY_LOG)
loggerEasyHaproxy = logging.getLogger(Functions.EASYHAPROXY_LOG)
loggerCertbot = logging.getLogger(Functions.CERTBOT_LOG)
Functions.setup_log(loggerInit)
Functions.setup_log(loggerHaproxy)
Functions.setup_log(loggerEasyHaproxy)
Functions.setup_log(loggerCertbot)

43
src/main.py
View file

@ -1,8 +1,10 @@
import os
import logging
from deepdiff import DeepDiff
from functions import Functions, DaemonizeHAProxy, Certbot, Consts
from functions import Functions, DaemonizeHAProxy, Certbot, Consts, loggerInit, loggerEasyHaproxy, loggerHaproxy, \
loggerCertbot
from processor import ProcessorInterface
@ -17,14 +19,13 @@ def start():
processor_obj.save_config(Consts.haproxy_config)
processor_obj.save_certs(Consts.certs_haproxy)
certbot_certs_found = processor_obj.get_certbot_hosts()
Functions.log(Functions.EASYHAPROXY_LOG, Functions.DEBUG,
'Found hosts: %s' % ", ".join(processor_obj.get_hosts())) # Needs to run after save_config
Functions.log(Functions.EASYHAPROXY_LOG, Functions.TRACE, 'Object Found: %s' % (processor_obj.get_parsed_object()))
loggerEasyHaproxy.info('Found hosts: %s' % ", ".join(processor_obj.get_hosts())) # Needs to run after save_config
loggerEasyHaproxy.debug('Object Found: %s' % (processor_obj.get_parsed_object()))
old_haproxy = None
haproxy = DaemonizeHAProxy()
current_custom_config_files = haproxy.get_custom_config_files()
haproxy.haproxy("start")
haproxy.haproxy(DaemonizeHAProxy.HAPROXY_START)
haproxy.sleep()
certbot = Certbot(Consts.certs_certbot)
@ -37,41 +38,39 @@ def start():
old_parsed = processor_obj.get_parsed_object()
processor_obj.refresh()
if certbot.check_certificates(certbot_certs_found) or DeepDiff(old_parsed, processor_obj.get_parsed_object()) != {} or not haproxy.is_alive() or DeepDiff(current_custom_config_files, haproxy.get_custom_config_files()) != {}:
Functions.log(Functions.EASYHAPROXY_LOG, Functions.DEBUG, 'New configuration found. Reloading...')
Functions.log(Functions.EASYHAPROXY_LOG, Functions.TRACE,
'Object Found: %s' % (processor_obj.get_parsed_object()))
loggerEasyHaproxy.info('New configuration found. Reloading...')
loggerEasyHaproxy.debug('Object Found: %s' % (processor_obj.get_parsed_object()))
processor_obj.save_config(Consts.haproxy_config)
processor_obj.save_certs(Consts.certs_haproxy)
certbot_certs_found = processor_obj.get_certbot_hosts()
Functions.log(Functions.EASYHAPROXY_LOG, Functions.DEBUG,
'Found hosts: %s' % ", ".join(processor_obj.get_hosts())) # Needs to after save_config
loggerEasyHaproxy.info('Found hosts: %s' % ", ".join(processor_obj.get_hosts())) # Needs to after save_config
old_haproxy = haproxy
haproxy = DaemonizeHAProxy()
current_custom_config_files = haproxy.get_custom_config_files()
haproxy.haproxy("reload")
haproxy.haproxy(DaemonizeHAProxy.HAPROXY_RELOAD)
old_haproxy.terminate()
except Exception as e:
Functions.log(Functions.EASYHAPROXY_LOG, Functions.FATAL, "Err: %s" % e)
loggerEasyHaproxy.fatal("Err: %s" % e)
Functions.log(Functions.EASYHAPROXY_LOG, Functions.DEBUG, 'Heartbeat')
loggerEasyHaproxy.info('Heartbeat')
haproxy.sleep()
def main():
Functions.run_bash(Functions.INIT_LOG, '/usr/sbin/haproxy -v')
Functions.run_bash(loggerInit, '/usr/sbin/haproxy -v')
Functions.log(Functions.INIT_LOG, Functions.INFO, " _ ")
Functions.log(Functions.INIT_LOG, Functions.INFO, " ___ __ _ ____ _ ___| |_ __ _ _ __ _ _ _____ ___ _ ")
Functions.log(Functions.INIT_LOG, Functions.INFO, "/ -_) _` (_-< || |___| ' \\/ _` | '_ \\ '_/ _ \\ \\ / || |")
Functions.log(Functions.INIT_LOG, Functions.INFO, "\\___\\__,_/__/\\_, | |_||_\\__,_| .__/_| \\___/_\\_\\_, |")
Functions.log(Functions.INIT_LOG, Functions.INFO, " |__/ |_| |__/ ")
loggerInit.info(" _ ")
loggerInit.info(" ___ __ _ ____ _ ___| |_ __ _ _ __ _ _ _____ ___ _ ")
loggerInit.info("/ -_) _` (_-< || |___| ' \\/ _` | '_ \\ '_/ _ \\ \\ / || |")
loggerInit.info("\\___\\__,_/__/\\_, | |_||_\\__,_| .__/_| \\___/_\\_\\_, |")
loggerInit.info(" |__/ |_| |__/ ")
Functions.log(Functions.INIT_LOG, Functions.INFO, "Release: %s" % (os.getenv("RELEASE_VERSION")))
Functions.log(Functions.INIT_LOG, Functions.DEBUG, 'Environment:')
loggerInit.info("Release: %s" % (os.getenv("RELEASE_VERSION")))
loggerInit.debug('Environment:')
for name, value in os.environ.items():
if "HAPROXY" in name:
Functions.log(Functions.INIT_LOG, Functions.DEBUG, "- {0}: {1}".format(name, value))
loggerInit.debug("- {0}: {1}".format(name, value))
start()

27
src/processor/__init__.py
View file

@ -1,5 +1,6 @@
import base64
import socket
from typing import Final
import docker
import yaml
@ -8,9 +9,15 @@ from kubernetes.client.rest import ApiException
from easymapping import HaproxyConfigGenerator
from functions import Functions, Consts, ContainerEnv
from functions import loggerEasyHaproxy
class ProcessorInterface:
STATIC: Final[str] = "static"
DOCKER: Final[str] = "docker"
SWARM: Final[str] = "swarm"
KUBERNETES: Final[str] = "kubernetes"
static_file = Consts.easyhaproxy_config
def __init__(self, filename=None):
@ -27,17 +34,16 @@ class ProcessorInterface:
@staticmethod
def factory(mode):
if mode == "static":
if mode == ProcessorInterface.STATIC:
return Static(ProcessorInterface.static_file)
elif mode == "docker":
elif mode == ProcessorInterface.DOCKER:
return Docker()
elif mode == "swarm":
elif mode == ProcessorInterface.SWARM:
return Swarm()
elif mode == "kubernetes":
elif mode == ProcessorInterface.KUBERNETES:
return Kubernetes()
else:
Functions.log("EASYHAPROXY", Functions.FATAL,
"Expected mode to be 'static', 'docker', 'swarm' or 'kubernetes'. I got '%s'" % mode)
loggerEasyHaproxy.fatal("Expected mode to be 'static', 'docker', 'swarm' or 'kubernetes'. I got '%s'" % mode)
return None
def refresh(self):
@ -244,11 +250,9 @@ class Kubernetes(ProcessorInterface):
ssl_hosts.extend(tls.hosts)
except Exception as e:
Functions.log("EASYHAPROXY", Functions.WARN,
"Ingress %s - Get secret failed: '%s'" % (ingress_name, e))
loggerEasyHaproxy.warn("Ingress %s - Get secret failed: '%s'" % (ingress_name, e))
Functions.log("EASYHAPROXY", Functions.TRACE,
"Ingress %s - SSL Hosts found '%s'" % (ingress_name, ssl_hosts))
loggerEasyHaproxy.debug("Ingress %s - SSL Hosts found '%s'" % (ingress_name, ssl_hosts))
for rule in ingress.spec.rules:
rule_data = {}
@ -275,8 +279,7 @@ class Kubernetes(ProcessorInterface):
cluster_ip = api_response.spec.cluster_ip
except ApiException as e:
cluster_ip = None
Functions.log("EASYHAPROXY", Functions.WARN,
"Ingress %s - Service %s - Failed: '%s'" % (ingress_name, service_name, e))
loggerEasyHaproxy.warn("Ingress %s - Service %s - Failed: '%s'" % (ingress_name, service_name, e))
if cluster_ip is not None:
if cluster_ip not in self.parsed_object.keys():

14
src/tests/test_daemonize.py
View file

@ -14,15 +14,15 @@ def test_daemonize_haproxy_check_config():
def test_daemonize_haproxy_get_haproxy_command_start():
daemon = DaemonizeHAProxy()
command = daemon.get_haproxy_command("start")
command = daemon.get_haproxy_command(DaemonizeHAProxy.HAPROXY_START)
assert command == "/usr/sbin/haproxy -W -f /etc/haproxy/haproxy.cfg -p /run/haproxy.pid -S /var/run/haproxy.sock"
def test_daemonize_haproxy_get_haproxy_command_reload():
daemon = DaemonizeHAProxy()
command = daemon.get_haproxy_command("reload")
command = daemon.get_haproxy_command(DaemonizeHAProxy.HAPROXY_RELOAD)
assert command == "/usr/sbin/haproxy -W -f /etc/haproxy/haproxy.cfg -p /run/haproxy.pid -x /var/run/haproxy.sock -sf "
def test_daemonize_haproxy_check_config():
def test_daemonize_haproxy2_check_config():
daemon = DaemonizeHAProxy(os.path.abspath(os.path.dirname(__file__)) + '/fixtures')
filed = daemon.get_custom_config_files()
assert filed == {
@ -30,19 +30,19 @@ def test_daemonize_haproxy_check_config():
os.path.dirname(__file__) + "/fixtures/10_haproxy.cfg": os.path.getmtime(os.path.dirname(__file__) + "/fixtures/10_haproxy.cfg")
}
def test_daemonize_haproxy_get_haproxy_command_start():
def test_daemonize_haproxy2_get_haproxy_command_start():
daemon = DaemonizeHAProxy(os.path.abspath(os.path.dirname(__file__)) + '/fixtures')
command = daemon.get_haproxy_command("start")
command = daemon.get_haproxy_command(DaemonizeHAProxy.HAPROXY_START)
assert command == "/usr/sbin/haproxy -W -f /etc/haproxy/haproxy.cfg -f %s -p /run/haproxy.pid -S /var/run/haproxy.sock" % (os.path.dirname(__file__) + "/fixtures")
def test_daemonize_haproxy_get_haproxy_command_reload():
def test_daemonize_haproxy2_get_haproxy_command_reload():
tmp_pid_file = "/tmp/tmp_pid.txt"
Functions.save(tmp_pid_file, "10")
try:
daemon = DaemonizeHAProxy(os.path.abspath(os.path.dirname(__file__)) + '/fixtures')
command = daemon.get_haproxy_command("reload", tmp_pid_file)
command = daemon.get_haproxy_command(DaemonizeHAProxy.HAPROXY_RELOAD, tmp_pid_file)
assert command == "/usr/sbin/haproxy -W -f /etc/haproxy/haproxy.cfg -f %s -p %s -x /var/run/haproxy.sock -sf %s" % (os.path.dirname(__file__) + "/fixtures", tmp_pid_file, 10)
finally:
os.remove(tmp_pid_file)

2
src/tests/test_docker.py
View file

@ -64,7 +64,7 @@ def test_processor_docker():
os.environ['EASYHAPROXY_CERTBOT_EMAIL'] = 'docker@example.org'
static = ProcessorInterface.factory("docker")
static = ProcessorInterface.factory(ProcessorInterface.DOCKER)
assert static.get_certbot_hosts() is None
assert {

120
src/tests/test_functions.py
View file

@ -1,26 +1,37 @@
import logging
import os
import random
import re
import string
from logging import Logger
from functions import Functions
from functions import Functions, loggerEasyHaproxy, loggerCertbot, loggerHaproxy
from io import StringIO
log_stream = StringIO() # Create StringIO object
log_handler = logging.StreamHandler(log_stream)
log_formatter = logging.Formatter('%(levelname)s - %(message)s')
log_handler.setFormatter(log_formatter)
loggerDebug = logging.getLogger(__name__)
loggerDebug.setLevel(logging.DEBUG)
loggerDebug.addHandler(log_handler)
def test_functions_check_local_level():
assert Functions.skip_log('CERTBOT', Functions.INFO) == False
assert Functions.skip_log('HAPOROXY', Functions.INFO) == False
assert Functions.skip_log('EASYHAPROXY', Functions.INFO) == False
assert Functions.setup_log(loggerCertbot) == logging.INFO
assert Functions.setup_log(loggerHaproxy) == logging.INFO
assert Functions.setup_log(loggerEasyHaproxy) == logging.INFO
os.environ['CERTBOT_LOG_LEVEL'] = 'warn'
assert Functions.skip_log('CERTBOT', Functions.INFO) == True
assert Functions.setup_log(loggerCertbot) == logging.WARNING
del os.environ['CERTBOT_LOG_LEVEL']
os.environ['HAPROXY_LOG_LEVEL'] = 'warn'
assert Functions.skip_log('HAPROXY', Functions.INFO) == True
assert Functions.setup_log(loggerHaproxy) == logging.WARNING
del os.environ['HAPROXY_LOG_LEVEL']
os.environ['EASYHAPROXY_LOG_LEVEL'] = 'warn'
assert Functions.skip_log('EASYHAPROXY', Functions.INFO) == True
assert Functions.setup_log(loggerEasyHaproxy) == logging.WARNING
del os.environ['EASYHAPROXY_LOG_LEVEL']
@ -35,122 +46,99 @@ def test_function_load_and_save():
finally:
os.unlink(filename)
def test_functions_check_log_sanity():
print()
Functions.log(Functions.EASYHAPROXY_LOG, Functions.INFO, "Test 1")
assert Functions.debug_log is None
Functions.debug_log = []
try:
Functions.log(Functions.EASYHAPROXY_LOG, Functions.INFO, "Test 2")
assert len(Functions.debug_log) == 1
assert re.match("\[EASYHAPROXY\] .* \[INFO\]: Test 2", Functions.debug_log[0])
os.environ['CERTBOT_LOG_LEVEL'] = 'DEBUG'
Functions.log(Functions.EASYHAPROXY_LOG, Functions.INFO, "Test 3")
assert len(Functions.debug_log) == 2
assert re.match("\[EASYHAPROXY\] .* \[INFO\]: Test 3", Functions.debug_log[1])
os.environ['EASYHAPROXY_LOG_LEVEL'] = 'warn'
Functions.log(Functions.EASYHAPROXY_LOG, Functions.INFO, "Test 4") # Should not log to debug
assert len(Functions.debug_log) == 2
finally:
del os.environ['EASYHAPROXY_LOG_LEVEL']
Functions.debug_log = None
def test_functions_run_bash_log_output():
print()
Functions.debug_log = []
try:
return_code, result = Functions.run_bash(Functions.EASYHAPROXY_LOG, "echo 'test run 1'", log_output=True,
return_code, result = Functions.run_bash(loggerDebug, "echo 'test run 1'", log_output=True,
return_result=False)
assert return_code == 0
assert result == []
assert len(Functions.debug_log) == 1
assert re.match("\[EASYHAPROXY\] .* \[INFO\]: test run 1", Functions.debug_log[0])
log_value = log_stream.getvalue()
assert len(log_value) > 0
assert log_value == "INFO - test run 1\n"
finally:
Functions.debug_log = None
log_stream.truncate(0)
def test_functions_run_bash_no_log_output():
print()
Functions.debug_log = []
try:
return_code, result = Functions.run_bash(Functions.EASYHAPROXY_LOG, "echo 'test run 2'", log_output=False,
return_code, result = Functions.run_bash(loggerDebug, "echo 'test run 2'", log_output=False,
return_result=False)
assert return_code == 0
assert result == []
assert len(Functions.debug_log) == 0
assert len(log_stream.getvalue()) == 0
finally:
Functions.debug_log = None
log_stream.truncate(0)
def test_functions_run_bash_return():
print()
Functions.debug_log = []
try:
return_code, result = Functions.run_bash(Functions.EASYHAPROXY_LOG, "echo 'test run 3'", log_output=False,
return_code, result = Functions.run_bash(loggerDebug, "echo 'test run 3'", log_output=False,
return_result=True)
assert return_code == 0
assert len(Functions.debug_log) == 0
assert len(log_stream.getvalue()) == 0
assert "".join(result) == 'test run 3'
finally:
Functions.debug_log = None
log_stream.truncate(0)
def test_functions_run_bash_log_and_return_output():
print()
Functions.debug_log = []
try:
return_code, result = Functions.run_bash(Functions.EASYHAPROXY_LOG, "echo 'test run 4'", log_output=True, return_result=True)
return_code, result = Functions.run_bash(loggerDebug, "echo 'test run 4'",
log_output=True,
return_result=True)
assert return_code == 0
assert "".join(result) == 'test run 4'
assert len(Functions.debug_log) == 1
assert re.match("\[EASYHAPROXY\] .* \[INFO\]: test run 4", Functions.debug_log[0])
log_value = log_stream.getvalue().strip("\x00")
assert len(log_value) > 0
assert log_value == "INFO - test run 4\n"
finally:
Functions.debug_log = None
log_stream.truncate(0)
def test_functions_run_bash_ok():
print()
Functions.debug_log = []
try:
return_code, result = Functions.run_bash(Functions.EASYHAPROXY_LOG, "%s/fixtures/run_bash.sh" % os.path.dirname(__file__), log_output=True,
return_code, result = Functions.run_bash(loggerDebug, "%s/fixtures/run_bash.sh" % os.path.dirname(__file__),
log_output=True,
return_result=False)
assert return_code == 0
assert result == []
assert len(Functions.debug_log) == 1
assert re.match("\[EASYHAPROXY\] .* \[INFO\]: Processing run_bash.sh", Functions.debug_log[0])
log_value = log_stream.getvalue().strip("\x00")
assert len(log_value) > 1
assert log_value == "INFO - Processing run_bash.sh\n"
finally:
Functions.debug_log = None
log_stream.truncate(0)
def test_functions_run_bash_fail():
print()
Functions.debug_log = []
try:
return_code, result = Functions.run_bash(Functions.EASYHAPROXY_LOG, "%s/fixtures/run_bash.sh 15" % os.path.dirname(__file__), log_output=True,
return_code, result = Functions.run_bash(loggerDebug, "%s/fixtures/run_bash.sh 15" % os.path.dirname(__file__),
log_output=True,
return_result=False)
assert return_code == 15
assert result == []
assert len(Functions.debug_log) == 1
assert re.match("\[EASYHAPROXY\] .* \[INFO\]: Processing run_bash.sh", Functions.debug_log[0])
log_value = log_stream.getvalue().strip("\x00")
assert len(log_value) > 0
assert log_value == "INFO - Processing run_bash.sh\n"
finally:
Functions.debug_log = None
log_stream.truncate(0)
def test_functions_run_command_not_found():
print()
Functions.debug_log = []
try:
return_code, result = Functions.run_bash(Functions.EASYHAPROXY_LOG, "no_command_here", log_output=True,
return_code, result = Functions.run_bash(loggerDebug, "no_command_here",
log_output=True,
return_result=False)
assert return_code == -99
assert str(result) == "[Errno 2] No such file or directory: 'no_command_here'"
assert len(Functions.debug_log) == 1
assert re.match("\[EASYHAPROXY\] .* \[ERROR\]: \[Errno 2\] No such file or directory: 'no_command_here'", Functions.debug_log[0])
log_value = log_stream.getvalue().strip("\x00")
assert len(log_value) > 0
assert log_value == "ERROR - [Errno 2] No such file or directory: 'no_command_here'\n"
finally:
Functions.debug_log = None
log_stream.truncate(0)

2
src/tests/test_static.py
View file

@ -6,7 +6,7 @@ from processor import ProcessorInterface
def test_processor_static():
ProcessorInterface.static_file = os.path.join(os.path.dirname(os.path.realpath(__file__)), "./fixtures/static.yml")
static = ProcessorInterface.factory("static")
static = ProcessorInterface.factory(ProcessorInterface.STATIC)
parsed_object = [
{