test/vpp_neighbor.py

   1 """
   2   Neighbour Entries
   3
   4   object abstractions for ARP and ND
   5 """
   6
   7 from socket import inet_pton, inet_ntop, AF_INET, AF_INET6
   8 from vpp_object import *
   9 from util import mactobinary
  10
  11
  12 def find_nbr(test, sw_if_index, ip_addr, is_static=0, inet=AF_INET):
  13     nbrs = test.vapi.ip_neighbor_dump(sw_if_index,
  14                                       is_ipv6=1 if AF_INET6 == inet else 0)
  15     if inet == AF_INET:
  16         s = 4
  17     else:
  18         s = 16
  19     nbr_addr = inet_pton(inet, ip_addr)
  20
  21     for n in nbrs:
  22         if nbr_addr == n.ip_address[:s] \
  23            and is_static == n.is_static:
  24             return True
  25     return False
  26
  27
  28 class VppNeighbor(VppObject):
  29     """
  30     ARP Entry
  31     """
  32
  33     def __init__(self, test, sw_if_index, mac_addr, nbr_addr,
  34                  af=AF_INET, is_static=False, is_no_fib_entry=0):
  35         self._test = test
  36         self.sw_if_index = sw_if_index
  37         self.mac_addr = mactobinary(mac_addr)
  38         self.af = af
  39         self.is_static = is_static
  40         self.is_no_fib_entry = is_no_fib_entry
  41         self.nbr_addr = nbr_addr
  42         self.nbr_addr_n = inet_pton(af, nbr_addr)
  43
  44     def add_vpp_config(self):
  45         self._test.vapi.ip_neighbor_add_del(
  46             self.sw_if_index,
  47             self.mac_addr,
  48             self.nbr_addr_n,
  49             is_add=1,
  50             is_ipv6=1 if AF_INET6 == self.af else 0,
  51             is_static=self.is_static,
  52             is_no_adj_fib=self.is_no_fib_entry)
  53         self._test.registry.register(self, self._test.logger)
  54
  55     def remove_vpp_config(self):
  56         self._test.vapi.ip_neighbor_add_del(
  57             self.sw_if_index,
  58             self.mac_addr,
  59             self.nbr_addr_n,
  60             is_ipv6=1 if AF_INET6 == self.af else 0,
  61             is_add=0,
  62             is_static=self.is_static)
  63
  64     def query_vpp_config(self):
  65         return find_nbr(self._test,
  66                         self.sw_if_index,
  67                         self.nbr_addr,
  68                         self.is_static,
  69                         self.af)
  70
  71     def __str__(self):
  72         return self.object_id()
  73
  74     def object_id(self):
  75         return ("%d:%s" % (self.sw_if_index, self.nbr_addr))