On Wed, May 15, 2019 at 01:44:04PM +0200, csfakian(a)redhat.com wrote:
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)
What are p_min and p_max parameters for? I only see them used to return
them back to the Controller and not actually used for anything...
> + 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
> _______________________________________________
> LNST-developers mailing list -- lnst-developers(a)lists.fedorahosted.org
> To unsubscribe send an email to lnst-developers-leave(a)lists.fedorahosted.org
> Fedora Code of Conduct:
https://getfedora.org/code-of-conduct.html
> List Guidelines:
https://fedoraproject.org/wiki/Mailing_list_guidelines
> List Archives:
https://lists.fedorahosted.org/archives/list/lnst-developers@lists.fedora...