KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use logline method in tox

Best Python code snippet using tox_python

logs.py

Source:logs.py

1import datetime2import re3BUFFER_BEGIN = re.compile("^--------- beginning of (.*)$")4BUFFER_SWITCH = re.compile("^--------- switch to (.*)$")5HEADER = re.compile("^\\[ (\\d\\d-\\d\\d \\d\\d:\\d\\d:\\d\\d.\\d\\d\\d) +(.+?): *(\\d+): *(\\d+) *([EWIDV])/(.*?) *\\]$")6HEADER_TYPE2 = re.compile("^(\\d\\d-\\d\\d \\d\\d:\\d\\d:\\d\\d.\\d\\d\\d) *(\\d+) *(\\d+) *([EWIDV]) ([^ :]*?): (.*?)$")7CHATTY_IDENTICAL = re.compile("^.* identical (\\d+) lines$")8STATE_BEGIN = 09STATE_BUFFER = 110STATE_HEADER = 211STATE_TEXT = 312STATE_BLANK = 413class LogLine(object):14  """Represents a line of android logs."""15  def __init__(self, buf=None, timestamp=None, uid=None, pid=None, tid=None, level=None,16      tag=None, text=""):17    self.buf = buf18    self.timestamp = timestamp19    self.uid = uid20    self.pid = pid21    self.tid = tid22    self.level = level23    self.tag = tag24    self.text = text25    self.process = None26  def __str__(self):27    return "{%s} {%s} {%s} {%s} {%s} {%s}/{%s}: {%s}" % (self.buf, self.timestamp, self.uid,28        self.pid, self.tid, self.level, self.tag, self.text)29  def __eq__(self, other):30      return (31            self.buf == other.buf32            and self.timestamp == other.timestamp 33            and self.uid == other.uid 34            and self.pid == other.pid 35            and self.tid == other.tid 36            and self.level == other.level 37            and self.tag == other.tag 38            and self.text == other.text 39          )40  def clone(self):41    logLine = LogLine(self.buf, self.timestamp, self.uid, self.pid, self.tid, self.level,42        self.tag, self.text)43    logLine.process = self.process44    return logLine45  def memory(self):46    """Return an estimate of how much memory is used for the log.47      32 bytes of header + 8 bytes for the pointer + the length of the tag and the text.48      This ignores the overhead of the list of log lines itself."""49    return 32 + 8 + len(self.tag) + 1 + len(self.text) + 150def ParseLogcat(f, processes, duration=None):51  previous = None52  for logLine in ParseLogcatInner(f, processes, duration):53    if logLine.tag == "chatty" and logLine.level == "I":54      m = CHATTY_IDENTICAL.match(logLine.text)55      if m:56        for i in range(int(m.group(1))):57          clone = previous.clone()58          clone.timestamp = logLine.timestamp59          yield clone60        continue61    previous = logLine62    yield logLine63def ParseLogcatInner(f, processes, duration=None):64  """Parses a file object containing log text and returns a list of LogLine objects."""65  result = []66  buf = None67  timestamp = None68  uid = None69  pid = None70  tid = None71  level = None72  tag = None73  state = STATE_BEGIN74  logLine = None75  previous = None76  if duration:77    endTime = datetime.datetime.now() + datetime.timedelta(seconds=duration)78  # TODO: use a nonblocking / timeout read so we stop if there are79  # no logs coming out (haha joke, right!)80  for line in f:81    if duration and endTime <= datetime.datetime.now():82      break83    if len(line) > 0 and line[-1] == '\n':84      line = line[0:-1]85    m = BUFFER_BEGIN.match(line)86    if m:87      if logLine:88        yield logLine89        logLine = None90      buf = m.group(1)91      state = STATE_BUFFER92      continue93    m = BUFFER_SWITCH.match(line)94    if m:95      if logLine:96        yield logLine97        logLine = None98      buf = m.group(1)99      state = STATE_BUFFER100      continue101    m = HEADER.match(line)102    if m:103      if logLine:104        yield logLine105      logLine = LogLine(106            buf=buf,107            timestamp=m.group(1),108            uid=m.group(2),109            pid=m.group(3),110            tid=m.group(4),111            level=m.group(5),112            tag=m.group(6)113          )114      previous = logLine115      logLine.process = processes.FindPid(logLine.pid, logLine.uid)116      state = STATE_HEADER117      continue118    m = HEADER_TYPE2.match(line)119    if m:120      if logLine:121        yield logLine122      logLine = LogLine(123            buf=buf,124            timestamp=m.group(1),125            uid="0",126            pid=m.group(2),127            tid=m.group(3),128            level=m.group(4),129            tag=m.group(5),130            text=m.group(6)131          )132      previous = logLine133      logLine.process = processes.FindPid(logLine.pid, logLine.uid)134      state = STATE_BEGIN135      continue136    if not len(line):137      if state == STATE_BLANK:138        if logLine:139          logLine.text += "\n"140      state = STATE_BLANK141      continue142    if logLine:143      if state == STATE_HEADER:144        logLine.text += line145      elif state == STATE_TEXT:146        logLine.text += "\n"147        logLine.text += line148      elif state == STATE_BLANK:149        if len(logLine.text):150          logLine.text += "\n"151        logLine.text += "\n"152        logLine.text += line153    state = STATE_TEXT154  if logLine:155    yield logLine...

