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         self._test = 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
  45 class VppDot1QSubint(VppSubInterface):
  46
  47     @property
  48     def vlan(self):
  49         """VLAN tag"""
  50         return self._vlan
  51
  52     def __init__(self, test, parent, sub_id, vlan=None):
  53         if vlan is None:
  54             vlan = sub_id
  55         super(VppDot1QSubint, self).__init__(test, parent, sub_id)
  56         self._vlan = vlan
  57         r = self.test.vapi.create_vlan_subif(parent.sw_if_index, self.vlan)
  58         self._sw_if_index = r.sw_if_index
  59         VppInterface.post_init_setup(self)
  60
  61     def create_arp_req(self):
  62         packet = VppPGInterface.create_arp_req(self)
  63         return self.add_dot1_layer(packet)
  64
  65     def create_ndp_req(self):
  66         packet = VppPGInterface.create_ndp_req(self)
  67         return self.add_dot1_layer(packet)
  68
  69     def add_dot1_layer(self, packet):
  70         payload = packet.payload
  71         packet.remove_payload()
  72         packet.add_payload(Dot1Q(vlan=self.sub_id) / payload)
  73         return packet
  74
  75     def remove_dot1_layer(self, packet):
  76         payload = packet.payload
  77         self.test.instance().assertEqual(type(payload), Dot1Q)
  78         self.test.instance().assertEqual(payload.vlan, self.vlan)
  79         payload = payload.payload
  80         packet.remove_payload()
  81         packet.add_payload(payload)
  82         return packet
  83
  84
  85 class VppDot1ADSubint(VppSubInterface):
  86
  87     @property
  88     def outer_vlan(self):
  89         """Outer VLAN tag"""
  90         return self._outer_vlan
  91
  92     @property
  93     def inner_vlan(self):
  94         """Inner VLAN tag"""
  95         return self._inner_vlan
  96
  97     def __init__(self, test, parent, sub_id, outer_vlan, inner_vlan):
  98         super(VppDot1ADSubint, self).__init__(test, parent, sub_id)
  99         self.DOT1AD_TYPE = 0x88A8
 100         self.DOT1Q_TYPE = 0x8100
 101         self._outer_vlan = outer_vlan
 102         self._inner_vlan = inner_vlan
 103         r = self.test.vapi.create_subif(
 104             parent.sw_if_index,
 105             self.sub_id,
 106             self.outer_vlan,
 107             self.inner_vlan,
 108             dot1ad=1,
 109             two_tags=1,
 110             exact_match=1)
 111         self._sw_if_index = r.sw_if_index
 112         VppInterface.post_init_setup(self)
 113
 114     def create_arp_req(self):
 115         packet = VppPGInterface.create_arp_req(self)
 116         return self.add_dot1_layer(packet)
 117
 118     def create_ndp_req(self):
 119         packet = VppPGInterface.create_ndp_req(self)
 120         return self.add_dot1_layer(packet)
 121
 122     def add_dot1_layer(self, packet):
 123         payload = packet.payload
 124         packet.remove_payload()
 125         packet.add_payload(Dot1Q(vlan=self.outer_vlan) /
 126                            Dot1Q(vlan=self.inner_vlan) / payload)
 127         packet.type = self.DOT1AD_TYPE
 128         return packet
 129
 130     def remove_dot1_layer(self, packet):
 131         self.test.instance().assertEqual(type(packet), Ether)
 132         self.test.instance().assertEqual(packet.type, self.DOT1AD_TYPE)
 133         packet.type = self.DOT1Q_TYPE
 134         packet = Ether(str(packet))
 135         payload = packet.payload
 136         self.test.instance().assertEqual(type(payload), Dot1Q)
 137         self.test.instance().assertEqual(payload.vlan, self.outer_vlan)
 138         payload = payload.payload
 139         self.test.instance().assertEqual(type(payload), Dot1Q)
 140         self.test.instance().assertEqual(payload.vlan, self.inner_vlan)
 141         payload = payload.payload
 142         packet.remove_payload()
 143         packet.add_payload(payload)
 144         return packet