apama.testplugin¶
Contains a PySys test plugin that provides a simple way to access Apama functionality from your testcases using self.apama.XXX()
.
ApamaPlugin¶
- class apama.testplugin.ApamaPlugin[source]¶
Bases:
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 asself.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).New in version Apama: 10.7.0
- startCorrelator(name='testCorrelator', arguments=[], java=None, Xclock=False, config=None, **xargs)[source]¶
Start a new correlator process on a dynamically allocated port, and wait until it is running.
- Parameters
name (str) – A logical name for this correlator, which will be used for the log filename, stdouterr and the component name.
java (bool) – If True then the correlator will be started with support for injecting Java applications.
Xclock (bool) – 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.config (list[str]|str) – path or list of paths to a initialization or connectivity configuration .yaml file or directory containing them.
arguments (list[str]) – Additional arguments to be passed to the correlator.
xargs – Additional keyword arguments that will be passed to
apama.correlator.CorrelatorHelper.start()
.
- Returns
An instance of
apama.correlator.CorrelatorHelper
that you can use to interact with your application.
- startIAF(name, configname, **xargs)[source]¶
Start a new IAF (adapter) process on a dynamically allocate port, and wait until it is running.
- Parameters
name (str) – A logical name for this IAF process, which will be used for the log filename, stdouterr and the component name.
configname (str) – The IAF configuration file or template name
xargs – Additional keyword arguments that will be passed to
apama.iaf.IAFHelper.start()
.
- Returns
An instance of
apama.iaf.IAFHelper
.
- 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 usingextractEventLoggerOutput()
.
- extractEventLoggerOutput(logFile)[source]¶
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
- Parameters
logFile (str) – The path of the correlator log file containing the logged events.
- Returns
A list[dict[str,obj]] where each item is a dictionary representing the fields of an Apama event.
- projectHelper(projectName, **kwargs)[source]¶
Create a new instance of the
apama.project.ProjectHelper
class that can be used for creating, manipulating and deploying Apama project directories.- Parameters
projectName (str) – The project directory.
kwargs – Additional arguments to be passed to the
apama.project.ProjectHelper
constructor.
- Returns
An instance of
apama.project.ProjectHelper
.
- antBuild(buildfile, **kwargs)[source]¶
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.
- Parameters
buildfile (str) – absolute path to the ant build.xml to run
kwargs – Additional keyword arguments; see
apama.build.antBuild()
for details.
- static JoinEPLStackLines()[source]¶
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()])
New in version 10.11.0.
- 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
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.New in version 10.11.0.
ApamaHelper¶
- class apama.testplugin.ApamaHelper(*args, **kwargs)[source]¶
Bases:
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 ofapama.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):
New in version Apama: 10.15.1
- apama¶
An
apama.testplugin.ApamaPlugin
instance which provides access to Apama helper methods and fields.This field is added by inheriting from the
ApamaHelper
class.