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