Using $(CURDIR) to fix the absolute path issue
[vpp.git] / test / template_bd.py
1 #!/usr/bin/env python
2
3 from abc import abstractmethod, ABCMeta
4
5 from scapy.layers.l2 import Ether, Raw
6 from scapy.layers.inet import IP, UDP
7
8 from util import ip4_range
9
10
11 class BridgeDomain(object):
12     """ Bridge domain abstraction """
13     __metaclass__ = ABCMeta
14
15     @property
16     def frame_request(self):
17         """ Ethernet frame modeling a generic request """
18         return (Ether(src='00:00:00:00:00:01', dst='00:00:00:00:00:02') /
19                 IP(src='1.2.3.4', dst='4.3.2.1') /
20                 UDP(sport=10000, dport=20000) /
21                 Raw('\xa5' * 100))
22
23     @property
24     def frame_reply(self):
25         """ Ethernet frame modeling a generic reply """
26         return (Ether(src='00:00:00:00:00:02', dst='00:00:00:00:00:01') /
27                 IP(src='4.3.2.1', dst='1.2.3.4') /
28                 UDP(sport=20000, dport=10000) /
29                 Raw('\xa5' * 100))
30
31     @abstractmethod
32     def encap_mcast(self, pkt, src_ip, src_mac, vni):
33         """ Encapsulate mcast packet """
34         pass
35
36     @abstractmethod
37     def encapsulate(self, pkt, vni):
38         """ Encapsulate packet """
39         pass
40
41     @abstractmethod
42     def decapsulate(self, pkt):
43         """ Decapsulate packet """
44         pass
45
46     @abstractmethod
47     def check_encapsulation(self, pkt, vni, local_only=False):
48         """ Verify the encapsulation """
49         pass
50
51     def assert_eq_pkts(self, pkt1, pkt2):
52         """ Verify the Ether, IP, UDP, payload are equal in both
53         packets
54         """
55         self.assertEqual(pkt1[Ether].src, pkt2[Ether].src)
56         self.assertEqual(pkt1[Ether].dst, pkt2[Ether].dst)
57         self.assertEqual(pkt1[IP].src, pkt2[IP].src)
58         self.assertEqual(pkt1[IP].dst, pkt2[IP].dst)
59         self.assertEqual(pkt1[UDP].sport, pkt2[UDP].sport)
60         self.assertEqual(pkt1[UDP].dport, pkt2[UDP].dport)
61         self.assertEqual(pkt1[Raw], pkt2[Raw])
62
63     def test_decap(self):
64         """ Decapsulation test
65         Send encapsulated frames from pg0
66         Verify receipt of decapsulated frames on pg1
67         """
68
69         encapsulated_pkt = self.encapsulate(self.frame_request,
70                                             self.single_tunnel_bd)
71
72         self.pg0.add_stream([encapsulated_pkt, ])
73
74         self.pg1.enable_capture()
75
76         self.pg_start()
77
78         # Pick first received frame and check if it's the non-encapsulated
79         # frame
80         out = self.pg1.get_capture(1)
81         pkt = out[0]
82         self.assert_eq_pkts(pkt, self.frame_request)
83
84     def test_encap(self):
85         """ Encapsulation test
86         Send frames from pg1
87         Verify receipt of encapsulated frames on pg0
88         """
89         self.pg1.add_stream([self.frame_reply])
90
91         self.pg0.enable_capture()
92
93         self.pg_start()
94
95         # Pick first received frame and check if it's corectly encapsulated.
96         out = self.pg0.get_capture(1)
97         pkt = out[0]
98         self.check_encapsulation(pkt, self.single_tunnel_bd)
99
100         payload = self.decapsulate(pkt)
101         self.assert_eq_pkts(payload, self.frame_reply)
102
103     def test_ucast_flood(self):
104         """ Unicast flood test
105         Send frames from pg3
106         Verify receipt of encapsulated frames on pg0
107         """
108         self.pg3.add_stream([self.frame_reply])
109
110         self.pg0.enable_capture()
111
112         self.pg_start()
113
114         # Get packet from each tunnel and assert it's corectly encapsulated.
115         out = self.pg0.get_capture(self.n_ucast_tunnels)
116         for pkt in out:
117             self.check_encapsulation(pkt, self.ucast_flood_bd, True)
118             payload = self.decapsulate(pkt)
119             self.assert_eq_pkts(payload, self.frame_reply)
120
121     def test_mcast_flood(self):
122         """ Multicast flood test
123         Send frames from pg2
124         Verify receipt of encapsulated frames on pg0
125         """
126         self.pg2.add_stream([self.frame_reply])
127
128         self.pg0.enable_capture()
129
130         self.pg_start()
131
132         # Pick first received frame and check if it's corectly encapsulated.
133         out = self.pg0.get_capture(1)
134         pkt = out[0]
135         self.check_encapsulation(pkt, self.mcast_flood_bd,
136                                  local_only=False, mcast_pkt=True)
137
138         payload = self.decapsulate(pkt)
139         self.assert_eq_pkts(payload, self.frame_reply)
140
141     def test_mcast_rcv(self):
142         """ Multicast receive test
143         Send 20 encapsulated frames from pg0 only 10 match unicast tunnels
144         Verify receipt of 10 decap frames on pg2
145         """
146         mac = self.pg0.remote_mac
147         ip_range_start = 10
148         ip_range_end = 30
149         mcast_stream = [
150             self.encap_mcast(self.frame_request, ip, mac, self.mcast_flood_bd)
151             for ip in ip4_range(self.pg0.remote_ip4,
152                                 ip_range_start, ip_range_end)]
153         self.pg0.add_stream(mcast_stream)
154         self.pg2.enable_capture()
155         self.pg_start()
156         out = self.pg2.get_capture(10)
157         for pkt in out:
158             self.assert_eq_pkts(pkt, self.frame_request)