resources/traffic_scripts/send_tcp_for_classifier_test.py

   1 #!/usr/bin/env python
   2 # Copyright (c) 2017 Cisco and/or its affiliates.
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at:
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 """
  16 Traffic script that sends an TCP packet
  17 from TG to DUT.
  18 """
  19 import sys
  20 import time
  21
  22 from scapy.layers.inet import IP, UDP, TCP
  23 from scapy.layers.inet6 import IPv6
  24 from scapy.all import Ether, Packet, Raw
  25 from scapy.all import sendp
  26
  27 from resources.libraries.python.SFC.VerifyPacket import *
  28 from resources.libraries.python.SFC.SFCConstants import SFCConstants as sfccon
  29 from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
  30
  31 from robot.api import logger
  32
  33 def main():
  34     """Send TCP packet from one traffic generator interface to DUT.
  35
  36     :raises: If the IP address is invalid.
  37     """
  38     args = TrafficScriptArg(
  39         ['src_mac', 'dst_mac', 'src_ip', 'dst_ip',
  40         'timeout', 'framesize', 'testtype'])
  41
  42     src_mac = args.get_arg('src_mac')
  43     dst_mac = args.get_arg('dst_mac')
  44     src_ip = args.get_arg('src_ip')
  45     dst_ip = args.get_arg('dst_ip')
  46     tx_if = args.get_arg('tx_if')
  47     rx_if = args.get_arg('rx_if')
  48     timeout = int(args.get_arg('timeout'))
  49     frame_size = int(args.get_arg('framesize'))
  50     test_type = args.get_arg('testtype')
  51
  52     protocol = TCP
  53     source_port = sfccon.DEF_SRC_PORT
  54     destination_port = sfccon.DEF_DST_PORT
  55
  56     ip_version = None
  57     if valid_ipv4(src_ip) and valid_ipv4(dst_ip):
  58         ip_version = IP
  59     elif valid_ipv6(src_ip) and valid_ipv6(dst_ip):
  60         ip_version = IPv6
  61     else:
  62         raise ValueError("Invalid IP version!")
  63
  64     pkt_header = (Ether(src=src_mac, dst=dst_mac) /
  65                ip_version(src=src_ip, dst=dst_ip) /
  66                protocol(sport=int(source_port), dport=int(destination_port)))
  67
  68     fsize_no_fcs = frame_size - 4
  69     pad_len = max(0, fsize_no_fcs - len(pkt_header))
  70     pad_data = "A" * pad_len
  71
  72     pkt_raw = pkt_header / Raw(load=pad_data)
  73
  74     sendp(pkt_raw, iface=tx_if, count=3)
  75
  76     time.sleep(timeout)
  77
  78     # let us begin to check the NSH SFC loopback  packet
  79     VerifyPacket.check_the_nsh_sfc_packet(frame_size, test_type)
  80
  81     # we check all the fields about the loopback packet, this test will pass
  82     sys.exit(0)
  83
  84
  85 if __name__ == "__main__":
  86     main()