Skip to content

Commit

Permalink
Refactor test file structure
Browse files Browse the repository at this point in the history 
Remove old test file for app engine device get samples data and refactor
test file structure:
- Move `test_app_engine.py` to
`app_engine/device/app_engine_device_get_samples_error_cases.py`
- Add resources file for test data

Signed-off-by: Osman Hadzic <[email protected]>
  • Loading branch information
@osmanhadzic
osmanhadzic committed Oct 31, 2024
1 parent a3fa8db commit f02686c
Show file tree
Hide file tree
Showing 3 changed files with 442 additions and 4 deletions.
290 changes: 290 additions & 0 deletions ...engine/device/app_engine_device_get_samples/test_app_engine_device_get_samples_default.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,290 @@
# SPDX-FileCopyrightText: 2024 SECO Mind Srl
#
# SPDX-License-Identifier: Apache-2.0

import subprocess
import re
import ast

from resources import (
expected_output_object,
list_of_params_endpoints,
list_of_nonparams_endpoints,
map_of_params_data,
map_of_nonparams_data,
)


def test_get_sample_data_for_device_individual_parametric_datastream(astarte_env_vars):
device_id = astarte_env_vars["device_test_1"]
astarte_url = astarte_env_vars["astarte_url"]
realm = astarte_env_vars["realm"]
jwt = astarte_env_vars["jwt"]

interface_name = "test.astarte-platform.device.individual.parametric.Datastream"

for path in list_of_params_endpoints:
arg_list = [
"astartectl",
"appengine",
"devices",
"get-samples",
device_id,
interface_name,
path,
"-t",
jwt,
"-u",
astarte_url,
"-r",
realm,
]
sample_data_result = subprocess.run(arg_list, capture_output=True, text=True)
sample_data_result.stdout = _remove_table_border(sample_data_result)

assert sample_data_result.stdout != ""

sample_data_result.stdout = re.sub(r"\s+", " ", sample_data_result.stdout).strip()
sample_data_result.stdout = sample_data_result.stdout.replace(" ", ",").split(",")
sample_data_result.stdout = [parse_element(el) for el in sample_data_result.stdout]

list = []
for element in sample_data_result.stdout:
if element.startswith("["):
list.append(element.replace("[", ""))
elif element.endswith("]"):
list.append(element.replace("]", ""))

list_length = len(sample_data_result.stdout) - 1

if list != []:
assert list == map_of_params_data[path]
else:
assert sample_data_result.stdout[list_length] == map_of_params_data[path]


def test_get_sample_data_for_device_individual_nonparametric_datastream(astarte_env_vars):
device_id = astarte_env_vars["device_test_1"]
astarte_url = astarte_env_vars["astarte_url"]
realm = astarte_env_vars["realm"]
jwt = astarte_env_vars["jwt"]

interface_name = "test.astarte-platform.device.individual.nonparametric.Datastream"

for path in list_of_nonparams_endpoints:
arg_list = [
"astartectl",
"appengine",
"devices",
"get-samples",
device_id,
interface_name,
path,
"-t",
jwt,
"-u",
astarte_url,
"-r",
realm,
]
sample_data_result = subprocess.run(arg_list, capture_output=True, text=True)
sample_data_result.stdout = _remove_table_border(sample_data_result)

assert sample_data_result.stdout != ""

sample_data_result.stdout = re.sub(r"\s+", " ", sample_data_result.stdout).strip()
sample_data_result.stdout = sample_data_result.stdout.replace(" ", ",").split(",")
sample_data_result.stdout = [parse_element(el) for el in sample_data_result.stdout]

list = []
for element in sample_data_result.stdout:
if element.startswith("["):
list.append(element.replace("[", ""))
elif element.endswith("]"):
list.append(element.replace("]", ""))

list_length = len(sample_data_result.stdout) - 1

if list != []:
assert list == map_of_nonparams_data[path]
else:
assert sample_data_result.stdout[list_length] == map_of_nonparams_data[path]


