From 94d8484e2c32c54f416045a4d825edb0cf3a49b8 Mon Sep 17 00:00:00 2001 From: Chris Reed Date: Wed, 30 Sep 2020 12:22:50 +0100 Subject: [PATCH] Apama Analytics EPL Apps tools for 10.7.0.1 --- .../apamax/eplapplications/basetest.html | 126 ++++++++++-------- .../apamax.eplapplications.basetest.html | 27 ++++ doc/pydoc/genindex.html | 15 +++ doc/pydoc/objects.inv | Bin 883 -> 904 bytes doc/pydoc/searchindex.js | 2 +- doc/pydoc/testing-epl.html | 6 +- doc/pydoc/using-pysys.html | 7 +- doc/testing-epl.rst | 6 +- doc/using-pysys.rst | 7 +- readme.rst | 11 +- scripts/eplapp.cmd | 18 ++- .../apamax/eplapplications/basetest.py | 124 +++++++++-------- 12 files changed, 220 insertions(+), 129 deletions(-) diff --git a/doc/pydoc/_modules/apamax/eplapplications/basetest.html b/doc/pydoc/_modules/apamax/eplapplications/basetest.html index 0bea56d..07715b2 100644 --- a/doc/pydoc/_modules/apamax/eplapplications/basetest.html +++ b/doc/pydoc/_modules/apamax/eplapplications/basetest.html @@ -76,7 +76,10 @@

Source code for apamax.eplapplications.basetest

<
[docs] def setup(self): super(ApamaC8YBaseTest, self).setup() - self.modelId = 0
+ self.modelId = 0 + self.TEST_DEVICE_PREFIX = "PYSYS_" + # connect to the platform + self.platform = CumulocityPlatform(self)
[docs] def createAppKey(self, url, username, password): """ @@ -112,12 +115,11 @@

Source code for apamax.eplapplications.basetest

< from apama.project import ProjectHelper except: self.abort(BLOCKED, "Could not load Apama extensions. Try running from an Apama installation") - #create the project and add required bundles - + # create the project and add required bundles apama_project = ProjectHelper(self, name) apama_project.create(existingProject) - #need to have a version independent addition or this will need to be maintained. + # need to have a version independent addition or this will need to be maintained. apama_project.addBundle("Automatic onApplicationInitialized") apama_project.addBundle("Cumulocity IoT > Cumulocity Client") apama_project.addBundle("Cumulocity IoT > Event Definitions for Cumulocity") @@ -196,13 +198,63 @@

Source code for apamax.eplapplications.basetest

< # Check file is a .mon file before appending if os.path.splitext(eplApp)[1] == '.mon': eplAppsPaths.append(os.path.join(self.project.EPL_APPS, eplApp)) - return eplAppsPaths
+ return eplAppsPaths
+ +
[docs] def prepareTenant(self): + """ + Prepares the tenant for a test by deleting all devices created by previous tests, and clearing all active alarms. + """ + self.log.info('Preparing tenant to run test(s)') + # Clear all active alarms + self._clearActiveAlarms() + # Delete devices that were created by tests + self._deleteTestDevices()
+ + def _clearActiveAlarms(self): + """ + Clears all active alarms as part of a pre-test tenant cleanup. + """ + self.log.info("Clearing active alarms") + self.platform.getC8YConnection().do_request_json('PUT', '/alarm/alarms?status=ACTIVE', {"status": "CLEARED"}) + + def _deleteTestDevices(self): + """ + Deletes all ManagedObjects that have name prefixed with "PYSYS_" and the 'c8y_isDevice' param as part of pre-test tenant cleanup. + """ + self.log.info("Deleting old test devices") + # Retrieving test devices + PAGE_SIZE = 100 # By default, pageSize = 5 for querying to C8y + resp = self.platform.getC8YConnection().do_get( + "/inventory/managedObjects" + + f"?query=has(c8y_IsDevice)+and+name+eq+'{self.TEST_DEVICE_PREFIX}*'" + + f"&pageSize={PAGE_SIZE}&currentPage=1&withTotalPages=true") + testDevices = resp['managedObjects'] + # Make sure we retrieve all pages from query + TOTAL_PAGES = resp['statistics']['totalPages'] + if TOTAL_PAGES > 1: + for currentPage in range(2, TOTAL_PAGES + 1): + resp = self.platform.getC8YConnection().do_get( + "/inventory/managedObjects" + + f"?query=has(c8y_IsDevice)+and+name+eq+'{self.TEST_DEVICE_PREFIX}*'" + + f"&pageSize={PAGE_SIZE}&currentPage={currentPage}") + testDevices = testDevices + resp['managedObjects'] + + # Deleting test devices + testDeviceIds = [device['id'] for device in testDevices] + for deviceId in testDeviceIds: + resp = self.platform.getC8YConnection().request('DELETE', f'/inventory/managedObjects/{deviceId}') +
[docs]class LocalCorrelatorSimpleTest(ApamaC8YBaseTest): """ Base test for running test with no run.py with local correlator connected to Cumulocity IoT. """ +
[docs] def setup(self): + super(LocalCorrelatorSimpleTest, self).setup() + # Prepare the tenant for the test by performing a pre-test cleanup of artifacts from previous tests + self.prepareTenant()
+
[docs] def execute(self): """ Runs all the tests in the Input directory against the applications configured in the EPL_APPS @@ -214,7 +266,6 @@

