1 from scapy.layers.l2 import Dot1Q
3 from vpp_pg_interface import VppPGInterface
4 from vpp_interface import VppInterface
5 from vpp_papi import VppEnum
20 class VppSubInterface(VppPGInterface, metaclass=abc.ABCMeta):
23 """Parent interface for this sub-interface"""
28 """Sub-interface ID"""
43 def __init__(self, test, parent, sub_id):
44 VppInterface.__init__(self, test)
46 self._parent.add_sub_if(self)
48 self.DOT1AD_TYPE = 0x88A8
49 self.DOT1Q_TYPE = 0x8100
51 def set_sw_if_index(self, sw_if_index):
52 super(VppSubInterface, self).set_sw_if_index(sw_if_index)
53 self.set_vtr(L2_VTR_OP.L2_DISABLED)
56 def create_arp_req(self):
60 def create_ndp_req(self, addr=None):
63 def resolve_arp(self):
64 super(VppSubInterface, self).resolve_arp(self.parent)
66 def resolve_ndp(self):
67 super(VppSubInterface, self).resolve_ndp(self.parent)
70 def add_dot1_layer(self, pkt):
73 def remove_vpp_config(self):
74 self.test.vapi.delete_subif(self.sw_if_index)
76 def _add_tag(self, packet, vlan, tag_type):
77 payload = packet.payload
78 inner_type = packet.type
79 packet.remove_payload()
80 packet.add_payload(Dot1Q(vlan=vlan) / payload)
81 packet.payload.type = inner_type
82 packet.payload.vlan = vlan
83 packet.type = tag_type
86 def _remove_tag(self, packet, vlan=None, tag_type=None):
88 self.test.instance().assertEqual(packet.type, tag_type)
90 payload = packet.payload
92 self.test.instance().assertEqual(payload.vlan, vlan)
93 inner_type = payload.type
94 payload = payload.payload
95 packet.remove_payload()
96 packet.add_payload(payload)
97 packet.type = inner_type
100 def add_dot1q_layer(self, packet, vlan):
101 return self._add_tag(packet, vlan, self.DOT1Q_TYPE)
103 def add_dot1ad_layer(self, packet, outer, inner):
104 p = self._add_tag(packet, inner, self.DOT1Q_TYPE)
105 return self._add_tag(p, outer, self.DOT1AD_TYPE)
107 def remove_dot1q_layer(self, packet, vlan=None):
108 return self._remove_tag(packet, vlan, self.DOT1Q_TYPE)
110 def remove_dot1ad_layer(self, packet, outer=None, inner=None):
111 p = self._remove_tag(packet, outer, self.DOT1AD_TYPE)
112 return self._remove_tag(p, inner, self.DOT1Q_TYPE)
114 def set_vtr(self, vtr, push1q=0, tag=None, inner=None, outer=None):
120 vtr == L2_VTR_OP.L2_PUSH_1
121 or vtr == L2_VTR_OP.L2_TRANSLATE_1_1
122 or vtr == L2_VTR_OP.L2_TRANSLATE_2_1
125 self._push1q = push1q
127 vtr == L2_VTR_OP.L2_PUSH_2
128 or vtr == L2_VTR_OP.L2_TRANSLATE_1_2
129 or vtr == L2_VTR_OP.L2_TRANSLATE_2_2
133 self._push1q = push1q
135 self.test.vapi.l2_interface_vlan_tag_rewrite(
136 sw_if_index=self.sw_if_index,
138 push_dot1q=self._push1q,
145 class VppDot1QSubint(VppSubInterface):
151 def __init__(self, test, parent, sub_id, vlan=None):
152 super(VppDot1QSubint, self).__init__(test, parent, sub_id)
156 r = test.vapi.create_vlan_subif(parent.sw_if_index, vlan)
157 self.set_sw_if_index(r.sw_if_index)
159 def create_arp_req(self):
160 packet = VppPGInterface.create_arp_req(self)
161 return self.add_dot1_layer(packet)
163 def create_ndp_req(self, addr=None):
164 packet = VppPGInterface.create_ndp_req(self, addr)
165 return self.add_dot1_layer(packet)
167 # called before sending packet
168 def add_dot1_layer(self, packet):
169 return self.add_dot1q_layer(packet, self.vlan)
171 # called on received packet to "reverse" the add call
172 def remove_dot1_layer(self, packet):
173 return self.remove_dot1q_layer(packet, self.vlan)
176 class VppDot1ADSubint(VppSubInterface):
178 def outer_vlan(self):
180 return self._outer_vlan
183 def inner_vlan(self):
185 return self._inner_vlan
187 def __init__(self, test, parent, sub_id, outer_vlan, inner_vlan):
188 super(VppDot1ADSubint, self).__init__(test, parent, sub_id)
190 VppEnum.vl_api_sub_if_flags_t.SUB_IF_API_FLAG_DOT1AD
191 | VppEnum.vl_api_sub_if_flags_t.SUB_IF_API_FLAG_TWO_TAGS
192 | VppEnum.vl_api_sub_if_flags_t.SUB_IF_API_FLAG_EXACT_MATCH
194 r = test.vapi.create_subif(
195 sw_if_index=parent.sw_if_index,
197 outer_vlan_id=outer_vlan,
198 inner_vlan_id=inner_vlan,
201 self.set_sw_if_index(r.sw_if_index)
202 self._outer_vlan = outer_vlan
203 self._inner_vlan = inner_vlan
205 def create_arp_req(self):
206 packet = VppPGInterface.create_arp_req(self)
207 return self.add_dot1_layer(packet)
209 def create_ndp_req(self, addr=None):
210 packet = VppPGInterface.create_ndp_req(self, addr)
211 return self.add_dot1_layer(packet)
213 def add_dot1_layer(self, packet):
214 return self.add_dot1ad_layer(packet, self.outer_vlan, self.inner_vlan)
216 def remove_dot1_layer(self, packet):
217 return self.remove_dot1ad_layer(packet, self.outer_vlan, self.inner_vlan)
220 class VppP2PSubint(VppSubInterface):
221 def __init__(self, test, parent, sub_id, remote_mac):
222 super(VppP2PSubint, self).__init__(test, parent, sub_id)
223 r = test.vapi.p2p_ethernet_add(parent.sw_if_index, remote_mac, sub_id)
224 self.set_sw_if_index(r.sw_if_index)
225 self.parent_sw_if_index = parent.sw_if_index
226 self.p2p_remote_mac = remote_mac
228 def add_dot1_layer(self, packet):
231 def remove_dot1_layer(self, packet):
234 def create_arp_req(self):
235 packet = VppPGInterface.create_arp_req(self)
238 def create_ndp_req(self, addr=None):
239 packet = VppPGInterface.create_ndp_req(self, addr)