4:49 a.m.
commit a75f9d32f3da1939ac823f268bed2c7f386a534d
Author: Radek Pazdera <rpazdera(a)redhat.com>
Date: Thu Apr 5 14:56:38 2012 +0200
TestMulticast.py: Adding test for multicast tools
This test integratest multicast test tools into LNST
and makes it possible to run and evaluate them from
the recipes.
Signed-off-by: Radek Pazdera <rpazdera(a)redhat.com>
Tests/TestMulticast.py | 128 ++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 128 insertions(+), 0 deletions(-)
---
diff --git a/Tests/TestMulticast.py b/Tests/TestMulticast.py
new file mode 100644
index 0000000..c757bca
--- /dev/null
+++ b/Tests/TestMulticast.py
@@ -0,0 +1,128 @@
+"""
+Wrapper for executing the multicast test tools in LNST
+
+Copyright 2012 Red Hat, Inc.
+Licensed under the GNU General Public License, version 2 as
+published by the Free Software Foundation; see COPYING for details.
+"""
+
+__author__ = """
+rpazdera(a)redhat.com (Radek Pazdera)
+"""
+
+import logging
+import re
+from Common.TestsCommon import TestGeneric
+from NetTest.NetTestCommand import CommandException
+from Common.ExecCmd import exec_cmd
+
+class TestMulticast(TestGeneric):
+ """ Wrapper for executing the multicast test tools in LNST
+
+ Running this test executes specified test setup from
+ multicast tools and evaluates the results. The behavior
+ is defined entirely by behavior of the selected setup.
+
+ """
+
+ _tools_location = "test_tools/multicast/"
+ _conditions = {}
+
+ @staticmethod
+ def _remove_mask(address_in_string):
+ """ Remove mask suffix from the IP address """
+ if address_in_string != None:
+ return address_in_string.split('/')[0]
+ else:
+ return address_in_string
+
+ def _compose_cmd(self):
+ """ Setup a command from the recipe options """
+ cmd = ""
+ opts = {}
+
+ setup = self._tools_location + self.get_mopt("setup")
+ opts["multicast_address"] =
self._remove_mask(self.get_opt("address"))
+ opts["port"] = self.get_opt("port")
+ opts["interface"] =
self._remove_mask(self.get_opt("interface"))
+ opts["duration"] = self.get_opt("duration")
+
+ # sender-specific
+ opts["loop"] = self.get_opt("loop")
+ opts["ttl"] = self.get_opt("ttl")
+ opts["delay"] = self.get_opt("delay")
+
+ # receiver-specific
+ opts["source"] = self._remove_mask(self.get_opt("source"))
+
+ cmd = "./{0} ".format(setup)
+
+ for optname, optval in opts.iteritems():
+ if optval != None:
+ cmd += "--{0} \"{1}\" ".format(optname, optval)
+
+ return cmd
+
+ def _evaluate_result(self, name, value):
+ """ Check if the result meets required conditions
"""
+ if name in self._conditions:
+ try:
+ float(value)
+ except ValueError:
+ value = '"' + value + '"'
+
+ result = eval(value + self._conditions[name])
+ logging.info("Condition evaluated {2}: {0}{1}".format(name,
+ self._conditions[name], str(result)))
+
+ return result
+ else:
+ return True
+
+ def _prepare_conditions(self):
+ """ Search for var names in conditions """
+ varname_r = r"[a-zA-Z_][a-zA-Z0-9_]*"
+
+ conds = self.get_multi_opt("condition")
+ logging.debug(conds)
+ for cond in conds:
+ if cond == None:
+ continue
+ logging.debug(cond)
+ match = re.match(varname_r, cond)
+ if match:
+ name = match.group(0)
+ self._conditions[name] = cond.replace(name, "")
+ else:
+ raise CommandException(self)
+
+
+ def run(self):
+ self._prepare_conditions()
+
+ setup_name = self.get_mopt("setup")
+ logging.info("Started Multicast test setup {0}".format(setup_name))
+
+ exec_cmd("cd " + self._tools_location + " && make")
+
+ cmd = self._compose_cmd()
+ data_stdout = exec_cmd(cmd, die_on_err=False)[0]
+
+ res = {}
+
+ # line format matches name=value pairs with optional
+ # double quotes around the value
+ line_format_r = r"([a-zA-Z0-9_ ]+)=\"?([a-zA-Z0-9_ ]*)\"?"
+
+ for line in data_stdout.split("\n"):
+ match = re.search(line_format_r, line)
+ if match:
+ name = match.group(1).strip()
+ value = match.group(2).strip()
+
+ res[name] = value
+ logging.info("Test result: {0} = {1}".format(name, value))
+ if not self._evaluate_result(name, value):
+ return self.set_fail("Conditions not met!", res)
+
+ return self.set_pass(res)