monitoring.py

Source:monitoring.py

1import time2import os3def main():4    # Set the filename and open the file5    filename = 'collect.log'6    file = open(filename, 'r')7    ct_dst_sport_ltm_list = []8    ct_dst_src_ltm_list = []9    ct_src_dport_ltm_list = []10    ct_dst_ltm_list = []11    ct_dst_sport_ltm_hash = 012    ct_dst_src_ltm_hash = 013    ct_src_dport_ltm_hash = 014    ct_dst_ltm_hash = 015    ct_state_ttl = 016    for i in range(100):17        ct_dst_sport_ltm_list.append(0)18        ct_dst_src_ltm_list.append(0)19        ct_src_dport_ltm_list.append(0)20        ct_dst_ltm_list.append(0)21    while 1:22        where = file.tell()23        line = file.readline()24        if not line:25            time.sleep(1)26            file.seek(where)27        else:28            logline = line[:-2]29            logline = logline.split(",")30            logline = removeNull(logline)31            #print(logline)32            ct_dst_sport_ltm_hash = hash(logline[1] + logline[2])33            ct_dst_sport_ltm_list.pop()34            ct_dst_sport_ltm_list.insert(0, ct_dst_sport_ltm_hash)35            ct_dst_src_ltm_hash = hash(logline[0] + logline[1])36            ct_dst_src_ltm_list.pop()37            ct_dst_src_ltm_list.insert(0, ct_dst_src_ltm_hash)38            ct_src_dport_ltm_hash = hash(logline[0] + logline[3])39            ct_src_dport_ltm_list.pop()40            ct_src_dport_ltm_list.insert(0, ct_src_dport_ltm_hash)41            ct_dst_ltm_hash = hash(logline[1])42            ct_dst_ltm_list.pop()43            ct_dst_ltm_list.insert(0, ct_dst_ltm_hash)44            ct_state_ttl = 045            if (logline[4] == 62 or logline[4] == 63 or logline[4] == 254 or logline[4] == 255) and (logline[5] == 252 or logline[5] == 253) and logline[6] == "FIN":46                ct_state_ttl = 147            elif (logline[4] == 0 or logline[4] == 62 or logline[4] == 254) and (logline[5] == 0) and logline[6] == "INT":48                ct_state_ttl = 249            elif(logline[4] == 62 or logline[4] == 254) and (logline[5] == 60 or logline[5] == 252 or logline[5] == 253) and logline[6] == "CON":50                ct_state_ttl = 351                52            elif(logline[4] == 254) and (logline[5] == 252) and logline[6] == "ACC":53                ct_state_ttl = 454                55            elif(logline[4] == 254) and (logline[5] == 252) and logline[6] == "CLO":56                ct_state_ttl = 557                58            elif(logline[4] == 254) and (logline[5] == 0) and logline[6] == "REQ":59                ct_state_ttl = 760                61            else:62                ct_state_ttl = 063            for i in range(7):64                logline.pop(0)65            dataline = logline + [str(ct_dst_ltm_list.count(ct_dst_ltm_hash))] + [str(ct_src_dport_ltm_list.count(ct_src_dport_ltm_hash))] + [66                str(ct_dst_sport_ltm_list.count(ct_dst_sport_ltm_hash))] + [str(ct_dst_src_ltm_list.count(ct_dst_src_ltm_hash))] + [str(ct_state_ttl)]67            #print("new data: ")68            #print(" ".join(dataline))  # already has newline69            write_data(dataline)70def write_data(line_data):71    with open("dataset.log", mode='a') as file:72        file.write(",".join(line_data) + "\n")73def removeNull(row):74    return ['0' if i=='' else i for i in row]75if __name__ == "__main__":...

test_signals.py

Source:test_signals.py

1from django.test import TestCase2from apps.store.models import Book3from apps.core.models import LogLine4class CreateLogLineTests(TestCase):5    def test_create_logline(self):6        self.assertEqual(Book.objects.count(), 0)7        self.assertEqual(LogLine.objects.count(), 0)8        self.book = Book.objects.create(9            book_title='Test book1',10            authors_info='Test author1',11            ISBN='111-1111111111',12            price=40,13        )14        self.assertEqual(Book.objects.count(), 1)15        self.assertEqual(LogLine.objects.count(), 1)16        self.assertEqual(LogLine.objects.first().logging_type, LogLine.CREATE)17        self.assertEqual(LogLine.objects.first().logging_text, f"New Book {self.book} is created!")18        self.book.book_title='Updated Test book'19        self.book.save()...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.