Mon, May 20, 2019 at 07:01:31PM CEST, 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: auto-import PacketAssert
Add PacketAssert to the auto-imported test classes.
Signed-off-by: Christos Sfakianakis <csfakian(a)redhat.com>
---
lnst/Tests/PacketAssert.py | 96 ++++++++++++++++++++++++++++++++++++++
lnst/Tests/__init__.py | 1 +
2 files changed, 97 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..20e3fb7
--- /dev/null
+++ b/lnst/Tests/PacketAssert.py
@@ -0,0 +1,96 @@
+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=[])
+ 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"]
These hardcoded values (link-type, capture size) can be different if ran on
other systems. They should be regex instead to be able to cover it.
+
+ 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):
I think that it's probably better to check the exit code of tcpdump
instead of checking the error output and make Pass/Fail decision on
that.
>+ 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
>+
>+ return True
>diff --git a/lnst/Tests/__init__.py b/lnst/Tests/__init__.py
>index a65890c..f7c6c90 100644
>--- a/lnst/Tests/__init__.py
>+++ b/lnst/Tests/__init__.py
>@@ -13,6 +13,7 @@ olichtne(a)redhat.com (Ondrej Lichtner)
> """
>
> from lnst.Tests.Ping import Ping
>+from lnst.Tests.PacketAssert import PacketAssert
> from lnst.Tests.Iperf import IperfClient, IperfServer
>
> #TODO add support for test classes from lnst-ctl.conf
>--
>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...