apamax.eplapplications.perf.basetest¶

ObjectCreator¶

class apamax.eplapplications.perf.basetest.ObjectCreator[source]¶

Bases: object

Base class for object creator implementation.

createObject(device, time)[source]¶

Creates an object to publish.

Parameters

device (str) – The ID of the device to create an object for.
time (str) – The source time to use for the object.

Returns

A new object instance to publish.

For example:

# Create and return a measurement object
return {
    'time': time,
    "type": 'my_measurement',
    "source": { "id": device },
    'my_fragment': {
        'my_series': {
            "value": random.uniform(0, 100)
        }
    }
}

ApamaC8YPerfBaseTest¶

class apamax.eplapplications.perf.basetest.ApamaC8YPerfBaseTest(descriptor, outsubdir: str, runner)[source]¶

Bases: apamax.eplapplications.basetest.ApamaC8YBaseTest

Base class for performance tests for EPL apps and smart rules.

Requires the following to be set on the project in the pysysproject.xml file (typically from the environment):

EPL_TESTING_SDK

Variables

eplapps (EPLApps) – The object for deploying and undeploying EPL apps.
smartRulesManager (SmartRulesManager) – The object for managing smart rules.

generateHTMLReport(description, testConfigurationDetails=None, extraPerformanceMetrics=None)[source]¶

Generates an HTML report of the performance result. The report is generated at the end of the test.

When testing for multiple variations, multiple HTML reports are combined into a single HTML report.

Parameters

description (str) – A brief description of the test.
testConfigurationDetails (dict, list, str, optional) – Details of the test configuration to include in the report.
extraPerformanceMetrics (dict, list, str, optional) – Extra application-specific performance metrics to include in the report.

prepareTenant(restartMicroservice=False, tenant=None)[source]¶

Prepares the tenant for a performance test by deleting all devices created by previous tests, deleting all EPL test applications, and clearing all active alarms.

This must be called by the test before the application is deployed.

Parameters

restartMicroservice (bool) – Restart the Apama-ctrl microservice.
tenant (CumulocityTenant, optional) – The Cumulocity IoT tenant. If no tenant is specified, the tenant configured in the pysysproject.xml file is prepared.

read_json(fileName, fileDirectory=None)[source]¶

Reads a JSON file and returns its content.

Parameters

fileName (str) – Name of the file.
fileDirectory (str, optional) – Directory of the file. Use test output directory if not specified.

Returns

The decoded content of the file.

restartApamaMicroservice()[source]¶: Restarts the Apama-ctrl microservice and waits for it to come back up.

setup()[source]¶

Contains setup actions to be executed before the test is executed.

The setup method may be overridden by individual test classes, or (more commonly) in a custom BaseTest subclass that provides common functionality for multiple individual tests. However before implementing a custom BaseTest subclass with its own setup() method, consider whether the PySys concept of test plugins would meet your needs.

If you do override this method, be sure to call super(BASETEST_CLASS_HERE, self).setup() to allow the setup commands from the base test to run.

If setup throws an exception, the cleanup method will still be called, to clean up any resources that were already allocated.

startAlarmSimulator(devices, perDeviceRate, creatorFile, creatorClassName, creatorParams, duration=None, processingMode='CEP', tenant=None)[source]¶

Starts an alarm simulator process to publish simulated alarms to Cumulocity IoT.

The simulator uses an instance of the provided alarm creator class to create new alarms to send, allowing the test to publish alarms of different types. The simulator looks up the specified class in the specified Python file and creates a new instance of the class using the provided parameters. The alarm creator class must extend the apamax.eplapplications.perf.basetest.ObjectCreator class.

Parameters

devices (list[str]) – List of device IDs to generate alarms for.
perDeviceRate (float) – The rate of alarms to publish per device.
creatorFile (str) – The path to the Python file containing the alarm creator class.
creatorClassName (str) – The name of the alarm creator class that extends the apamax.eplapplications.perf.basetest.ObjectCreator class.
creatorParams (list) – The list of parameters to pass to the constructor of the alarm creator class to create a new instance.
processingMode (str) – Cumulocity IoT processing mode. Possible values are CEP, PERSISTENT, TRANSIENT, and QUIESCENT.
duration (float, optional) – The duration (in seconds) to run the simulator for. If no duration is specified, then the simulator runs until either stopped or the end of the test.
tenant (CumulocityTenant) – The Cumulocity IoT tenant. If no tenant is specified, alarms are published to the tenant configured in the pysysproject.xml file.

Returns

The process handle of the simulator process.

Return type

pysys.process.Process

For example:

# In a 'creator.py' file in test input directory.
class MyAlarmCreator(ObjectCreator):
    def createObject(self, device, time):
        return {
                    'source': {
                        'id': device
                    },
                    'type': 'my_alarm',
                    'text': 'My alarm',
                    'severity': 'MAJOR',
                    'status': 'ACTIVE',
                    'time': time,
                }

...

# In the test
self.startAlarmSimulator(
        ['12345'],                  # device IDs
        0.01,                       # rate of alarms to publish
        f'{self.input}/creator.py', # Python file path
        'MyAlarmCreator',           # class name
        [],                         # constructor parameters for MyAlarmCreator class
    )

startEventSimulator(devices, perDeviceRate, creatorFile, creatorClassName, creatorParams, duration=None, processingMode='CEP', tenant=None)[source]¶

