1
0
mirror of https://github.com/RIOT-OS/RIOT.git synced 2024-12-29 04:50:03 +01:00

riotctrl_shell.gnrc: fix some more ping6 parsing errors

Found during integration of release tests:

- if the payload is too small, RTTs are not shown
- duplicates are now shown properly, if duplicates exist, the stats are
  now parsed instead of ignored
This commit is contained in:
Martine S. Lenders 2020-07-09 03:16:55 +02:00
parent 9b5b20e3b9
commit 667ba11732
No known key found for this signature in database
GPG Key ID: CCD317364F63286F
2 changed files with 46 additions and 14 deletions

View File

@ -22,9 +22,18 @@ class GNRCICMPv6EchoParser(ShellInteractionParser):
def _add_reply(res, reply):
reply["seq"] = int(reply["seq"])
reply["ttl"] = int(reply["ttl"])
reply["rtt"] = float(reply["rtt"])
if reply.get("rtt") is not None:
reply["rtt"] = float(reply["rtt"])
else:
reply.pop("rtt", None)
if reply.get("dup"):
reply["dup"] = True
else:
reply.pop("dup", None)
if reply.get("rssi") is not None:
reply["rssi"] = int(reply["rssi"])
else:
reply.pop("rssi", None)
if "replies" in res:
res["replies"].append(reply)
else:
@ -35,6 +44,10 @@ class GNRCICMPv6EchoParser(ShellInteractionParser):
stats["packet_loss"] = int(stats["packet_loss"])
stats["rx"] = int(stats["rx"])
stats["tx"] = int(stats["tx"])
if stats.get("dup") is not None:
stats["dup"] = int(stats["dup"])
else:
stats.pop("dup")
res["stats"] = stats
@staticmethod
@ -94,11 +107,13 @@ class GNRCICMPv6EchoParser(ShellInteractionParser):
"""
res = {}
c_reply = re.compile(r"\d+ bytes from (?P<source>[0-9a-f:]+(%\S+)?): "
r"icmp_seq=(?P<seq>\d+) ttl=(?P<ttl>\d+) "
r"(rssi=(?P<rssi>-?\d+) dBm )?"
r"time=(?P<rtt>\d+.\d+) ms")
r"icmp_seq=(?P<seq>\d+) ttl=(?P<ttl>\d+)"
r"( rssi=(?P<rssi>-?\d+) dBm)?"
r"( time=(?P<rtt>\d+.\d+) ms)?"
r"(?P<dup> \(DUP\))?")
c_stats = re.compile(r"(?P<tx>\d+) packets transmitted, "
r"(?P<rx>\d+) packets received, "
r"((?P<dup>\d+) duplicates, )?"
r"(?P<packet_loss>\d+)% packet loss")
c_rtts = re.compile(r"round-trip min/avg/max = (?P<min>\d+.\d+)/"
r"(?P<avg>\d+.\d+)/(?P<max>\d+.\d+) ms")
@ -116,9 +131,7 @@ class GNRCICMPv6EchoParser(ShellInteractionParser):
m = c_rtts.match(line)
if m is not None:
self._set_rtts(res, m.groupdict())
return res
# Unable to parse RTTs, so something went wrong.
return None
return res
class GNRCPktbufStatsResults(dict):

View File

@ -19,7 +19,7 @@ def test_ping6():
assert " -i 100 " in res
def test_ping6_parser_success():
def test_ping6_parser_success1():
parser = riotctrl_shell.gnrc.GNRCICMPv6EchoParser()
ping_res = parser.parse("""
12 bytes from ::1: icmp_seq=0 ttl=64 time=0.435 ms
@ -29,27 +29,46 @@ def test_ping6_parser_success():
--- ::1 PING statistics ---
3 packets transmitted, 3 packets received, 0% packet loss
round-trip min/avg/max = 0.432/0.433/0.435 ms""")
assert ping_res is not None
assert ping_res
assert "rtts" in ping_res
assert "avg" in ping_res["rtts"]
def test_ping6_parser_success2():
parser = riotctrl_shell.gnrc.GNRCICMPv6EchoParser()
ping_res = parser.parse("""
12 bytes from ::1: icmp_seq=0 ttl=64
12 bytes from ::1: icmp_seq=1 ttl=64
12 bytes from ::1: icmp_seq=1 ttl=64 (DUP)
--- ::1 PING statistics ---
2 packets transmitted, 3 packets received, 1 duplicates, 0% packet loss
round-trip min/avg/max = 0.432/0.433/0.435 ms""")
assert ping_res
assert "rtts" in ping_res
assert "avg" in ping_res["rtts"]
assert len(ping_res["replies"]) == 3
assert ping_res["replies"][2]["dup"]
def test_ping6_parser_empty():
parser = riotctrl_shell.gnrc.GNRCICMPv6EchoParser()
ping_res = parser.parse("")
assert ping_res is None
assert not ping_res
def test_ping6_parser_missing_rtts():
parser = riotctrl_shell.gnrc.GNRCICMPv6EchoParser()
ping_res = parser.parse("""
12 bytes from ::1: icmp_seq=0 ttl=64 time=0.553 ms
12 bytes from ::1: icmp_seq=1 ttl=64 time=0.496 ms
12 bytes from ::1: icmp_seq=2 ttl=64 time=0.496 ms
12 bytes from ::1: icmp_seq=0 ttl=64
12 bytes from ::1: icmp_seq=1 ttl=64
12 bytes from ::1: icmp_seq=2 ttl=64
--- ::1 PING statistics ---
3 packets transmitted, 3 packets received, 0% packet loss""")
assert ping_res is None
assert ping_res
assert "rtts" not in ping_res
assert len(ping_res["replies"]) == 3
def test_pktbuf():