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

test: ft case for deleting subscribing app

Browse Source
This commit is contained in:
eeff
2023-12-04 17:29:13 +08:00
parent 43a87bcdcd
commit b10b964ab3
2 changed files with 22 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/ft/app/test_app.py
+17
View File
@@ -24,6 +24,23 @@ class TestDriver:
assert 200 == response.status_code
assert error.NEU_ERR_SUCCESS == response.json()['error']
@description(given="app node with subscriptions", when="delete node", then="should success")
@pytest.mark.parametrize('app,plugin', [('mqtt', config.PLUGIN_MQTT)])
def test_delete_app(self, app, plugin):
driver = 'modbus'
group = 'group'
try:
api.add_node_check(node=app, plugin=plugin)
api.add_node_check(node=driver, plugin=config.PLUGIN_MODBUS_TCP)
api.modbus_tcp_node_setting(node=driver, port=502)
api.add_tags_check(node=driver, group=group, tags=hold_int16)
api.subscribe_group_check(app=app, driver=driver, group=group)
response = api.del_node(app)
assert 200 == response.status_code
assert error.NEU_ERR_SUCCESS == response.json()['error']
finally:
api.del_node(driver)
@description(given="modbus tcp node", when="setting modbus tcp node and groups/tags", then="setting success")
def test_modbus_tcp_setting(self):
response = api.modbus_tcp_node_setting(node='modbus-tcp-1', port=502)
tests/ft/neuron/api.py
+5 -2
View File
@@ -284,12 +284,15 @@ def del_template_tags(tmpl, group, tags):
def get_template_tags(tmpl, group, name):
return requests.get(url=config.BASE_URL + "/api/v2/template/tag", headers={"Authorization": config.default_jwt}, params={"template": tmpl, "group": group, "name": name})
def subscribe_group(app, driver, group):
return requests.post(url=config.BASE_URL + "/api/v2/subscribe", headers={"Authorization": config.default_jwt}, json={"app": app, "driver": driver, "group": group})
@gen_check
def subscribe_group(app, driver, group, params={}):
return requests.post(url=config.BASE_URL + "/api/v2/subscribe", headers={"Authorization": config.default_jwt}, json={"app": app, "driver": driver, "group": group, "params": params})
@gen_check
def subscribe_groups(app, groups):
return requests.post(url=config.BASE_URL + "/api/v2/subscribes", headers={"Authorization": config.default_jwt}, json={"app": app, "groups": groups})
@gen_check
def unsubscribe_group(app, driver, group):
return requests.delete(url=config.BASE_URL + "/api/v2/subscribe", headers={"Authorization": config.default_jwt}, json={"app": app, "driver": driver, "group": group})