Testing a plugin

Testing a plugin is really important! This allow to:

  • check if the plugin works as designed
  • check if the last update didn’t break anything: this is called non regression tests

Some libraries have been created in the Domogik project to help you to create the test scripts. These libraries can:

  • create some devices
  • configure, start, stop and do some basic checks on the plugin
  • help you to test xPL dialogs

Be careful : executing the tests may delete your existing devices and so you can loose some data!!!!

When should the tests be launched ?

Well, after each plugin update! But as this could be time consuming, you can automate this step thanks to the testrunner.py tool and Travis, the continuous integration service.

File tree

The following files are mandatory for the tests:

tests/
  # the 0* files are just helpers for the developers
  001_configure.py     # this python file is used by the developers to quickly configure the plugin
  002_create_device.py # this python file is used by the developers to quickly create some test devices

  # all the other files are related to the plugin tests
  tests.json          # this is a file which describe all the test files. It is used for tests automation

Then, depending on your plugin, you can have only one test file:

tests/
  ..
  tests.py

Or several files:

tests/
  ..
  test_feature_A.py
  test_feature_B.py
  test_feature_C.py

tests.json

This file is very important! It will be used by the testrunner.py tool.

Example:

{
    "tests" : {
        "alter_configuration_or_setup" : true,
        "need_hardware" : false,
        "criticity" : "high"
    }
}
  • alter_configuration_or_setup: true if the test need to alter the plugin configuration or some devices. The test should not be run on a production environment! false if the test doesn’t alter anything and can be run safely on a production environment.
  • need_hardware: true if some hardware is needed by the test. Please note that the testrunner.py tool will never run the tests that need some hardware.
  • criticity: high, medium or low.

Each test file must be listed in the tests.json file.

A test file

A test file is made of 2 parts:

  • a class which inherits from PluginTestCase. This class will contain all the test cases related to the plugin.
  • the main part which will do some actions and launch the test cases.

Here is a sample file from the teleinfo plugin. This sample file has only one dummy test defined. This is the minimal file you must prepare before creating the tests. This test file will test only global features :

  • deletion and creation of devices
  • plugin configuration
  • plugin startup
  • xpl hbeat
  • plugin stop successfully
#!/usr/bin/python
# -*- coding: utf-8 -*-

from domogik.xpl.common.plugin import XplPlugin
from domogik.tests.common.plugintestcase import PluginTestCase
from domogik.tests.common.testplugin import TestPlugin
from domogik.tests.common.testdevice import TestDevice
from domogik.tests.common.testsensor import TestSensor
from domogik.common.utils import get_sanitized_hostname
from datetime import datetime
import unittest
import sys
import os
import traceback

class TeleinfoTestCase(PluginTestCase):

    def test_0100_dummy(self):
        self.assertTrue(True)

