You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
210 lines
8.8 KiB
210 lines
8.8 KiB
import unittest
|
|
import os
|
|
import inspect
|
|
import tools.path_tool
|
|
import basic.program
|
|
import test.constants
|
|
import basic.constants as B
|
|
import basic.componentHandling
|
|
import tools.file_tool
|
|
import basic.message
|
|
|
|
HOME_PATH = test.constants.HOME_PATH
|
|
PYTHON_CMD = "python"
|
|
TEST_FUNCTIONS = ["test_00init", "test_04logLevel", "test_05loginigArgs", "test_10set", "test_11log", "test_20close"]
|
|
#TEST_FUNCTIONS = ["test_00init", "test_04logLevel"]
|
|
|
|
PROGRAM_NAME = "clean_workspace"
|
|
NOT_TO_LOG = ["xx1xx", "xx4xx"]
|
|
TO_LOG = ["yy1yy", "yy2yy", "yy3yy", "yy4yy", "yy5yy", "yy6yy", "xx0xx", "xx8xx"]
|
|
NOT_TO_DEBUG = ["xx4xx", "xx1xx", "xx1xx", "xx1xx", "xx1xx", "xx1xx", "xx1xx"]
|
|
TO_DEBUG = ["xx9xx", "xx22xx", "xx35xx"]
|
|
class MyTestCase(unittest.TestCase):
|
|
mymsg = "--------------------------------------------------------------"
|
|
|
|
def test_00init(self):
|
|
global mymsg
|
|
global msgObject
|
|
global jobObject
|
|
actfunction = str(inspect.currentframe().f_code.co_name)
|
|
cnttest = 0
|
|
if actfunction not in TEST_FUNCTIONS:
|
|
return
|
|
# simple job instantiate - without parameter and only simple messaging
|
|
args = {}
|
|
args["par"] = {}
|
|
args["par"]["mode"] = "unit"
|
|
args["par"]["wsdir"] = os.path.join(B.HOME_PATH, "workspace")
|
|
job = basic.program.SimpleJob(PROGRAM_NAME, "unittest", args)
|
|
jobObject = job
|
|
print(str(job.__dict__))
|
|
tlogTime = "20220101_123456"
|
|
# temporary Message - a simple implementation at initialization of the job
|
|
tmsg = basic.message.TempMessage(job, tlogTime)
|
|
print(str(tmsg.__dict__))
|
|
self.checkSimpleMessage(tmsg, "debug", tlogTime)
|
|
cnttest += 3
|
|
self.assertEqual(basic.message.LIMIT_DEBUG, getattr(tmsg, "level"))
|
|
setattr(job, "m", tmsg)
|
|
msg = basic.message.Message(job, "trace", tlogTime, None)
|
|
setattr(job, "m", msg)
|
|
print(str(msg.__dict__))
|
|
self.checkSimpleMessage(msg, "debug", tlogTime)
|
|
cnttest += 3
|
|
self.checkSimpleMessage(msg, "log", tlogTime)
|
|
cnttest += 3
|
|
self.assertEqual(basic.message.LIMIT_TRACE, getattr(msg, "level"))
|
|
cnttest += 3
|
|
msgObject = msg #
|
|
jobObject = job
|
|
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest)
|
|
|
|
def test_04logLevel(self):
|
|
global mymsg
|
|
global msgObject
|
|
global jobObject
|
|
actfunction = str(inspect.currentframe().f_code.co_name)
|
|
cnttest = 0
|
|
if actfunction not in TEST_FUNCTIONS:
|
|
return
|
|
res = msgObject.getLogLevel()
|
|
self.assertEqual(0, res)
|
|
res = msgObject.getLogLevel("msg_tool")
|
|
self.assertEqual(0, res)
|
|
setattr(jobObject.par, "tool", "job_tool")
|
|
res = msgObject.getLogLevel("msg_tool")
|
|
self.assertEqual(0, res)
|
|
setattr(jobObject.par, "tool", "msg_tool")
|
|
res = msgObject.getLogLevel("msg_tool")
|
|
self.assertEqual(2, res)
|
|
cnttest += 4
|
|
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest)
|
|
|
|
|
|
def test_05loginigArgs(self):
|
|
global mymsg
|
|
actfunction = str(inspect.currentframe().f_code.co_name)
|
|
cnttest = 0
|
|
if actfunction not in TEST_FUNCTIONS:
|
|
return
|
|
res = msgObject.getLoggingArgs(12, 1, "text")
|
|
self.assertEqual(res["mlevel"], 12)
|
|
self.assertEqual(res["mprio"], 1)
|
|
self.assertEqual(res["mtext"], "text")
|
|
res = msgObject.getLoggingArgs(12, "text", "")
|
|
self.assertEqual(res["mlevel"], 12)
|
|
self.assertEqual(res["mprio"], 0)
|
|
self.assertEqual(res["mtext"], "text")
|
|
res = msgObject.getLoggingArgs(12, "text", "ERROR: ")
|
|
self.assertEqual(res["mlevel"], 12)
|
|
self.assertEqual(res["mprio"], 0)
|
|
self.assertEqual(res["mtext"], "ERROR: text")
|
|
cnttest += 9
|
|
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest)
|
|
|
|
def test_10set(self):
|
|
global mymsg
|
|
actfunction = str(inspect.currentframe().f_code.co_name)
|
|
cnttest = 0
|
|
if actfunction not in TEST_FUNCTIONS:
|
|
return
|
|
print("test_set "+str(msgObject.__dict__))
|
|
msgObject.setMsg("yy1yy result-msg")
|
|
self.assertEqual(basic.message.RC_MSG, msgObject.getFinalRc())
|
|
self.assertEqual(msgObject.topmessage, "MSG: yy1yy result-msg")
|
|
self.assertEqual(True, msgObject.isRc(basic.message.RC_MSG))
|
|
self.assertEqual(True, msgObject.isRc(basic.message.MTEXT_MSG))
|
|
self.assertEqual(True, msgObject.isRc(basic.message.MTEXT_INFO))
|
|
self.assertEqual(True, msgObject.isRc(basic.message.RC_WARN))
|
|
self.assertEqual(True, msgObject.isRc(basic.message.MTEXT_WARN))
|
|
msgObject.setWarn("yy2yy warn-msg")
|
|
self.assertEqual(basic.message.RC_WARN, msgObject.getFinalRc())
|
|
self.assertEqual("WARN: yy2yy warn-msg", msgObject.topmessage)
|
|
self.assertEqual(False, msgObject.isRc(basic.message.RC_MSG))
|
|
msgObject.setError("yy3yy error-msg")
|
|
self.assertEqual(basic.message.RC_ERROR, msgObject.getFinalRc())
|
|
self.assertEqual("ERROR: yy3yy error-msg", msgObject.topmessage)
|
|
msgObject.setWarn("yy4yy warn-msg")
|
|
self.assertEqual(basic.message.RC_ERROR, msgObject.getFinalRc())
|
|
self.assertEqual("ERROR: yy3yy error-msg", msgObject.topmessage)
|
|
msgObject.setError("yy5yy error-msg")
|
|
self.assertEqual(basic.message.RC_ERROR, msgObject.getFinalRc())
|
|
self.assertEqual("ERROR: yy3yy error-msg", msgObject.topmessage)
|
|
msgObject.setFatal("yy6yy fatal-msg")
|
|
self.assertEqual(basic.message.RC_FATAL, msgObject.getFinalRc())
|
|
self.assertEqual("FATAL: yy6yy fatal-msg", msgObject.topmessage)
|
|
cnttest += 6
|
|
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest)
|
|
|
|
|
|
def test_11log(self):
|
|
global mymsg
|
|
actfunction = str(inspect.currentframe().f_code.co_name)
|
|
cnttest = 0
|
|
if actfunction not in TEST_FUNCTIONS:
|
|
return
|
|
print("test_log "+str(msgObject.__dict__))
|
|
i = 0
|
|
# auf verschiedenen Ebenen loggen ohne und mit Weitergabe von Prio
|
|
for level in [basic.message.LIMIT_INFO, basic.message.LIMIT_DEBUG, basic.message.LIMIT_TRACE, basic.message.LIMIT_XTRACE]:
|
|
setattr(msgObject, "level", level)
|
|
msgObject.logInfo("xx"+str(i)+"xx info at level "+str(msgObject.level))
|
|
i += 1
|
|
msgObject.logDebug("xx" + str(i) + "xx debug at level " + str(msgObject.level))
|
|
i += 1
|
|
msgObject.logTrace("xx" + str(i) + "xx trace at level " + str(msgObject.level))
|
|
i += 1
|
|
msgObject.logXTrace("xx" + str(i) + "xx xtrace at level " + str(msgObject.level))
|
|
i += 1
|
|
msgObject.logInfo(2, "xx"+str(i)+"xx info++ at level "+str(msgObject.level))
|
|
i += 1
|
|
msgObject.logDebug(2, "xx" + str(i) + "xx debug++ at level " + str(msgObject.level))
|
|
i += 1
|
|
msgObject.logTrace(2, "xx" + str(i) + "xx trace++ at level " + str(msgObject.level))
|
|
i += 1
|
|
msgObject.logXTrace(2, "xx" + str(i) + "xx xtrace++ at level " + str(msgObject.level))
|
|
i += 1
|
|
msgObject.logInfo(-2, "xx"+str(i)+"xx info-- at level "+str(msgObject.level))
|
|
i += 1
|
|
msgObject.logDebug(-2, "xx" + str(i) + "xx debug-- at level " + str(msgObject.level))
|
|
i += 1
|
|
msgObject.logTrace(-2, "xx" + str(i) + "xx trace-- at level " + str(msgObject.level))
|
|
i += 1
|
|
msgObject.logXTrace(-2, "xx" + str(i) + "xx xtrace-- at level " + str(msgObject.level))
|
|
i += 1
|
|
|
|
def test_20close(self):
|
|
"""
|
|
|
|
:return:
|
|
"""
|
|
global mymsg
|
|
actfunction = str(inspect.currentframe().f_code.co_name)
|
|
cnttest = 0
|
|
if actfunction not in TEST_FUNCTIONS:
|
|
return
|
|
print("test_close "+str(msgObject.__dict__))
|
|
self.assertEqual(False, msgObject.logfile.closed)
|
|
self.assertEqual(False, msgObject.debugfile.closed)
|
|
msgObject.closeMessage()
|
|
self.assertEqual(True, msgObject.logfile.closed)
|
|
self.assertEqual(True, msgObject.debugfile.closed)
|
|
# now check what is logged
|
|
logtext = tools.file_tool.readFileText(jobObject, msgObject.logpath, None)
|
|
debugtext = tools.file_tool.readFileText(jobObject, msgObject.debugpath, None)
|
|
for x in TO_LOG:
|
|
regex = r".*" + x + ""
|
|
self.assertIn(x, logtext)
|
|
self.assertIn(x, debugtext)
|
|
|
|
def checkSimpleMessage(self, msg, prefix, logTime):
|
|
for x in [prefix+"path", prefix+"file", "level"]:
|
|
print(x)
|
|
self.assertIn(x, msg.__dict__)
|
|
self.assertIn(logTime, getattr(msg, "debugpath"))
|
|
|
|
def test_zzz(self):
|
|
print(MyTestCase.mymsg)
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|
|
|