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