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
8 class VppSubInterface(VppPGInterface):
9 __metaclass__ = ABCMeta
13 """Parent interface for this sub-interface"""
18 """Sub-interface ID"""
33 def __init__(self, test, parent, sub_id):
34 VppInterface.__init__(self, test)
36 self._parent.add_sub_if(self)
38 self.DOT1AD_TYPE = 0x88A8
39 self.DOT1Q_TYPE = 0x8100
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)
46 def create_arp_req(self):
50 def create_ndp_req(self):
53 def resolve_arp(self):
54 super(VppSubInterface, self).resolve_arp(self.parent)
56 def resolve_ndp(self):
57 super(VppSubInterface, self).resolve_ndp(self.parent)
60 def add_dot1_layer(self, pkt):
63 def remove_vpp_config(self):
64 self.test.vapi.delete_subif(self.sw_if_index)
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
76 def _remove_tag(self, packet, vlan=None, tag_type=None):
78 self.test.instance().assertEqual(packet.type, tag_type)
80 payload = packet.payload
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
90 def add_dot1q_layer(self, packet, vlan):
91 return self._add_tag(packet, vlan, self.DOT1Q_TYPE)
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)
97 def remove_dot1q_layer(self, packet, vlan=None):
98 return self._remove_tag(packet, vlan, self.DOT1Q_TYPE)
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)
104 def set_vtr(self, vtr, push1q=0, tag=None, inner=None, outer=None):
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):
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):
119 self._push1q = push1q
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)
127 class VppDot1QSubint(VppSubInterface):
134 def __init__(self, test, parent, sub_id, vlan=None):
135 super(VppDot1QSubint, self).__init__(test, parent, sub_id)
139 r = test.vapi.create_vlan_subif(parent.sw_if_index, vlan)
140 self.set_sw_if_index(r.sw_if_index)
142 def create_arp_req(self):
143 packet = VppPGInterface.create_arp_req(self)
144 return self.add_dot1_layer(packet)
146 def create_ndp_req(self):
147 packet = VppPGInterface.create_ndp_req(self)
148 return self.add_dot1_layer(packet)
150 # called before sending packet
151 def add_dot1_layer(self, packet):
152 return self.add_dot1q_layer(packet, self.vlan)
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)
159 class VppDot1ADSubint(VppSubInterface):
162 def outer_vlan(self):
164 return self._outer_vlan
167 def inner_vlan(self):
169 return self._inner_vlan
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,
176 self.set_sw_if_index(r.sw_if_index)
177 self._outer_vlan = outer_vlan
178 self._inner_vlan = inner_vlan
180 def create_arp_req(self):
181 packet = VppPGInterface.create_arp_req(self)
182 return self.add_dot1_layer(packet)
184 def create_ndp_req(self):
185 packet = VppPGInterface.create_ndp_req(self)
186 return self.add_dot1_layer(packet)
188 def add_dot1_layer(self, packet):
189 return self.add_dot1ad_layer(packet, self.outer_vlan, self.inner_vlan)
191 def remove_dot1_layer(self, packet):
192 return self.remove_dot1ad_layer(packet, self.outer_vlan,
196 class VppP2PSubint(VppSubInterface):
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,
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
206 def add_dot1_layer(self, packet):
209 def remove_dot1_layer(self, packet):
212 def create_arp_req(self):
213 packet = VppPGInterface.create_arp_req(self)
216 def create_ndp_req(self):
217 packet = VppPGInterface.create_ndp_req(self)