tests/dcaegen2-pmmapper/files-processing-config-pmmapper/libraries/LogReader.py

   1 import re
   2
   3 class LogReader:
   4
   5   def filter_unique(self, merged_logs_output, testname):
   6     logs = merged_logs_output.splitlines()
   7     del_logs = list(filter(lambda line: "|DEL|" in line, logs))
   8     nrs_set = set()
   9     ret_logs = set()
  10     for log in del_logs:
  11       filename = re.findall(testname + "\d", log)
  12       if len(filename) > 0 and filename[0] not in nrs_set:
  13         ret_logs.add(log)
  14         nrs_set.add(filename[0])
  15     return ret_logs
  16
  17   def get_number_of_dropped_messages(self, logs_output):
  18     return len(list(filter(lambda line: "|429|" in line, logs_output)))
  19
  20   def get_log_files_list(self, fileNames):
  21     files = fileNames.split()
  22     return files