From: Christos Sfakianakis <csfakian(a)redhat.com>
Add a test module for capturing packets using tcpdump. Its
structure follows that of Ping, however it is coded to wait
for an appropriate interrupt while running, since its duration
is not known a priori.
Signed-off-by: Christos Sfakianakis <csfakian(a)redhat.com>
---
lnst/Tests/PacketAssert.py | 101 +++++++++++++++++++++++++++++++++++++
1 file changed, 101 insertions(+)
create mode 100644 lnst/Tests/PacketAssert.py
diff --git a/lnst/Tests/PacketAssert.py b/lnst/Tests/PacketAssert.py
new file mode 100644
index 0000000..183a583
--- /dev/null
+++ b/lnst/Tests/PacketAssert.py
@@ -0,0 +1,101 @@
+import re
+import logging
+import subprocess
+from lnst.Common.Parameters import StrParam, ListParam, DeviceParam, IntParam, BoolParam
+from lnst.Devices.Device import Device
+from lnst.Common.Utils import is_installed
+from lnst.Tests.BaseTestModule import BaseTestModule
+from lnst.Common.LnstError import LnstError
+
+class PacketAssert(BaseTestModule):
+ interface = DeviceParam(mandatory=True)
+ p_filter = StrParam(default='')
+ grep_for = ListParam(default=[])
+ p_min = IntParam(default=0)
+ p_max = IntParam(default=0)
+ promiscuous = BoolParam(default=False)
+ _grep_exprs = []
+ _p_recv = 0
+
+ def _prepare_grep_exprs(self):
+ for expr in self.params.grep_for:
+ if expr is not None:
+ self._grep_exprs.append(expr)
+
+ def _compose_cmd(self):
+ cmd = "tcpdump"
+ if not self.params.promiscuous:
+ cmd += " -p"
+ iface = self.params.interface.name
+ filt = self.params.p_filter
+ cmd += " -nn -i %s \"%s\"" % (iface, filt)
+
+ return cmd
+
+ def _check_line(self, line):
+ if line != "":
+ for exp in self._grep_exprs:
+ if not re.search(exp, line):
+ return
+ self._p_recv += 1
+
+ def _is_real_err(self, err):
+
+ ignore_exprs = [r"tcpdump: verbose output suppressed, use -v or -vv for full
protocol decode",
+ r"listening on %s, link-type EN10MB \(Ethernet\), capture
size 262144 bytes" %
+ self.params.interface.name, r"\d+ packets captured",
+ r"\d+ packets received by filter", r"\d+ packets
dropped by kernel"]
+
+ for line in err.split('\n'):
+ if not line:
+ continue
+ match = False
+ for expr in ignore_exprs:
+ if re.search(expr, line):
+ match = True
+ break
+ if not match:
+ return True
+ return False
+
+ def run(self):
+ self._res_data = {}
+ if not is_installed("tcpdump"):
+ self._res_data["msg"] = "tcpdump is not installed on this
machine!"
+ logging.error(self._res_data["msg"])
+ return False
+
+ self._prepare_grep_exprs()
+ cmd = self._compose_cmd()
+ logging.debug("compiled command: {}".format(cmd))
+
+ packet_assert_process = subprocess.Popen(cmd, shell=True,
stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE, close_fds=True)
+
+ try:
+ self.wait_for_interrupt()
+ except:
+ raise LnstError("Could not handle interrupt properly.")
+
+ with packet_assert_process.stdout, packet_assert_process.stderr:
+ stderr=packet_assert_process.stderr.read()
+ stdout=packet_assert_process.stdout.read()
+
+ self._res_data["stderr"] = stderr
+
+ if self._is_real_err(stderr):
+ self._res_data["msg"] = "errors reported by tcpdump"
+ logging.error(self._res_data["msg"])
+ logging.error(self._res_data["stderr"])
+ return False
+
+ for line in stdout.split("\n"):
+ self._check_line(line)
+
+ logging.debug("Capturing finised. Received %d packets." %
self._p_recv)
+ self._res_data["p_recv"] = self._p_recv
+ self._res_data["p_min"] = self.params.p_min
+ if self.params.p_max:
+ self._res_data["p_max"] = self.params.p_max
+
+ return True
--
2.17.1