if __name__ == "__main__":
    ### global variables
    device = "/dev/teleinfo"
    interval = 60

    # set up the xpl features
    xpl_plugin = XplPlugin(name = 'test',
                           daemonize = False,
                           parser = None,
                           nohub = True,
                           test  = True)

    # set up the plugin name
    name = "teleinfo"

    # set up the configuration of the plugin
    # configuration is done in test_0010_configure_the_plugin with the cfg content
    # notice that the old configuration is deleted before
    cfg = { 'configured' : True }

    ### start tests
    # load the test devices class
    td = TestDevice()

    # delete existing devices for this plugin on this host
    client_id = "{0}-{1}.{2}".format("plugin", name, get_sanitized_hostname())
    try:
        td.del_devices_by_client(client_id)
    except:
        print(u"Error while deleting all the test device for the client id '{0}' : {1}".format(client_id, traceback.format_exc()))
        sys.exit(1)

    # create a test device
    try:
        #device_id = td.create_device(client_id, "test_device_teleinfo", "teleinfo.electric_meter")

        params = td.get_params(client_id, "teleinfo.electric_meter")

        # fill in the params
        params["device_type"] = "teleinfo.electric_meter"
        params["name"] = "test_device_teleinfo"
        params["reference"] = "reference"
        params["description"] = "description"
        # global params
        for the_param in params['global']:
            if the_param['key'] == "interval":
                the_param['value'] = interval
            if the_param['key'] == "device":
                the_param['value'] = device
        print params['global']
        # xpl params
        pass # there are no xpl params for this plugin
        # create
        td.create_device(params)

    except:
        print(u"Error while creating the test devices : {0}".format(traceback.format_exc()))
        sys.exit(1)

    ### prepare and run the test suite
    suite = unittest.TestSuite()
    # check domogik is running, configure the plugin
    suite.addTest(TeleinfoTestCase("test_0001_domogik_is_running", xpl_plugin, name, cfg))
    suite.addTest(TeleinfoTestCase("test_0010_configure_the_plugin", xpl_plugin, name, cfg))

    # start the plugin
    suite.addTest(TeleinfoTestCase("test_0050_start_the_plugin", xpl_plugin, name, cfg))


    # do the specific plugin tests
    suite.addTest(TeleinfoTestCase("test_0100_dummy", xpl_plugin, name, cfg))

    # do some tests comon to all the plugins
    suite.addTest(TeleinfoTestCase("test_9900_hbeat", xpl_plugin, name, cfg))
    suite.addTest(TeleinfoTestCase("test_9990_stop_the_plugin", xpl_plugin, name, cfg))

    # quit
    res = unittest.TextTestRunner().run(suite)
    if res.wasSuccessful() == True:
        rc = 0   # tests are ok so the shell return code is 0
    else:
        rc = 1   # tests are ok so the shell return code is != 0
    xpl_plugin.force_leave(return_code = rc)

A test file : the class which inherits from PluginTestCase

In this class, you will define the tests that will be executed on the plugin.

There is a norm to name the functions in this class: test_9999_thetestname.

  • test_0xxx_xxx : these functions are reserved, declared in PluginTestCase and related to the preparation of the plugin.
  • test_9xxx_xxx : these functions are reserved, declared in PluginTestCase and related to the end of the plugin tests (hbeat test, plugin stop).
  • test_1xxx_xxx to test_8xxx_xxx : these functions are free for use.

Dummy example

Here is an example of a dummy test which is always good:

class DiskfreeTestCase(PluginTestCase):

    def test_0100_dummy(self):
        self.assertTrue(True)

Useful functions

The following functions can be used for your tests.

Wait for a xPL message

self.assertTrue(self.wait_for_xpl(xpltype = "xpl-stat",
                                  xplschema = "sensor.basic",
                                  xplsource = "domogik-{0}.{1}".format(self.name, get_sanitized_hostname()),
                                  data = {"type" : "total_space",
                                          "device" : path,
                                          "current" : du_total},
                                  timeout = interval * 60))

The function self.wait_for_xpl is a blocking function : it will wait until the required xPL message is reveived or until the timeout is reached.

  • xpltype : the type of the xPL message waited. Available values are xpl-stat, xpl-trig, xpl-cmnd.
  • xplschema : the required xPL schema.
  • xplsource : the xPL source of the plugin to test. Always use this value for a plugin : “domogik-{0}.{1}”.format(self.name, get_sanitized_hostname())
  • data : a dictionnary with the required content of the xPL message. If you know exactly the value you are waiting for, just add it in the dictionnary. If you can’t guess the value, just put all the known values (device address, ...) in the dictionnary and check after the value with self.xpl_data.data.
  • timeout : the timeout in seconds. Once reached, the function will return False as no expected message has been received. Please notice that a 5% margin is allowed, so il you set a timeout of 100 seconds, the function will really use a timeout of 105 seconds : this allows to avoid some errors due to some processing which could, for example, add 1 or 2 seconds between 2 messages.

Once a xPL message is received, its content is stored in self.xpl_data.data. For exemple, to get the value of the key current, you can do:

current_value = self.xpl_data.data['current']

Check the value in inserted in database

It is important to check that the values of the received messages are stored in database : a plugin can successfully send some xPL messages but the info.json file can be wrong and this is this file which defines the way to store values in database.

