Pylint fixes

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

diff --git a/resources/traffic_scripts/send_icmp_wait_for_reply.py b/resources/traffic_scripts/send_icmp_wait_for_reply.py
index 3c8b71a..a77f15e 100755 (executable)
--- a/resources/traffic_scripts/send_icmp_wait_for_reply.py
+++ b/resources/traffic_scripts/send_icmp_wait_for_reply.py
@@ -12,22 +12,33 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
 # See the License for the specific language governing permissions and
 # limitations under the License.
 

-"""Traffic script that sends an IP ICMPv4."""
+"""Traffic script that sends an IP ICMPv4 or ICMPv6."""

 
 import sys
 
 import sys

+import ipaddress

 
 from scapy.layers.inet import ICMP, IP
 
 from scapy.layers.inet import ICMP, IP

+from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest

 from scapy.all import Ether
 
 from resources.libraries.python.PacketVerifier import RxQueue, TxQueue
 from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
 
 
 from scapy.all import Ether
 
 from resources.libraries.python.PacketVerifier import RxQueue, TxQueue
 from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
 
 

-def is_icmp_reply(pkt):
+def is_icmp_reply(pkt, ipformat):

     """Return True if pkt is echo reply, else return False. If exception occurs
     """Return True if pkt is echo reply, else return False. If exception occurs

-    return False."""
+    return False.
+
+    :param pkt: Packet.
+    :param ipformat: Dictionary of names to distinguish IPv4 and IPv6.
+    :type pkt: dict
+    :type ipformat: dict
+    :rtype: bool
+    """
+    #pylint: disable=bare-except

     try:
     try:

-        if pkt['IP']['ICMP'].type == 0:  # 0 - echo-reply
+        if pkt[ipformat['IPType']][ipformat['ICMP_rep']].type == \
+                ipformat['Type']:

             return True
         else:
             return False
             return True
         else:
             return False


@@ -35,16 +46,59 @@ def is_icmp_reply(pkt):
         return False
 
 
         return False
 
 

-def address_check(request, reply):
+def address_check(request, reply, ipformat):

     """Compare request packet source address with reply destination address
     """Compare request packet source address with reply destination address

-    and vice versa. If exception occurs return False."""
+    and vice versa. If exception occurs return False.
+
+    :param request: Sent packet containing request.
+    :param reply: Received packet containing reply.
+    :param ipformat: Dictionary of names to distinguish IPv4 and IPv6.
+    :type request: dict
+    :type reply: dict
+    :type ipformat: dict
+    :rtype: bool
+    """
+    #pylint: disable=bare-except

     try:
     try:

-        return reply['IP'].src == request['IP'].dst \
-               and reply['IP'].dst == request['IP'].src
+        r_src = reply[ipformat['IPType']].src == request[ipformat['IPType']].dst
+        r_dst = reply[ipformat['IPType']].dst == request[ipformat['IPType']].src
+        return r_src and r_dst

     except:
         return False
 
 
     except:
         return False
 
 

+def valid_ipv4(ip):
+    """Check if IP address has the correct IPv4 address format.
+
+    :param ip: IP address.
+    :type ip: str
+    :return: True in case of correct IPv4 address format,
+             otherwise return False.
+    :rtype: bool
+    """
+    try:
+        ipaddress.IPv4Address(unicode(ip))
+        return True
+    except (AttributeError, ipaddress.AddressValueError):
+        return False
+
+
+def valid_ipv6(ip):
+    """Check if IP address has the correct IPv6 address format.
+
+    :param ip: IP address.
+    :type ip: str
+    :return: True in case of correct IPv6 address format,
+             otherwise return False.
+    :rtype: bool
+    """
+    try:
+        ipaddress.IPv6Address(unicode(ip))
+        return True
+    except (AttributeError, ipaddress.AddressValueError):
+        return False
+
+

 def main():
     """Send ICMP echo request and wait for ICMP echo reply. It ignores all other
     packets."""
 def main():
     """Send ICMP echo request and wait for ICMP echo reply. It ignores all other
     packets."""


@@ -65,21 +119,33 @@ def main():
     sent_packets = []
 
     # Create empty ip ICMP packet
     sent_packets = []
 
     # Create empty ip ICMP packet

-    icmp_request = (Ether(src=src_mac, dst=dst_mac) /
-                    IP(src=src_ip, dst=dst_ip) /
-                    ICMP())
+    if valid_ipv4(src_ip) and valid_ipv4(dst_ip):
+        icmp_request = (Ether(src=src_mac, dst=dst_mac) /
+                        IP(src=src_ip, dst=dst_ip) /
+                        ICMP())
+        ip_format = {'IPType': 'IP', 'ICMP_req': 'ICMP',
+                     'ICMP_rep': 'ICMP', 'Type': 0}
+    elif valid_ipv6(src_ip) and valid_ipv6(dst_ip):
+        icmp_request = (Ether(src=src_mac, dst=dst_mac) /
+                        IPv6(src=src_ip, dst=dst_ip) /
+                        ICMPv6EchoRequest())
+        ip_format = {'IPType': 'IPv6', 'ICMP_req': 'ICMPv6 Echo Request',
+                     'ICMP_rep': 'ICMPv6 Echo Reply', 'Type': 129}
+    else:
+        raise ValueError("IP not in correct format")
+

     # Send created packet on the interface
     sent_packets.append(icmp_request)
     txq.send(icmp_request)
 
     for _ in range(1000):
     # Send created packet on the interface
     sent_packets.append(icmp_request)
     txq.send(icmp_request)
 
     for _ in range(1000):

-        icmp_reply = rxq.recv(wait_step)
+        icmp_reply = rxq.recv(wait_step, ignore=sent_packets)

         if icmp_reply is None:
             timeout -= wait_step
             if timeout < 0:
                 raise RuntimeError("ICMP echo Rx timeout")
         if icmp_reply is None:
             timeout -= wait_step
             if timeout < 0:
                 raise RuntimeError("ICMP echo Rx timeout")

-        elif is_icmp_reply(icmp_reply):
-            if address_check(icmp_request, icmp_reply):
+        elif is_icmp_reply(icmp_reply, ip_format):
+            if address_check(icmp_request, icmp_reply, ip_format):

                 break
     else:
         raise RuntimeError("Max packet count limit reached")
                 break
     else:
         raise RuntimeError("Max packet count limit reached")