5 from scapy.layers.l2 import Ether
6 from scapy.packet import Raw
7 from scapy.layers.inet import IP, UDP
10 class BridgeDomain(metaclass=abc.ABCMeta):
11 """Bridge domain abstraction"""
14 def frame_request(self):
15 """Ethernet frame modeling a generic request"""
17 Ether(src="00:00:00:00:00:01", dst="00:00:00:00:00:02")
18 / IP(src="1.2.3.4", dst="4.3.2.1")
19 / UDP(sport=10000, dport=20000)
24 def frame_reply(self):
25 """Ethernet frame modeling a generic reply"""
27 Ether(src="00:00:00:00:00:02", dst="00:00:00:00:00:01")
28 / IP(src="4.3.2.1", dst="1.2.3.4")
29 / UDP(sport=20000, dport=10000)
34 def ip_range(self, start, end):
35 """range of remote ip's"""
39 def encap_mcast(self, pkt, src_ip, src_mac, vni):
40 """Encapsulate mcast packet"""
44 def encapsulate(self, pkt, vni):
45 """Encapsulate packet"""
49 def decapsulate(self, pkt):
50 """Decapsulate packet"""
54 def check_encapsulation(self, pkt, vni, local_only=False):
55 """Verify the encapsulation"""
58 def assert_eq_pkts(self, pkt1, pkt2):
59 """Verify the Ether, IP, UDP, payload are equal in both
62 self.assertEqual(pkt1[Ether].src, pkt2[Ether].src)
63 self.assertEqual(pkt1[Ether].dst, pkt2[Ether].dst)
64 self.assertEqual(pkt1[IP].src, pkt2[IP].src)
65 self.assertEqual(pkt1[IP].dst, pkt2[IP].dst)
66 self.assertEqual(pkt1[UDP].sport, pkt2[UDP].sport)
67 self.assertEqual(pkt1[UDP].dport, pkt2[UDP].dport)
68 self.assertEqual(pkt1[Raw], pkt2[Raw])
72 Send encapsulated frames from pg0
73 Verify receipt of decapsulated frames on pg1
76 encapsulated_pkt = self.encapsulate(self.frame_request, self.single_tunnel_vni)
84 self.pg1.enable_capture()
88 # Pick first received frame and check if it's the non-encapsulated
90 out = self.pg1.get_capture(1)
92 self.assert_eq_pkts(pkt, self.frame_request)
97 Verify receipt of encapsulated frames on pg0
99 self.pg1.add_stream([self.frame_reply])
101 self.pg0.enable_capture()
105 # Pick first received frame and check if it's correctly encapsulated.
106 out = self.pg0.get_capture(1)
108 self.check_encapsulation(pkt, self.single_tunnel_vni)
110 payload = self.decapsulate(pkt)
111 self.assert_eq_pkts(payload, self.frame_reply)
113 def test_ucast_flood(self):
114 """Unicast flood test
116 Verify receipt of encapsulated frames on pg0
118 self.pg3.add_stream([self.frame_reply])
120 self.pg0.enable_capture()
124 # Get packet from each tunnel and assert it's correctly encapsulated.
125 out = self.pg0.get_capture(self.n_ucast_tunnels)
127 self.check_encapsulation(pkt, self.ucast_flood_bd, True)
128 payload = self.decapsulate(pkt)
129 self.assert_eq_pkts(payload, self.frame_reply)
131 def test_mcast_flood(self):
132 """Multicast flood test
134 Verify receipt of encapsulated frames on pg0
136 self.pg2.add_stream([self.frame_reply])
138 self.pg0.enable_capture()
142 # Pick first received frame and check if it's correctly encapsulated.
143 out = self.pg0.get_capture(1)
145 self.check_encapsulation(
146 pkt, self.mcast_flood_bd, local_only=False, mcast_pkt=True
149 payload = self.decapsulate(pkt)
150 self.assert_eq_pkts(payload, self.frame_reply)
152 def test_mcast_rcv(self):
153 """Multicast receive test
154 Send 20 encapsulated frames from pg0 only 10 match unicast tunnels
155 Verify receipt of 10 decap frames on pg2
157 mac = self.pg0.remote_mac
161 self.encap_mcast(self.frame_request, ip, mac, self.mcast_flood_bd)
162 for ip in self.ip_range(ip_range_start, ip_range_end)
164 self.pg0.add_stream(mcast_stream)
165 self.pg2.enable_capture()
167 out = self.pg2.get_capture(10)
169 self.assert_eq_pkts(pkt, self.frame_request)