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