test/vpp_sub_interface.py

   1 from scapy.layers.l2 import Ether, Dot1Q
   2 from abc import abstractmethod, ABCMeta
   3 from vpp_interface import VppInterface
   4
   5
   6 class VppSubInterface(VppInterface):
   7     __metaclass__ = ABCMeta
   8
   9     @property
  10     def parent(self):
  11         """Parent interface for this sub-interface"""
  12         return self._parent
  13
  14     @property
  15     def sub_id(self):
  16         """Sub-interface ID"""
  17         return self._sub_id
  18
  19     def __init__(self, test, parent, sub_id):
  20         self._test = test
  21         self._parent = parent
  22         self._parent.add_sub_if(self)
  23         self._sub_id = sub_id
  24
  25     @abstractmethod
  26     def create_arp_req(self):
  27         pass
  28
  29     @abstractmethod
  30     def create_ndp_req(self):
  31         pass
  32
  33     def resolve_arp(self):
  34         super(VppSubInterface, self).resolve_arp(self.parent)
  35
  36     def resolve_ndp(self):
  37         super(VppSubInterface, self).resolve_ndp(self.parent)
  38
  39     @abstractmethod
  40     def add_dot1_layer(self, pkt):
  41         pass
  42
  43
  44 class VppDot1QSubint(VppSubInterface):
  45
  46     @property
  47     def vlan(self):
  48         """VLAN tag"""
  49         return self._vlan
  50
  51     def __init__(self, test, parent, sub_id, vlan=None):
  52         if vlan is None:
  53             vlan = sub_id
  54         super(VppDot1QSubint, self).__init__(test, parent, sub_id)
  55         self._vlan = vlan
  56         r = self.test.vapi.create_vlan_subif(parent.sw_if_index, self.vlan)
  57         self._sw_if_index = r.sw_if_index
  58         self.post_init_setup()
  59
  60     def create_arp_req(self):
  61         packet = VppInterface.create_arp_req(self)
  62         return self.add_dot1_layer(packet)
  63
  64     def create_ndp_req(self):
  65         packet = VppInterface.create_ndp_req(self)
  66         return self.add_dot1_layer(packet)
  67
  68     def add_dot1_layer(self, packet):
  69         payload = packet.payload
  70         packet.remove_payload()
  71         packet.add_payload(Dot1Q(vlan=self.sub_id) / payload)
  72         return packet
  73
  74     def remove_dot1_layer(self, packet):
  75         payload = packet.payload
  76         self.test.instance().assertEqual(type(payload), Dot1Q)
  77         self.test.instance().assertEqual(payload.vlan, self.vlan)
  78         payload = payload.payload
  79         packet.remove_payload()
  80         packet.add_payload(payload)
  81         return packet
  82
  83
  84 class VppDot1ADSubint(VppSubInterface):
  85
  86     @property
  87     def outer_vlan(self):
  88         """Outer VLAN tag"""
  89         return self._outer_vlan
  90
  91     @property
  92     def inner_vlan(self):
  93         """Inner VLAN tag"""
  94         return self._inner_vlan
  95
  96     def __init__(self, test, parent, sub_id, outer_vlan, inner_vlan):
  97         super(VppDot1ADSubint, self).__init__(test, parent, sub_id)
  98         self.DOT1AD_TYPE = 0x88A8
  99         self.DOT1Q_TYPE = 0x8100
 100         self._outer_vlan = outer_vlan
 101         self._inner_vlan = inner_vlan
 102         r = self.test.vapi.create_subif(
 103             parent.sw_if_index,
 104             self.sub_id,
 105             self.outer_vlan,
 106             self.inner_vlan,
 107             dot1ad=1,
 108             two_tags=1,
 109             exact_match=1)
 110         self._sw_if_index = r.sw_if_index
 111         self.post_init_setup()
 112
 113     def create_arp_req(self):
 114         packet = VppInterface.create_arp_req(self)
 115         return self.add_dot1_layer(packet)
 116
 117     def create_ndp_req(self):
 118         packet = VppInterface.create_ndp_req(self)
 119         return self.add_dot1_layer(packet)
 120
 121     def add_dot1_layer(self, packet):
 122         payload = packet.payload
 123         packet.remove_payload()
 124         packet.add_payload(Dot1Q(vlan=self.outer_vlan) /
 125                            Dot1Q(vlan=self.inner_vlan) / payload)
 126         packet.type = self.DOT1AD_TYPE
 127         return packet
 128
 129     def remove_dot1_layer(self, packet):
 130         self.test.instance().assertEqual(type(packet), Ether)
 131         self.test.instance().assertEqual(packet.type, self.DOT1AD_TYPE)
 132         packet.type = self.DOT1Q_TYPE
 133         packet = Ether(str(packet))
 134         payload = packet.payload
 135         self.test.instance().assertEqual(type(payload), Dot1Q)
 136         self.test.instance().assertEqual(payload.vlan, self.outer_vlan)
 137         payload = payload.payload
 138         self.test.instance().assertEqual(type(payload), Dot1Q)
 139         self.test.instance().assertEqual(payload.vlan, self.inner_vlan)
 140         payload = payload.payload
 141         packet.remove_payload()
 142         packet.add_payload(payload)
 143         return packet