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