test/vpp_sub_interface.py

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