mirror of
https://github.com/emqx/neuron.git
synced 2026-10-18 21:50:42 +08:00
fengzero
228 lines
7.7 KiB
Python
228 lines
7.7 KiB
Python
import neuron.api as api
|
|
import neuron.error as error
|
|
import neuron.config as config
|
|
from neuron.common import *
|
|
|
|
hold_int16 = [{"name": "hold_int16", "address": "1!400001",
|
|
"attribute": config.NEU_TAG_ATTRIBUTE_RW, "type": config.NEU_TYPE_INT16}]
|
|
|
|
|
|
class TestAppsExportImport:
|
|
|
|
@description(given="mqtt node subscribed to modbus group",
|
|
when="export apps",
|
|
then="returns node config with subscriptions")
|
|
def test_export_app_with_subscriptions(self):
|
|
driver = 'modbus-export-test'
|
|
app = 'mqtt-export-test'
|
|
group = 'group1'
|
|
|
|
try:
|
|
api.add_node_check(node=driver, plugin=config.PLUGIN_MODBUS_TCP)
|
|
api.modbus_tcp_node_setting(node=driver, port=502)
|
|
api.add_group_check(node=driver, group=group)
|
|
api.add_tags_check(node=driver, group=group, tags=hold_int16)
|
|
|
|
api.add_node_check(node=app, plugin=config.PLUGIN_MQTT)
|
|
api.subscribe_group_check(app=app, driver=driver, group=group)
|
|
|
|
response = api.get_apps()
|
|
assert 200 == response.status_code
|
|
assert 'attachment' in response.headers.get('Content-Disposition', '')
|
|
|
|
data = response.json()
|
|
assert 'apps' in data
|
|
|
|
app_data = None
|
|
for node in data['apps']:
|
|
if node['name'] == app:
|
|
app_data = node
|
|
break
|
|
|
|
assert app_data is not None
|
|
assert app_data['plugin'] == config.PLUGIN_MQTT
|
|
assert 'groups' in app_data
|
|
assert len(app_data['groups']) >= 1
|
|
|
|
sub = app_data['groups'][0]
|
|
assert sub['driver'] == driver
|
|
assert sub['group'] == group
|
|
|
|
finally:
|
|
api.del_node(app)
|
|
api.del_node(driver)
|
|
|
|
@description(given="app node does not exist",
|
|
when="import new app node",
|
|
then="node created successfully")
|
|
def test_import_new_app_node(self):
|
|
app = 'mqtt-import-new'
|
|
|
|
try:
|
|
api.del_node(app)
|
|
|
|
apps = [{
|
|
"name": app,
|
|
"plugin": config.PLUGIN_MQTT,
|
|
"params": {
|
|
"client-id": "neuron_test_import",
|
|
"qos": 0,
|
|
"format": 0,
|
|
"write-req-topic": f"/neuron/{app}/write/req",
|
|
"write-resp-topic": f"/neuron/{app}/write/resp",
|
|
"offline-cache": False,
|
|
"host": "broker.emqx.io",
|
|
"port": 1883,
|
|
"ssl": False
|
|
},
|
|
"groups": [],
|
|
"tags": ""
|
|
}]
|
|
|
|
response = api.put_apps(apps)
|
|
assert 200 == response.status_code
|
|
assert error.NEU_ERR_SUCCESS == response.json()['error']
|
|
|
|
response = api.get_nodes_state(app)
|
|
assert 200 == response.status_code
|
|
|
|
finally:
|
|
api.del_node(app)
|
|
|
|
@description(given="app node already exists",
|
|
when="import same app node",
|
|
then="old node deleted and new node created")
|
|
def test_import_existing_app_node_overwrite(self):
|
|
app = 'mqtt-import-overwrite'
|
|
|
|
try:
|
|
api.add_node_check(node=app, plugin=config.PLUGIN_MQTT)
|
|
|
|
response = api.get_nodes_state(app)
|
|
assert 200 == response.status_code
|
|
|
|
apps = [{
|
|
"name": app,
|
|
"plugin": config.PLUGIN_MQTT,
|
|
"params": {
|
|
"client-id": "neuron_overwrite_test",
|
|
"qos": 1,
|
|
"format": 0,
|
|
"write-req-topic": f"/neuron/{app}/write/req",
|
|
"write-resp-topic": f"/neuron/{app}/write/resp",
|
|
"offline-cache": False,
|
|
"host": "broker.emqx.io",
|
|
"port": 1883,
|
|
"ssl": False
|
|
},
|
|
"groups": [],
|
|
"tags": ""
|
|
}]
|
|
|
|
response = api.put_apps(apps)
|
|
assert 200 == response.status_code
|
|
assert error.NEU_ERR_SUCCESS == response.json()['error']
|
|
|
|
response = api.get_nodes_state(app)
|
|
assert 200 == response.status_code
|
|
|
|
response = api.get_apps(names=[app])
|
|
assert 200 == response.status_code
|
|
data = response.json()
|
|
assert len(data['apps']) == 1
|
|
assert 'neuron_overwrite_test' in str(data['apps'][0].get('params', ''))
|
|
|
|
finally:
|
|
api.del_node(app)
|
|
|
|
@description(given="subscription driver does not exist",
|
|
when="import app with subscription",
|
|
then="returns subscription error")
|
|
def test_import_app_subscription_driver_not_exist(self):
|
|
app = 'mqtt-import-sub-error'
|
|
non_exist_driver = 'non-exist-driver-12345'
|
|
|
|
try:
|
|
api.del_node(app)
|
|
|
|
apps = [{
|
|
"name": app,
|
|
"plugin": config.PLUGIN_MQTT,
|
|
"params": {
|
|
"client-id": "neuron_sub_test",
|
|
"qos": 0,
|
|
"format": 0,
|
|
"write-req-topic": f"/neuron/{app}/write/req",
|
|
"write-resp-topic": f"/neuron/{app}/write/resp",
|
|
"offline-cache": False,
|
|
"host": "broker.emqx.io",
|
|
"port": 1883,
|
|
"ssl": False
|
|
},
|
|
"tags": "",
|
|
"groups": [{
|
|
"driver": non_exist_driver,
|
|
"group": "group1"
|
|
}]
|
|
}]
|
|
|
|
response = api.put_apps(apps)
|
|
assert response.json()['error'] != error.NEU_ERR_SUCCESS
|
|
|
|
finally:
|
|
api.del_node(app)
|
|
|
|
@description(given="configured app node",
|
|
when="export then delete then import",
|
|
then="node restored with same config")
|
|
def test_export_import_roundtrip(self):
|
|
driver = 'modbus-roundtrip'
|
|
app = 'mqtt-roundtrip'
|
|
group = 'group1'
|
|
|
|
try:
|
|
api.add_node_check(node=driver, plugin=config.PLUGIN_MODBUS_TCP)
|
|
api.modbus_tcp_node_setting(node=driver, port=502)
|
|
api.add_group_check(node=driver, group=group)
|
|
api.add_tags_check(node=driver, group=group, tags=hold_int16)
|
|
|
|
api.add_node_check(node=app, plugin=config.PLUGIN_MQTT)
|
|
api.subscribe_group_check(app=app, driver=driver, group=group)
|
|
|
|
response = api.get_apps(names=[app])
|
|
assert 200 == response.status_code
|
|
exported_data = response.json()
|
|
|
|
assert 'apps' in exported_data
|
|
assert len(exported_data['apps']) == 1
|
|
app_config = exported_data['apps'][0]
|
|
assert app_config['name'] == app
|
|
|
|
api.del_node(app)
|
|
response = api.get_nodes_state(app)
|
|
assert response.json()['error'] == error.NEU_ERR_NODE_NOT_EXIST
|
|
|
|
response = api.put_apps(exported_data['apps'])
|
|
assert 200 == response.status_code
|
|
assert error.NEU_ERR_SUCCESS == response.json()['error']
|
|
|
|
response = api.get_nodes_state(app)
|
|
assert 200 == response.status_code
|
|
|
|
response = api.get_subscribe_group(app)
|
|
assert 200 == response.status_code
|
|
groups = response.json().get('groups', [])
|
|
assert len(groups) >= 1
|
|
|
|
found = False
|
|
for g in groups:
|
|
if g['driver'] == driver and g['group'] == group:
|
|
found = True
|
|
break
|
|
assert found, "Subscription not restored"
|
|
|
|
finally:
|
|
api.del_node(app)
|
|
api.del_node(driver)
|
|
|