How to print debug file to console?

Robot Framework
5

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