101 lines
2.8 KiB
Python
Executable File
101 lines
2.8 KiB
Python
Executable File
import unittest
|
|
from serviceapp import service as serviceapp
|
|
import time
|
|
import config
|
|
import service
|
|
|
|
class Mock_ServiceApp_Service() :
|
|
def __init__(self, *args, **kwargs) :
|
|
self.args = args
|
|
self.kwargs = kwargs
|
|
|
|
def send_stat(self, *args, **kwargs) :
|
|
return
|
|
|
|
class Mock_Logger() :
|
|
def __init__(self) :
|
|
self.lines = []
|
|
self.info = self.log
|
|
self.warn = self.log
|
|
self.warning = self.log
|
|
self.debug = self.log
|
|
self.error = self.log
|
|
|
|
def log(self, *args, **kwargs) :
|
|
self.lines.append("{}, {}".format(args, kwargs))
|
|
print(self.lines[-1])
|
|
|
|
def Mock_Sleep(t) :
|
|
return
|
|
|
|
def Mock_Get_Healthy(*args, **kwargs) :
|
|
return 0, 1
|
|
|
|
def Mock_Distribute_Configs(*args, **kwargs) :
|
|
return True
|
|
|
|
def Mock_Is_Valid(*args, **kwargs) :
|
|
return True
|
|
|
|
def ignore_warnings(test_func):
|
|
import warnings
|
|
def do_test(self, *args, **kwargs):
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore")
|
|
test_func(self, *args, **kwargs)
|
|
return do_test
|
|
|
|
class Test_Service(unittest.TestCase) :
|
|
def setUp(self) :
|
|
self.mock_serviceapp_service = Mock_ServiceApp_Service
|
|
self.was_k = serviceapp.check_kairosdb_alert
|
|
self.was_p = serviceapp.check_prometheus_alert
|
|
self.was_service = service.service
|
|
self.was_sleep = time.sleep
|
|
self.was_get_healthy = config.get_healthy_nodes_and_index
|
|
self.was_distribute = config.distribute_configs
|
|
self.was_is_valid = config.is_valid
|
|
serviceapp.check_kairosdb_alert = self.mock_serviceapp_service
|
|
serviceapp.check_prometheus_alert = self.mock_serviceapp_service
|
|
config.get_healthy_nodes_and_index = Mock_Get_Healthy
|
|
config.distribute_configs = Mock_Distribute_Configs
|
|
config.is_valid = Mock_Is_Valid
|
|
serviceapp.sleep = Mock_Sleep
|
|
service.sleep = Mock_Sleep
|
|
time.sleep = Mock_Sleep
|
|
|
|
def tearDown(self) :
|
|
serviceapp = self.was_service
|
|
serviceapp.check_kairosdb_alert = self.was_k
|
|
serviceapp.check_prometheus_alert = self.was_p
|
|
config.get_healthy_nodes_and_index = self.was_get_healthy
|
|
config.distribute_configs = self.was_distribute
|
|
config.is_valid = self.was_is_valid
|
|
time.sleep = self.was_sleep
|
|
serviceapp.sleep = self.was_sleep
|
|
service.sleep = self.was_sleep
|
|
|
|
@ignore_warnings
|
|
def test(self) :
|
|
import service
|
|
logger = Mock_Logger()
|
|
s = service.Service(logger, 100, "HOST", {
|
|
"alert_folder": "./testdata",
|
|
"alert_routing_config": {},
|
|
})
|
|
global first
|
|
first = True
|
|
def f() :
|
|
global first
|
|
is_first = first
|
|
first = False
|
|
return is_first
|
|
def purge_stale(*args) :
|
|
return
|
|
s.is_running = f
|
|
s.purge_stale = purge_stale
|
|
s.start()
|
|
|
|
if __name__ == "__main__" :
|
|
unittest.main()
|