diff options
author | initard <solo@softwareag.com> | 2022-04-08 08:46:49 +0100 |
---|---|---|
committer | initard <solo@softwareag.com> | 2022-04-13 09:48:40 +0100 |
commit | 6fa381f99a8f5b16bcda963afb4034a31e086db6 (patch) | |
tree | a77841a36e53ea0f3da3f67d0f9ca72dc5abed19 /tests | |
parent | 924980f9f850ee6c08f207d71de983cf54a46f32 (diff) |
moving python tedge mapper tests to rust #958
- removed tedge_mapper_c8y_negative
- removed tedge_mapper_c8y_positive
- added the same tests in rust using mqtt_tests
Signed-off-by: initard <solo@softwareag.com>
Diffstat (limited to 'tests')
4 files changed, 0 insertions, 227 deletions
diff --git a/tests/PySys/tedge_mapper_c8y/tedge_mapper_c8y_negative/pysystest.xml b/tests/PySys/tedge_mapper_c8y/tedge_mapper_c8y_negative/pysystest.xml deleted file mode 100644 index 36f4212b..00000000 --- a/tests/PySys/tedge_mapper_c8y/tedge_mapper_c8y_negative/pysystest.xml +++ /dev/null @@ -1,28 +0,0 @@ -<?xml version="1.0" encoding="utf-8"?> -<pysystest type="auto"> - <description> - <title> - Validate the Cumulocity Mapper that publishes an error message on tedge/errors - when it receives an invalid JSON message on tedge/measurements. - </title> - <purpose> - <![CDATA[ ]]> - </purpose> - </description> - <classification> - <groups inherit="true"> - <group> - </group> - </groups> - <modes inherit="true"> - </modes> - </classification> - <data> - <class name="TedgeMapperC8y" module="run" /> - </data> - <traceability> - <requirements> - <requirement id="" /> - </requirements> - </traceability> -</pysystest> diff --git a/tests/PySys/tedge_mapper_c8y/tedge_mapper_c8y_negative/run.py b/tests/PySys/tedge_mapper_c8y/tedge_mapper_c8y_negative/run.py deleted file mode 100644 index e7f9f25d..00000000 --- a/tests/PySys/tedge_mapper_c8y/tedge_mapper_c8y_negative/run.py +++ /dev/null @@ -1,70 +0,0 @@ -from pysys.basetest import BaseTest - -import time - -""" -Validate an invalid JSON message that is published -on tedge/errors by tedge-mapper c8y - -Given a configured system -When we start the tedge-mapper-c8y systemctl service -When we start tedge sub with sudo in the background -When we publish an invalid Thin Edge JSON message -Then we kill tedge sub with sudo as it is running with a different user account -Then we validate the error message in the output of tedge sub -Then we stop the tedge-mapper-c8y systemctl service - -""" - - -class TedgeMapperC8y(BaseTest): - def setup(self): - self.tedge = "/usr/bin/tedge" - self.sudo = "/usr/bin/sudo" - - mapper = self.startProcess( - command=self.sudo, - arguments=["systemctl", "start", "tedge-mapper-c8y"], - stdouterr="tedge_mapper_c8y", - ) - - self.addCleanupFunction(self.mapper_cleanup) - - def execute(self): - sub = self.startProcess( - command=self.sudo, - arguments=[self.tedge, "mqtt", "sub", "--no-topic", "tedge/errors"], - stdouterr="tedge_sub", - background=True, - ) - - # Wait for a small amount of time to give tedge sub time - # to initialize. This is a heuristic measure. - # Without an additional wait we observe failures in 1% of the test - # runs. - time.sleep(0.1) - - pub = self.startProcess( - command=self.sudo, - arguments=[self.tedge, "mqtt", "pub", "tedge/measurements", "{"], - stdouterr="tedge_temp", - ) - - # Kill the subscriber process explicitly with sudo as PySys does - # not have the rights to do it - kill = self.startProcess( - command=self.sudo, - arguments=["killall", "tedge"], - stdouterr="kill_out", - ) - - def validate(self): - self.assertGrep("tedge_sub.out", "Invalid JSON", contains=True) - - def mapper_cleanup(self): - self.log.info("mapper_cleanup") - mapper = self.startProcess( - command=self.sudo, - arguments=["systemctl", "stop", "tedge-mapper-c8y"], - stdouterr="tedge_mapper_c8y", - ) diff --git a/tests/PySys/tedge_mapper_c8y/tedge_mapper_c8y_positive/pysystest.xml b/tests/PySys/tedge_mapper_c8y/tedge_mapper_c8y_positive/pysystest.xml deleted file mode 100644 index 97128c57..00000000 --- a/tests/PySys/tedge_mapper_c8y/tedge_mapper_c8y_positive/pysystest.xml +++ /dev/null @@ -1,29 +0,0 @@ -<?xml version="1.0" encoding="utf-8"?> -<pysystest type="auto"> - <description> - <title> - Validate the Cumulocity Mapper that publishes a Thin Edge JSON message - on c8y/measurement/measurements/create - when it receives a valid JSON message on tedge/measurements. - </title> - <purpose> - <![CDATA[ ]]> - </purpose> - </description> - <classification> - <groups inherit="true"> - <group> - </group> - </groups> - <modes inherit="true"> - </modes> - </classification> - <data> - <class name="TedgeMapperC8y" module="run" /> - </data> - <traceability> - <requirements> - <requirement id="" /> - </requirements> - </traceability> -</pysystest> diff --git a/tests/PySys/tedge_mapper_c8y/tedge_mapper_c8y_positive/run.py b/tests/PySys/tedge_mapper_c8y/tedge_mapper_c8y_positive/run.py deleted file mode 100644 index d2a9f492..00000000 --- a/tests/PySys/tedge_mapper_c8y/tedge_mapper_c8y_positive/run.py +++ /dev/null @@ -1,100 +0,0 @@ -from pysys.basetest import BaseTest - -import time -import json - -""" -Validate a tedge-mapper-c8y message that is published -on c8y/measurement/measurements/create - -Given a configured system -When we start the tedge-mapper-c8y systemctl service -When we start tedge sub with sudo in the background -When we publish a Thin Edge JSON message -Then we kill tedge sub with sudo as it is running with a different user account -Then we validate the JSON message in the output of tedge sub -Then we stop the tedge-mapper-c8y systemctl service - -""" - - -class TedgeMapperC8y(BaseTest): - def setup(self): - self.tedge = "/usr/bin/tedge" - self.sudo = "/usr/bin/sudo" - - mapper = self.startProcess( - command=self.sudo, - arguments=["systemctl", "start", "tedge-mapper-c8y"], - stdouterr="tedge_mapper_c8y", - ) - - self.addCleanupFunction(self.mapper_cleanup) - - def execute(self): - sub = self.startProcess( - command=self.sudo, - arguments=[ - self.tedge, - "mqtt", - "sub", - "--no-topic", - "c8y/measurement/measurements/create", - ], - stdouterr="tedge_sub", - background=True, - ) - - # Wait for a small amount of time to give tedge sub time - # to initialize. This is a heuristic measure. - # Without an additional wait we observe failures in 1% of the test - # runs. - time.sleep(0.1) - - pub = self.startProcess( - command=self.sudo, - arguments=[ - self.tedge, - "mqtt", - "pub", - "tedge/measurements", - '{"temperature": 12, "time": "2021-06-15T17:01:15.806181503+02:00"}', - ], - stdouterr="tedge_temp", - ) - - # Kill the subscriber process explicitly with sudo as PySys does - # not have the rights to do it - kill = self.startProcess( - command=self.sudo, - arguments=["killall", "tedge"], - stdouterr="kill_out", - ) - - def validate(self): - f = open(self.output + "/tedge_sub.out", "r") - c8y_json = json.load(f) - - self.assertThat( - "actual == expected", - actual=c8y_json["type"], - expected="ThinEdgeMeasurement", - ) - self.assertThat( - "actual == expected", - actual=c8y_json["temperature"]["temperature"]["value"], - expected=12, - ) - self.assertThat( - "actual == expected", - actual=c8y_json["time"], - expected="2021-06-15T17:01:15.806181503+02:00", - ) - - def mapper_cleanup(self): - self.log.info("mapper_cleanup") - mapper = self.startProcess( - command=self.sudo, - arguments=["systemctl", "stop", "tedge-mapper-c8y"], - stdouterr="tedge_mapper_c8y", - ) |