To do this, you must create a TestSensor instance for the device id and the sensor reference. Then, compare the last value of this sensor to the value from the xPL message.

Example:

print(u"Check that the value of the xPL message has been inserted in database")
sensor = TestSensor(device_id, "get_total_space")
self.assertTrue(sensor.get_last_value()[1] == self.xpl_data.data['current'])

Check the time between two xPL messages

When a plugin feature should send a xPL message each N seconds, you have to test if the interval is correct. To do this, wait for a first message, get the current time. Wait for another message, get the current time again and check if the difference is correct.

Example (for a message each 2 minutes) :

# get the first message
self.assertTrue(self.wait_for_xpl(xpltype = "xpl-stat",
                                  ...
                                  timeout = 2 * 60))
# get the time of the first message
msg1_time = datetime.now()

# get the second message
self.assertTrue(self.wait_for_xpl(xpltype = "xpl-stat",
                                  ...
                                  timeout = 2 * 60))
# get the time of the second message
msg2_time = datetime.now()

# check if the interval between the 2 messages is OK
self.assertTrue(self.is_interval_of(2 * 60, msg2_time - msg1_time))

The function self.is_interval_of() takes 2 parameters:

  • the expected interval
  • the mesured interval

Notice that the function will allow a margin of 5%: if you expect for 100 seconds and there are 105 seconds in the reality, the test will be ok.

Send a xPL message

Todo

Explain

Example 1 : wait for a xPL message and check its content

The following example if one function from the diskfree plugin tests. It will wait for a given xPL message, check if the value in the xPL message is correct and is inserted in database, wait for a second message, check that the interval between the 2 messages is ok.

Example:

def test_0110_total_space(self):
    """ check if the xpl messages about total space are OK
        Sample message :
        xpl-stat
        {
        hop=1
        source=domogik-diskfree.darkstar
        target=*
        }
        sensor.basic
        {
        device=/home
        type=total_space
        current=19465224
        }
    """
    global interval
    global path
    global device_id


    # get the current total space on the device
    du = os.statvfs(path)
    du_total = (du.f_blocks * du.f_frsize) / 1024

    # do the test
    print(u"Check that a message about total space is sent. The message must be received each {0} minute(s)".format(interval))

    self.assertTrue(self.wait_for_xpl(xpltype = "xpl-stat",
                                      xplschema = "sensor.basic",
                                      xplsource = "domogik-{0}.{1}".format(self.name, get_sanitized_hostname()),
                                      data = {"type" : "total_space",
                                              "device" : path,
                                              "current" : du_total},
                                      timeout = interval * 60))
    print(u"Check that the value of the xPL message has been inserted in database")
    sensor = TestSensor(device_id, "get_total_space")
    self.assertTrue(sensor.get_last_value()[1] == self.xpl_data.data['current'])
    msg1_time = datetime.now()

    print(u"Check there is a second message is sent and the interval between them")
    self.assertTrue(self.wait_for_xpl(xpltype = "xpl-stat",
                                      xplschema = "sensor.basic",
                                      xplsource = "domogik-{0}.{1}".format(self.name, get_sanitized_hostname()),
                                      data = {"type" : "total_space",
                                              "device" : path,
                                              "current" : du_total},
                                      timeout = interval * 60))
    msg2_time = datetime.now()
    self.assertTrue(self.is_interval_of(interval * 60, msg2_time - msg1_time))

Example 2 : send a xPL command and check for its response

Todo

Example

A test file : the main part