Source code for apamax.eplapplications.basetest

< if not os.path.isdir(self.project.EPL_TESTING_SDK): self.abort(BLOCKED, f'EPL_TESTING_SDK is not valid ({self.project.EPL_TESTING_SDK}). Please set the EPL_TESTING_SDK environment variable.') - self.log.info(f"Connecting to Cumulocity platform at {self.project.CUMULOCITY_SERVER_URL} as user {self.project.CUMULOCITY_USERNAME}") from apama.correlator import CorrelatorHelper # Create test project and add C8Y properties and EPL Apps project = self.createProject("test-project") @@ -301,38 +352,29 @@

Source code for apamax.eplapplications.basetest

< """
[docs] def setup(self): - super(ApamaC8YBaseTest, self).setup() + super(EPLAppsSimpleTest, self).setup() + # connect to the platform self.tests = None self.apps = None - self.platform = None self.eplapps = None self.addCleanupFunction(lambda: self.shutdown()) - self.EPL_APP_PREFIX = "PYSYS_" + self.EPL_APP_PREFIX = self.TEST_DEVICE_PREFIX self.EPL_TEST_APP_PREFIX = self.EPL_APP_PREFIX + "TEST_" - self.TEST_DEVICE_PREFIX = self.EPL_APP_PREFIX # Prefix test device names the same way as EPL apps - # Check EPL_TESTING_SDK env is set if not os.path.isdir(self.project.EPL_TESTING_SDK): self.abort(BLOCKED, f'EPL_TESTING_SDK is not valid ({self.project.EPL_TESTING_SDK}). Please set the EPL_TESTING_SDK environment variable.') - # connect to the platform - self.platform = CumulocityPlatform(self) self.eplapps = EPLApps(self.platform.getC8YConnection()) + self.prepareTenant()
- self._prepareTenant()
- - def _prepareTenant(self): +
[docs] def prepareTenant(self): """ - Prepares the tenant for a test by deleting all EPL apps uploaded - and devices created by previous tests, and clearing all active alarms. + 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 there's no possibility existing test apps raising alarms or creating devices """ - self.log.info('Preparing tenant to run test(s)') - # Delete all EPL Apps which have been uploaded by tests - # This is done first so that there's no possibility existing test apps raising alarms or creating devices self._deleteTestEPLApps() - # Clear all active alarms - self._clearActiveAlarms() - # Delete devices that were created by tests - self._deleteTestDevices() + super(EPLAppsSimpleTest, self).prepareTenant()
+ def _deleteTestEPLApps(self): """ @@ -350,40 +392,6 @@

Source code for apamax.eplapplications.basetest

< for name in appsToDelete: self.eplapps.delete(name) - def _clearActiveAlarms(self): - """ - Clears all active alarms as part of a pre-test tenant cleanup. - """ - self.log.info("Clearing active alarms...") - self.platform.getC8YConnection().do_request_json('PUT', '/alarm/alarms?status=ACTIVE', {"status": "CLEARED"}) - - def _deleteTestDevices(self): - """ - Deletes all ManagedObjects that have name prefixed with "PYSYS_" and the 'c8y_isDevice' param - as part of pre-test tenant cleanup. - """ - # Retrieving test devices - PAGE_SIZE = 100 # By default, pageSize = 5 for querying to C8y - resp = self.platform.getC8YConnection().do_get( - "/inventory/managedObjects" + - f"?query=has(c8y_IsDevice)+and+name+eq+'{self.TEST_DEVICE_PREFIX}*'" + - f"&pageSize={PAGE_SIZE}&currentPage=1&withTotalPages=true") - testDevices = resp['managedObjects'] - # Make sure we retrieve all pages from query - TOTAL_PAGES = resp['statistics']['totalPages'] - if TOTAL_PAGES > 1: - for currentPage in range(2, TOTAL_PAGES + 1): - resp = self.platform.getC8YConnection().do_get( - "/inventory/managedObjects" + - f"?query=has(c8y_IsDevice)+and+name+eq+'{self.TEST_DEVICE_PREFIX}*'" + - f"&pageSize={PAGE_SIZE}&currentPage={currentPage}") - testDevices = testDevices + resp['managedObjects'] - - # Deleting test devices - testDeviceIds = [device['id'] for device in testDevices] - for deviceId in testDeviceIds: - resp = self.platform.getC8YConnection().request('DELETE', f'/inventory/managedObjects/{deviceId}') -
[docs] def execute(self): """ Runs all the tests in the Input directory against the applications configured in the EPL_APPS diff --git a/doc/pydoc/autodocgen/apamax.eplapplications.basetest.html b/doc/pydoc/autodocgen/apamax.eplapplications.basetest.html index efb20f2..e0d3364 100644 --- a/doc/pydoc/autodocgen/apamax.eplapplications.basetest.html +++ b/doc/pydoc/autodocgen/apamax.eplapplications.basetest.html @@ -117,6 +117,12 @@

