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