diff --git a/pytest_cafy/logupdate.py b/pytest_cafy/logupdate.py new file mode 100644 index 0000000..38ff7d2 --- /dev/null +++ b/pytest_cafy/logupdate.py @@ -0,0 +1,409 @@ +import re +import sys +from jinja2 import Template + +class HtmlWriter: + def tag(self,tag,data,klass=None,**kwargs): + optstr = "" + if klass is not None: + optstr = "class='{klass}'".format(klass=klass) + for key,value in kwargs.items(): + optstr = optstr + "{key}={value}".format(key=key,value=value) + print("<{tag} {optstr}>{data}".format(tag=tag,optstr=optstr,data=data)) + + def span(self,data,**kwargs): + return self.tag("span",data,**kwargs) + + def div(self,data,**kwargs): + return self.tag("div",data,**kwargs) +htmlwriter = HtmlWriter() + + +print(""" + + + + + + + + + + + all.log + + + + + + + +
+
+
+

+ + + + + + +""") + + +class LogParser: + class LogObject: + def __init__(self, timestamp, message, thread, logger): + self.data = { + 'timestamp' : timestamp, + 'message' : message, + 'thread' : thread, + 'etype' : self.__class__.__name__.lower(), + 'logger' : logger + } + + def xaccept(self): + pass + def accept(self): + print(""" +
+
{etype}
+
{timestamp}
+
{logger}
+
{message}
+
+ """.format(**self.data)) + + class Info(LogObject): + pass + + class Warning(LogObject): + pass + + class Debug(LogObject): + pass + + class Out(LogObject): + def accept(self): + print(""" +
+
{etype}
+
{timestamp}
+
{logger}
+
+
+ """.format(**self.data)) + + + + class Error(LogObject): + pass + + class LogList(LogObject): + def __init__(self,name): + self.logs = list() + + def add(self,obj): + self.logs.append(obj) + + def accept(self): + pass + + class ThreadLog(LogList): + def __init__(self,name): + self.thread = name + self.thread_logs = dict() + self.id = 1 + + def add(self,thread, obj): + if thread not in self.thread_logs: + self.thread_logs[thread]=list() + self.thread_logs[thread].append(obj) + + + def accept(self): + self.id = self.id+1 + thread_id = 0 + thread_id_lookup = dict() + + print(""" +
+
+ Threaded logs begin from here +
+
+ """) + + print("""
""".format(id=self.id)) + extra = "show active" + for thread, logs in self.thread_logs.items(): + print(""" +
+ """.format(id=self.id,thread=self.thread,thread_id=thread_id_lookup[thread],extra = extra)) + extra = "" + for log in logs: + log.accept() + print("""
""") + + + print("""
""") + print("""""") + print("""
""") + + + + + + + class TestcaseLog(LogList): + TID = 1 + def __init__(self,testcase): + self.testcase = testcase + self.logs = list() + self.id = LogParser.TestcaseLog.TID + if self.id == 1: + self.show = "show" + self.expanded = "true" + else: + self.show = "" + self.expanded = "false" + LogParser.TestcaseLog.TID = LogParser.TestcaseLog.TID + 1 + + + + def accept(self): + + + print(""" +
+
+
+ +
+
+ +
+
+ + + """.format(id=self.id, testcase=self.testcase, show=self.show, expanded=self.expanded)) + for log in self.logs: + + log.accept() + + print("""
+
+
""") + + MatchClass = { + 'Info': Info, + 'Warning': Warning, + 'Debug' : Debug, + 'Error' : Error, + 'Out' : Out + } + + def __init__(self,logfile): + self.logfile = logfile + self.id = 1 + self.testcases = list() + + def bootstrap_begin_accordion(self): + print("""
""") + + def bootstrap_end_accordion(self): + print("
") + + + + + + def parse(self): + f = open(self.logfile, 'r') + last_output = None + logsep = re.compile("-(\w+)\-*?(\d{4}-\d{2}-\d{2}T\d{2}-\d{2}-\d{2})\[(.*?)\]\[(.*?)\]>(.*$)") + current_tag = None + current_testcase = self.TestcaseLog("setup_module") + thread_logger = None + output = "" + self.bootstrap_begin_accordion() + + for line in f: + result = logsep.match(line) + if result: + if current_tag is not None: + klass = LogParser.MatchClass.get(current_tag, None) + if klass is not None: + data = klass(timestamp=timestamp, message=output, thread=thread, logger=logger) + if thread_logger is not None: + thread_logger.add(thread,data) + else: + current_testcase.add(data) + + current_tag = result.group(1) + timestamp = result.group(2) + logger = result.group(4) + thread = result.group(3) + message = result.group(5) + + #if thread != "MainThread": + #if thread_logger is not None None: + # thread_logger = self.ThreadLog(thread) + # current_testcase.add(thread_logger) + + if 'BEGIN THREAD' in message or 'verification(parallel) starts' in message : + thread_logger = self.ThreadLog(thread) + current_testcase.add(thread_logger) + + if 'END THREAD' in message or 'verification(parallel) ends' in message : + thread_logger = None + + + + tcmatch = re.compile("\s*Start test\:\s+(.*)\s*$") + if current_tag == "Title": + testcase = tcmatch.match(message) + if testcase: + # print("Found: %s" % testcase.group(1)) + #import ipdb; + #ipdb.set_trace() + + if current_testcase is not None: + current_testcase.accept() + current_testcase = self.TestcaseLog(testcase.group(1)) + + output = "
{line}
".format(line=result.group(5).strip()) + else: + output = output + "
{line}
".format(line=line.strip()) + f.close() + if current_testcase is not None: + current_testcase.accept() + self.bootstrap_end_accordion() + +logparser = LogParser(sys.argv[1]) +print("""
""") + +logparser.parse() +print("""
""") + +mapButtons = { + 'info' : 'info', + 'warning' : 'warning', + 'out' : 'secondary', + 'error': 'danger', + 'fail' : 'danger', + 'success' : 'success', + 'debug' : 'primary' + +} + +navbar = Template(""" +
+
+
+ + + + + + + + + + + +""") + + +print(navbar.render(mapButtons=mapButtons)) + +""" +last_testclass = None +for tc in logparser.testcases: + id = tc['id'] + testcase = tc['testcase'] + tci = testcase.split(".") + if len(tci) > 1: + testclass = tci[0] + testname = tci[1] + else: + testclass = "*" + testname = tci[0] + + if testclass != last_testclass: + + {testclass}.format(id=id,testclass=testclass)) + last_testclass = testclass +       {testcase}.format(id=id,testcase=testname)) + # + + +""" diff --git a/pytest_cafy/plugin.py b/pytest_cafy/plugin.py index 74ff9c3..247b7a7 100755 --- a/pytest_cafy/plugin.py +++ b/pytest_cafy/plugin.py @@ -1316,6 +1316,7 @@ def invoke_rc_on_failed_testcase(self, params, headers): def pytest_terminal_summary(self, terminalreporter): '''this hook is the execution point of email plugin''' self._generate_email_report(terminalreporter) + self._generate_all_star_log_html() self._generate_all_log_html() if self.CAFY_REPO: option = terminalreporter.config.option @@ -1381,6 +1382,16 @@ def _parse_all_log(self, input_file_handler): if processed_log_line: log_grouping.append_log_line(LogLine(html.escape(processed_log_line.group(2)), processed_log_line.group(1).upper())) return all_log_groupings + + def _generate_all_star_log_html(self): + log_file_name = os.path.join(CafyLog.work_dir, "all.log") + try: + output_file_name = os.path.join(CafyLog.work_dir, "star.log.html") + log_update_file = "logupdate.py" + cmd = "python {log_update_file} {input_file} > {output_file}".format(log_update_file = log_update_file, input_file = log_file_name, output_file = output_file_name) + subprocess.Popen(cmd, shell=True) + except FileNotFoundError: + return def _generate_all_log_html(self): log_file_name = os.path.join(CafyLog.work_dir, "all.log")