resources/libraries/python/TLDK/UdpTest.py

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