test/vpp_sub_interface.py

   1 from scapy.layers.l2 import Dot1Q
   2 from abc import abstractmethod, ABCMeta
   3 from vpp_interface import VppInterface
   4 from vpp_pg_interface import VppPGInterface
   5 from vpp_papi_provider import L2_VTR_OP
   6
   7
   8 class VppSubInterface(VppPGInterface):
   9     __metaclass__ = ABCMeta
  10
  11     @property
  12     def parent(self):
  13         """Parent interface for this sub-interface"""
  14         return self._parent
  15
  16     @property
  17     def sub_id(self):
  18         """Sub-interface ID"""
  19         return self._sub_id
  20
  21     @property
  22     def tag1(self):
  23         return self._tag1
  24
  25     @property
  26     def tag2(self):
  27         return self._tag2
  28
  29     @property
  30     def vtr(self):
  31         return self._vtr
  32
  33     def __init__(self, test, parent, sub_id):
  34         VppInterface.__init__(self, test)
  35         self._parent = parent
  36         self._parent.add_sub_if(self)
  37         self._sub_id = sub_id
  38         self.set_vtr(L2_VTR_OP.L2_DISABLED)
  39         self.DOT1AD_TYPE = 0x88A8
  40         self.DOT1Q_TYPE = 0x8100
  41
  42     @abstractmethod
  43     def create_arp_req(self):
  44         pass
  45
  46     @abstractmethod
  47     def create_ndp_req(self):
  48         pass
  49
  50     def resolve_arp(self):
  51         super(VppSubInterface, self).resolve_arp(self.parent)
  52
  53     def resolve_ndp(self):
  54         super(VppSubInterface, self).resolve_ndp(self.parent)
  55
  56     @abstractmethod
  57     def add_dot1_layer(self, pkt):
  58         pass
  59
  60     def remove_vpp_config(self):
  61         self.test.vapi.delete_subif(self._sw_if_index)
  62
  63     def _add_tag(self, packet, vlan, tag_type):
  64         payload = packet.payload
  65         inner_type = packet.type
  66         packet.remove_payload()
  67         packet.add_payload(Dot1Q(vlan=vlan) / payload)
  68         packet.payload.type = inner_type
  69         packet.payload.vlan = vlan
  70         packet.type = tag_type
  71         return packet
  72
  73     def _remove_tag(self, packet, vlan=None, tag_type=None):
  74         if tag_type:
  75             self.test.instance().assertEqual(packet.type, tag_type)
  76
  77         payload = packet.payload
  78         if vlan:
  79             self.test.instance().assertEqual(payload.vlan, vlan)
  80         inner_type = payload.type
  81         payload = payload.payload
  82         packet.remove_payload()
  83         packet.add_payload(payload)
  84         packet.type = inner_type
  85         return packet
  86
  87     def add_dot1q_layer(self, packet, vlan):
  88         return self._add_tag(packet, vlan, self.DOT1Q_TYPE)
  89
  90     def add_dot1ad_layer(self, packet, outer, inner):
  91         p = self._add_tag(packet, inner, self.DOT1Q_TYPE)
  92         return self._add_tag(p, outer, self.DOT1AD_TYPE)
  93
  94     def remove_dot1q_layer(self, packet, vlan=None):
  95         return self._remove_tag(packet, vlan, self.DOT1Q_TYPE)
  96
  97     def remove_dot1ad_layer(self, packet, outer=None, inner=None):
  98         p = self._remove_tag(packet, outer, self.DOT1AD_TYPE)
  99         return self._remove_tag(p, inner, self.DOT1Q_TYPE)
 100
 101     def set_vtr(self, vtr, push1q=0, tag=None, inner=None, outer=None):
 102         self._tag1 = 0
 103         self._tag2 = 0
 104         self._push1q = 0
 105
 106         if (vtr == L2_VTR_OP.L2_PUSH_1 or
 107             vtr == L2_VTR_OP.L2_TRANSLATE_1_1 or
 108                 vtr == L2_VTR_OP.L2_TRANSLATE_2_1):
 109             self._tag1 = tag
 110             self._push1q = push1q
 111         if (vtr == L2_VTR_OP.L2_PUSH_2 or
 112             vtr == L2_VTR_OP.L2_TRANSLATE_1_2 or
 113                 vtr == L2_VTR_OP.L2_TRANSLATE_2_2):
 114             self._tag1 = outer
 115             self._tag2 = inner
 116             self._push1q = push1q
 117
 118         self.test.vapi.sw_interface_set_l2_tag_rewrite(
 119             self.sw_if_index, vtr, push=self._push1q,
 120             tag1=self._tag1, tag2=self._tag2)
 121         self._vtr = vtr
 122
 123
 124 class VppDot1QSubint(VppSubInterface):
 125
 126     @property
 127     def vlan(self):
 128         """VLAN tag"""
 129         return self._vlan
 130
 131     def __init__(self, test, parent, sub_id, vlan=None):
 132         if vlan is None:
 133             vlan = sub_id
 134         self._vlan = vlan
 135         r = test.vapi.create_vlan_subif(parent.sw_if_index, vlan)
 136         self._sw_if_index = r.sw_if_index
 137         super(VppDot1QSubint, self).__init__(test, parent, sub_id)
 138
 139     def create_arp_req(self):
 140         packet = VppPGInterface.create_arp_req(self)
 141         return self.add_dot1_layer(packet)
 142
 143     def create_ndp_req(self):
 144         packet = VppPGInterface.create_ndp_req(self)
 145         return self.add_dot1_layer(packet)
 146
 147     # called before sending packet
 148     def add_dot1_layer(self, packet):
 149         return self.add_dot1q_layer(packet, self.vlan)
 150
 151     # called on received packet to "reverse" the add call
 152     def remove_dot1_layer(self, packet):
 153         return self.remove_dot1q_layer(packet, self.vlan)
 154
 155
 156 class VppDot1ADSubint(VppSubInterface):
 157
 158     @property
 159     def outer_vlan(self):
 160         """Outer VLAN tag"""
 161         return self._outer_vlan
 162
 163     @property
 164     def inner_vlan(self):
 165         """Inner VLAN tag"""
 166         return self._inner_vlan
 167
 168     def __init__(self, test, parent, sub_id, outer_vlan, inner_vlan):
 169         r = test.vapi.create_subif(parent.sw_if_index, sub_id, outer_vlan,
 170                                    inner_vlan, dot1ad=1, two_tags=1,
 171                                    exact_match=1)
 172         self._sw_if_index = r.sw_if_index
 173         self._outer_vlan = outer_vlan
 174         self._inner_vlan = inner_vlan
 175         super(VppDot1ADSubint, self).__init__(test, parent, sub_id)
 176
 177     def create_arp_req(self):
 178         packet = VppPGInterface.create_arp_req(self)
 179         return self.add_dot1_layer(packet)
 180
 181     def create_ndp_req(self):
 182         packet = VppPGInterface.create_ndp_req(self)
 183         return self.add_dot1_layer(packet)
 184
 185     def add_dot1_layer(self, packet):
 186         return self.add_dot1ad_layer(packet, self.outer_vlan, self.inner_vlan)
 187
 188     def remove_dot1_layer(self, packet):
 189         return self.remove_dot1ad_layer(packet, self.outer_vlan,
 190                                         self.inner_vlan)
 191
 192
 193 class VppP2PSubint(VppSubInterface):
 194
 195     def __init__(self, test, parent, sub_id, remote_mac):
 196         r = test.vapi.create_p2pethernet_subif(parent.sw_if_index,
 197                                                remote_mac, sub_id)
 198         self._sw_if_index = r.sw_if_index
 199         self.parent_sw_if_index = parent.sw_if_index
 200         self.p2p_remote_mac = remote_mac
 201         super(VppP2PSubint, self).__init__(test, parent, sub_id)
 202
 203     def add_dot1_layer(self, packet):
 204         return packet
 205
 206     def remove_dot1_layer(self, packet):
 207         return packet
 208
 209     def create_arp_req(self):
 210         packet = VppPGInterface.create_arp_req(self)
 211         return packet
 212
 213     def create_ndp_req(self):
 214         packet = VppPGInterface.create_ndp_req(self)
 215         return packet