test/vpp_sub_interface.py

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