-
Notifications
You must be signed in to change notification settings - Fork 32
Creating Your Own Test Modules
LNST comes with a set of predefined modules, however, these certainly doesn't cover everything. They are useful especially with more complex test scenarios. We write them in Python, so using one might be much easier than trying to implement your test in your recipe's task using a series of shell commands. This page explains how to write your own test module for the LNST framework.
After installing LNST, you will find the test modules, distributed with LNST, in the /usr/share/lnst/test_modules directory. This is the only directory LNST searches by default. However you can change the search locations by changing the option test_module_dirs in your controller configuration file. You can use as many directories as you like, we recommend you keep the default system-wide directory and use at least one other directory with your own modules. For more details regarding LNST configuration see the Configure LNST page.
For the purpose of this document let's assume that you're going to implement a test with name MyNetworkTest. The class implementing the test needs to be called MyNetworkTest and the file holding the code should be named MyNetworkTest.py.
So, let's start with implementation. Change to one of the user-specific test module directories and create a file MyNetworkTest.py and open it with your favorite editor.
Every class implementing an LNST test inherits from the TestGeneric class located in the TestsCommon module:
from lnst.Common.TestsCommon import TestGeneric
class MyNetworkTest(TestGeneric):
def run(self):
logging.info("Started MyNetworkTest ...")
...The only method you need to implement is the run() method and this is the code that will be executed whenever the test is referenced from the recipe.
TestGeneric class provides following set of methods to get the test module options and their values that user specifies in the recipe.
get_opt()get_mopt()get_multi_opt()get_multi_mopt()
The get_opt() and get_multi_opt() are used to get optional options. To make an option mandatory use their mopt variants, get_mopt() and get_multi_mopt().
Let's assume that your test requires an option containing an IP address to connect to. Let's name it remote_ip. Additionally you want to let user specify an optional option saying how many messages the test should send. Let's name it message_count.
So the code for the mandatory option remote_ip would be:
rip = self.get_mopt("remote_ip")For the optional one, message_count, the code looks like this:
mc = self.get_opt("message_count", default=10)Note that you can specify a default value for the option in case the user does not specify it in the recipe.
Putting it all together the whole class implementation would look like following,
class MyNetworkTest(TestGeneric):
def do_some_stuff_with_parameters(self, remote_ip, count):
s = connect(remote_ip)
for n in range(count):
s.send_message("data%s" % n)
s.close()
def run(self):
rip = self.get_mopt("remote_ip")
mc = self.get_opt("message_count", default=10)
do_some_stuff_with_parameters(rip, mc)And here is an example how to run your test from the recipe.
<run module="MyNetworkTest" host="1">
<options>
<option name="remote_ip" value="192.168.100.10" />
<option name="message_count" value="50" />
</options>
</run>Or from a Python Task:
module = ctl.get_module("MyNetworkTest",
options={
"remote_ip": "192.168.100.10",
"message_count": "50"
})
host1 = ctl.get_host("1")
host1.run(module)The multi class method variants let you specify multi-value options. Let's consider the following example. You'd like to specify multiple remote targets for your test. Without the multi method variant you would have to run the test multiple times from the recipe in the background. Using this method you can write the following command:
<run module="MyNetworkTest" host="1">
<options>
<option name="remote_target" value="192.168.100.10" />
<option name="remote_target" value="192.168.100.20" />
<option name="remote_target" value="192.168.100.30" />
</options>
</run>And you can use following code to use all of the values:
class MyNetworkTest(TestGeneric):
def do_some_stuff_with_target(self)
s = connect(t)
s.send_message("hello")
s.close()
def run(self):
targets = self.get_multi_opt("remote_target)
for t in targets:
self.do_some_stuff_with_target(t)Method get_multi_mopt() is the same except that it's mandatory to specify at least one value for the option.
For what reason do we have tests if they don't tell us their result?
The TestGeneric class provides two methods related to reporting the test results.
set_fail( res_data )set_pass( res_data )
If you don't call any of these methods from your test the result will always be a success (pass). Both methods take an optional parameter res_data that can be used to report the result in more detail, e.g. what was the transfer rate, how many connections have been established, etc. The res_data object must be a serializable type which for us means nested dictionary and list structures. We recommend that the top level type is a dictionary. This result data will be formatted and you will be able to see it in your logs after the run method returns.
Let us now enhance our example above a bit:
class MyNetworkTest(TestGeneric):
def do_some_stuff_with_target(self)
s = connect(t)
if not s:
return False
s.send_message("hello")
s.close()
def run(self):
targets = self.get_multi_opt("remote_target)
for t in targets:
rc = self.do_some_stuff_with_target(t)
if not rc:
self.set_fail("Could not connect to target %s" % t)
# if we're not reporting anything interesting, you can omit the
# following line
self.set_pass()As was mentioned earlier the res_data structure that you pass to the set_pass() and set_fail() functions is automatically formatted and logged when the test module returns. By default the formatting is pretty basic leading to a simple tree like output. This looks good for most cases but in case this doesn't work for your custom test module you have the ability to define your own formatting function.
To do this you can override the method format_res_data(res_data) that is inherited from the TestGeneric class. You can rewrite it however you want to the only restrictions are that the first parameter must represent the result data structure that you passed to the set_pass()/set_fail() method, and that the function returns a string.
There are two approaches how to do this depending on the desired behaviour.
This approach is used if you need to block the execution of a test. The TestGeneric class provides the following two methods to support interrupt handling:
set_handle_intr()wait_on_interrupt()
If set_handle_intr() method is called from the test code it simply tells the framework that the test is interested in delivering the interrupt signal. The test is then suspended until the delivery of this signal using the wait_on_interrupt() method.
So, let's assume following task:
<run module="IntrExample" host="1" bg_id="1" /> <!-- (1) -->
<ctl_wait seconds="30" /> <!-- (2) -->
<intr host="1" bg_id="1" /> <!-- (3) -->
<wait host="1" bg_id="1"/> <!-- (4) -->We're telling the framework that we want to run IntrExample test in the background (1)
then wait for 30 seconds (2)
and finally interrupt the test (3)
and wait for it's exit (4).
The python code would look like the following:
IntrExample(TestGeneric):
def run():
self.set_handle_intr()
...
# parse options
...
# spawn workers or whatever that runs in background
...
self.wait_on_interrupt()
# we're blocked until <intr> command is executedIf you plan to use more complex interrupt signal handling you have to code it directly into your test code. As an example you can look at the code in PacketAssert.py under /usr/share/lnst/test_modules directory.
Basically you need to register a method for the interrupt signal. The following code should do it:
IntrExample2(TestGeneric):
def _interrupt_handler(self):
self.do_whatever_needs_to_be_done_upon_signal_delivery()
def run(self):
signal.signal(signal.SIGINT, self._interrupt_handler)In your recipe you will just execute the <intr> command with the proper bg_id parameter.