GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-27 04:19:50 +08:00
Code Issues Packages Projects Releases Wiki Activity

test: read group query ft

Browse Source
This commit is contained in:
eeff
2024-01-03 17:32:15 +08:00
parent 40025265f2
commit 2c09db89c0
2 changed files with 45 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/ft/driver/test_modbus.py
+39 -5
View File
@@ -68,15 +68,15 @@ coil_bit_5 = [{"name": "coil_bit_5", "address": "1!000010",
"attribute": config.NEU_TAG_ATTRIBUTE_RW, "type": config.NEU_TYPE_BIT}]
input_bit_1 = [{"name": "input_bit_1", "address": "1!100001",
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT}]
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT, "description": "bit_1"}]
input_bit_2 = [{"name": "input_bit_2", "address": "1!100002",
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT}]
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT, "description": "bit_2"}]
input_bit_3 = [{"name": "input_bit_3", "address": "1!100003",
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT}]
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT, "description": "bit_3"}]
input_bit_4 = [{"name": "input_bit_4", "address": "1!100005",
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT}]
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT, "description": "bit_4"}]
input_bit_5 = [{"name": "input_bit_5", "address": "1!100010",
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT}]
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT, "description": "bit_5"}]
input_register_bit = [{"name": "input_register_bit", "address": "1!30101.15",
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT}]
@@ -494,6 +494,40 @@ class TestModbus:
assert 0 == api.read_tag(
node=param[0], group='group', tag=input_bit_5[0]['name'])
@description(given="created modbus node and tags", when="read tags fuzz", then="should return correct result")
def test_read_tags_fuzz(self, param):
response = api.read_tags(
node=param[0], group='group', query={"name": "input_bit"})
assert 200 == response.status_code
assert 5 == len(response.json()["tags"])
response = api.read_tags(
node=param[0], group='group', query={"name": "input_bit_1"})
assert 200 == response.status_code
assert 1 == len(response.json()["tags"])
response = api.read_tags(
node=param[0], group='group', query={"description": "bit"})
assert 200 == response.status_code
assert 5 == len(response.json()["tags"])
response = api.read_tags(
node=param[0], group='group', query={"description": "bit_1"})
assert 200 == response.status_code
assert 1 == len(response.json()["tags"])
response = api.read_tags(
node=param[0], group='group', query={"name": "input_bit", "description": "bit"})
assert 200 == response.status_code
assert 5 == len(response.json()["tags"])
response = api.read_tags(
node=param[0], group='group', query={"name": "input_bit_1", "description": "bit_1"})
assert 200 == response.status_code
assert 1 == len(response.json()["tags"])
@description(given="created modbus node and tags", when="read tags synchronously", then="read fail")
def test_read_tags_sync(self, param):
assert error.NEU_ERR_PLUGIN_NOT_SUPPORT_READ_SYNC == api.read_tag_error(
tests/ft/neuron/api.py
+6 -3
View File
@@ -145,12 +145,15 @@ def get_tags(node, group):
return requests.get(url=config.BASE_URL + "/api/v2/tags", headers={"Authorization": config.default_jwt}, params={"node": node, "group": group})
def read_tags(node, group, sync=False):
return requests.post(url=config.BASE_URL + "/api/v2/read", headers={"Authorization": config.default_jwt}, json={"node": node, "group": group, "sync": sync})
def read_tags(node, group, sync=False, query=None):
body = {"node": node, "group": group, "sync": sync}
if query:
body["query"] = query
return requests.post(url=config.BASE_URL + "/api/v2/read", headers={"Authorization": config.default_jwt}, json=body)
def read_tag(node, group, tag, sync=False):
response = read_tags(node, group, sync)
response = read_tags(node, group, sync, query={"name": tag})
assert 200 == response.status_code
x = list(filter(lambda x: x['name'] == tag,
response.json()['tags']))