mirror of
https://github.com/emqx/neuron.git
synced 2026-10-17 04:40:42 +08:00
eeff
398 lines
11 KiB
Python
398 lines
11 KiB
Python
import json
|
|
import time
|
|
|
|
import pytest
|
|
import pynng
|
|
|
|
import neuron.api as api
|
|
import neuron.error as error
|
|
import neuron.config as config
|
|
from neuron.common import (
|
|
case_time,
|
|
class_setup_and_teardown,
|
|
compare_float,
|
|
description,
|
|
process,
|
|
random_port,
|
|
)
|
|
|
|
|
|
NODE = "ekuiper"
|
|
PLUGIN = "eKuiper"
|
|
DRIVER = "modbus"
|
|
DRIVER_PORT = random_port()
|
|
GROUP = "group"
|
|
INTERVAL = 100
|
|
TAGS = [
|
|
{
|
|
"name": "tag0",
|
|
"address": "1!400001.",
|
|
"attribute": config.NEU_TAG_ATTRIBUTE_RW,
|
|
"type": config.NEU_TYPE_INT16,
|
|
},
|
|
{
|
|
"name": "tag1",
|
|
"address": "1!000001",
|
|
"attribute": config.NEU_TAG_ATTRIBUTE_RW,
|
|
"type": config.NEU_TYPE_BIT,
|
|
},
|
|
{
|
|
"name": "tag2",
|
|
"address": "1!400017",
|
|
"attribute": config.NEU_TAG_ATTRIBUTE_RW,
|
|
"type": config.NEU_TYPE_FLOAT,
|
|
},
|
|
{
|
|
"name": "tag3",
|
|
"address": "1!400020.10",
|
|
"attribute": config.NEU_TAG_ATTRIBUTE_RW,
|
|
"type": config.NEU_TYPE_STRING,
|
|
},
|
|
{
|
|
"name": "tag4",
|
|
"address": "1!400040.4",
|
|
"attribute": config.NEU_TAG_ATTRIBUTE_RW,
|
|
"type": config.NEU_TYPE_BYTES,
|
|
},
|
|
]
|
|
|
|
|
|
@pytest.fixture(autouse=True, scope="class")
|
|
def ekuiper_node():
|
|
api.add_node_check(NODE, PLUGIN)
|
|
yield NODE
|
|
api.del_node(NODE)
|
|
|
|
|
|
@pytest.fixture(autouse=True, scope="class")
|
|
def modbus_node(ekuiper_node):
|
|
try:
|
|
api.add_node_check(DRIVER, "Modbus TCP")
|
|
api.add_group_check(DRIVER, GROUP, INTERVAL)
|
|
api.add_tags_check(DRIVER, GROUP, tags=TAGS)
|
|
api.subscribe_group_check(ekuiper_node, DRIVER, GROUP)
|
|
p = process.start_simulator(
|
|
["./modbus_simulator", "tcp", f"{DRIVER_PORT}", "ip_v4"]
|
|
)
|
|
response = api.modbus_tcp_node_setting(DRIVER, DRIVER_PORT)
|
|
assert 200 == response.status_code
|
|
assert error.NEU_ERR_SUCCESS == response.json()["error"]
|
|
yield
|
|
finally:
|
|
# api.del_node(DRIVER)
|
|
process.stop_simulator(p)
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def conf_base():
|
|
return {
|
|
"host": "0.0.0.0",
|
|
"port": 7081,
|
|
}
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def conf_empty_host(conf_base):
|
|
return {
|
|
**conf_base,
|
|
"host": "",
|
|
}
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def conf_bad_host(conf_base):
|
|
return {
|
|
**conf_base,
|
|
"host": "8.8.8.8",
|
|
}
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def conf_bad_port(conf_base):
|
|
return {
|
|
**conf_base,
|
|
"port": 1_000_000,
|
|
}
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def mocker():
|
|
mock = Mocker(timeout=1000)
|
|
yield mock
|
|
mock.disconnect()
|
|
|
|
|
|
class Mocker:
|
|
def __init__(self, timeout=1000):
|
|
self.timeout = timeout
|
|
self.sock_ = None
|
|
|
|
@property
|
|
def sock(self):
|
|
if self.sock_:
|
|
return self.sock_
|
|
self.sock_ = pynng.Pair0(
|
|
recv_timeout=self.timeout, send_timeout=self.timeout
|
|
)
|
|
return self.sock_
|
|
|
|
@sock.setter
|
|
def sock(self, value):
|
|
if self.sock_:
|
|
self.sock_.close()
|
|
self.sock_ = value
|
|
|
|
def connect(self, host, port):
|
|
self.sock.dial(f"tcp://{host}:{port}", block=True)
|
|
|
|
def disconnect(self):
|
|
self.sock = None
|
|
|
|
def recv(self):
|
|
msg = self.sock.recv()
|
|
return json.loads(msg.decode())
|
|
|
|
def send(self, data):
|
|
msg = json.dumps(data).encode()
|
|
self.sock.send(msg)
|
|
|
|
|
|
class TesteKuiper:
|
|
@description(
|
|
given="eKuiper node and {conf}",
|
|
when="config node",
|
|
then="should fail",
|
|
)
|
|
@pytest.mark.parametrize(
|
|
"conf",
|
|
[
|
|
"conf_empty_host",
|
|
"conf_bad_host",
|
|
"conf_bad_port",
|
|
],
|
|
)
|
|
def test_ekuiper_conf_bad(self, request, ekuiper_node, conf):
|
|
conf = request.getfixturevalue(conf)
|
|
response = api.node_setting(ekuiper_node, conf)
|
|
assert 400 == response.status_code
|
|
assert error.NEU_ERR_NODE_SETTING_INVALID == response.json()["error"]
|
|
|
|
@description(
|
|
given="eKuiper node",
|
|
when="config node with already in use address",
|
|
then="should fail",
|
|
)
|
|
def test_ekuiper_conf_address_in_use(
|
|
self, mocker, ekuiper_node, conf_base
|
|
):
|
|
api.node_setting_check(ekuiper_node, conf_base)
|
|
|
|
conf_new = {**conf_base, "port": conf_base["port"] + 1}
|
|
node = ekuiper_node + "-2"
|
|
api.add_node_check(node, PLUGIN)
|
|
api.node_setting_check(node, conf_new)
|
|
|
|
response = api.node_setting(ekuiper_node, conf_new)
|
|
assert 400 == response.status_code
|
|
assert error.NEU_ERR_NODE_SETTING_INVALID == response.json()["error"]
|
|
|
|
# should be able to connect with old conf
|
|
mocker.connect(**conf_base)
|
|
|
|
@description(
|
|
given="eKuiper node",
|
|
when="configured",
|
|
then="should allow connection from eKuiper client",
|
|
)
|
|
def test_ekuiper_connect(self, mocker, ekuiper_node, conf_base):
|
|
api.node_setting_check(ekuiper_node, conf_base)
|
|
api.node_ctl(ekuiper_node, config.NEU_CTL_START)
|
|
|
|
mocker.connect(**conf_base)
|
|
|
|
@description(
|
|
given="eKuiper node",
|
|
when="stop/start node",
|
|
then="should disconnect/connect broker",
|
|
)
|
|
def test_ekuiper_start_stop(self, mocker, ekuiper_node, conf_base):
|
|
api.node_setting_check(ekuiper_node, conf_base)
|
|
api.node_ctl_check(ekuiper_node, config.NEU_CTL_STOP)
|
|
with pytest.raises(Exception):
|
|
mocker.connect(**conf_base)
|
|
api.node_ctl_check(ekuiper_node, config.NEU_CTL_START)
|
|
mocker.connect(**conf_base)
|
|
|
|
@description(
|
|
given="eKuiper node",
|
|
when="configured",
|
|
then="should upload data to eKuiper client",
|
|
)
|
|
def test_ekuiper_upload(self, mocker, ekuiper_node, conf_base):
|
|
api.node_setting_check(ekuiper_node, conf_base)
|
|
api.node_ctl(ekuiper_node, config.NEU_CTL_START)
|
|
mocker.connect(**conf_base)
|
|
|
|
time.sleep(INTERVAL * 5 / 1000)
|
|
msg = mocker.recv()
|
|
assert msg is not None
|
|
assert DRIVER == msg["node_name"]
|
|
assert GROUP == msg["group_name"]
|
|
assert "timestamp" in msg
|
|
assert len(TAGS) == len(msg["values"]) + len(msg["errors"])
|
|
for tag in TAGS:
|
|
assert tag["name"] in msg["values"] or tag["name"] in msg["errors"]
|
|
|
|
@description(
|
|
given="eKuiper node",
|
|
when="subscribed driver renamed",
|
|
then="should upload data to eKuiper client",
|
|
)
|
|
def test_ekuiper_upload_when_driver_rename(
|
|
self, mocker, ekuiper_node, conf_base
|
|
):
|
|
api.node_setting_check(ekuiper_node, conf_base)
|
|
api.node_ctl(ekuiper_node, config.NEU_CTL_START)
|
|
|
|
new_name = DRIVER + "-new"
|
|
api.update_node_check(DRIVER, new_name)
|
|
|
|
try:
|
|
mocker.connect(**conf_base)
|
|
time.sleep(INTERVAL * 5 / 1000)
|
|
msg = mocker.recv()
|
|
assert msg is not None
|
|
assert new_name == msg["node_name"]
|
|
assert GROUP == msg["group_name"]
|
|
assert "timestamp" in msg
|
|
assert len(TAGS) == len(msg["values"]) + len(msg["errors"])
|
|
for tag in TAGS:
|
|
assert (
|
|
tag["name"] in msg["values"]
|
|
or tag["name"] in msg["errors"]
|
|
)
|
|
finally:
|
|
api.update_node_check(new_name, DRIVER)
|
|
|
|
@description(
|
|
given="eKuiper node",
|
|
when="subscribed driver group renamed",
|
|
then="should upload data to eKuiper client",
|
|
)
|
|
def test_ekuiper_upload_when_group_rename(
|
|
self, mocker, ekuiper_node, conf_base
|
|
):
|
|
api.node_setting_check(ekuiper_node, conf_base)
|
|
api.node_ctl(ekuiper_node, config.NEU_CTL_START)
|
|
|
|
new_name = GROUP + "-new"
|
|
api.update_group_check(DRIVER, GROUP, new_name)
|
|
|
|
try:
|
|
mocker.connect(**conf_base)
|
|
time.sleep(INTERVAL * 5 / 1000)
|
|
msg = mocker.recv()
|
|
assert msg is not None
|
|
assert DRIVER == msg["node_name"]
|
|
assert new_name == msg["group_name"]
|
|
assert "timestamp" in msg
|
|
assert len(TAGS) == len(msg["values"]) + len(msg["errors"])
|
|
for tag in TAGS:
|
|
assert (
|
|
tag["name"] in msg["values"]
|
|
or tag["name"] in msg["errors"]
|
|
)
|
|
finally:
|
|
api.update_group_check(DRIVER, new_name, GROUP)
|
|
|
|
@description(
|
|
given="eKuiper node",
|
|
when="driver group unsubscribed",
|
|
then="should upload no data",
|
|
)
|
|
def test_ekuiper_upload_when_unsubscribe(
|
|
self, mocker, ekuiper_node, conf_base
|
|
):
|
|
api.node_setting_check(ekuiper_node, conf_base)
|
|
api.node_ctl(ekuiper_node, config.NEU_CTL_START)
|
|
|
|
api.unsubscribe_group_check(ekuiper_node, DRIVER, GROUP)
|
|
|
|
mocker.connect(**conf_base)
|
|
with pytest.raises(Exception):
|
|
mocker.recv()
|
|
api.subscribe_group_check(ekuiper_node, DRIVER, GROUP)
|
|
|
|
@description(
|
|
given="eKuiper node",
|
|
when="subscribed driver deleted",
|
|
then="should upload no data",
|
|
)
|
|
def test_gewu_upload_when_driver_deleted(
|
|
self, mocker, ekuiper_node, conf_base
|
|
):
|
|
api.node_setting_check(ekuiper_node, conf_base)
|
|
api.node_ctl(ekuiper_node, config.NEU_CTL_START)
|
|
|
|
api.del_node(DRIVER)
|
|
|
|
mocker.connect(**conf_base)
|
|
with pytest.raises(Exception):
|
|
mocker.recv()
|
|
|
|
api.add_node_check(DRIVER, "Modbus TCP")
|
|
api.add_group_check(DRIVER, GROUP, INTERVAL)
|
|
api.add_tags_check(DRIVER, GROUP, tags=TAGS)
|
|
api.subscribe_group_check(ekuiper_node, DRIVER, GROUP)
|
|
api.modbus_tcp_node_setting(DRIVER, DRIVER_PORT)
|
|
|
|
@description(
|
|
given="eKuiper node",
|
|
when="send bad write req",
|
|
then="should fail",
|
|
)
|
|
def test_ekuiper_write_bad_req(self, mocker, ekuiper_node, conf_base):
|
|
api.node_setting_check(ekuiper_node, conf_base)
|
|
api.node_ctl(ekuiper_node, config.NEU_CTL_START)
|
|
|
|
req = {
|
|
"node_name": DRIVER,
|
|
"group_name": GROUP,
|
|
}
|
|
mocker.connect(**conf_base)
|
|
mocker.send(req)
|
|
|
|
@description(
|
|
given="eKuiper node",
|
|
when="write a driver tag",
|
|
then="should success",
|
|
)
|
|
def test_ekuiper_write_tag(self, mocker, ekuiper_node, conf_base):
|
|
api.node_setting_check(ekuiper_node, conf_base)
|
|
api.node_ctl(ekuiper_node, config.NEU_CTL_START)
|
|
|
|
req = {
|
|
"node_name": DRIVER,
|
|
"group_name": GROUP,
|
|
}
|
|
req_tags = {
|
|
TAGS[0]["name"]: 1234,
|
|
TAGS[1]["name"]: True,
|
|
TAGS[2]["name"]: 3.14,
|
|
TAGS[3]["name"]: "hello",
|
|
TAGS[4]["name"]: [1, 2, 3, 4],
|
|
}
|
|
|
|
mocker.connect(**conf_base)
|
|
for name, value in req_tags.items():
|
|
mocker.send({**req, "tag_name": name, "value": value})
|
|
|
|
time.sleep(INTERVAL * 5 / 1000)
|
|
resp = api.read_tags(DRIVER, GROUP)
|
|
assert 200 == resp.status_code
|
|
for tag in resp.json()["tags"]:
|
|
if isinstance(tag["value"], float):
|
|
assert compare_float(req_tags[tag["name"]], tag["value"])
|
|
else:
|
|
assert req_tags[tag["name"]] == tag["value"]
|