Here are the actions that can be done in the main part:

  • if needed, define some global variables (polling interval, ...). Example:

    ### global variables
    interval = 1
    path = "/home"
    
  • set up the xpl features for the test file. A XplPlugin instance will be created with some special parameters (please always use these parameters, even the generic name). Example:

    # set up the xpl features
    xpl_plugin = XplPlugin(name = 'test',
                           daemonize = False,
                           parser = None,
                           nohub = True,
                           test  = True)
    
  • set up the plugin name:

    # set up the plugin name
    name = "diskfree"
    
  • define the configuration of the plugin. If no configuration is required for the plugin, at least you must set up the configured key to True. Example:

    # set up the configuration of the plugin
    # configuration is done in test_0010_configure_the_plugin with the cfg content
    # notice that the old configuration is deleted before
    cfg = { 'configured' : True }
    
  • start the common tests by setting up the TestDevice class which helps to manage the devices. Example:

    ### start tests
    # load the test devices class
    td = TestDevice()
    
  • if needed, delete all the existing devices of the plugin on the current host. If you do this, you must set alter_configuration_or_setup to True in the json. Example:

    # delete existing devices for this plugin on this host
    client_id = "{0}-{1}.{2}".format("plugin", name, get_sanitized_hostname())
    try:
        td.del_devices_by_client(client_id)
    except:
        print(u"Error while deleting all the test device for the client id '{0}' : {1}".format(client_id, traceback.format_exc()))
        sys.exit(1)
    
  • if needed, create some devices. If you do this, you must set alter_configuration_or_setup to True in the json. Notice that the device parameters should come from the global variables defined before. Example (here 2 global parameters are defined but no xpl parameters are defined):

    # create a test device
    try:
        #device_id = td.create_device(client_id, "test_device_teleinfo", "teleinfo.electric_meter")
    
        params = td.get_params(client_id, "teleinfo.electric_meter")
    
        # fill in the params
        params["device_type"] = "teleinfo.electric_meter"
        params["name"] = "test_device_teleinfo"
        params["reference"] = "reference"
        params["description"] = "description"
        # global params
        for the_param in params['global']:
            if the_param['key'] == "interval":
                the_param['value'] = interval
            if the_param['key'] == "device":
                the_param['value'] = device
        print params['global']
        # xpl params
        pass # there are no xpl params for this plugin
        # create
        td.create_device(params)
    
    except:
        print(u"Error while creating the test devices : {0}".format(traceback.format_exc()))
        sys.exit(1)
    
  • then, call the common tests related to the plugin. These tests are common to all plugins and are defined in the class PluginTestCase. The first one will just check if Domogik is running (if not, the plugin will not be able to start). The second one will configure the plugin and the last one will start the plugin. Example:

    ### prepare and run the test suite
    suite = unittest.TestSuite()
    # check domogik is running, configure the plugin
    suite.addTest(DiskfreeTestCase("test_0001_domogik_is_running", xpl_plugin, name, cfg))
    suite.addTest(DiskfreeTestCase("test_0010_configure_the_plugin", xpl_plugin, name, cfg))
    
    # start the plugin
    suite.addTest(DiskfreeTestCase("test_0050_start_the_plugin", xpl_plugin, name, cfg))
    
  • launch all the tests you created in the YourpluginTestCase class. Example:

    # do the specific plugin tests
    suite.addTest(DiskfreeTestCase("test_0110_total_space", xpl_plugin, name, cfg))
    suite.addTest(DiskfreeTestCase("test_0120_free_space", xpl_plugin, name, cfg))
    suite.addTest(DiskfreeTestCase("test_0130_used_space", xpl_plugin, name, cfg))
    suite.addTest(DiskfreeTestCase("test_0140_percent_used", xpl_plugin, name, cfg))
    
  • launch some common tests related to the plugin stopping process. The first one will check that the plugin sends hbeat messages and can take several minutes! The second one will try to stop the plugin and check if the plugin can be stopped. Example:

    # do some tests common to all the plugins
    suite.addTest(DiskfreeTestCase("test_9900_hbeat", xpl_plugin, name, cfg))
    suite.addTest(DiskfreeTestCase("test_9990_stop_the_plugin", xpl_plugin, name, cfg))
    
  • and finally get the status of the tests. If there were some errors, the python test file will return 1. This is very important for the continuous integration tools and testrunner. Example:

    # quit
    res = unittest.TextTestRunner().run(suite)
    if res.wasSuccessful() == True:
        rc = 0   # tests are ok so the shell return code is 0
    else:
        rc = 1   # tests are ok so the shell return code is != 0
    xpl_plugin.force_leave(return_code = rc)
    

