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