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