ApamaC8YBaseTest +
+prepareTenant()[source]
+

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

+
+
setup()[source]
@@ -161,6 +167,20 @@

LocalCorrelatorSimpleTesthttp://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 allow for clean up resources +that were already allocated.

+
+
validate()[source]
@@ -184,6 +204,13 @@

EPLAppsSimpleTest +
+prepareTenant()[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 there’s no possibility existing test apps raising alarms or creating devices

+

+
setup()[source]
diff --git a/doc/pydoc/genindex.html b/doc/pydoc/genindex.html index d9530b2..f16523d 100644 --- a/doc/pydoc/genindex.html +++ b/doc/pydoc/genindex.html @@ -47,6 +47,7 @@

Index

| G | L | M + | P | R | S | U @@ -221,6 +222,18 @@

M

+

P

+ + +
+

R

diff --git a/doc/pydoc/objects.inv b/doc/pydoc/objects.inv index 36258197b9747381c2088052e6e540ea1601fe82..8dbc8796b530cc520ded4bc9a1a2fad12331e614 100644 GIT binary patch delta 783 zcmV+q1MvLw28aichJQP#D}n+ek5tq zh(Qagwct5*TuFPz$)0vkr`aR5;J~^_?_0B2nC!Y*pI!d?f`5WlgPRL#ExCw>9fFrp z#_)Avlm_EC*pQI=I&Q8y>}l~XZO#(#jIPMPr%{3LL7zcc|E?`Ap8W~{P1%>euSy^ zm5UZ~|Mc$y4?_$Y zgI0HHJzyeuD+>5U*1bk1;m~TYU%^o>>=aZ6el{Gz-nSvvu`862^y0*yU}RA)OG`E& zG>#tx{eO` zj?vzLK+dWqy&G^Gk#^T6uj?fJt0Rs>r>4jmr34*u$sm-OppPIbl&v)Qm{*U+*iSUYxRuTqJPBxBD>ZWsL3~7!ypSb%doznYe=dJ zNx4jOp*-uFc+2HF)uk<8ZrcFCJfGGd~ delta 761 zcmV$aYPxl~d>7Vd?_riS zVl}4;B8GXQ&sFf5*ZO7+pHH zeJ6}eD@B2&`$)9CVLVhizbxk+;Qlpth&uo<<@%n|EAkz_5^#aPJ!QiEqt$8x1`Yh% zA&2MK+-<*3QW!tc^Tx` zUm7^RmVfl0#^NwLGaQ^zN+QR#NGi`JJ?NvE#2oZ-j1C3_IIEWQZoqLw+Fcu7)=BzT zM;wPvO~DzZgd7RQF_f8P4=5{?oizAZRJX}qt)bG=*i{;=wU#!a^W=zAQ_gk^!RjuY zSHXR2^^DoFr2eA1))rZjZM%j+7Hn2weaSYMRB{zE6Ka-8?RnS4TcI{HQ`++X$yIntroduction

  • Creating device simulators.

  • Sending mock data to Cumulocity IoT.

    • -

  • Listening for events that the EPL app should produce; and/or

    • +

  • Listening for events that the EPL app should produce.

  • Querying Cumulocity IoT for any objects created by the EPL app.

    • -

    This can all be done using additional EPL apps that run parallel to the EPL app that we wish to test for correctness. This document will demonstrate some of the common processes involved in writing these additional EPL apps that test your exising EPL apps, while outlining some of the conventions for writing tests that best utilize the PySys test framework provided in the SDK.

    +

    This can all be done using additional EPL apps that run parallel to the EPL app that we wish to test for correctness. This document will demonstrate some of the common processes involved in writing these additional EPL apps that test your existing EPL apps, while outlining some of the conventions for writing tests that best utilize the PySys test framework provided in the SDK.

    Creating device simulators

    All measurements and alarms in Cumulocity IoT must be associated with a source. Devices in Cumulocity IoT are represented by managed objects, each of which has a unique identifier. When sending measurement or alarm events, the source field of these events must be set to a identifier of a managed object in Cumulocity IoT. Therefore, in order to send measurements from our test EPL app, it must create a ManagedObject device simulator to be the source of these measurements.

    -

    If you are using the PySys framework to run tests in the cloud, any devices created by your tests should be named with prefix “PYSYS_”, and have the c8y_IsDevice property. These indicators are what the framework uses to identify which devices should be deleted following a test. Note that deleting a device in Cumulocity IoT will also delete any alarms or measurements associated with that device so the cleanup from a test is done when another test is next run in the cloud again.

    +

    If you are using the PySys framework to run tests in the cloud, any devices created by your tests should be named with prefix “PYSYS_”, and have the c8y_IsDevice property. These indicators are what the framework uses to identify which devices should be deleted following a test. Note that deleting a device in Cumulocity IoT will also delete any alarms or measurements associated with that device so the cleanup from a test is done when another test is next run.

    To see how this can be done, have a look at the createNewDevice action below:

    action createNewDevice(string name) returns integer
 {
diff --git a/doc/pydoc/using-pysys.html b/doc/pydoc/using-pysys.html
index 5a30f01..ddccc6f 100644
--- a/doc/pydoc/using-pysys.html
+++ b/doc/pydoc/using-pysys.html
@@ -53,7 +53,7 @@ 
    Using PySys to test your EPL apps
 
    Introduction
    
 
    PySys is a testing framework that provides a way to test your applications.
    
-
    See PySys Documentation for details on how the framework can be used and the facilities it contains.
    
+
    See PySys Documentation for details on installation and how the framework can be used and the facilities it contains.

    Testing in the Cumulocity IoT cloud

    @@ -163,6 +163,11 @@

    Running the test
    pysys run TestLocalCorrelator

    +

    Whenever you run a test using a local correlator, before the test is executed:

    +
      +

    • All active Alarms in your Cumulocity IoT tenant are cleared.

      • +

    • Any devices created by previous tests (which are identified by the device name having prefix “PYSYS_”) are deleted from your tenant.

      • +
    diff --git a/doc/testing-epl.rst b/doc/testing-epl.rst index ec7a713..47e7b77 100644 --- a/doc/testing-epl.rst +++ b/doc/testing-epl.rst @@ -10,16 +10,16 @@ The behavior of most EPL apps usually consists of receiving data, sending measur + Creating device simulators. + Sending mock data to Cumulocity IoT. -+ Listening for events that the EPL app should produce; and/or ++ Listening for events that the EPL app should produce. + Querying Cumulocity IoT for any objects created by the EPL app. -This can all be done using additional EPL apps that run parallel to the EPL app that we wish to test for correctness. This document will demonstrate some of the common processes involved in writing these additional EPL apps that test your exising EPL apps, while outlining some of the conventions for writing tests that best utilize the PySys test framework provided in the SDK. +This can all be done using additional EPL apps that run parallel to the EPL app that we wish to test for correctness. This document will demonstrate some of the common processes involved in writing these additional EPL apps that test your existing EPL apps, while outlining some of the conventions for writing tests that best utilize the PySys test framework provided in the SDK. Creating device simulators =========================== All measurements and alarms in Cumulocity IoT must be associated with a source. Devices in Cumulocity IoT are represented by managed objects, each of which has a unique identifier. When sending measurement or alarm events, the ``source`` field of these events must be set to a identifier of a managed object in Cumulocity IoT. Therefore, in order to send measurements from our test EPL app, it must create a ``ManagedObject`` device simulator to be the source of these measurements. -If you are using the PySys framework to :ref:`run tests in the cloud `, any devices created by your tests should be named with prefix "PYSYS\_", and have the ``c8y_IsDevice`` property. These indicators are what the framework uses to identify which devices should be deleted following a test. Note that deleting a device in Cumulocity IoT will also delete any alarms or measurements associated with that device so the cleanup from a test is done when another test is next run in the cloud again. +If you are using the PySys framework to :ref:`run tests in the cloud `, any devices created by your tests should be named with prefix "PYSYS\_", and have the ``c8y_IsDevice`` property. These indicators are what the framework uses to identify which devices should be deleted following a test. Note that deleting a device in Cumulocity IoT will also delete any alarms or measurements associated with that device so the cleanup from a test is done when another test is next run. To see how this can be done, have a look at the ``createNewDevice`` action below:: diff --git a/doc/using-pysys.rst b/doc/using-pysys.rst index 96c9f6d..070f170 100644 --- a/doc/using-pysys.rst +++ b/doc/using-pysys.rst @@ -8,7 +8,7 @@ Introduction PySys is a testing framework that provides a way to test your applications. -See `PySys Documentation `_ for details on how the framework can be used and the facilities it contains. +See `PySys Documentation `_ for details on installation and how the framework can be used and the facilities it contains. .. _test-in-cloud: @@ -158,6 +158,11 @@ Run the test from within the samples directory by using the following command: pysys run TestLocalCorrelator +Whenever you run a test using a local correlator, before the test is executed: + ++ All active Alarms in your Cumulocity IoT tenant are cleared. ++ Any devices created by previous tests (which are identified by the device name having prefix "PYSYS\_") are deleted from your tenant. + Advanced tests ============== diff --git a/readme.rst b/readme.rst index 03a55ec..6e72cda 100644 --- a/readme.rst +++ b/readme.rst @@ -9,8 +9,6 @@ you to script uploads of your EPL apps and manage them for CI/CD use cases. It also provides extensions to the PySys test framework to allow you to simply write tests for your EPL apps and to run them automatically. -See for further docs. - Disclaimer ---------- These tools are provided as-is and without warranty or support. They do not @@ -39,6 +37,15 @@ To find out how to write a test for your EPL apps please look at `Writing tests In order to view documentation on classes for PySys helpers for Apama EPL Apps please see: `PySys helpers `_. +See for further docs. + +System requirements +------------------- +This SDK requires an installation of Python 3.7+ and will run on either Windows or Linux. + +The EPL apps test framework requires you to have an installation of PySys. See `PySys Documentation `_ for details. If you want the option of running tests locally, you will also need an installation of Apama, which can be obtained from . If you choose to install Apama, you may skip the manual installation of Python and PySys as both are shipped with Apama. + + Repository structure ==================== diff --git a/scripts/eplapp.cmd b/scripts/eplapp.cmd index 32ca10f..77e7419 100644 --- a/scripts/eplapp.cmd +++ b/scripts/eplapp.cmd @@ -15,7 +15,23 @@ setlocal set THIS_SCRIPT=%~$PATH:0 call :getpath %THIS_SCRIPT -py.exe "%THIS_DIR%\eplapp.py" %* +if not defined APAMA_HOME (goto APAMA_HOME_UNDEFINED) + +set "PATH=%APAMA_HOME%\third_party\python;%PATH%" +python.exe "%THIS_DIR%\eplapp.py" %* +goto CHECK + +:APAMA_HOME_UNDEFINED +WHERE python >NUL 2>&1 +if errorlevel 1 (goto PY_UNDEFINED) +python.exe "%THIS_DIR%\eplapp.py" %* +goto CHECK + +:PY_UNDEFINED +echo Please add an appropriate Python 3 version to your path. Please refer to the System requirements section of the documentaion. +goto END + +:CHECK if NOT %errorlevel% == 0 ( EXIT /B %errorlevel% ) diff --git a/testframework/apamax/eplapplications/basetest.py b/testframework/apamax/eplapplications/basetest.py index 368fb00..d936f50 100755 --- a/testframework/apamax/eplapplications/basetest.py +++ b/testframework/apamax/eplapplications/basetest.py @@ -37,7 +37,10 @@ class ApamaC8YBaseTest(BaseTest): def setup(self): super(ApamaC8YBaseTest, self).setup() - self.modelId = 0 + self.modelId = 0 + self.TEST_DEVICE_PREFIX = "PYSYS_" + # connect to the platform + self.platform = CumulocityPlatform(self) def createAppKey(self, url, username, password): """ @@ -73,12 +76,11 @@ def createProject(self, name, existingProject=None): from apama.project import ProjectHelper except: self.abort(BLOCKED, "Could not load Apama extensions. Try running from an Apama installation") - #create the project and add required bundles - + # create the project and add required bundles apama_project = ProjectHelper(self, name) apama_project.create(existingProject) - #need to have a version independent addition or this will need to be maintained. + # need to have a version independent addition or this will need to be maintained. apama_project.addBundle("Automatic onApplicationInitialized") apama_project.addBundle("Cumulocity IoT > Cumulocity Client") apama_project.addBundle("Cumulocity IoT > Event Definitions for Cumulocity") @@ -159,11 +161,61 @@ def getTestSubjectEPLApps(self): eplAppsPaths.append(os.path.join(self.project.EPL_APPS, eplApp)) return eplAppsPaths + def prepareTenant(self): + """ + Prepares the tenant for a test by deleting all devices created by previous tests, and clearing all active alarms. + """ + self.log.info('Preparing tenant to run test(s)') + # Clear all active alarms + self._clearActiveAlarms() + # Delete devices that were created by tests + self._deleteTestDevices() + + def _clearActiveAlarms(self): + """ + Clears all active alarms as part of a pre-test tenant cleanup. + """ + self.log.info("Clearing active alarms") + self.platform.getC8YConnection().do_request_json('PUT', '/alarm/alarms?status=ACTIVE', {"status": "CLEARED"}) + + def _deleteTestDevices(self): + """ + Deletes all ManagedObjects that have name prefixed with "PYSYS_" and the 'c8y_isDevice' param as part of pre-test tenant cleanup. + """ + self.log.info("Deleting old test devices") + # Retrieving test devices + PAGE_SIZE = 100 # By default, pageSize = 5 for querying to C8y + resp = self.platform.getC8YConnection().do_get( + "/inventory/managedObjects" + + f"?query=has(c8y_IsDevice)+and+name+eq+'{self.TEST_DEVICE_PREFIX}*'" + + f"&pageSize={PAGE_SIZE}¤tPage=1&withTotalPages=true") + testDevices = resp['managedObjects'] + # Make sure we retrieve all pages from query + TOTAL_PAGES = resp['statistics']['totalPages'] + if TOTAL_PAGES > 1: + for currentPage in range(2, TOTAL_PAGES + 1): + resp = self.platform.getC8YConnection().do_get( + "/inventory/managedObjects" + + f"?query=has(c8y_IsDevice)+and+name+eq+'{self.TEST_DEVICE_PREFIX}*'" + + f"&pageSize={PAGE_SIZE}¤tPage={currentPage}") + testDevices = testDevices + resp['managedObjects'] + + # Deleting test devices + testDeviceIds = [device['id'] for device in testDevices] + for deviceId in testDeviceIds: + resp = self.platform.getC8YConnection().request('DELETE', f'/inventory/managedObjects/{deviceId}') + + class LocalCorrelatorSimpleTest(ApamaC8YBaseTest): """ Base test for running test with no run.py with local correlator connected to Cumulocity IoT. """ + def setup(self): + super(LocalCorrelatorSimpleTest, self).setup() + # Prepare the tenant for the test by performing a pre-test cleanup of artifacts from previous tests + self.prepareTenant() + def execute(self): """ Runs all the tests in the Input directory against the applications configured in the EPL_APPS @@ -175,7 +227,6 @@ def execute(self): if not os.path.isdir(self.project.EPL_TESTING_SDK): self.abort(BLOCKED, f'EPL_TESTING_SDK is not valid ({self.project.EPL_TESTING_SDK}). Please set the EPL_TESTING_SDK environment variable.') - self.log.info(f"Connecting to Cumulocity platform at {self.project.CUMULOCITY_SERVER_URL} as user {self.project.CUMULOCITY_USERNAME}") from apama.correlator import CorrelatorHelper # Create test project and add C8Y properties and EPL Apps project = self.createProject("test-project") @@ -262,38 +313,29 @@ class EPLAppsSimpleTest(ApamaC8YBaseTest): """ def setup(self): - super(ApamaC8YBaseTest, self).setup() + super(EPLAppsSimpleTest, self).setup() + # connect to the platform self.tests = None self.apps = None - self.platform = None self.eplapps = None self.addCleanupFunction(lambda: self.shutdown()) - self.EPL_APP_PREFIX = "PYSYS_" + self.EPL_APP_PREFIX = self.TEST_DEVICE_PREFIX self.EPL_TEST_APP_PREFIX = self.EPL_APP_PREFIX + "TEST_" - self.TEST_DEVICE_PREFIX = self.EPL_APP_PREFIX # Prefix test device names the same way as EPL apps - # Check EPL_TESTING_SDK env is set if not os.path.isdir(self.project.EPL_TESTING_SDK): self.abort(BLOCKED, f'EPL_TESTING_SDK is not valid ({self.project.EPL_TESTING_SDK}). Please set the EPL_TESTING_SDK environment variable.') - # connect to the platform - self.platform = CumulocityPlatform(self) self.eplapps = EPLApps(self.platform.getC8YConnection()) + self.prepareTenant() - self._prepareTenant() - - def _prepareTenant(self): + def prepareTenant(self): """ - Prepares the tenant for a test by deleting all EPL apps uploaded - and devices created by previous tests, and clearing all active alarms. + 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 there's no possibility existing test apps raising alarms or creating devices """ - self.log.info('Preparing tenant to run test(s)') - # Delete all EPL Apps which have been uploaded by tests - # This is done first so that there's no possibility existing test apps raising alarms or creating devices self._deleteTestEPLApps() - # Clear all active alarms - self._clearActiveAlarms() - # Delete devices that were created by tests - self._deleteTestDevices() + super(EPLAppsSimpleTest, self).prepareTenant() + def _deleteTestEPLApps(self): """ @@ -311,40 +353,6 @@ def _deleteTestEPLApps(self): for name in appsToDelete: self.eplapps.delete(name) - def _clearActiveAlarms(self): - """ - Clears all active alarms as part of a pre-test tenant cleanup. - """ - self.log.info("Clearing active alarms...") - self.platform.getC8YConnection().do_request_json('PUT', '/alarm/alarms?status=ACTIVE', {"status": "CLEARED"}) - - def _deleteTestDevices(self): - """ - Deletes all ManagedObjects that have name prefixed with "PYSYS_" and the 'c8y_isDevice' param - as part of pre-test tenant cleanup. - """ - # Retrieving test devices - PAGE_SIZE = 100 # By default, pageSize = 5 for querying to C8y - resp = self.platform.getC8YConnection().do_get( - "/inventory/managedObjects" + - f"?query=has(c8y_IsDevice)+and+name+eq+'{self.TEST_DEVICE_PREFIX}*'" + - f"&pageSize={PAGE_SIZE}¤tPage=1&withTotalPages=true") - testDevices = resp['managedObjects'] - # Make sure we retrieve all pages from query - TOTAL_PAGES = resp['statistics']['totalPages'] - if TOTAL_PAGES > 1: - for currentPage in range(2, TOTAL_PAGES + 1): - resp = self.platform.getC8YConnection().do_get( - "/inventory/managedObjects" + - f"?query=has(c8y_IsDevice)+and+name+eq+'{self.TEST_DEVICE_PREFIX}*'" + - f"&pageSize={PAGE_SIZE}¤tPage={currentPage}") - testDevices = testDevices + resp['managedObjects'] - - # Deleting test devices - testDeviceIds = [device['id'] for device in testDevices] - for deviceId in testDeviceIds: - resp = self.platform.getC8YConnection().request('DELETE', f'/inventory/managedObjects/{deviceId}') - def execute(self): """ Runs all the tests in the Input directory against the applications configured in the EPL_APPS