Implement support to report error,warning logs after every test case run.
Change-Id: Id5710e9ee3497b4bc1030d7fa69c182d2e1348c6
diff --git a/src/test/utils/CordLogger.py b/src/test/utils/CordLogger.py
new file mode 100644
index 0000000..2d9242c
--- /dev/null
+++ b/src/test/utils/CordLogger.py
@@ -0,0 +1,26 @@
+from OnosLog import OnosLog
+from scapy.all import log
+import unittest
+
+class CordLogger(unittest.TestCase):
+
+ def setUp(self):
+ '''Read the log buffer'''
+ try:
+ onosLog = OnosLog()
+ st, output = onosLog.get_log()
+ onosLog.close()
+ except: pass
+
+ def tearDown(self):
+ '''Dump the log buffer for ERRORS/warnings'''
+ try:
+ onosLog = OnosLog()
+ st, output = onosLog.get_log( ('ERROR','WARN') )
+ onosLog.close()
+ if st and output:
+ log.info('\nTest %s has errors and warnings\n' %self._testMethodName)
+ log.info('%s' %output)
+ else:
+ log.info('\nTest %s has no errors and warnings in the logs' %self._testMethodName)
+ except: pass
diff --git a/src/test/utils/OnosLog.py b/src/test/utils/OnosLog.py
new file mode 100644
index 0000000..e876c67
--- /dev/null
+++ b/src/test/utils/OnosLog.py
@@ -0,0 +1,72 @@
+import os
+from SSHTestAgent import SSHTestAgent
+
+class OnosLog(object):
+ CLI_USER = 'karaf'
+ CLI_PASSWD = 'karaf'
+ CLI_PORT = 8101
+ HOST = os.getenv('ONOS_CONTROLLER_IP', '172.17.0.2')
+ last_snapshot_map = {}
+
+ def __init__(self, host = HOST):
+ self.ssh_agent = SSHTestAgent(host = host, user = self.CLI_USER,
+ password = self.CLI_PASSWD, port = self.CLI_PORT)
+ if not OnosLog.last_snapshot_map.has_key(host):
+ OnosLog.last_snapshot_map[host] = []
+
+ #should be the last function to call for tearing down the log session
+ def close(self):
+ self.ssh_agent.client.close()
+
+ @classmethod
+ def get_last_snapshot(cls, host):
+ if cls.last_snapshot_map.has_key(host):
+ return cls.last_snapshot_map[host]
+ return []
+
+ @classmethod
+ def update_last_snapshot(cls, host, res):
+ cls.last_snapshot_map[host] = res
+
+ def get_log(self, search_terms = None):
+ """Run the command on the test host"""
+ cmd = 'cat /root/onos/apache-karaf-3.0.5/data/log/karaf.log'
+ st, output = self.ssh_agent.run_cmd(cmd)
+ if st is False:
+ return output
+ last_snapshot = self.get_last_snapshot(self.ssh_agent.host)
+ lines = output.splitlines()
+ if search_terms:
+ if type(search_terms) is str:
+ terms = [ search_terms ]
+ else:
+ terms = search_terms
+ match_lines = []
+ last_len = len(last_snapshot)
+ for i in xrange(0, len(lines)):
+ if i < last_len and lines[i] in last_snapshot:
+ ##skip lines matching the last snapshot
+ continue
+ for t in terms:
+ if lines[i].find(t) >= 0:
+ match_lines.append(lines[i])
+
+ output = '\n'.join(match_lines)
+
+ #update the last snapshot
+ self.update_last_snapshot(self.ssh_agent.host, lines)
+ return st, output
+
+if __name__ == '__main__':
+ onos = os.getenv('ONOS_CONTROLLER_IP', '172.17.0.2')
+ onos_log = OnosLog(host = onos)
+ print('Checking for INFO')
+ st, output = onos_log.get_log('INFO')
+ print(st, output)
+ print('\n\nChecking for ERROR\n\n')
+ st, output = onos_log.get_log('ERROR')
+ print(st, output)
+ print('Checking for ERROR and INFO')
+ st, output = onos_log.get_log(('ERROR', 'INFO'))
+ print(st, output)
+ onos_log.close()
diff --git a/src/test/utils/SSHTestAgent.py b/src/test/utils/SSHTestAgent.py
index 6e89ffb..12d3530 100644
--- a/src/test/utils/SSHTestAgent.py
+++ b/src/test/utils/SSHTestAgent.py
@@ -9,25 +9,29 @@
user = 'ubuntu'
password = None
- def __init__(self, user = user, password = password):
+ def __init__(self, host = host, user = user, password = password, port = 22):
+ self.host = host
self.user = user
self.password = password
+ self.port = port
self.client = SSHClient()
- self.client.load_system_host_keys()
self.client.set_missing_host_key_policy(AutoAddPolicy())
def run_cmd(self, cmd, timeout = 5):
"""Run the command on the test host"""
try:
self.client.connect(self.host, username = self.user, password = self.password,
- key_filename = self.key_file, timeout=timeout)
+ key_filename = self.key_file, timeout=timeout, port = self.port)
except:
log.error('Unable to connect to test host %s' %self.host)
return False, None
channel = self.client.get_transport().open_session()
channel.exec_command(cmd)
- status = channel.recv_exit_status()
+ if channel.exit_status_ready():
+ status = channel.recv_exit_status()
+ else:
+ status = 0
output = None
st = status == 0
if st: