6 from scapy.packet import Raw
7 from scapy.layers.l2 import Ether
8 from scapy.layers.inet import IP, UDP
11 from framework import VppTestCase, VppTestRunner
15 class TestSpan(VppTestCase):
16 """ SPAN Test Case """
19 hosts_nr = 10 # Number of hosts
20 pkts_per_burst = 257 # Number of packets per burst
24 super(TestSpan, cls).setUpClass()
27 super(TestSpan, self).setUp()
29 # create 3 pg interfaces
30 self.create_pg_interfaces(range(3))
32 # packet flows mapping pg0 -> pg1, pg2 -> pg3, etc.
34 self.flows[self.pg0] = [self.pg1]
37 self.pg_if_packet_sizes = [64, 512] #, 1518, 9018]
39 self.interfaces = list(self.pg_interfaces)
41 # Create host MAC and IPv4 lists
42 # self.MY_MACS = dict()
43 # self.MY_IP4S = dict()
44 self.create_host_lists(TestSpan.hosts_nr)
46 # Create bi-directional cross-connects between pg0 and pg1
47 self.vapi.sw_interface_set_l2_xconnect(
48 self.pg0.sw_if_index, self.pg1.sw_if_index, enable=1)
49 self.vapi.sw_interface_set_l2_xconnect(
50 self.pg1.sw_if_index, self.pg0.sw_if_index, enable=1)
52 # setup all interfaces
53 for i in self.interfaces:
58 # Enable SPAN on pg0 (mirrored to pg2)
59 self.vapi.sw_interface_span_enable_disable(self.pg0.sw_if_index, self.pg2.sw_if_index)
62 super(TestSpan, self).tearDown()
64 def create_host_lists(self, count):
65 """ Method to create required number of MAC and IPv4 addresses.
66 Create required number of host MAC addresses and distribute them among
67 interfaces. Create host IPv4 address for every host MAC address too.
69 :param count: Number of hosts to create MAC and IPv4 addresses for.
71 # mapping between packet-generator index and lists of test hosts
72 self.hosts_by_pg_idx = dict()
74 for pg_if in self.pg_interfaces:
75 # self.MY_MACS[i.sw_if_index] = []
76 # self.MY_IP4S[i.sw_if_index] = []
77 self.hosts_by_pg_idx[pg_if.sw_if_index] = []
78 hosts = self.hosts_by_pg_idx[pg_if.sw_if_index]
79 for j in range(0, count):
81 "00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
82 "172.17.1%02x.%u" % (pg_if.sw_if_index, j))
85 def create_stream(self, src_if, packet_sizes):
87 for i in range(0, TestSpan.pkts_per_burst):
88 dst_if = self.flows[src_if][0]
89 dst_host = random.choice(self.hosts_by_pg_idx[dst_if.sw_if_index])
90 src_host = random.choice(self.hosts_by_pg_idx[src_if.sw_if_index])
91 pkt_info = self.create_packet_info(
92 src_if.sw_if_index, dst_if.sw_if_index)
93 payload = self.info_to_payload(pkt_info)
94 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
95 IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4) /
96 UDP(sport=1234, dport=1234) /
98 pkt_info.data = p.copy()
99 size = packet_sizes[(i / 2) % len(packet_sizes)]
100 self.extend_packet(p, size)
104 def verify_capture(self, dst_if, capture_pg1, capture_pg2):
106 for i in self.interfaces:
107 last_info[i.sw_if_index] = None
108 dst_sw_if_index = dst_if.sw_if_index
109 if len(capture_pg1) != len(capture_pg2):
110 error("Diffrent number of outgoing and mirrored packets : %u != %u"
111 % (len(capture_pg1), len(capture_pg2)))
113 for pkt_pg1, pkt_pg2 in zip(capture_pg1, capture_pg2):
119 if pkt_pg1[Ether] != pkt_pg2[Ether]:
120 error("Diffrent ethernet header of outgoing and mirrored packet")
122 if ip1 != pkt_pg2[IP]:
123 error("Diffrent ip header of outgoing and mirrored packet")
125 if udp1 != pkt_pg2[UDP]:
126 error("Diffrent udp header of outgoing and mirrored packet")
128 if raw1 != pkt_pg2[Raw]:
129 error("Diffrent raw data of outgoing and mirrored packet")
132 payload_info = self.payload_to_info(str(raw1))
133 packet_index = payload_info.index
134 self.assertEqual(payload_info.dst, dst_sw_if_index)
135 debug("Got packet on port %s: src=%u (id=%u)" %
136 (dst_if.name, payload_info.src, packet_index))
137 next_info = self.get_next_packet_info_for_interface2(
138 payload_info.src, dst_sw_if_index,
139 last_info[payload_info.src])
140 last_info[payload_info.src] = next_info
141 self.assertTrue(next_info is not None)
142 self.assertEqual(packet_index, next_info.index)
143 saved_packet = next_info.data
144 # Check standard fields
145 self.assertEqual(ip1.src, saved_packet[IP].src)
146 self.assertEqual(ip1.dst, saved_packet[IP].dst)
147 self.assertEqual(udp1.sport, saved_packet[UDP].sport)
148 self.assertEqual(udp1.dport, saved_packet[UDP].dport)
150 error("Unexpected or invalid packet:")
154 for i in self.interfaces:
155 remaining_packet = self.get_next_packet_info_for_interface2(
156 i, dst_sw_if_index, last_info[i.sw_if_index])
157 self.assertTrue(remaining_packet is None,
158 "Port %u: Packet expected from source %u didn't"
159 " arrive" % (dst_sw_if_index, i.sw_if_index))
166 3 interfaces, pg0 l2xconnected with pg1
167 2. sending l2 eth packets between 2 interfaces (pg0, pg1) and mirrored to pg2
168 64B, 512B, 1518B, 9018B (ether_size)
169 burst of packets per interface
172 # Create incoming packet streams for packet-generator interfaces
173 pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes)
174 self.pg0.add_stream(pkts)
176 # Enable packet capturing and start packet sending
177 self.pg_enable_capture(self.pg_interfaces)
180 # Verify packets outgoing packet streams on mirrored interface (pg2)
181 info("Verifying capture on interfaces %s and %s" % (self.pg1.name, self.pg2.name))
182 self.verify_capture(self.pg1, self.pg1.get_capture(), self.pg2.get_capture())
185 if __name__ == '__main__':
186 unittest.main(testRunner=VppTestRunner)