aboutsummaryrefslogtreecommitdiffstats
path: root/test/vpp_bond_interface.py
blob: 0db04e135b1dc90b306907d210dcd990dda0ae94 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
from vpp_object import VppObject
from vpp_interface import VppInterface


class VppBondInterface(VppInterface):
    """VPP bond interface."""

    def __init__(self, test, mode, lb=0, numa_only=0,
                 use_custom_mac=0, mac_address=''):

        """ Create VPP Bond interface """
        super(VppBondInterface, self).__init__(test)
        self.mode = mode
        self.lb = lb
        self.numa_only = numa_only
        self.use_custom_mac = use_custom_mac
        self.mac_address = mac_address

    def add_vpp_config(self):
        r = self.test.vapi.bond_create(self.mode,
                                       self.lb,
                                       self.numa_only,
                                       self.use_custom_mac,
                                       self.mac_address)
        self.set_sw_if_index(r.sw_if_index)

    def remove_vpp_config(self):
        self.test.vapi.bond_delete(self.sw_if_index)

    def enslave_vpp_bond_interface(self,
                                   sw_if_index,
                                   is_passive=0,
                                   is_long_timeout=0):
        self.test.vapi.bond_enslave(sw_if_index,
                                    self.sw_if_index,
                                    is_passive,
                                    is_long_timeout)

    def detach_vpp_bond_interface(self,
                                  sw_if_index):
        self.test.vapi.bond_detach_slave(sw_if_index)

    def is_interface_config_in_dump(self, dump):
        for i in dump:
            if i.sw_if_index == self.sw_if_index:
                return True
        else:
            return False
iting, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """ Traffic script that sends an TCP packet from TG to DUT. """ import sys from scapy.all import Ether, Packet, Raw from scapy.layers.inet import IP, TCP from scapy.layers.inet6 import IPv6, ICMPv6ND_NS from resources.libraries.python.SFC.VerifyPacket import * from resources.libraries.python.SFC.SFCConstants import SFCConstants as SfcCon from resources.libraries.python.TrafficScriptArg import TrafficScriptArg from resources.libraries.python.PacketVerifier import RxQueue, TxQueue def main(): """Send TCP packet from one traffic generator interface to DUT. :raises: If the IP address is invalid. """ args = TrafficScriptArg(['src_mac', 'dst_mac', 'src_ip', 'dst_ip', 'timeout', 'framesize', 'testtype']) 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') tx_if = args.get_arg('tx_if') rx_if = args.get_arg('rx_if') timeout = int(args.get_arg('timeout')) frame_size = int(args.get_arg('framesize')) test_type = args.get_arg('testtype') rxq = RxQueue(rx_if) txq = TxQueue(tx_if) sent_packets = [] protocol = TCP source_port = SfcCon.DEF_SRC_PORT destination_port = SfcCon.DEF_DST_PORT if valid_ipv4(src_ip) and valid_ipv4(dst_ip): ip_version = IP elif valid_ipv6(src_ip) and valid_ipv6(dst_ip): ip_version = IPv6 else: raise ValueError("Invalid IP version!") pkt_header = (Ether(src=src_mac, dst=dst_mac) / ip_version(src=src_ip, dst=dst_ip) / protocol(sport=int(source_port), dport=int(destination_port))) fsize_no_fcs = frame_size - 4 pad_len = max(0, fsize_no_fcs - len(pkt_header)) pad_data = "A" * pad_len pkt_raw = pkt_header / Raw(load=pad_data) # Send created packet on one interface and receive on the other sent_packets.append(pkt_raw) txq.send(pkt_raw) while True: ether = rxq.recv(timeout) if ether is None: raise RuntimeError('No packet is received!') if ether.haslayer(ICMPv6ND_NS): # read another packet in the queue if the current one is ICMPv6ND_NS continue else: # otherwise process the current packet break # let us begin to check the NSH SFC loopback packet VerifyPacket.check_the_nsh_sfc_packet(ether, frame_size, test_type) # we check all the fields about the loopback packet, this test will pass sys.exit(0) if __name__ == "__main__": main()