Source code for apama.basetest

#!/usr/bin/env python3
# Copyright (c) 2018-2019 Software AG, Darmstadt, Germany and/or Software AG USA Inc., Reston, VA, USA, and/or its subsidiaries and/or its affiliates and/or their licensors.
# Use, reproduction, transfer, publication or disclosure is prohibited except as specifically provided for in your License Agreement with Software AG

from pysys.basetest import BaseTest
from os import path
import logging, json

class ApamaPlugin(object):
	"""PySys test pseudo-plugin class for Apama, which provides access to Apama functionality from individual testcases.

	 (Replaced in PAM 10.7.0 with a pysys plugin, but pysys 1.5 doesn't have that feature)
	"""
	def __init__(self, testObj):
		self.testObj = testObj
		self.__verboseAssertionsLog = logging.getLogger('pysys.assertions.apama.ApamaPlugin')

	TEST_EVENT_LOGGER_REGEX = '^[-0-9]* [0-9:.]* INFO  .[0-9]*. - apama.test.TestEventLogger .[0-9]*. -- Got test event: (.*)'
	"""
	The regular expression that identifies each event written to the correlator log by 
	`apama.correlator.CorrelatorHelper.injectTestEventLogger()` and extracted using `extractEventLoggerOutput()`. 
	"""
	
	def extractEventLoggerOutput(self, logFile):
		"""
		Uses log messages generated by `apama.correlator.CorrelatorHelper.injectTestEventLogger()` to extract a list of events 
		from the specified correlator log file in a form that's easy to manipulate from Python in your test's 
		validate() method. 
		
		Top-level events are each represented as a dictionary with an item for each event field, and a special key 
		named ``.eventType`` whose value identifies the event type. Note that the order of the event fields is not 
		preserved, however all dictionaries in the returned value will be sorted alphabetically by key to ensure 
		consistent results. 
		
		This method is often used with `pysys.basetest.BaseTest.assertThat`, for example::
		
			sensor1_temps = [ 
				# Extract only the field value(s) we care about (allows us to ignore unimportant information, timestamps, etc):
				(evt['temperature']) 
				for evt in self.apama.extractEventLoggerOutput('testCorrelator.log')
				
				# Filter to include the desired subset of events:
				if evt['.eventType']=='apamax.myapp.Alert' and evt['sensorId']=='TempSensor001'
				]
			self.assertThat('sensor1_temps == expected', sensor1_temps=sensor1_temps, expected=[
					111.0,
					120,
					145.2,
				])
		
		For debugging purposes, a readable multi-line JSON representation of the events is logged when run using::
		
			pysys run -v assertions.apama=debug 
		
		:param str logFile: The path of the correlator log file containing the logged events. 
		:return: A list[dict[str,obj]] where each item is a dictionary representing the fields of an Apama event. 
		"""
		# do a double load since otherwise the dictionaries are sorted randomly which is a really bad idea in testcases
		events = [json.loads(json.dumps(json.loads(evt), sort_keys=True, allow_nan=True)) 
			for evt in self.testObj.getExprFromFile(logFile, self.TEST_EVENT_LOGGER_REGEX, 
				returnAll=True, encoding='utf-8')]
		
		if self.__verboseAssertionsLog.isEnabledFor(logging.DEBUG):
			self.__verboseAssertionsLog.debug('extractEventLoggerOutput from %s: \n%s', logFile, json.dumps(events, ensure_ascii=False, allow_nan=True, indent='  '))
		
		return events


class ApamaBaseTest(BaseTest):
	"""
	A custom PySys basetest for Apama tests.

	"""

	def __init__(self, descriptor, outsubdir, runner):
		super(ApamaBaseTest, self).__init__(descriptor, outsubdir, runner)
		self.apama = ApamaPlugin(self)

	def getDefaultFileEncoding(self, file, **xargs):
		filename, file_extension = path.splitext(file)

		# give first priority to super-class since for projects created with 
		# PySys 1.4.0+ these will be configured through the pysysproject.xml 
		# instead; this code is only for pre-Apama 10.5 project configs
		delegated = super(ApamaBaseTest, self).getDefaultFileEncoding(file, **xargs)
		if delegated: return delegated
		
		if file_extension in ['xml', 'yaml', 'json', 'log']:
			return 'utf-8'
		else: 
			return None