def test_get_sample_data_for_device_object_nonparametric_datastream(astarte_env_vars):
device_id = astarte_env_vars["device_test_1"]
astarte_url = astarte_env_vars["astarte_url"]
realm = astarte_env_vars["realm"]
jwt = astarte_env_vars["jwt"]

interface_name = "test.astarte-platform.device.object.nonparametric.Datastream"
path = "/the"

arg_list = [
"astartectl",
"appengine",
"devices",
"get-samples",
device_id,
interface_name,
path,
"-t",
jwt,
"-u",
astarte_url,
"-r",
realm,
]
sample_data_result = subprocess.run(arg_list, capture_output=True, text=True)
sample_data_result.stdout = _remove_table_border(sample_data_result)

sample_data_result.stdout = re.sub(r"\s+", " ", sample_data_result.stdout).strip()
sample_data_result.stdout = sample_data_result.stdout.replace(" ", ",").split(",")
sample_data_result.stdout = [parse_element(el) for el in sample_data_result.stdout]

assert sample_data_result.stdout == expected_output_object


def test_get_sample_data_for_device_object_parametric_datastream(astarte_env_vars):
device_id = astarte_env_vars["device_test_1"]
astarte_url = astarte_env_vars["astarte_url"]
realm = astarte_env_vars["realm"]
jwt = astarte_env_vars["jwt"]

interface_name = "test.astarte-platform.device.object.parametric.Datastream"
path = "/a"

arg_list = [
"astartectl",
"appengine",
"devices",
"get-samples",
device_id,
interface_name,
path,
"-t",
jwt,
"-u",
astarte_url,
"-r",
realm,
]
sample_data_result = subprocess.run(arg_list, capture_output=True, text=True)
sample_data_result.stdout = _remove_table_border(sample_data_result)

sample_data_result.stdout = re.sub(r"\s+", " ", sample_data_result.stdout).strip()
sample_data_result.stdout = sample_data_result.stdout.replace(" ", ",").split(",")
sample_data_result.stdout = [parse_element(el) for el in sample_data_result.stdout]

assert sample_data_result.stdout == expected_output_object


def test_get_sample_data_for_server_individual_parametric_datastream(astarte_env_vars):
device_id = astarte_env_vars["device_test_1"]
astarte_url = astarte_env_vars["astarte_url"]
realm = astarte_env_vars["realm"]
jwt = astarte_env_vars["jwt"]

interface_name = "test.astarte-platform.server.individual.parametric.Datastream"

for path in list_of_params_endpoints:
arg_list = [
"astartectl",
"appengine",
"devices",
"get-samples",
device_id,
interface_name,
path,
"-t",
jwt,
"-u",
astarte_url,
"-r",
realm,
]
sample_data_result = subprocess.run(arg_list, capture_output=True, text=True)
sample_data_result.stdout = _remove_table_border(sample_data_result)

assert sample_data_result.stdout != ""

sample_data_result.stdout = re.sub(r"\s+", " ", sample_data_result.stdout).strip()
sample_data_result.stdout = sample_data_result.stdout.replace(" ", ",").split(",")
sample_data_result.stdout = [parse_element(el) for el in sample_data_result.stdout]

parsed_elements_list = []
for element in sample_data_result.stdout:
if element.startswith("["):
parsed_elements_list.append(element.replace("[", ""))
elif element.endswith("]"):
parsed_elements_list.append(element.replace("]", ""))

list_length = len(sample_data_result.stdout) - 1

if parsed_elements_list != []:
assert parsed_elements_list == map_of_params_data[path]
else:
assert sample_data_result.stdout[list_length] == map_of_params_data[path]


def test_get_sample_data_for_server_individual_nonparametric_datastream(astarte_env_vars):
device_id = astarte_env_vars["device_test_1"]
astarte_url = astarte_env_vars["astarte_url"]
realm = astarte_env_vars["realm"]
jwt = astarte_env_vars["jwt"]

