I wrote a little test runner in python (my first script) to run my robot framework tests. It has a console and prints whether a test suite passed or failed, but I want more detail without having to open the logs. I want the name of each test case in real time that is running and whether it passed or failed just like RIDE does. Is that the debug console? How does RIDE print the debug file to console? I use RIDE for editing and creating tests, but it get’s buggy if I run too many tests and I wanted a simpler layout. Like a slimmed down mini RIDE.
You have to develop a listener.
Or wait until RIDE have a better version 
Thank you! I’ll google how to do it and give it a try.
Ok I searched and searched for how to create a Listener that sends the debug output to the console, but cannot find or figure out how to do it. It needs to be done as each test step runs. I found results that show how to create Listeners for every other thing like time/date or create a file on test_startup and those work fine so I think I understand how to write a Listener now. Just not how to have it print the debug to console while the tests are running. Can someone please tell me how to do this? I’m not experienced enough in Python to figure this out. Thank you for any help that you can provide.
Hi Greg,
It took me a little while to figure out listeners too, the trick is to create a function with the name that matches the names listed in 4.3.3 Listener interface methods within your listener class, the next trick is to figure out which functions you need to implement in order to get the information you want.
In your case I think it’s the end_test method you want to implement, so something like this as a starting point:
yourListener.py
class yourListener:
ROBOT_LISTENER_API_VERSION = 3
def end_keyword(self, name, attributes):
print("name :", name)
print("attributes :", attributes)
Obviously you’ll want to modify and format the output to your needs.
Since it’s open source, below is the listener that is used by the rfswram agent to send the keyword results back to the rfswram manager, so you can use it as a working example if you need.
Dave.
Listener2.py:
import os
import tempfile
import sys
import socket
from datetime import datetime
import time
import requests
import inspect
import threading
class RFSListener2:
ROBOT_LISTENER_API_VERSION = 2
msg = None
swarmmanager = "http://localhost:8138/"
excludelibraries = ["BuiltIn","String","OperatingSystem","perftest"]
debuglevel = 0
index = 0
robot = 0
iter = 0
seq = 0
def start_suite(self, name, attrs):
if 'RFS_DEBUGLEVEL' in attrs['metadata']:
self.debuglevel = int(attrs['metadata']['RFS_DEBUGLEVEL'])
self.debugmsg(6, 'debuglevel: ', self.debuglevel)
if 'RFS_INDEX' in attrs['metadata']:
self.index = attrs['metadata']['RFS_INDEX']
self.debugmsg(6, 'index: ', self.index)
if 'RFS_ITERATION' in attrs['metadata']:
self.iter = attrs['metadata']['RFS_ITERATION']
self.debugmsg(6, 'iter: ', self.iter)
if 'RFS_ROBOT' in attrs['metadata']:
self.robot = attrs['metadata']['RFS_ROBOT']
self.debugmsg(6, 'robot: ', self.robot)
if 'RFS_SWARMMANAGER' in attrs['metadata']:
self.swarmmanager = attrs['metadata']['RFS_SWARMMANAGER']
self.debugmsg(6, 'swarmmanager: ', self.swarmmanager)
if 'RFS_EXCLUDELIBRARIES' in attrs['metadata']:
self.excludelibraries = attrs['metadata']['RFS_EXCLUDELIBRARIES'].split(",")
self.debugmsg(6, 'excludelibraries: ', self.excludelibraries)
def log_message(self, message):
if message['message'][0:2] != '${':
self.msg = None
self.msg = message
def end_keyword(self, name, attrs):
self.debugmsg(3, 'name: ', name)
self.debugmsg(6, 'attrs: ', attrs)
self.debugmsg(5, 'attrs[doc]: ', attrs['doc'])
self.debugmsg(5, 'self.msg: ', self.msg)
if self.msg is not None:
self.debugmsg(8, 'self.msg: attrs[libname]: ', attrs['libname'], ' excludelibraries:', self.excludelibraries)
if attrs['libname'] not in self.excludelibraries:
self.seq += 1
self.debugmsg(8, 'self.seq: ', self.seq)
startdate = datetime.strptime(attrs['starttime'], '%Y%m%d %H:%M:%S.%f')
enddate = datetime.strptime(attrs['endtime'], '%Y%m%d %H:%M:%S.%f')
self.debugmsg(6, 'ResultName: self.msg[message]: ', self.msg['message'])
payload = {
'AgentName': 'DavesMBPSG',
'ResultName': self.msg['message'],
'Result': attrs['status'],
'ElapsedTime': (attrs['elapsedtime']/1000),
'StartTime': startdate.timestamp(),
'EndTime': enddate.timestamp(),
'ScriptIndex': self.index,
'Robot': self.robot,
'Iteration': self.iter,
'Sequence': self.seq
}
self.debugmsg(8, 'payload: ', payload)
t = threading.Thread(target=self.send_result, args=(payload,))
t.start()
else:
self.debugmsg(5, attrs['libname'], 'is an excluded library')
elif 'doc' in attrs and len(attrs['doc'])>0:
self.debugmsg(8, 'attrs[doc]: attrs[libname]: ', attrs['libname'], ' excludelibraries:', self.excludelibraries)
if attrs['libname'] not in self.excludelibraries:
self.seq += 1
self.debugmsg(8, 'self.seq: ', self.seq)
startdate = datetime.strptime(attrs['starttime'], '%Y%m%d %H:%M:%S.%f')
enddate = datetime.strptime(attrs['endtime'], '%Y%m%d %H:%M:%S.%f')
self.debugmsg(8, 'attrs: ', attrs)
self.debugmsg(6, 'ResultName: attrs[doc]: ', attrs['doc'])
payload = {
'AgentName': 'DavesMBPSG',
'ResultName': attrs['doc'],
'Result': attrs['status'],
'ElapsedTime': (attrs['elapsedtime']/1000),
'StartTime': startdate.timestamp(),
'EndTime': enddate.timestamp(),
'ScriptIndex': self.index,
'Robot': self.robot,
'Iteration': self.iter,
'Sequence': self.seq
}
self.debugmsg(8, 'payload: ', payload)
t = threading.Thread(target=self.send_result, args=(payload,))
t.start()
else:
self.debugmsg(5, attrs['libname'], 'is an excluded library')
self.msg = None
def debugmsg(self, lvl, *msg):
msglst = []
prefix = ""
if self.debuglevel >= lvl:
try:
if self.debuglevel >= 4:
stack = inspect.stack()
the_class = stack[1][0].f_locals["self"].__class__.__name__
the_method = stack[1][0].f_code.co_name
prefix = "{}: {}: [{}:{}] ".format(str(the_class), the_method, self.debuglevel, lvl)
if len(prefix.strip())<32:
prefix = "{} ".format(prefix)
if len(prefix.strip())<24:
prefix = "{} ".format(prefix)
msglst.append(str(prefix))
for itm in msg:
msglst.append(str(itm))
print(" ".join(msglst))
except:
pass
def send_result(self, payload):
exceptn = None
retry = True
count = 100
uri = self.swarmmanager + 'Result'
while retry and count>0:
try:
r = requests.post(uri, json=payload, timeout=600)
self.debugmsg(7, 'send_result: ',r.status_code, r.text)
if (r.status_code != requests.codes.ok):
exceptn = r.status_code
else:
retry = False
except Exception as e:
exceptn = e
time.sleep(1)
count -= 1
if retry:
self.debugmsg(0, 'send_result: while attempting to send result to', uri)
self.debugmsg(0, 'send_result: with payload:', payload)
self.debugmsg(0, 'send_result: Exception:', exceptn)
1 Like