def local_addr(self):
""" BFD session local address (VPP address) """
if self._local_addr is None:
- return self._interface.local_ip4
+ if self.af == AF_INET:
+ return self._interface.local_ip4
+ elif self.af == AF_INET6:
+ return self._interface.local_ip6
+ else:
+ raise Exception("Unexpected af %s' % af" % self.af)
return self._local_addr
@property
def local_addr_n(self):
""" BFD session local address (VPP address) - raw, suitable for API """
if self._local_addr is None:
- return self._interface.local_ip4n
+ if self.af == AF_INET:
+ return self._interface.local_ip4n
+ elif self.af == AF_INET6:
+ return self._interface.local_ip6n
+ else:
+ raise Exception("Unexpected af %s' % af" % self.af)
return self._local_addr_n
@property
session = s
break
if session is None:
- raise Exception(
- "Could not find BFD session in VPP response: %s" % repr(result))
+ raise Exception("Could not find BFD session in VPP response: %s" %
+ repr(result))
return session.state
@property
def detect_mult(self):
return self._detect_mult
- def __init__(self, test, interface, peer_addr, local_addr=None, af=AF_INET):
+ def __init__(self, test, interface, peer_addr, local_addr=None, af=AF_INET,
+ desired_min_tx=100000, required_min_rx=100000, detect_mult=3):
self._test = test
self._interface = interface
self._af = af
self._peer_addr = peer_addr
self._peer_addr_n = socket.inet_pton(af, peer_addr)
self._bs_index = None
- self._desired_min_tx = 200000 # 0.2s
- self._required_min_rx = 200000 # 0.2s
- self._detect_mult = 3 # 3 packets need to be missed
+ self._desired_min_tx = desired_min_tx
+ self._required_min_rx = required_min_rx
+ self._detect_mult = detect_mult
def add_vpp_config(self):
is_ipv6 = 1 if AF_INET6 == self.af else 0
self.peer_addr_n,
is_ipv6=is_ipv6)
self._bs_index = result.bs_index
+ self._test.registry.register(self, self.test.logger)
def query_vpp_config(self):
result = self.test.vapi.bfd_udp_session_dump()
return True
def remove_vpp_config(self):
- if hasattr(self, '_bs_index'):
+ if self._bs_index is not None:
is_ipv6 = 1 if AF_INET6 == self._af else 0
self.test.vapi.bfd_udp_del(
self._interface.sw_if_index,
def object_id(self):
return "bfd-udp-%d" % self.bs_index
+ def __str__(self):
+ return self.object_id()
+
def admin_up(self):
self.test.vapi.bfd_session_set_flags(self.bs_index, 1)