aboutsummaryrefslogtreecommitdiffstats
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/bfd.py133
-rw-r--r--test/test_bfd.py72
-rw-r--r--test/vpp_papi_provider.py86
3 files changed, 97 insertions, 194 deletions
diff --git a/test/bfd.py b/test/bfd.py
index a297dab2527..e096284fbd7 100644
--- a/test/bfd.py
+++ b/test/bfd.py
@@ -9,6 +9,8 @@ from scapy.fields import BitField, BitEnumField, XByteField, FlagsField,\
ConditionalField, StrField
from vpp_object import VppObject
from util import NumericConstant
+from vpp_ip import VppIpAddress
+from vpp_papi import VppEnum
class BFDDiagCode(NumericConstant):
@@ -202,7 +204,8 @@ class VppBFDAuthKey(VppObject):
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):
@@ -217,7 +220,7 @@ class VppBFDAuthKey(VppObject):
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
@@ -232,13 +235,11 @@ class VppBFDUDPSession(VppObject):
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
@@ -274,29 +275,12 @@ class VppBFDUDPSession(VppObject):
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 """
@@ -305,14 +289,12 @@ class VppBFDUDPSession(VppObject):
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
@@ -357,28 +339,26 @@ class VppBFDUDPSession(VppObject):
""" 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,
@@ -391,28 +371,26 @@ class VppBFDUDPSession(VppObject):
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):
@@ -420,11 +398,9 @@ class VppBFDUDPSession(VppObject):
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,
@@ -434,18 +410,15 @@ class VppBFDUDPSession(VppObject):
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())
diff --git a/test/test_bfd.py b/test/test_bfd.py
index 9a37eb8fed7..d3616c7e241 100644
--- a/test/test_bfd.py
+++ b/test/test_bfd.py
@@ -25,9 +25,11 @@ from util import ppp
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
@@ -273,7 +275,8 @@ class BFDAPITestCase(VppTestCase):
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)
@@ -287,7 +290,7 @@ class BFDAPITestCase(VppTestCase):
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()
@@ -298,9 +301,9 @@ class BFDAPITestCase(VppTestCase):
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()
@@ -625,20 +628,11 @@ def verify_event(test, event, expected_state):
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)
@@ -726,7 +720,7 @@ class BFD4TestCase(VppTestCase):
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
@@ -1158,7 +1152,8 @@ class BFD4TestCase(VppTestCase):
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):
@@ -1203,7 +1198,8 @@ class BFD4TestCase(VppTestCase):
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):
@@ -1241,7 +1237,8 @@ class BFD4TestCase(VppTestCase):
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)
@@ -1273,7 +1270,8 @@ class BFD4TestCase(VppTestCase):
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)
@@ -1304,7 +1302,8 @@ class BFD4TestCase(VppTestCase):
""" 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
@@ -1357,7 +1356,8 @@ class BFD4TestCase(VppTestCase):
""" 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
@@ -1538,7 +1538,7 @@ class BFD6TestCase(VppTestCase):
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
@@ -1654,7 +1654,8 @@ class BFD6TestCase(VppTestCase):
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):
@@ -1736,7 +1737,7 @@ class BFDFIBTestCase(VppTestCase):
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()
@@ -1929,7 +1930,7 @@ class BFDSHA1TestCase(VppTestCase):
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()
@@ -2166,7 +2167,7 @@ class BFDAuthOnOffTestCase(VppTestCase):
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()
@@ -2376,7 +2377,7 @@ class BFDCLITestCase(VppTestCase):
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
@@ -2391,7 +2392,11 @@ class BFDCLITestCase(VppTestCase):
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")
@@ -2765,5 +2770,6 @@ class BFDCLITestCase(VppTestCase):
self.cli_verify_response("show bfd echo-source",
"UDP echo source is not set.")
+
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)
diff --git a/test/vpp_papi_provider.py b/test/vpp_papi_provider.py
index f9bb470ec1d..663f688e1db 100644
--- a/test/vpp_papi_provider.py
+++ b/test/vpp_papi_provider.py
@@ -27,6 +27,10 @@ defaultmapping = {
'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, },
@@ -272,7 +276,7 @@ class VppPapiProvider(object):
for i, o in enumerate(fields[3:]):
try:
d[o] = a[i]
- except:
+ except BaseException:
break
# Default override
@@ -762,86 +766,6 @@ class VppPapiProvider(object):
'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,