3 from abc import abstractmethod, ABCMeta
5 from scapy.layers.l2 import Ether, Raw
6 from scapy.layers.inet import IP, UDP
9 class BridgeDomain(object):
10 """ Bridge domain abstraction """
11 __metaclass__ = ABCMeta
14 def frame_pg0_to_pg1(self):
15 """ Ethernet frame sent from pg0 and expected to arrive at pg1 """
16 return (Ether(src='00:00:00:00:00:01', dst='00:00:00:00:00:02') /
17 IP(src='1.2.3.4', dst='4.3.2.1') /
18 UDP(sport=10000, dport=20000) /
22 def frame_pg1_to_pg0(self):
23 """ Ethernet frame sent from pg1 and expected to arrive at pg0 """
24 return (Ether(src='00:00:00:00:00:02', dst='00:00:00:00:00:01') /
25 IP(src='4.3.2.1', dst='1.2.3.4') /
26 UDP(sport=20000, dport=10000) /
30 def encapsulate(self, pkt):
31 """ Encapsulate packet """
35 def decapsulate(self, pkt):
36 """ Decapsulate packet """
40 def check_encapsulation(self, pkt):
41 """ Verify the encapsulation """
45 """ Decapsulation test
46 Send encapsulated frames from pg0
47 Verify receipt of decapsulated frames on pg1
50 encapsulated_pkt = self.encapsulate(self.frame_pg0_to_pg1)
52 self.pg0.add_stream([encapsulated_pkt, ])
54 self.pg1.enable_capture()
58 # Pick first received frame and check if it's the non-encapsulated frame
59 out = self.pg1.get_capture()
60 self.assertEqual(len(out), 1,
61 'Invalid number of packets on '
62 'output: {}'.format(len(out)))
65 # TODO: add error messages
66 self.assertEqual(pkt[Ether].src, self.frame_pg0_to_pg1[Ether].src)
67 self.assertEqual(pkt[Ether].dst, self.frame_pg0_to_pg1[Ether].dst)
68 self.assertEqual(pkt[IP].src, self.frame_pg0_to_pg1[IP].src)
69 self.assertEqual(pkt[IP].dst, self.frame_pg0_to_pg1[IP].dst)
70 self.assertEqual(pkt[UDP].sport, self.frame_pg0_to_pg1[UDP].sport)
71 self.assertEqual(pkt[UDP].dport, self.frame_pg0_to_pg1[UDP].dport)
72 self.assertEqual(pkt[Raw], self.frame_pg0_to_pg1[Raw])
75 """ Encapsulation test
77 Verify receipt of encapsulated frames on pg0
79 self.pg1.add_stream([self.frame_pg1_to_pg0])
81 self.pg0.enable_capture()
85 # Pick first received frame and check if it's corectly encapsulated.
86 out = self.pg0.get_capture()
87 self.assertEqual(len(out), 1,
88 'Invalid number of packets on '
89 'output: {}'.format(len(out)))
91 self.check_encapsulation(pkt)
93 payload = self.decapsulate(pkt)
94 # TODO: add error messages
95 self.assertEqual(payload[Ether].src, self.frame_pg1_to_pg0[Ether].src)
96 self.assertEqual(payload[Ether].dst, self.frame_pg1_to_pg0[Ether].dst)
97 self.assertEqual(payload[IP].src, self.frame_pg1_to_pg0[IP].src)
98 self.assertEqual(payload[IP].dst, self.frame_pg1_to_pg0[IP].dst)
99 self.assertEqual(payload[UDP].sport, self.frame_pg1_to_pg0[UDP].sport)
100 self.assertEqual(payload[UDP].dport, self.frame_pg1_to_pg0[UDP].dport)
101 self.assertEqual(payload[Raw], self.frame_pg1_to_pg0[Raw])