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 :)