test/vpp_sub_interface.py

   1 from scapy.layers.l2 import Ether, Dot1Q
   2 from abc import abstractmethod, ABCMeta
   3 from vpp_interface import VppInterface
   4 from vpp_pg_interface import VppPGInterface
   5
   6
   7 class VppSubInterface(VppPGInterface):
   8     __metaclass__ = ABCMeta
   9
  10     @property
  11     def parent(self):
  12         """Parent interface for this sub-interface"""
  13         return self._parent
  14
  15     @property
  16     def sub_id(self):
  17         """Sub-interface ID"""
  18         return self._sub_id
  19
  20     def __init__(self, test, parent, sub_id):
  21         VppInterface.__init__(self, test)
  22         self._parent = parent
  23         self._parent.add_sub_if(self)
  24         self._sub_id = sub_id
  25
  26     @abstractmethod
  27     def create_arp_req(self):
  28         pass
  29
  30     @abstractmethod
  31     def create_ndp_req(self):
  32         pass
  33
  34     def resolve_arp(self):
  35         super(VppSubInterface, self).resolve_arp(self.parent)
  36
  37     def resolve_ndp(self):
  38         super(VppSubInterface, self).resolve_ndp(self.parent)
  39
  40     @abstractmethod
  41     def add_dot1_layer(self, pkt):
  42         pass
  43
  44     def remove_vpp_config(self):
  45         self.test.vapi.delete_subif(self._sw_if_index)
  46
  47
  48 class VppDot1QSubint(VppSubInterface):
  49
  50     @property
  51     def vlan(self):
  52         """VLAN tag"""
  53         return self._vlan
  54
  55     def __init__(self, test, parent, sub_id, vlan=None):
  56         if vlan is None:
  57             vlan = sub_id
  58         self._vlan = vlan
  59         r = test.vapi.create_vlan_subif(parent.sw_if_index, vlan)
  60         self._sw_if_index = r.sw_if_index
  61         super(VppDot1QSubint, self).__init__(test, parent, sub_id)
  62
  63     def create_arp_req(self):
  64         packet = VppPGInterface.create_arp_req(self)
  65         return self.add_dot1_layer(packet)
  66
  67     def create_ndp_req(self):
  68         packet = VppPGInterface.create_ndp_req(self)
  69         return self.add_dot1_layer(packet)
  70
  71     def add_dot1_layer(self, packet):
  72         payload = packet.payload
  73         packet.remove_payload()
  74         packet.add_payload(Dot1Q(vlan=self.sub_id) / payload)
  75         return packet
  76
  77     def remove_dot1_layer(self, packet):
  78         payload = packet.payload
  79         self.test.instance().assertEqual(type(payload), Dot1Q)
  80         self.test.instance().assertEqual(payload.vlan, self.vlan)
  81         payload = payload.payload
  82         packet.remove_payload()
  83         packet.add_payload(payload)
  84         return packet
  85
  86
  87 class VppDot1ADSubint(VppSubInterface):
  88
  89     @property
  90     def outer_vlan(self):
  91         """Outer VLAN tag"""
  92         return self._outer_vlan
  93
  94     @property
  95     def inner_vlan(self):
  96         """Inner VLAN tag"""
  97         return self._inner_vlan
  98
  99     def __init__(self, test, parent, sub_id, outer_vlan, inner_vlan):
 100         r = test.vapi.create_subif(parent.sw_if_index, sub_id, outer_vlan,
 101                                    inner_vlan, dot1ad=1, two_tags=1,
 102                                    exact_match=1)
 103         self._sw_if_index = r.sw_if_index
 104         super(VppDot1ADSubint, self).__init__(test, parent, sub_id)
 105         self.DOT1AD_TYPE = 0x88A8
 106         self.DOT1Q_TYPE = 0x8100
 107         self._outer_vlan = outer_vlan
 108         self._inner_vlan = inner_vlan
 109
 110     def create_arp_req(self):
 111         packet = VppPGInterface.create_arp_req(self)
 112         return self.add_dot1_layer(packet)
 113
 114     def create_ndp_req(self):
 115         packet = VppPGInterface.create_ndp_req(self)
 116         return self.add_dot1_layer(packet)
 117
 118     def add_dot1_layer(self, packet):
 119         payload = packet.payload
 120         packet.remove_payload()
 121         packet.add_payload(Dot1Q(vlan=self.outer_vlan) /
 122                            Dot1Q(vlan=self.inner_vlan) / payload)
 123         packet.type = self.DOT1AD_TYPE
 124         return packet
 125
 126     def remove_dot1_layer(self, packet):
 127         self.test.instance().assertEqual(type(packet), Ether)
 128         self.test.instance().assertEqual(packet.type, self.DOT1AD_TYPE)
 129         packet.type = self.DOT1Q_TYPE
 130         packet = Ether(str(packet))
 131         payload = packet.payload
 132         self.test.instance().assertEqual(type(payload), Dot1Q)
 133         self.test.instance().assertEqual(payload.vlan, self.outer_vlan)
 134         payload = payload.payload
 135         self.test.instance().assertEqual(type(payload), Dot1Q)
 136         self.test.instance().assertEqual(payload.vlan, self.inner_vlan)
 137         payload = payload.payload
 138         packet.remove_payload()
 139         packet.add_payload(payload)
 140         return packet