apamax.eplapplications.basetest¶

ApamaC8YBaseTest¶

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

Bases: pysys.basetest.BaseTest

Base test for EPL applications tests.

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

EPL_TESTING_SDK
APAMA_HOME - Only if running a local correlator.

addC8YPropertiesToProject(apamaProject, params=None)[source]¶

Adds the connection parameters into a project.

Parameters

apamaProject – The ProjectHelper object for a project.

params –

The dictionary of parameters to override and add to those defined for the project:

<property name="CUMULOCITY_USERNAME" value="my-user"/>
<property name="CUMULOCITY_PASSWORD" value="my-password"/>
<property name="CUMULOCITY_SERVER_URL" value="https://my-url/"/>
<property name="CUMULOCITY_AUTHORITY_FILE" value=""/>
<property name="CUMULOCITY_TENANT" value=""/>
<property name="CUMULOCITY_MEASUREMENT_FORMAT" value=""/>

copyWithReplace(sourceFile, targetFile, replacementDict, marker='@')[source]¶

Copies the source file to the target file and replaces the placeholder strings with the actual values.

Parameters

sourceFile (str) – The path to the source file to copy.
targetFile (str) – The path to the target file.
replacementDict (dict[str, str]) – A dictionary containing placeholder strings and their actual values to replace.
marker (str, optional) – Marker string used to surround replacement strings in the source file to disambiguate from normal strings. For example, @.

createAppKey(url, username, password)[source]¶

Checks if the tenant has an external application defined for us and if not, creates it.

Parameters

url – The URL to the Cumulocity IoT tenant.
username – The user to authenticate to the tenant.
password – The password to authenticate to the tenant.

Returns

An app key suitable for connecting a test correlator to the tenant.

createProject(name, existingProject=None)[source]¶

Creates a ProjectHelper object which mimics the Cumulocity IoT EPL applications environment.

Adds all the required bundles and adds the properties to connect and authenticate to the configured Cumulocity IoT tenant.

Parameters

name – The name of the project.
existingProject – If provided the path to an existing project. The environment will be added to that project instead of a new one.

Returns

A ProjectHelper object.

createTestDevice(name, type='PySysTestDevice', children=None, tenant=None)[source]¶

Creates a Cumulocity IoT device for testing.

Parameters

name (str) – The name of the device. The name of the device is prefixed with PYSYS_ so that the framework can identify and clean up test devices.
type (str, optional) – The type of the device.
children (list[str], optional) – The list of device IDs to add them as children to the created device.
tenant (CumulocityTenant, optional) – The Cumulocity IoT tenant. If no tenant is specified, the tenant configured in the pysysproject.xml file is used.

Returns

The ID of the device created.

Return type

str

getAlarms(source=None, type=None, status=None, dateFrom=None, dateTo=None, tenant=None, **kwargs)[source]¶

Gets all alarms with matching parameters.

For example:

self.getAlarms(type='my_alarms', dateFrom='2021-04-15 11:00:00.000Z', 
                                dateTo='2021-04-15 11:30:00.000Z')

Parameters

source (str, optional) – The source object of the alarm. Get alarms for all objects if not specified.
type (str, optional) – The type of alarm to get. Get alarms of all types if not specified.
status (str, optional) – The status of the alarms to get. Get alarms of all status if not specified.
dateFrom (str, optional) – The start time of the alarm in the ISO format. If specified, only alarms that are created on or after this time are fetched.
dateTo (str, optional) – The end time of the alarm in the ISO format. If specified, only alarms that are created on or before this time are fetched.
tenant (CumulocityTenant, optional) – The Cumulocity IoT tenant. If no tenant is specified, the tenant configured in the pysysproject.xml file is used.
**kwargs – All additional keyword arguments are treated as extra parameters for filtering alarms.

Returns

List of alarms.

Return type

list[object]

getOperations(deviceId=None, fragmentType=None, dateFrom=None, dateTo=None, tenant=None, **kwargs)[source]¶

Gets all operations with matching parameters.

For example:

self.getOperations(fragmentType='my_ops', dateFrom='2021-04-15 11:00:00.000Z', 
                                        dateTo='2021-04-15 11:30:00.000Z')

Parameters

deviceId (str, optional) – The device ID of the alarm. Get operations for all devices if not specified.
fragmentType (str, optional) – The type of fragment that must be part of the operation.
dateFrom (str, optional) – The start time of the operation in the ISO format. If specified, only operations that are created on or after this time are fetched.
dateTo (str, optional) – The end time of the operation in the ISO format. If specified, only operations that are created on or before this time are fetched.
tenant (CumulocityTenant, optional) – The Cumulocity IoT tenant. If no tenant is specified, the tenant configured in the pysysproject.xml file is used.
**kwargs – All additional keyword arguments are treated as extra parameters for filtering operations.

Returns

List of operations.

Return type

list[object]

getTestSubjectEPLApps()[source]¶

Retrieves a list of paths to the EPL apps being tested.

If the user defines the <user-data name="EPLApp" value="EPLAppToBeTested"/> tag in the pysystest.xml file, then we just return the EPL app defined by the tag’s value. If this tag is not defined (or the value is an empty string) then all the mon files in the project.EPL_APPS directory are returned.

getUTCTime(timestamp=None)[source]¶

Gets a Cumulocity IoT-compliant UTC timestamp string for the current time or the specified time.

Parameters: timestamp (float, optional) – The epoc timestamp to get timestamp string for. Use current time if not specified.
Returns: Timestamp string.
Return type: str

prepareTenant(tenant=None)[source]¶

Prepares the tenant for a test by deleting all devices created by previous tests and clearing all active alarms. However, you can disable the default behavior of clearing all active alarms by setting the clearAllActiveAlarmsDuringTenantPreparation property to false in the PySys project configuration.

Parameters: tenant (CumulocityTenant, optional) – The Cumulocity IoT tenant. If no tenant is specified, the tenant configured in the pysysproject.xml file is prepared.

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.

LocalCorrelatorSimpleTest¶

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

Bases: apamax.eplapplications.basetest.ApamaC8YBaseTest

Base test for running test with no run.py with local correlator connected to Cumulocity IoT.

addEPLAppsToProject(eplApps, project)[source]¶: Adds the EPL app(s) being tested to a project.

execute()[source]¶: Runs all the tests in the Input directory against the applications configured in the EPL_APPS directory or with the EPLApps directive.

getMonitorsFromInjectedFile(correlator, file)[source]¶: Retrieves a list of active monitors in a correlator, added from a particular mon file using a GET request to http://correlator.host:correlator.port.

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.

validate()[source]¶: Checks that no errors were logged to the correlator log file.

EPLAppsSimpleTest¶

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

Bases: apamax.eplapplications.basetest.ApamaC8YBaseTest

Base test for running test with no run.py on EPL apps running in Cumulocity IoT.

execute()[source]¶: Runs all the tests in the Input directory against the applications configured in the EPL_APPS directory or with the EPLApps directive using EPL apps to run each test.

prepareTenant(tenant=None)[source]¶

Prepares the tenant for a test by deleting all devices created by previous tests, deleting all EPL apps which have been uploaded by tests, and clearing all active alarms.

This is done first so that it is not possible for existing test apps to raise alarms or create devices.

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.

shutdown()[source]¶: Deactivates all uploaded EPL apps when the test terminates.

validate()[source]¶: Ensures that no tests failed.