neuron/tests/ft/driver/test_driver.py

import neuron.api as api
import neuron.error as error
import neuron.config as config
from neuron.common import *


class TestDriver:

    @description(given="running neuron", when="add driver node", then="add success")
    @pytest.mark.parametrize('node,plugin', [('modbus-tcp-1', config.PLUGIN_MODBUS_TCP),
                                             ('modbus-rtu-1', config.PLUGIN_MODBUS_RTU)])
    def test_create_node(self, node, plugin):
        response = api.add_node(node=node,
                                plugin=plugin)
        assert 200 == response.status_code
        assert error.NEU_ERR_SUCCESS == response.json()['error']

    @description(given="created node", when="get node state", then="state is init")
    @pytest.mark.parametrize('node', ['modbus-tcp-1', 'modbus-rtu-1'])
    def test_node_init(self, node):
        response = api.get_nodes_state(node=node)
        assert 200 == response.status_code
        assert config.NEU_NODE_STATE_INIT == response.json()['running']

    @description(given="created node", when="stop node", then="stop failed")
    @pytest.mark.parametrize('node', ['modbus-tcp-1', 'modbus-rtu-1'])
    def test_init_node_stop(self, node):
        response = api.node_ctl(node=node, ctl=config.NEU_CTL_STOP)
        assert 409 == response.status_code

    @description(given="modbus tcp node", when="setting modbus tcp node", then="setting success")
    def test_modbus_tcp_setting(self):
        response = api.modbus_tcp_node_setting(node='modbus-tcp-1', port=502)
        assert 200 == response.status_code
        assert error.NEU_ERR_SUCCESS == response.json()['error']

    @description(given="modbus rtu node", when="setting modbus rtu node", then="setting success")
    def test_modbus_rtu_setting(self):
        response = api.modbus_rtu_node_setting(node='modbus-rtu-1', port=502)
        assert 200 == response.status_code
        assert error.NEU_ERR_SUCCESS == response.json()['error']

    def test_wait_state(self):
        time.sleep(4)

    @description(given="configured node", when="get node state", then="state is running")
    @pytest.mark.parametrize('node', ['modbus-tcp-1', 'modbus-rtu-1'])
    def test_node_ready(self, node):
        response = api.get_nodes_state(node=node)
        assert 200 == response.status_code
        assert config.NEU_NODE_STATE_RUNNING == response.json()['running']

    @description(given="running modbus tcp node", when="setting modbus tcp node", then="state not change")
    def test_running_modbus_tcp_setting(self):
        response = api.modbus_tcp_node_setting(node='modbus-tcp-1', port=502)
        assert 200 == response.status_code

        response = api.get_nodes_state(node='modbus-tcp-1')
        assert 200 == response.status_code
        assert config.NEU_NODE_STATE_RUNNING == response.json()['running']

    @description(given="running modbus rtu node", when="setting modbus rtu node", then="state not change")
    def test_running_modbus_rtu_setting(self):
        response = api.modbus_rtu_node_setting(node='modbus-rtu-1', port=502)
        assert 200 == response.status_code

        response = api.get_nodes_state(node='modbus-tcp-1')
        assert 200 == response.status_code
        assert config.NEU_NODE_STATE_RUNNING == response.json()['running']

    @description(given="running node", when="stop node", then="state is stopped")
    @pytest.mark.parametrize('node', ['modbus-tcp-1', 'modbus-rtu-1'])
    def test_node_stop(self, node):
        response = api.node_ctl(node=node, ctl=config.NEU_CTL_STOP)
        assert 200 == response.status_code
        response = api.get_nodes_state(node=node)
        assert 200 == response.status_code
        assert config.NEU_NODE_STATE_STOPPED == response.json()['running']

    @description(given="stopped modbus tcp node", when="setting modbus tcp node", then="state not change")
    def test_stopped_modbus_tcp_setting(self):
        response = api.modbus_tcp_node_setting(node='modbus-tcp-1', port=502)
        assert 200 == response.status_code

        response = api.get_nodes_state(node='modbus-tcp-1')
        assert 200 == response.status_code
        assert config.NEU_NODE_STATE_STOPPED == response.json()['running']

    @description(given="stopped modbus rtu node", when="setting modbus rtu node", then="state not change")
    def test_stopped_modbus_rtu_setting(self):
        response = api.modbus_rtu_node_setting(node='modbus-rtu-1', port=502)
        assert 200 == response.status_code

        response = api.get_nodes_state(node='modbus-tcp-1')
        assert 200 == response.status_code
        assert config.NEU_NODE_STATE_STOPPED == response.json()['running']

    @description(given="stopped node", when="stop node", then="stop failed")
    @pytest.mark.parametrize('node', ['modbus-tcp-1', 'modbus-rtu-1'])
    def test_stop_node_stop(self, node):
        response = api.node_ctl(node=node, ctl=config.NEU_CTL_STOP)
        assert 409 == response.status_code

    @description(given="stopped node", when="start node", then="state is running")
    @pytest.mark.parametrize('node', ['modbus-tcp-1', 'modbus-rtu-1'])
    def test_node_start(self, node):
        response = api.node_ctl(node=node, ctl=config.NEU_CTL_START)
        assert 200 == response.status_code
        response = api.get_nodes_state(node=node)
        assert 200 == response.status_code
        assert config.NEU_NODE_STATE_RUNNING == response.json()['running']

    @description(given="running node", when="start node", then="start failed")
    @pytest.mark.parametrize('node', ['modbus-tcp-1', 'modbus-rtu-1'])
    def test_start_node_start(self, node):
        response = api.node_ctl(node=node, ctl=config.NEU_CTL_START)
        assert 409 == response.status_code

    @description(given="create node", when="del node", then="del success")
    @pytest.mark.parametrize('node', ['modbus-tcp-1', 'modbus-rtu-1'])
    def test_del_node(self, node):
        response = api.del_node(node=node)
        assert 200 == response.status_code