test/vpp_sub_interface.py

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