test/template_bd.py

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