#!/usr/bin/env python3
# Copyright (c) 2020-2022 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
""" Contains a PySys test plugin that provides a simple way to access Apama functionality from your testcases using ``self.apama.XXX()``.
"""
import sys, os, logging, types
import json
import pysys
from pysys.constants import *
import apama
import apama.correlator
import apama.iaf
import apama.build
import apama.project
[docs]class ApamaPlugin(object):
"""PySys test plugin class for Apama, which provides access to Apama functionality from individual testcases.
The recommended way to use this plugin is for tests to inherit from `apama.testplugin.ApamaHelper` (see that class
for an example), which exposes the plugin as ``self.apama``::
correlator = self.apama.startCorrelator('testCorrelator', ...)
(Alternative approaches such as subclassing ApamaPlugin directly or using a ``<test-plugin>`` project configuration
are not recommended for new projects).
.. versionadded:: Apama 10.7.0
"""
def setup(self, testObj):
self.owner = self.testObj = testObj
self.log = logging.getLogger('pysys.apama.ApamaPlugin')
self.__verboseAssertionsLog = logging.getLogger('pysys.assertions.apama.ApamaPlugin')
# nb: this needs to always create a new list, as we don't want people accidentally mutating the project property
self.defaultLogIgnores = testObj.project.getProperty('apamaDefaultLogIgnores', [])+[
# Tests running in parallel can get de-scheduled for some time, so ignore this:
' WARN .*Took [,.0-9]*s for an invocation of .*',
]
[docs] def startCorrelator(self, name='testCorrelator', arguments=[], java=None, Xclock=False, config=None, **xargs):
"""Start a new correlator process on a dynamically allocated port, and wait until it is running.
:param str name: A logical name for this correlator, which will be used for the log filename, stdouterr and
the component name.
:param bool java: If True then the correlator will be started with support for injecting Java applications.
:param bool Xclock: If True then the correlator will be started in externally clocked mode, meaning that time
is controlled by sending in ``&TIME(...)`` ticks from a .evt file rather than from the system clock.
:param list[str]|str config: path or list of paths to a initialization or connectivity configuration .yaml file or directory
containing them.
:param list[str] arguments: Additional arguments to be passed to the correlator.
:param xargs: Additional keyword arguments that will be passed to `apama.correlator.CorrelatorHelper.start()`.
:return: An instance of `apama.correlator.CorrelatorHelper` that you can use to interact with your application.
"""
c = apama.correlator.CorrelatorHelper(self.testObj, name=name)
xargs = xargs.copy()
xargs.setdefault('abortOnError', True)
xargs.setdefault('ignoreExitStatus', False)
xargs.setdefault('stdouterr', name)
xargs.setdefault('logfile', name+'.log')
c.start(arguments=arguments, java=java, Xclock=Xclock, config=config, **xargs)
return c
[docs] def startIAF(self, name, configname, **xargs):
"""Start a new IAF (adapter) process on a dynamically allocate port, and wait until it is running.
:param str name: A logical name for this IAF process, which will be used for the log filename, stdouterr and
the component name.
:param str configname: The IAF configuration file or template name
:param xargs: Additional keyword arguments that will be passed to `apama.iaf.IAFHelper.start()`.
:return: An instance of `apama.iaf.IAFHelper`.
"""
i = apama.iaf.IAFHelper(self.testObj, name=name)
xargs = xargs.copy()
xargs.setdefault('abortOnError', True)
xargs.setdefault('ignoreExitStatus', False)
xargs.setdefault('stdouterr', name)
xargs.setdefault('logfile', name+'.log')
i.start(configname=configname, **xargs)
return i
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()`.
"""
[docs] def projectHelper(self, projectName, **kwargs):
"""Create a new instance of the `apama.project.ProjectHelper` class that can be used for creating,
manipulating and deploying Apama project directories.
:param str projectName: The project directory.
:param kwargs: Additional arguments to be passed to the `apama.project.ProjectHelper` constructor.
:return: An instance of `apama.project.ProjectHelper`.
"""
return apama.project.projectHelper(self.testObj, name=projectName, **kwargs)
[docs] def antBuild(self, buildfile, **kwargs):
"""
Run an ant build file with Apama environment properties set,
typically to generate a project artifact such as a Java plugin or adapter.
Runs in a 'build' subdirectory of the parent's output directory.
Be careful to ensure that the ant build file generates its output
under its working directory, or under an explicitly specified directory
that is located inside the test output directory.
:param str buildfile: absolute path to the ant build.xml to run
:param kwargs: Additional keyword arguments; see `apama.build.antBuild()` for details.
"""
return apama.build.antBuild(self.testObj, buildfile=buildfile, **kwargs)
[docs] @staticmethod
def JoinEPLStackLines():
"""
Mapper that joins the lines of an EPL stack dump from a correlator log file
into a single line, for easier grepping and more self-contained test outcome failure reasons.
See `pysys.mappers.JoinLines` for similar mappers for other languages.
This mapper also converts ``[thread_identifier]`` to the placeholder ``...``.
Here is an example of using this mapper for checking no errors were logged::
self.assertGrep('testCorrelator.log', '(ERROR|FATAL|Failed to parse) .*', contains=False,
mappers=[self.apama.JoinEPLStackLines(), pysys.mappers.JoinLines.JavaStackTrace()])
.. versionadded:: 10.11.0
"""
def combiner(lines):
# remove the thread id
firstline = re.sub(r' \[[^\\]+\] - ', ' ... - ', lines[0], count=1)
return pysys.mappers.JoinLines.defaultCombiner([firstline]+
[l[l.find(' - ')+3:].strip() for l in lines[1:]])
return pysys.mappers.JoinLines(
startAt=' (ERROR|FATAL) ',
# Stop when the indenting stops
continueWhile=r'(.* Stack dump:|\] +- \t)',
combiner=combiner,
)
defaultLogIgnores = []
"""
A list of regular expressions indicating lines to be ignored when checking Apama log files for errors and warnings.
For example to use this with `~pysys.basetest.BaseTest.assertGrep`::
self.assertGrep('testCorrelator.log', 'WARN .*', contains=False,
ignores=self.apama.defaultLogIgnores+['An extra regex to ignore in this testcase'])
Project-specific ignores can be added to this list by setting the ``apamaDefaultLogIgnores`` project property
to a value delimited by either commas or newlines.
.. versionadded:: 10.11.0
"""
[docs]class ApamaHelper(object):
""" Helper class that can be subclassed by tests for easy access to Apama functionality such as starting correlators.
Inheriting from this class adds a ``self.apama`` member, which is an instance of `apama.testplugin.ApamaPlugin`.
When adding this helper to your inheritance list, be sure to leave the BaseTest as the last class, for example::
from apama.testplugin import ApamaHelper
class PySysTest(ApamaHelper, pysys.basetest.BaseTest):
.. versionadded:: Apama 10.15.1
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
for pluginClass, pluginAlias, pluginProperties in self.runner.project.testPlugins:
# don't introduce a duplicate alias, as that would give an error if a 10.15.1+ ApamaHelper was used with a pre-10.15.1 default project config
if pluginAlias == 'apama': return
self.apama = ApamaPlugin()
""" An `apama.testplugin.ApamaPlugin` instance which provides access to Apama helper methods and fields.
This field is added by inheriting from the `ApamaHelper` class.
"""
self.apama.setup(self)