Starts an event simulator process to publish simulated events to Cumulocity IoT.

The simulator uses an instance of the provided event creator class to create new events to send, allowing the test to publish events of different types and sizes. The simulator looks up the specified class in the specified Python file and creates a new instance of the class using the provided parameters. The event creator class must extend the apamax.eplapplications.perf.basetest.ObjectCreator class.

Parameters

devices (list[str]) – List of device IDs to generate events for.
perDeviceRate (float) – The rate of events to publish per device.
creatorFile (str) – The path to the Python file containing the event creator class.
creatorClassName (str) – The name of the event creator class that extends the apamax.eplapplications.perf.basetest.ObjectCreator class.
creatorParams (list) – The list of parameters to pass to the constructor of the event creator class to create a new instance.
processingMode (str) – Cumulocity IoT processing mode. Possible values are CEP, PERSISTENT, TRANSIENT, and QUIESCENT.
duration (float, optional) – The duration (in seconds) to run the simulator for. If no duration is specified, then the simulator runs until either stopped or the end of the test.
tenant (CumulocityTenant) – The Cumulocity IoT tenant. If no tenant is specified, events are published to the tenant configured in the pysysproject.xml file.

Returns

The process handle of the simulator process.

Return type

pysys.process.Process

For example:

# In a 'creator.py' file in test input directory.
class MyEventCreator(ObjectCreator):
    def createObject(self, device, time):
        return {
                    'time': time,
                    'type': 'pos_update_event',
                    'text': 'Position update',
                    'source': {
                        'id': device
                    },
                    'c8y_Position': {
                        'lng': random.uniform(0, 10),
                        'lat': random.uniform(0, 10)
                    }
                }

...

# In the test
self.startEventSimulator(
        ['12345'],                  # device IDs
        1,                          # rate of events to publish
        f'{self.input}/creator.py', # Python file path
        'MyEventCreator',           # class name
        [],                         # constructor parameters for MyEventCreator class
    )

startMeasurementSimulator(devices, perDeviceRate, creatorFile, creatorClassName, creatorParams, duration=None, processingMode='CEP', tenant=None)[source]¶

Starts a measurement simulator process to publish simulated measurements to Cumulocity IoT.

The simulator uses an instance of the provided measurement creator class to create new measurements to send, allowing the test to publish measurements of different types and sizes. The simulator looks up the specified class in the specified Python file and creates a new instance of the class using the provided parameters. The measurement creator class must extend the apamax.eplapplications.perf.basetest.ObjectCreator class.

Parameters

devices (list[str]) – List of device IDs to generate measurements for.
perDeviceRate (float) – The rate of measurements to publish per device.
creatorFile (str) – The path to the Python file containing the measurement creator class.
creatorClassName (str) – The name of the measurement creator class that extends the apamax.eplapplications.perf.basetest.ObjectCreator class.
creatorParams (list) – The list of parameters to pass to the constructor of the measurement creator class to create a new instance.
processingMode (str) – Cumulocity IoT processing mode. Possible values are CEP, PERSISTENT, TRANSIENT, and QUIESCENT.
duration (float, optional) – The duration (in seconds) to run the simulator for. If no duration is specified, then the simulator runs until either stopped or the end of the test.
tenant (CumulocityTenant) – The Cumulocity IoT tenant. If no tenant is specified, measurements are published to the tenant configured in the pysysproject.xml file.

Returns

The process handle of the simulator process.

Return type

pysys.process.Process

For example:

# In a 'creator.py' file in test input directory.
class MyMeasurementCreator(ObjectCreator):
    def __init__(lowerBound, upperBound):
        self.lowerBound = lowerBound
        self.upperBound = upperBound
    def createObject(self, device, time):
        return {
            'time': time,
            "type": 'my_measurement',
            "source": { "id": device },
            'my_fragment': {
                'my_series': {
                    "value": random.uniform(self.lowerBound, self.upperBound)
                }
            }
        }

...

# In the test
self.startMeasurementSimulator(
        ['12345'],                  # device IDs
        1,                          # rate of measurements to publish
        f'{self.input}/creator.py', # Python file path
        'MyMeasurementCreator',     # class name
        [10, 50],                   # constructor parameters for MyMeasurementCreator class
    )

startPerformanceMonitoring(pollingInterval=2)[source]¶

Starts a performance monitoring thread that periodically gathers and logs various metrics and publishes performance statistics at the end.

Parameters: pollingInterval (float, optional) – The polling interval to get performance data. Defaults to 2 seconds.
Returns: The background thread.
Return type: pysys.utils.threadutils.BackgroundThread

validate()[source]¶

Performs standard validations of the test run.

The following validations are performed.

Ensures no errors logged from EPL apps under test
Microservice did not terminate because of high memory usage.
Microservice’s memory usage remained below 90% of available memory.
Correlator was not swapping memory.

The test should define its own validate method for performing any application-specific validation. Ensure that the test calls the super implementation of the validate method, using super(PySysTest, self).validate().

EPLAppsPerfTest¶

class apamax.eplapplications.perf.basetest.EPLAppsPerfTest(descriptor, outsubdir: str, runner)[source]¶

Bases: apamax.eplapplications.perf.basetest.ApamaC8YPerfBaseTest

Base class for EPL applications performance tests.

The class ApamaC8YPerfBaseTest supersedes this class and should be used when writing new tests.