resources/libraries/python/TLDK/UdpTest.py

   1 # Copyright (c) 2017 Cisco and/or its affiliates.
   2 # Licensed under the Apache License, Version 2.0 (the "License");
   3 # you may not use this file except in compliance with the License.
   4 # You may obtain a copy of the License at:
   5 #
   6 #     http://www.apache.org/licenses/LICENSE-2.0
   7 #
   8 # Unless required by applicable law or agreed to in writing, software
   9 # distributed under the License is distributed on an "AS IS" BASIS,
  10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  11 # See the License for the specific language governing permissions and
  12 # limitations under the License.
  13
  14
  15 """
  16 This module exists to provide the UDP test for TLDK on topology nodes.
  17 """
  18
  19 from scapy.utils import rdpcap
  20 from scapy.layers.inet import IP
  21 from scapy.layers.inet6 import IPv6
  22
  23 from robot.libraries.BuiltIn import BuiltIn
  24
  25 from resources.libraries.python.ssh import SSH
  26 from resources.libraries.python.TLDK.TLDKConstants import TLDKConstants as con
  27 from resources.libraries.python.topology import Topology
  28
  29 class UdpTest:
  30     """Test the TLDK UDP function."""
  31
  32     @staticmethod
  33     def get_pcap_info(file_prefix):
  34         """Get the Dest IP from the RX pcap file
  35
  36         :param file_prefix: the test case pcap file prefix
  37         :type file_prefix: str
  38         :returns: packet counts, dest ip, is or not ipv4
  39         :rtype: tuple(int, str, bool).
  40         """
  41         exec_dir = BuiltIn().get_variable_value("${EXECDIR}")
  42
  43         rx_pcapfile = '{0}/{1}/{2}_rx.pcap' \
  44             .format(exec_dir, con.TLDK_TESTCONFIG, file_prefix)
  45         packets = rdpcap(rx_pcapfile)
  46         count = len(packets)
  47
  48         ### the first packet
  49         pkt = packets[0]
  50         if pkt.type == 0x0800:
  51             ### this is a IPv4 packet
  52             dest_ip = pkt[IP].dst
  53             is_ipv4 = True
  54         elif pkt.type == 0x86dd:
  55             ### this is a IPv6 packet
  56             dest_ip = pkt[IPv6].dst
  57             is_ipv4 = False
  58
  59         return count, dest_ip, is_ipv4
  60
  61     @staticmethod
  62     def exec_the_udpfwd_test(dut_node, dut_if, file_prefix, \
  63             dest_ip, is_ipv4=True):
  64         """Execute the udpfwd on the dut_node.
  65
  66         :param dut_node: Will execute the udpfwd on this node.
  67         :param dut_if: DUT interface name.
  68         :param file_prefix: The test case config file prefix.
  69         :param dest_ip: The UDP packet dest IP.
  70         :param is_ipv4: Execute the IPv4 or IPv6 test.
  71         :type dut_node: dict
  72         :type dut_if: str
  73         :type file_prefix: str
  74         :type dest_ip: str
  75         :type is_ipv4: bool
  76         :raises RuntimeError: If failed to execute udpfwd test on the dut node.
  77         """
  78         pci_address = Topology.get_interface_pci_addr(dut_node, dut_if)
  79         ssh = SSH()
  80         ssh.connect(dut_node)
  81         if is_ipv4:
  82             cmd = 'cd {0}/{4} && ./run_tldk.sh {0}/{5}/{2}_rx.pcap ' \
  83                 '{0}/{5}/{2}_tx.pcap {1} {0}/{5}/{2}_fe.cfg ' \
  84                 '{0}/{5}/{2}_be.cfg {3} NONE' \
  85                 .format(con.REMOTE_FW_DIR, pci_address, file_prefix, \
  86                 dest_ip, con.TLDK_SCRIPTS, con.TLDK_TESTCONFIG)
  87         else:
  88             cmd = 'cd {0}/{4} && ./run_tldk.sh {0}/{5}/{2}_rx.pcap ' \
  89                 '{0}/{5}/{2}_tx.pcap {1} {0}/{5}/{2}_fe.cfg ' \
  90                 '{0}/{5}/{2}_be.cfg NONE {3}' \
  91                 .format(con.REMOTE_FW_DIR, pci_address, file_prefix, \
  92                 dest_ip, con.TLDK_SCRIPTS, con.TLDK_TESTCONFIG)
  93
  94         (ret_code, _, _) = ssh.exec_command(cmd, timeout=600)
  95         if ret_code != 0:
  96             raise RuntimeError('Failed to execute udpfwd test at node {0}'
  97                                .format(dut_node['host']))
  98
  99     @staticmethod
 100     def get_the_test_result(dut_node, file_prefix):
 101         """
 102         After execute the udpfwd cmd, use this to get the test result.
 103
 104         :param dut_node: will get the test result in this dut node
 105         :param dut_if: the dut interface name
 106         :param file_prefix: the test case output file prefix
 107         :type dut_node: dice
 108         :type dut_if: str
 109         :type file_prefix: str
 110         :returns: str.
 111         :rtype: str
 112         :raises RuntimeError: If failed to get the test result.
 113         """
 114         ssh = SSH()
 115         ssh.connect(dut_node)
 116         cmd = 'cd {0}; sudo /usr/sbin/tcpdump -nnnn -vvv -r ./{2}/{1}_tx.pcap' \
 117               ' | grep \'udp sum ok\' | wc -l' \
 118             .format(con.REMOTE_FW_DIR, file_prefix, con.TLDK_TESTCONFIG)
 119
 120         (ret_code, stdout, _) = ssh.exec_command(cmd, timeout=100)
 121         if ret_code != 0:
 122             raise RuntimeError('Failed to get test result at node {0}'
 123                                .format(dut_node['host']))
 124
 125         return stdout