Python3: resources and libraries

[csit.git] / resources / traffic_scripts / policer.py

diff --git a/resources/traffic_scripts/policer.py b/resources/traffic_scripts/policer.py
index ee9fa29..c222c76 100755 (executable)
--- a/resources/traffic_scripts/policer.py
+++ b/resources/traffic_scripts/policer.py
@@ -1,6 +1,6 @@
-#!/usr/bin/env python
+#!/usr/bin/env python3
 
-# Copyright (c) 2016 Cisco and/or its affiliates.
+# Copyright (c) 2019 Cisco and/or its affiliates.
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at:
@@ -18,11 +18,15 @@
 import sys
 import logging
 
+from ipaddress import ip_address
 # pylint: disable=no-name-in-module
 # pylint: disable=import-error
 logging.getLogger("scapy.runtime").setLevel(logging.ERROR)
-from scapy.all import Ether, IP, IPv6, TCP
-from ipaddress import ip_address
+
+from scapy.layers.l2 import Ether
+from scapy.layers.inet import IP, TCP
+from scapy.layers.inet6 import IPv6, ICMPv6ND_NS
+from scapy.packet import Raw
 
 from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
 from resources.libraries.python.PacketVerifier import RxQueue, TxQueue
@@ -38,17 +42,14 @@ def check_ipv4(pkt_recv, dscp):
     :raises RuntimeError: If received packet is invalid.
     """
     if not pkt_recv.haslayer(IP):
-        raise RuntimeError(
-            'Not an IPv4 packet received: {0}'.format(pkt_recv.__repr__()))
+        raise RuntimeError(f"Not an IPv4 packet received: {pkt_recv!r}")
 
-    rx_dscp = pkt_recv['IP'].tos >> 2
+    rx_dscp = pkt_recv[IP].tos >> 2
     if rx_dscp != dscp:
-        raise RuntimeError(
-            'Invalid DSCP {0} should be {1}'.format(rx_dscp, dscp))
+        raise RuntimeError(f"Invalid DSCP {rx_dscp} should be {dscp}")
 
     if not pkt_recv.haslayer(TCP):
-        raise RuntimeError(
-            'Not a TCP packet received: {0}'.format(pkt_recv.__repr__()))
+        raise RuntimeError(f"Not a TCP packet received: {pkt_recv!r}")
 
 
 def check_ipv6(pkt_recv, dscp):
@@ -61,65 +62,66 @@ def check_ipv6(pkt_recv, dscp):
     :raises RuntimeError: If received packet is invalid.
     """
     if not pkt_recv.haslayer(IPv6):
-        raise RuntimeError(
-            'Not an IPv6 packet received: {0}'.format(pkt_recv.__repr__()))
+        raise RuntimeError(f"Not an IPv6 packet received: {pkt_recv!r}")
 
-    rx_dscp = pkt_recv['IPv6'].tc >> 2
+    rx_dscp = pkt_recv[IPv6].tc >> 2
     if rx_dscp != dscp:
-        raise RuntimeError(
-            'Invalid DSCP {0} should be {1}'.format(rx_dscp, dscp))
+        raise RuntimeError(f"Invalid DSCP {rx_dscp} should be {dscp}")
 
     if not pkt_recv.haslayer(TCP):
-        raise RuntimeError(
-            'Not a TCP packet received: {0}'.format(pkt_recv.__repr__()))
+        raise RuntimeError(f"Not a TCP packet received: {pkt_recv!r}")
 
 
 # pylint: disable=too-many-locals
 # pylint: disable=too-many-statements
 def main():
     """Send and receive TCP packet."""
-    args = TrafficScriptArg(['src_mac', 'dst_mac', 'src_ip', 'dst_ip', 'dscp'])
+    args = TrafficScriptArg(
+        [u"src_mac", u"dst_mac", u"src_ip", u"dst_ip", u"dscp"]
+    )
 
-    rxq = RxQueue(args.get_arg('rx_if'))
-    txq = TxQueue(args.get_arg('tx_if'))
+    rxq = RxQueue(args.get_arg(u"rx_if"))
+    txq = TxQueue(args.get_arg(u"tx_if"))
 
-    src_mac = args.get_arg('src_mac')
-    dst_mac = args.get_arg('dst_mac')
-    src_ip = args.get_arg('src_ip')
-    dst_ip = args.get_arg('dst_ip')
-    dscp = int(args.get_arg('dscp'))
+    src_mac = args.get_arg(u"src_mac")
+    dst_mac = args.get_arg(u"dst_mac")
+    src_ip = args.get_arg(u"src_ip")
+    dst_ip = args.get_arg(u"dst_ip")
+    dscp = int(args.get_arg(u"dscp"))
 
-    if 6 == ip_address(unicode(src_ip)).version:
-        is_ipv4 = False
-    else:
-        is_ipv4 = True
+    ip_layer = IPv6 if ip_address(src_ip).version == 6 else IP
 
-    sent_packets = []
-
-    if is_ipv4:
-        ip_pkt = IP(src=src_ip, dst=dst_ip) / \
-                 TCP()
-    else:
-        ip_pkt = IPv6(src=src_ip, dst=dst_ip) / \
-                 TCP()
-
-    pkt_send = Ether(src=src_mac, dst=dst_mac) / \
-               ip_pkt
+    sent_packets = list()
+    pkt_send = (Ether(src=src_mac, dst=dst_mac) /
+                ip_layer(src=src_ip, dst=dst_ip) /
+                TCP())
 
+    pkt_send /= Raw()
     sent_packets.append(pkt_send)
     txq.send(pkt_send)
 
-    pkt_recv = rxq.recv(2, sent_packets)
+    while True:
+        pkt_recv = rxq.recv(2, sent_packets)
+        if pkt_recv is None:
+            raise RuntimeError(u"ICMPv6 echo reply Rx timeout")
+
+        if pkt_recv.haslayer(ICMPv6ND_NS):
+            # read another packet in the queue if the current one is ICMPv6ND_NS
+            continue
+        else:
+            # otherwise process the current packet
+            break
 
     if pkt_recv is None:
-        raise RuntimeError('Rx timeout')
+        raise RuntimeError(u"Rx timeout")
 
-    if is_ipv4:
+    if ip_layer == IP:
         check_ipv4(pkt_recv, dscp)
     else:
         check_ipv6(pkt_recv, dscp)
 
     sys.exit(0)
 
-if __name__ == "__main__":
+
+if __name__ == u"__main__":
     main()