apama.basetest

ApamaPlugin

class apama.basetest.ApamaPlugin(testObj)

Bases: 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)

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().

extractEventLoggerOutput(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 

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.

ApamaBaseTest

class apama.basetest.ApamaBaseTest(descriptor, outsubdir, runner)

Bases: pysys.basetest.BaseTest

A custom PySys basetest for Apama tests.

getDefaultFileEncoding(file, **xargs)

Specifies what encoding should be used to read or write the specified text file. The default implementation for BaseTest delegates to the runner, which in turn gets its defaults from the pysyproject.xml configuration.

See pysys.process.user.ProcessUser.getDefaultFileEncoding for more details.