How to use the serial mock

A serial mock can be used to simulate some serial devices and so test the plugin without any hardware!

Test file

First, in the test file, define the test_folder variable.

Before

if __name__ == "__main__":

    ### global variables

After

if __name__ == "__main__":

    test_folder = os.path.dirname(os.path.realpath(__file__))

    ### global variables

Then, in the test file, add the following configuration elements in the main part:

# specific configuration for test mdode (handled by the manager for plugin startup)
cfg['test_mode'] = True
cfg['test_option'] = "{0}/tests_hchp_data.json".format(test_folder)

These are configuration options which will be handled by the plugin itself. The first one will activate the mock and the second one will give the json file which describe the fake device behavior.

Of course, the plugin will need to be adapted to handle these options and the serial mock!

Plugin binary part

Add the following parameter when you instantiate your plugin library class:

  • self.options.test_option

Example for the teleinfo plugin during the development phasis.

Before

teleinfo_list[device] = Teleinfo(self.log, self.send_xpl, self.get_stop(), device, interval)

After

teleinfo_list[device] = Teleinfo(self.log, self.send_xpl, self.get_stop(), device, interval, self.options.test_option)

Plugin library part

Import both serial and serial mock libraries:

import serial as serial
import domogik.tests.common.testserial as testserial

Handle this parameter in your library class:

Example for the teleinfo plugin during the development phases.

Before

class Teleinfo:

    def __init__(self, log, callback, stop, device, interval):
        ...
        self._device = device
        ...

After

class Teleinfo:

    def __init__(self, log, callback, stop, device, interval, fake_device):
         ...
         self._device = device
         self._fake_device = fake_device
         ...

Then, when you create the serial device, if this is a fake one, use the serial mock library.

Before

self._ser = serial.Serial(self._device, 1200, bytesize=7,
                          parity = 'E',stopbits=1)

After

if self._fake_device != None:
    self._ser = testserial.Serial(self._fake_device, baudrate=1200, bytesize=7,
                              parity = 'E',stopbits=1)
else:
    self._ser = serial.Serial(self._device, baudrate=1200, bytesize=7,
                              parity = 'E',stopbits=1)

Create a json serial mock file

Now you will have to create the json file. This json file is made of 3 parts:

{
    "history" : [ ],
    "responses" : {},
    "loop" : []
}

Note

Currently, the responses part is not yet implemented

The history part will contain some data coming from the fake real device. All history elements are played only one time, in the order they are defined in the json file. Then, when the history if finished, the loop part will be processed forever.

Here is an example for the history part:

{
    "history" : [
                    ...
                    { "description" : "Send a start frame flag",
                      "action" : "data-hex",
                      "data" : "02"
                    },
                    { "description" : "Send a HP frame",
                      "action" : "data",
                      "data" : "\nADCO 030928084432 B\r"
                    },
                    ...
                    { "description" : "wait",
                      "action" : "wait",
                      "delay" : 5
                    },
                    ...
                ],
    "responses" : {},
    "loop" : [...]
}

As you can see in the example, each history step is made of 3 parts:

  • description : a small description of the step. It will be used for display, but it is mainly used for the developer information.
  • action : the type of the action.
    • action = “data” : send a string on the fake serial device
    • action = “data-hex” : send some hexadecimal data on the fake serial device. Hexadecimal data is written in a human readable mode : 0F44DC...
    • action == wait” : send nothing for a while on the fake serial device. The wait time is defined in seconds
  • data (for action = data, data-hex) : the data to send
  • delay (for action = wait) : the wait time in seconds

Note

The fake serial device will not send data in the first 30 seconds. It allows the plugin to be fully started (MQ ready) so the xpl checks will not missed any xpl message sent before the MQ is ready and send an active status for the plugin (for example).

The loop part is to be defined in the same way as the history part. The only difference is that when the last step of the loop part is reached, then following step will be the first step of the loop part... Just a loop :)