Use consistent API types.
Type: fix
Signed-off-by: Jakub Grajciar <jgrajcia@cisco.com>
Change-Id: I4b91ebe8aadbec8b95716ed1cde4b0a13eee3a3f
Signed-off-by: Jakub Grajciar <jgrajcia@cisco.com>
* limitations under the License.
*/
-option version = "1.0.0";
+option version = "2.0.0";
+
+import "vnet/interface_types.api";
+import "vnet/ip/ip_types.api";
+
+enum bfd_state {
+ BFD_STATE_API_ADMIN_DOWN = 0,
+ BFD_STATE_API_DOWN = 1,
+ BFD_STATE_API_INIT = 2,
+ BFD_STATE_API_UP = 3,
+};
/** \brief Set BFD echo source
@param client_index - opaque cookie to identify the sender
{
u32 client_index;
u32 context;
- u32 sw_if_index;
+ vl_api_interface_index_t sw_if_index;
};
/** \brief Delete BFD echo source
{
u32 context;
i32 retval;
- u32 sw_if_index;
- u8 is_set;
- u8 have_usable_ip4;
- u8 ip4_addr[4];
- u8 have_usable_ip6;
- u8 ip6_addr[16];
+ vl_api_interface_index_t sw_if_index;
+ bool is_set;
+ bool have_usable_ip4;
+ vl_api_ip4_address_t ip4_addr;
+ bool have_usable_ip6;
+ vl_api_ip6_address_t ip6_addr;
};
/** \brief Add UDP BFD session on interface
{
u32 client_index;
u32 context;
- u32 sw_if_index;
+ vl_api_interface_index_t sw_if_index;
u32 desired_min_tx;
u32 required_min_rx;
- u8 local_addr[16];
- u8 peer_addr[16];
- u8 is_ipv6;
+ vl_api_address_t local_addr;
+ vl_api_address_t peer_addr;
u8 detect_mult;
- u8 is_authenticated;
+ bool is_authenticated;
u8 bfd_key_id;
u32 conf_key_id;
};
{
u32 client_index;
u32 context;
- u32 sw_if_index;
+ vl_api_interface_index_t sw_if_index;
u32 desired_min_tx;
u32 required_min_rx;
- u8 local_addr[16];
- u8 peer_addr[16];
- u8 is_ipv6;
+ vl_api_address_t local_addr;
+ vl_api_address_t peer_addr;
u8 detect_mult;
};
{
u32 client_index;
u32 context;
- u32 sw_if_index;
- u8 local_addr[16];
- u8 peer_addr[16];
- u8 is_ipv6;
+ vl_api_interface_index_t sw_if_index;
+ vl_api_address_t local_addr;
+ vl_api_address_t peer_addr;
};
/** \brief Get all BFD sessions
define bfd_udp_session_details
{
u32 context;
- u32 sw_if_index;
- u8 local_addr[16];
- u8 peer_addr[16];
- u8 is_ipv6;
- u8 state;
- u8 is_authenticated;
+ vl_api_interface_index_t sw_if_index;
+ vl_api_address_t local_addr;
+ vl_api_address_t peer_addr;
+ vl_api_bfd_state_t state;
+ bool is_authenticated;
u8 bfd_key_id;
u32 conf_key_id;
u32 required_min_rx;
@param local_addr - local address
@param peer_addr - peer address
@param is_ipv6 - local_addr, peer_addr are IPv6 if non-zero, otherwise IPv4
- @param admin_up_down - set the admin state, 1 = up, 0 = down
+ @param flags - set the admin state, 1 = up, 0 = down
*/
autoreply define bfd_udp_session_set_flags
{
u32 client_index;
u32 context;
- u32 sw_if_index;
- u8 local_addr[16];
- u8 peer_addr[16];
- u8 is_ipv6;
- u8 admin_up_down;
+ vl_api_interface_index_t sw_if_index;
+ vl_api_address_t local_addr;
+ vl_api_address_t peer_addr;
+ vl_api_if_status_flags_t flags;
};
/** \brief Register for BFD events
{
u32 client_index;
u32 context;
- u32 enable_disable;
+ bool enable_disable;
u32 pid;
};
{
u32 client_index;
u32 context;
- u32 sw_if_index;
- u8 local_addr[16];
- u8 peer_addr[16];
- u8 is_ipv6;
- u8 is_delayed;
+ vl_api_interface_index_t sw_if_index;
+ vl_api_address_t local_addr;
+ vl_api_address_t peer_addr;
+ bool is_delayed;
u8 bfd_key_id;
u32 conf_key_id;
};
{
u32 client_index;
u32 context;
- u32 sw_if_index;
- u8 local_addr[16];
- u8 peer_addr[16];
- u8 is_ipv6;
- u8 is_delayed;
+ vl_api_interface_index_t sw_if_index;
+ vl_api_address_t local_addr;
+ vl_api_address_t peer_addr;
+ bool is_delayed;
};
/*
#include <vnet/api_errno.h>
#include <vnet/bfd/bfd_main.h>
#include <vnet/bfd/bfd_api.h>
+#include <vnet/ip/ip_types_api.h>
#include <vnet/vnet_msg_enum.h>
#define BFD_UDP_API_PARAM_COMMON_CODE \
ip46_address_t local_addr; \
- clib_memset (&local_addr, 0, sizeof (local_addr)); \
ip46_address_t peer_addr; \
- clib_memset (&peer_addr, 0, sizeof (peer_addr)); \
- if (mp->is_ipv6) \
- { \
- clib_memcpy_fast (&local_addr.ip6, mp->local_addr, sizeof (local_addr.ip6)); \
- clib_memcpy_fast (&peer_addr.ip6, mp->peer_addr, sizeof (peer_addr.ip6)); \
- } \
- else \
- { \
- clib_memcpy_fast (&local_addr.ip4, mp->local_addr, sizeof (local_addr.ip4)); \
- clib_memcpy_fast (&peer_addr.ip4, mp->peer_addr, sizeof (peer_addr.ip4)); \
- }
+ ip_address_decode(&mp->local_addr, &local_addr); \
+ ip_address_decode(&mp->peer_addr, &peer_addr);
#define BFD_UDP_API_PARAM_FROM_MP(mp) \
clib_net_to_host_u32 (mp->sw_if_index), &local_addr, &peer_addr
clib_memset (mp, 0, sizeof (*mp));
mp->_vl_msg_id = ntohs (VL_API_BFD_UDP_SESSION_DETAILS);
mp->context = context;
- mp->state = bs->local_state;
+ mp->state = clib_host_to_net_u32 (bs->local_state);
bfd_udp_session_t *bus = &bs->udp;
bfd_udp_key_t *key = &bus->key;
mp->sw_if_index = clib_host_to_net_u32 (key->sw_if_index);
- mp->is_ipv6 = !(ip46_address_is_ip4 (&key->local_addr));
if ((!bs->auth.is_delayed && bs->auth.curr_key) ||
(bs->auth.is_delayed && bs->auth.next_key))
{
- mp->is_authenticated = 1;
+ mp->is_authenticated = true;
}
if (bs->auth.is_delayed && bs->auth.next_key)
{
mp->bfd_key_id = bs->auth.curr_bfd_key_id;
mp->conf_key_id = clib_host_to_net_u32 (bs->auth.curr_key->conf_key_id);
}
- if (mp->is_ipv6)
- {
- clib_memcpy_fast (mp->local_addr, &key->local_addr,
- sizeof (key->local_addr));
- clib_memcpy_fast (mp->peer_addr, &key->peer_addr,
- sizeof (key->peer_addr));
- }
- else
- {
- clib_memcpy_fast (mp->local_addr, key->local_addr.ip4.data,
- sizeof (key->local_addr.ip4.data));
- clib_memcpy_fast (mp->peer_addr, key->peer_addr.ip4.data,
- sizeof (key->peer_addr.ip4.data));
- }
+ ip_address_encode (&key->local_addr, IP46_TYPE_ANY, &mp->local_addr);
+ ip_address_encode (&key->peer_addr, IP46_TYPE_ANY, &mp->peer_addr);
mp->required_min_rx =
clib_host_to_net_u32 (bs->config_required_min_rx_usec);
BFD_UDP_API_PARAM_COMMON_CODE;
rv = bfd_udp_session_set_flags (BFD_UDP_API_PARAM_FROM_MP (mp),
- mp->admin_up_down);
+ clib_net_to_host_u32 (mp->flags) &
+ IF_STATUS_API_FLAG_ADMIN_UP);
REPLY_MACRO (VL_API_BFD_UDP_SESSION_SET_FLAGS_REPLY);
}
rmp->sw_if_index = ntohl (sw_if_index);
if (is_set)
{
- rmp->is_set = 1;
+ rmp->is_set = true;
rmp->sw_if_index = clib_host_to_net_u32 (sw_if_index);
if (have_usable_ip4)
{
- rmp->have_usable_ip4 = 1;
- clib_memcpy_fast (rmp->ip4_addr, &ip4, sizeof (ip4));
+ rmp->have_usable_ip4 = true;
+ ip4_address_encode(&ip4, rmp->ip4_addr);
}
else
{
- rmp->have_usable_ip4 = 0;
+ rmp->have_usable_ip4 = false;
}
if (have_usable_ip6)
{
- rmp->have_usable_ip6 = 1;
- clib_memcpy_fast (rmp->ip6_addr, &ip6, sizeof (ip6));
+ rmp->have_usable_ip6 = true;
+ ip6_address_encode(&ip6, rmp->ip6_addr);
}
else
{
- rmp->have_usable_ip6 = 0;
+ rmp->have_usable_ip6 = false;
}
}
else
{
- rmp->is_set = 0;
- rmp->have_usable_ip4 = 0;
- rmp->have_usable_ip6 = 0;
+ rmp->is_set = false;
+ rmp->have_usable_ip4 = false;
+ rmp->have_usable_ip6 = false;
}
}))
/* *INDENT-ON* */
ConditionalField, StrField
from vpp_object import VppObject
from util import NumericConstant
+from vpp_ip import VppIpAddress
+from vpp_papi import VppEnum
class BFDDiagCode(NumericConstant):
def add_vpp_config(self):
self.test.vapi.bfd_auth_set_key(
- self._conf_key_id, self._auth_type, self._key)
+ conf_key_id=self._conf_key_id, auth_type=self._auth_type,
+ key=self._key, key_len=len(self._key))
self._test.registry.register(self, self.test.logger)
def get_bfd_auth_keys_dump_entry(self):
return self.get_bfd_auth_keys_dump_entry() is not None
def remove_vpp_config(self):
- self.test.vapi.bfd_auth_del_key(self._conf_key_id)
+ self.test.vapi.bfd_auth_del_key(conf_key_id=self._conf_key_id)
def object_id(self):
return "bfd-auth-key-%s" % self._conf_key_id
self._test = test
self._interface = interface
self._af = af
- self._local_addr = local_addr
- if local_addr is not None:
- self._local_addr_n = inet_pton(af, local_addr)
+ if local_addr:
+ self._local_addr = VppIpAddress(local_addr)
else:
- self._local_addr_n = None
- self._peer_addr = peer_addr
- self._peer_addr_n = inet_pton(af, peer_addr)
+ self._local_addr = None
+ self._peer_addr = VppIpAddress(peer_addr)
self._desired_min_tx = desired_min_tx
self._required_min_rx = required_min_rx
self._detect_mult = detect_mult
return self._interface.local_ip6
else:
raise Exception("Unexpected af '%s'" % 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:
- 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'" % self.af)
- return self._local_addr_n
+ return self._local_addr.address
@property
def peer_addr(self):
""" BFD session peer address """
- return self._peer_addr
-
- @property
- def peer_addr_n(self):
- """ BFD session peer address - raw, suitable for API """
- return self._peer_addr_n
+ return self._peer_addr.address
def get_bfd_udp_session_dump_entry(self):
""" get the namedtuple entry from bfd udp session dump """
self.test.logger.debug("session entry: %s" % str(s))
if s.sw_if_index == self.interface.sw_if_index:
if self.af == AF_INET \
- and s.is_ipv6 == 0 \
- and self.interface.local_ip4n == s.local_addr[:4] \
- and self.interface.remote_ip4n == s.peer_addr[:4]:
+ and self.interface.local_ip4 == str(s.local_addr) \
+ and self.interface.remote_ip4 == str(s.peer_addr):
return s
if self.af == AF_INET6 \
- and s.is_ipv6 == 1 \
- and self.interface.local_ip6n == s.local_addr \
- and self.interface.remote_ip6n == s.peer_addr:
+ and self.interface.local_ip6 == str(s.local_addr) \
+ and self.interface.remote_ip6 == str(s.peer_addr):
return s
return None
""" activate authentication for this session """
self._bfd_key_id = bfd_key_id if bfd_key_id else randint(0, 255)
self._sha1_key = key
- is_ipv6 = 1 if AF_INET6 == self.af else 0
conf_key_id = self._sha1_key.conf_key_id
is_delayed = 1 if delayed else 0
- self.test.vapi.bfd_udp_auth_activate(self._interface.sw_if_index,
- self.local_addr_n,
- self.peer_addr_n,
- is_ipv6=is_ipv6,
- bfd_key_id=self._bfd_key_id,
- conf_key_id=conf_key_id,
- is_delayed=is_delayed)
+ self.test.vapi.bfd_udp_auth_activate(
+ sw_if_index=self._interface.sw_if_index,
+ local_addr=self.local_addr.encode(),
+ peer_addr=self.peer_addr.encode(),
+ bfd_key_id=self._bfd_key_id,
+ conf_key_id=conf_key_id,
+ is_delayed=is_delayed)
def deactivate_auth(self, delayed=False):
""" deactivate authentication """
self._bfd_key_id = None
self._sha1_key = None
is_delayed = 1 if delayed else 0
- is_ipv6 = 1 if AF_INET6 == self.af else 0
- self.test.vapi.bfd_udp_auth_deactivate(self._interface.sw_if_index,
- self.local_addr_n,
- self.peer_addr_n,
- is_ipv6=is_ipv6,
- is_delayed=is_delayed)
+ self.test.vapi.bfd_udp_auth_deactivate(
+ sw_if_index=self._interface.sw_if_index,
+ local_addr=self.local_addr.encode(),
+ peer_addr=self.peer_addr.encode(),
+ is_delayed=is_delayed)
def modify_parameters(self,
detect_mult=None,
self._desired_min_tx = desired_min_tx
if required_min_rx:
self._required_min_rx = required_min_rx
- is_ipv6 = 1 if AF_INET6 == self.af else 0
- self.test.vapi.bfd_udp_mod(self._interface.sw_if_index,
- self.desired_min_tx,
- self.required_min_rx,
- self.detect_mult,
- self.local_addr_n,
- self.peer_addr_n,
- is_ipv6=is_ipv6)
+ self.test.vapi.bfd_udp_mod(sw_if_index=self._interface.sw_if_index,
+ desired_min_tx=self.desired_min_tx,
+ required_min_rx=self.required_min_rx,
+ detect_mult=self.detect_mult,
+ local_addr=self.local_addr.encode(),
+ peer_addr=self.peer_addr.encode())
def add_vpp_config(self):
- is_ipv6 = 1 if AF_INET6 == self.af else 0
bfd_key_id = self._bfd_key_id if self._sha1_key else None
conf_key_id = self._sha1_key.conf_key_id if self._sha1_key else None
- self.test.vapi.bfd_udp_add(self._interface.sw_if_index,
- self.desired_min_tx,
- self.required_min_rx,
- self.detect_mult,
- self.local_addr_n,
- self.peer_addr_n,
- is_ipv6=is_ipv6,
+ is_authenticated = True if self._sha1_key else False
+ self.test.vapi.bfd_udp_add(sw_if_index=self._interface.sw_if_index,
+ desired_min_tx=self.desired_min_tx,
+ required_min_rx=self.required_min_rx,
+ detect_mult=self.detect_mult,
+ local_addr=self.local_addr.encode(),
+ peer_addr=self.peer_addr.encode(),
bfd_key_id=bfd_key_id,
- conf_key_id=conf_key_id)
+ conf_key_id=conf_key_id,
+ is_authenticated=is_authenticated)
self._test.registry.register(self, self.test.logger)
def query_vpp_config(self):
return session is not None
def remove_vpp_config(self):
- is_ipv6 = 1 if AF_INET6 == self._af else 0
self.test.vapi.bfd_udp_del(self._interface.sw_if_index,
- self.local_addr_n,
- self.peer_addr_n,
- is_ipv6=is_ipv6)
+ local_addr=self.local_addr.encode(),
+ peer_addr=self.peer_addr.encode())
def object_id(self):
return "bfd-udp-%s-%s-%s-%s" % (self._interface.sw_if_index,
def admin_up(self):
""" set bfd session admin-up """
- is_ipv6 = 1 if AF_INET6 == self._af else 0
- self.test.vapi.bfd_udp_session_set_flags(1,
- self._interface.sw_if_index,
- self.local_addr_n,
- self.peer_addr_n,
- is_ipv6=is_ipv6)
+ self.test.vapi.bfd_udp_session_set_flags(
+ flags=VppEnum.vl_api_if_status_flags_t.IF_STATUS_API_FLAG_ADMIN_UP,
+ sw_if_index=self._interface.sw_if_index,
+ local_addr=self.local_addr.encode(),
+ peer_addr=self.peer_addr.encode())
def admin_down(self):
""" set bfd session admin-down """
- is_ipv6 = 1 if AF_INET6 == self._af else 0
- self.test.vapi.bfd_udp_session_set_flags(0,
- self._interface.sw_if_index,
- self.local_addr_n,
- self.peer_addr_n,
- is_ipv6=is_ipv6)
+ self.test.vapi.bfd_udp_session_set_flags(
+ flags=0, sw_if_index=self._interface.sw_if_index,
+ local_addr=self.local_addr.encode(),
+ peer_addr=self.peer_addr.encode())
from vpp_ip import DpoProto
from vpp_ip_route import VppIpRoute, VppRoutePath
from vpp_lo_interface import VppLoInterface
-from vpp_papi_provider import UnexpectedApiReturnValueError
+from vpp_papi_provider import UnexpectedApiReturnValueError, \
+ CliFailedCommandError
from vpp_pg_interface import CaptureTimeoutError, is_ipv6_misc
from vpp_gre_interface import VppGreInterface
+from vpp_papi import VppEnum
USEC_IN_SEC = 1000000
self.assertFalse(echo_source.have_usable_ip4)
self.assertFalse(echo_source.have_usable_ip6)
- self.vapi.bfd_udp_set_echo_source(self.loopback0.sw_if_index)
+ self.vapi.bfd_udp_set_echo_source(
+ sw_if_index=self.loopback0.sw_if_index)
echo_source = self.vapi.bfd_udp_get_echo_source()
self.assertTrue(echo_source.is_set)
self.assertEqual(echo_source.sw_if_index, self.loopback0.sw_if_index)
self.assertTrue(echo_source.is_set)
self.assertEqual(echo_source.sw_if_index, self.loopback0.sw_if_index)
self.assertTrue(echo_source.have_usable_ip4)
- self.assertEqual(echo_source.ip4_addr, echo_ip4)
+ self.assertEqual(echo_source.ip4_addr.packed, echo_ip4)
self.assertFalse(echo_source.have_usable_ip6)
self.loopback0.config_ip6()
self.assertTrue(echo_source.is_set)
self.assertEqual(echo_source.sw_if_index, self.loopback0.sw_if_index)
self.assertTrue(echo_source.have_usable_ip4)
- self.assertEqual(echo_source.ip4_addr, echo_ip4)
+ self.assertEqual(echo_source.ip4_addr.packed, echo_ip4)
self.assertTrue(echo_source.have_usable_ip6)
- self.assertEqual(echo_source.ip6_addr, echo_ip6)
+ self.assertEqual(echo_source.ip6_addr.packed, echo_ip6)
self.vapi.bfd_udp_del_echo_source()
echo_source = self.vapi.bfd_udp_get_echo_source()
test.assert_equal(e.sw_if_index,
test.vpp_session.interface.sw_if_index,
"BFD interface index")
- is_ipv6 = 0
- if test.vpp_session.af == AF_INET6:
- is_ipv6 = 1
- test.assert_equal(e.is_ipv6, is_ipv6, "is_ipv6")
- if test.vpp_session.af == AF_INET:
- test.assert_equal(e.local_addr[:4], test.vpp_session.local_addr_n,
- "Local IPv4 address")
- test.assert_equal(e.peer_addr[:4], test.vpp_session.peer_addr_n,
- "Peer IPv4 address")
- else:
- test.assert_equal(e.local_addr, test.vpp_session.local_addr_n,
- "Local IPv6 address")
- test.assert_equal(e.peer_addr, test.vpp_session.peer_addr_n,
- "Peer IPv6 address")
+
+ test.assert_equal(str(e.local_addr), test.vpp_session.local_addr,
+ "Local IPv6 address")
+ test.assert_equal(str(e.peer_addr), test.vpp_session.peer_addr,
+ "Peer IPv6 address")
test.assert_equal(e.state, expected_state, BFDState)
self.vpp_session.add_vpp_config()
self.vpp_session.admin_up()
self.test_session = BFDTestSession(self, self.pg0, AF_INET)
- except:
+ except BaseException:
self.vapi.want_bfd_events(enable_disable=0)
raise
self.vpp_session.required_min_rx,
"BFD required min rx interval")
self.test_session.send_packet()
- self.vapi.bfd_udp_set_echo_source(self.loopback0.sw_if_index)
+ self.vapi.bfd_udp_set_echo_source(
+ sw_if_index=self.loopback0.sw_if_index)
echo_seen = False
# should be turned on - loopback echo packets
for dummy in range(3):
self.test_session.send_packet()
detection_time = self.test_session.detect_mult *\
self.vpp_session.required_min_rx / USEC_IN_SEC
- self.vapi.bfd_udp_set_echo_source(self.loopback0.sw_if_index)
+ self.vapi.bfd_udp_set_echo_source(
+ sw_if_index=self.loopback0.sw_if_index)
# echo function should be used now, but we will drop the echo packets
verified_diag = False
for dummy in range(3):
bfd_session_up(self)
self.test_session.update(required_min_echo_rx=150000)
self.test_session.send_packet()
- self.vapi.bfd_udp_set_echo_source(self.loopback0.sw_if_index)
+ self.vapi.bfd_udp_set_echo_source(
+ sw_if_index=self.loopback0.sw_if_index)
# wait for first echo packet
while True:
p = self.pg0.wait_for_packet(1)
bfd_session_up(self)
self.test_session.update(required_min_echo_rx=150000)
self.test_session.send_packet()
- self.vapi.bfd_udp_set_echo_source(self.loopback0.sw_if_index)
+ self.vapi.bfd_udp_set_echo_source(
+ sw_if_index=self.loopback0.sw_if_index)
# wait for first echo packet
while True:
p = self.pg0.wait_for_packet(1)
""" stale echo packets don't keep a session up """
bfd_session_up(self)
self.test_session.update(required_min_echo_rx=150000)
- self.vapi.bfd_udp_set_echo_source(self.loopback0.sw_if_index)
+ self.vapi.bfd_udp_set_echo_source(
+ sw_if_index=self.loopback0.sw_if_index)
self.test_session.send_packet()
# should be turned on - loopback echo packets
echo_packet = None
""" echo packets with invalid checksum don't keep a session up """
bfd_session_up(self)
self.test_session.update(required_min_echo_rx=150000)
- self.vapi.bfd_udp_set_echo_source(self.loopback0.sw_if_index)
+ self.vapi.bfd_udp_set_echo_source(
+ sw_if_index=self.loopback0.sw_if_index)
self.test_session.send_packet()
# should be turned on - loopback echo packets
timeout_at = None
self.vpp_session.admin_up()
self.test_session = BFDTestSession(self, self.pg0, AF_INET6)
self.logger.debug(self.vapi.cli("show adj nbr"))
- except:
+ except BaseException:
self.vapi.want_bfd_events(enable_disable=0)
raise
self.vpp_session.required_min_rx,
"BFD required min rx interval")
self.test_session.send_packet()
- self.vapi.bfd_udp_set_echo_source(self.loopback0.sw_if_index)
+ self.vapi.bfd_udp_set_echo_source(
+ sw_if_index=self.loopback0.sw_if_index)
echo_seen = False
# should be turned on - loopback echo packets
for dummy in range(3):
def tearDown(self):
if not self.vpp_dead:
- self.vapi.want_bfd_events(enable_disable=0)
+ self.vapi.want_bfd_events(enable_disable=False)
super(BFDFIBTestCase, self).tearDown()
def tearDown(self):
if not self.vpp_dead:
- self.vapi.want_bfd_events(enable_disable=0)
+ self.vapi.want_bfd_events(enable_disable=False)
self.vapi.collect_events() # clear the event queue
super(BFDSHA1TestCase, self).tearDown()
def tearDown(self):
if not self.vpp_dead:
- self.vapi.want_bfd_events(enable_disable=0)
+ self.vapi.want_bfd_events(enable_disable=False)
self.vapi.collect_events() # clear the event queue
super(BFDAuthOnOffTestCase, self).tearDown()
def tearDown(self):
try:
- self.vapi.want_bfd_events(enable_disable=0)
+ self.vapi.want_bfd_events(enable_disable=False)
except UnexpectedApiReturnValueError:
# some tests aren't subscribed, so this is not an issue
pass
def cli_verify_response(self, cli, expected):
""" execute a CLI, asserting that the response matches expectation """
- self.assert_equal(self.vapi.cli(cli).strip(),
+ try:
+ reply = self.vapi.cli(cli)
+ except CliFailedCommandError as cli_error:
+ reply = str(cli_error)
+ self.assert_equal(reply.strip(),
expected,
"CLI command response")
self.cli_verify_response("show bfd echo-source",
"UDP echo source is not set.")
+
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)
'acl_interface_list_dump': {'sw_if_index': 4294967295, },
'app_namespace_add_del': {'sw_if_index': 4294967295, },
'bd_ip_mac_add_del': {'is_add': 1, },
+ 'bfd_udp_add': {'is_authenticated': False, 'bfd_key_id': None,
+ 'conf_key_id': None},
+ 'bfd_udp_auth_activate': {'bfd_key_id': None, 'conf_key_id': None,
+ 'is_delayed': False},
'bier_disp_entry_add_del': {'next_hop_rpf_id': -1, 'next_hop_is_ip4': 1,
'is_add': 1, },
'bier_disp_table_add_del': {'is_add': 1, },
for i, o in enumerate(fields[3:]):
try:
d[o] = a[i]
- except:
+ except BaseException:
break
# Default override
'mt_paths': paths,
}})
- def bfd_udp_add(self, sw_if_index, desired_min_tx, required_min_rx,
- detect_mult, local_addr, peer_addr, is_ipv6=0,
- bfd_key_id=None, conf_key_id=None):
- if bfd_key_id is None:
- return self.api(self.papi.bfd_udp_add,
- {
- 'sw_if_index': sw_if_index,
- 'desired_min_tx': desired_min_tx,
- 'required_min_rx': required_min_rx,
- 'local_addr': local_addr,
- 'peer_addr': peer_addr,
- 'is_ipv6': is_ipv6,
- 'detect_mult': detect_mult,
- })
- else:
- return self.api(self.papi.bfd_udp_add,
- {
- 'sw_if_index': sw_if_index,
- 'desired_min_tx': desired_min_tx,
- 'required_min_rx': required_min_rx,
- 'local_addr': local_addr,
- 'peer_addr': peer_addr,
- 'is_ipv6': is_ipv6,
- 'detect_mult': detect_mult,
- 'is_authenticated': 1,
- 'bfd_key_id': bfd_key_id,
- 'conf_key_id': conf_key_id,
- })
-
- def bfd_udp_mod(self, sw_if_index, desired_min_tx, required_min_rx,
- detect_mult, local_addr, peer_addr, is_ipv6=0):
- return self.api(self.papi.bfd_udp_mod,
- {
- 'sw_if_index': sw_if_index,
- 'desired_min_tx': desired_min_tx,
- 'required_min_rx': required_min_rx,
- 'local_addr': local_addr,
- 'peer_addr': peer_addr,
- 'is_ipv6': is_ipv6,
- 'detect_mult': detect_mult,
- })
-
- def bfd_udp_auth_activate(self, sw_if_index, local_addr, peer_addr,
- is_ipv6=0, bfd_key_id=None, conf_key_id=None,
- is_delayed=False):
- return self.api(self.papi.bfd_udp_auth_activate,
- {
- 'sw_if_index': sw_if_index,
- 'local_addr': local_addr,
- 'peer_addr': peer_addr,
- 'is_ipv6': is_ipv6,
- 'is_delayed': 1 if is_delayed else 0,
- 'bfd_key_id': bfd_key_id,
- 'conf_key_id': conf_key_id,
- })
-
- def bfd_udp_session_set_flags(self, admin_up_down, sw_if_index, local_addr,
- peer_addr, is_ipv6=0):
- return self.api(self.papi.bfd_udp_session_set_flags, {
- 'admin_up_down': admin_up_down,
- 'sw_if_index': sw_if_index,
- 'local_addr': local_addr,
- 'peer_addr': peer_addr,
- 'is_ipv6': is_ipv6,
- })
-
- def want_bfd_events(self, enable_disable=1):
- return self.api(self.papi.want_bfd_events, {
- 'enable_disable': enable_disable,
- 'pid': os.getpid(),
- })
-
- def bfd_auth_set_key(self, conf_key_id, auth_type, key):
- return self.api(self.papi.bfd_auth_set_key, {
- 'conf_key_id': conf_key_id,
- 'auth_type': auth_type,
- 'key': key,
- 'key_len': len(key),
- })
-
def classify_add_del_table(
self,
is_add,
Code Review / vpp.git / commitdiff