5:13 a.m.
From: Ondrej Lichtner <olichtne(a)redhat.com>
This package will contain all upstream tracked test module classes. At
the moment it only contains the IcmpPing class which is a simple port
from the old IcmpPing test_module. In the future it should be heavily
refactored to support both ipv4 and ipv6 (instead of having 2 different
modules) but for early testing of Python Recipes this is good enough.
Signed-off-by: Ondrej Lichtner <olichtne(a)redhat.com>
---
lnst/Tests/IcmpPing.py | 76 ++++++++++++++++++++++++++++++++++++++++++++++++++
lnst/Tests/__init__.py | 1 +
2 files changed, 77 insertions(+)
create mode 100644 lnst/Tests/IcmpPing.py
create mode 100644 lnst/Tests/__init__.py
diff --git a/lnst/Tests/IcmpPing.py b/lnst/Tests/IcmpPing.py
new file mode 100644
index 0000000..8b07a2c
--- /dev/null
+++ b/lnst/Tests/IcmpPing.py
@@ -0,0 +1,76 @@
+import re
+import logging
+from lnst.Devices import Device
+from lnst.Common.Parameters import IntParam, Param, FloatParam, IpParam
+from lnst.Common.TestModule import BaseTestModule, TestModuleError
+from lnst.Common.ExecCmd import exec_cmd
+
+class IcmpPing(BaseTestModule):
+ """Port of old IcmpPing test modules"""
+ dst = IpParam(mandatory=True)
+ count = IntParam(default=10)
+ interval = FloatParam(default=1.0)
+ iface = Param()
+ size = IntParam()
+
+ limit_rate = IntParam(default=80)
+
+ def __init__(self, **kwargs):
+ super(IcmpPing, self).__init__(**kwargs)
+
+ if self.iface.set:
+ if not isinstance(self.iface.val, Device) and\
+ not isinstance(self.iface.val, str):
+ raise TestModuleError("Invalid 'iface' parameter.")
+
+ def _compose_cmd(self):
+ cmd = "ping %s" % self.params.dst.val
+ cmd += " -c %d" % self.params.count
+ cmd += " -i %f" % self.params.interval
+ if self.params.iface.set:
+ if isinstance(self.params.iface.val, str):
+ cmd += " -I %s" % self.params.iface
+ elif isinstance(self.params.iface.val, Device):
+ pass
+ # cmd += " -I %s" % iface.val.devname
+ if self.params.size.set:
+ cmd += " -s %d" % self.params.size
+ return cmd
+
+ def run(self):
+ cmd = self._compose_cmd()
+
+ limit_rate = self.params.limit_rate
+
+ data_stdout = exec_cmd(cmd, die_on_err=False)[0]
+ stat_pttr1 = r'(\d+) packets transmitted, (\d+) received'
+ stat_pttr2 = r'rtt min/avg/max/mdev =
(\d+\.\d+)/(\d+\.\d+)/(\d+\.\d+)/(\d+\.\d+) ms'
+
+ match = re.search(stat_pttr1, data_stdout)
+ if not match:
+ self._res_data = {"msg": "expected pattern not found"}
+ return False
+
+ trans_pkts, recv_pkts = match.groups()
+ rate = int(round((float(recv_pkts) / float(trans_pkts)) * 100))
+ logging.debug("Transmitted \"%s\", received \"%s\",
"
+ "rate \"%d%%\", limit_rate \"%d%%\""
+ % (trans_pkts, recv_pkts, rate, limit_rate))
+
+ self._res_data = {"rate": rate,
+ "limit_rate": limit_rate}
+
+ match = re.search(stat_pttr2, data_stdout)
+ if match:
+ tmin, tavg, tmax, tmdev = [float(x) for x in match.groups()]
+ logging.debug("rtt min \"%.3f\", avg \"%.3f\", max
\"%.3f\", "
+ "mdev \"%.3f\"" % (tmin, tavg, tmax,
tmdev))
+
+ self._res_data["rtt_min"] = tmin
+ self._res_data["rtt_max"] = tmax
+
+ if rate < limit_rate:
+ self._res_data["msg"] = "rate is lower than limit"
+ return False
+
+ return True
diff --git a/lnst/Tests/__init__.py b/lnst/Tests/__init__.py
new file mode 100644
index 0000000..60da773
--- /dev/null
+++ b/lnst/Tests/__init__.py
@@ -0,0 +1 @@
+from lnst.Tests.IcmpPing import IcmpPing
--
2.13.0