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