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 62c397d..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,14 +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
+from scapy.layers.l2 import Ether
 from scapy.layers.inet import IP, TCP
 from scapy.layers.inet6 import IPv6, ICMPv6ND_NS
-from ipaddress import ip_address
+from scapy.packet import Raw
 
 from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
 from resources.libraries.python.PacketVerifier import RxQueue, TxQueue
@@ -41,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
     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):
@@ -64,58 +62,48 @@ 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
     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'])
-
-    rxq = RxQueue(args.get_arg('rx_if'))
-    txq = TxQueue(args.get_arg('tx_if'))
+    args = TrafficScriptArg(
+        [u"src_mac", u"dst_mac", u"src_ip", u"dst_ip", u"dscp"]
+    )
 
-    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'))
+    rxq = RxQueue(args.get_arg(u"rx_if"))
+    txq = TxQueue(args.get_arg(u"tx_if"))
 
-    if 6 == ip_address(unicode(src_ip)).version:
-        is_ipv4 = False
-    else:
-        is_ipv4 = True
-
-    sent_packets = []
+    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 is_ipv4:
-        ip_pkt = (IP(src=src_ip, dst=dst_ip) /
-                  TCP())
-    else:
-        ip_pkt = (IPv6(src=src_ip, dst=dst_ip) /
-                  TCP())
+    ip_layer = IPv6 if ip_address(src_ip).version == 6 else IP
 
+    sent_packets = list()
     pkt_send = (Ether(src=src_mac, dst=dst_mac) /
-                ip_pkt)
+                ip_layer(src=src_ip, dst=dst_ip) /
+                TCP())
 
+    pkt_send /= Raw()
     sent_packets.append(pkt_send)
     txq.send(pkt_send)
 
     while True:
         pkt_recv = rxq.recv(2, sent_packets)
         if pkt_recv is None:
-            raise RuntimeError('ICMPv6 echo reply Rx timeout')
+            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
@@ -125,9 +113,9 @@ def main():
             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)
@@ -135,5 +123,5 @@ def main():
     sys.exit(0)
 
 
-if __name__ == "__main__":
+if __name__ == u"__main__":
     main()