interface_name = "test.astarte-platform.server.individual.nonparametric.Datastream"

for path in list_of_nonparams_endpoints:
arg_list = [
"astartectl",
"appengine",
"devices",
"get-samples",
device_id,
interface_name,
path,
"-t",
jwt,
"-u",
astarte_url,
"-r",
realm,
]
sample_data_result = subprocess.run(arg_list, capture_output=True, text=True)
sample_data_result.stdout = _remove_table_border(sample_data_result)

assert sample_data_result.stdout != ""

sample_data_result.stdout = re.sub(r"\s+", " ", sample_data_result.stdout).strip()
sample_data_result.stdout = sample_data_result.stdout.replace(" ", ",").split(",")
sample_data_result.stdout = [parse_element(el) for el in sample_data_result.stdout]

parsed_elements_list = []
for element in sample_data_result.stdout:
if element.startswith("["):
parsed_elements_list.append(element.replace("[", ""))
elif element.endswith("]"):
parsed_elements_list.append(element.replace("]", ""))

list_length = len(sample_data_result.stdout) - 1

if parsed_elements_list != []:
assert parsed_elements_list == map_of_nonparams_data[path]
else:
assert sample_data_result.stdout[list_length] == map_of_nonparams_data[path]


def _remove_table_border(table):
table = table.stdout.replace("\n", "")

# Remove all characters with ASCII code in range 9472 to 9584
table = "".join(ch for ch in table if not (9472 <= ord(ch) <= 9584))
return table


def parse_element(element):
element = element.strip()
if element.startswith("[") and element.endswith("]"):
return ast.literal_eval(element)
return element
45 changes: 41 additions & 4 deletions tests/test_app_engine.py → ..._engine_device_get_samples_error_cases.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

import subprocess


def test_get_sample_data_for_nonexisting_device(astarte_env_vars):
device_id = "nonexisting_device"
astarte_url = astarte_env_vars["astarte_url"]
Expand Down Expand Up @@ -34,14 +35,15 @@ def test_get_sample_data_for_nonexisting_device(astarte_env_vars):

assert "Device not found" in sample_data_result.stderr

def test_get_sample_data_for_existing_device(astarte_env_vars):

def test_get_sample_data_for_nonexisting_interface(astarte_env_vars):
device_id = astarte_env_vars["device_test_1"]
astarte_url = astarte_env_vars["astarte_url"]
realm = astarte_env_vars["realm"]
jwt = astarte_env_vars["jwt"]

interface_name = "test.astarte-platform.device.object.parametric.Datastream"
path = "/a"
interface_name = "nonexisting_interface"
path = "any_path"

arg_list = [
"astartectl",
Expand All @@ -60,6 +62,41 @@ def test_get_sample_data_for_existing_device(astarte_env_vars):
"-o",
"json",
]

sample_data_result = subprocess.run(arg_list, capture_output=True, text=True)
output = f"Device {device_id} has no interface named nonexisting_interface"
assert output in sample_data_result.stderr


def test_get_sample_data_for_nonexisting_path(astarte_env_vars):
device_id = astarte_env_vars["device_test_1"]
astarte_url = astarte_env_vars["astarte_url"]
realm = astarte_env_vars["realm"]
jwt = astarte_env_vars["jwt"]

interface_name = "test.astarte-platform.device.object.parametric.Datastream"
path = "nonexisting_path"

arg_list = [
"astartectl",
"appengine",
"devices",
"get-samples",
device_id,
interface_name,
path,
"-t",
jwt,
"-u",
astarte_url,
"-r",
realm,
"-o",
"json",
]

assert "44" in sample_data_result.stdout
sample_data_result = subprocess.run(arg_list, capture_output=True, text=True)
assert (
"nonexisting_path does not match valid query paths for endpoint /%{parameter}/stringarray"
in sample_data_result.stderr
)
Loading

0 comments on commit f02686c

Please sign in to comment.