summaryrefslogtreecommitdiffstats
path: root/test
diff options
context:
space:
mode:
authorJuraj Sloboda <jsloboda@cisco.com>2018-02-01 15:18:49 +0100
committerOle Trøan <otroan@employees.org>2018-03-16 20:07:51 +0000
commitc03742346f0f3f08b83b0d2944bf72e19ea26191 (patch)
tree124cb6dc5cd45d070d2649bcf64916b27bf7bd01 /test
parentff92efe1074e338f91e59d9c27125d102516e7bf (diff)
IPv6 ND Router discovery control plane (VPP-1095)
Change-Id: I4b5b60e7c6f618bb935eab1e96a2e79bbb14f58f Signed-off-by: Juraj Sloboda <jsloboda@cisco.com>
Diffstat (limited to 'test')
-rw-r--r--test/test_ip6.py175
-rw-r--r--test/vpp_interface.py29
-rw-r--r--test/vpp_papi_provider.py7
3 files changed, 204 insertions, 7 deletions
diff --git a/test/test_ip6.py b/test/test_ip6.py
index 70ebc36dcc0..c9efcbbc51c 100644
--- a/test/test_ip6.py
+++ b/test/test_ip6.py
@@ -1005,6 +1005,181 @@ class TestIPv6RD(TestIPv6ND):
self.verify_prefix_info(ev.prefixes[1], prefix_info_2)
+class TestIPv6RDControlPlane(TestIPv6ND):
+ """ IPv6 Router Discovery Control Plane Test Case """
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPv6RDControlPlane, cls).setUpClass()
+
+ def setUp(self):
+ super(TestIPv6RDControlPlane, self).setUp()
+
+ # create 1 pg interface
+ self.create_pg_interfaces(range(1))
+
+ self.interfaces = list(self.pg_interfaces)
+
+ # setup all interfaces
+ for i in self.interfaces:
+ i.admin_up()
+ i.config_ip6()
+
+ def tearDown(self):
+ super(TestIPv6RDControlPlane, self).tearDown()
+
+ @staticmethod
+ def create_ra_packet(pg, routerlifetime=None):
+ src_ip = pg.remote_ip6_ll
+ dst_ip = pg.local_ip6
+ if routerlifetime is not None:
+ ra = ICMPv6ND_RA(routerlifetime=routerlifetime)
+ else:
+ ra = ICMPv6ND_RA()
+ p = (Ether(dst=pg.local_mac, src=pg.remote_mac) /
+ IPv6(dst=dst_ip, src=src_ip) / ra)
+ return p
+
+ @staticmethod
+ def get_default_routes(fib):
+ list = []
+ for entry in fib:
+ if entry.address_length == 0:
+ for path in entry.path:
+ if path.sw_if_index != 0xFFFFFFFF:
+ defaut_route = {}
+ defaut_route['sw_if_index'] = path.sw_if_index
+ defaut_route['next_hop'] = path.next_hop
+ list.append(defaut_route)
+ return list
+
+ @staticmethod
+ def get_interface_addresses(fib, pg):
+ list = []
+ for entry in fib:
+ if entry.address_length == 128:
+ path = entry.path[0]
+ if path.sw_if_index == pg.sw_if_index:
+ list.append(entry.address)
+ return list
+
+ def test_all(self):
+ """ Test handling of SLAAC addresses and default routes """
+
+ fib = self.vapi.ip6_fib_dump()
+ default_routes = self.get_default_routes(fib)
+ initial_addresses = set(self.get_interface_addresses(fib, self.pg0))
+ self.assertEqual(default_routes, [])
+ router_address = self.pg0.remote_ip6n_ll
+
+ self.vapi.ip6_nd_address_autoconfig(self.pg0.sw_if_index, 1, 1)
+
+ self.sleep(0.1)
+
+ # send RA
+ packet = (self.create_ra_packet(self.pg0) / ICMPv6NDOptPrefixInfo(
+ prefix="1::",
+ prefixlen=64,
+ validlifetime=2,
+ preferredlifetime=2,
+ L=1,
+ A=1,
+ ) / ICMPv6NDOptPrefixInfo(
+ prefix="7::",
+ prefixlen=20,
+ validlifetime=1500,
+ preferredlifetime=1000,
+ L=1,
+ A=0,
+ ))
+ self.pg0.add_stream([packet])
+ self.pg_start()
+
+ self.sleep(0.1)
+
+ fib = self.vapi.ip6_fib_dump()
+
+ # check FIB for new address
+ addresses = set(self.get_interface_addresses(fib, self.pg0))
+ new_addresses = addresses.difference(initial_addresses)
+ self.assertEqual(len(new_addresses), 1)
+ prefix = list(new_addresses)[0][:8] + '\0\0\0\0\0\0\0\0'
+ self.assertEqual(inet_ntop(AF_INET6, prefix), '1::')
+
+ # check FIB for new default route
+ default_routes = self.get_default_routes(fib)
+ self.assertEqual(len(default_routes), 1)
+ dr = default_routes[0]
+ self.assertEqual(dr['sw_if_index'], self.pg0.sw_if_index)
+ self.assertEqual(dr['next_hop'], router_address)
+
+ # send RA to delete default route
+ packet = self.create_ra_packet(self.pg0, routerlifetime=0)
+ self.pg0.add_stream([packet])
+ self.pg_start()
+
+ self.sleep(0.1)
+
+ # check that default route is deleted
+ fib = self.vapi.ip6_fib_dump()
+ default_routes = self.get_default_routes(fib)
+ self.assertEqual(len(default_routes), 0)
+
+ self.sleep(0.1)
+
+ # send RA
+ packet = self.create_ra_packet(self.pg0)
+ self.pg0.add_stream([packet])
+ self.pg_start()
+
+ self.sleep(0.1)
+
+ # check FIB for new default route
+ fib = self.vapi.ip6_fib_dump()
+ default_routes = self.get_default_routes(fib)
+ self.assertEqual(len(default_routes), 1)
+ dr = default_routes[0]
+ self.assertEqual(dr['sw_if_index'], self.pg0.sw_if_index)
+ self.assertEqual(dr['next_hop'], router_address)
+
+ # send RA, updating router lifetime to 1s
+ packet = self.create_ra_packet(self.pg0, 1)
+ self.pg0.add_stream([packet])
+ self.pg_start()
+
+ self.sleep(0.1)
+
+ # check that default route still exists
+ fib = self.vapi.ip6_fib_dump()
+ default_routes = self.get_default_routes(fib)
+ self.assertEqual(len(default_routes), 1)
+ dr = default_routes[0]
+ self.assertEqual(dr['sw_if_index'], self.pg0.sw_if_index)
+ self.assertEqual(dr['next_hop'], router_address)
+
+ self.sleep(1)
+
+ # check that default route is deleted
+ fib = self.vapi.ip6_fib_dump()
+ default_routes = self.get_default_routes(fib)
+ self.assertEqual(len(default_routes), 0)
+
+ # check FIB still contains the SLAAC address
+ addresses = set(self.get_interface_addresses(fib, self.pg0))
+ new_addresses = addresses.difference(initial_addresses)
+ self.assertEqual(len(new_addresses), 1)
+ prefix = list(new_addresses)[0][:8] + '\0\0\0\0\0\0\0\0'
+ self.assertEqual(inet_ntop(AF_INET6, prefix), '1::')
+
+ self.sleep(1)
+
+ # check that SLAAC address is deleted
+ fib = self.vapi.ip6_fib_dump()
+ addresses = set(self.get_interface_addresses(fib, self.pg0))
+ new_addresses = addresses.difference(initial_addresses)
+ self.assertEqual(len(new_addresses), 0)
+
+
class IPv6NDProxyTest(TestIPv6ND):
""" IPv6 ND ProxyTest Case """
diff --git a/test/vpp_interface.py b/test/vpp_interface.py
index 8891dde6afe..194a0c4fb32 100644
--- a/test/vpp_interface.py
+++ b/test/vpp_interface.py
@@ -55,6 +55,16 @@ class VppInterface(object):
return socket.inet_pton(socket.AF_INET6, self.local_ip6)
@property
+ def remote_ip6(self):
+ """IPv6 address of remote peer "connected" to this interface."""
+ return self._remote_hosts[0].ip6
+
+ @property
+ def remote_ip6n(self):
+ """IPv6 address of remote peer - raw, suitable as API parameter"""
+ return socket.inet_pton(socket.AF_INET6, self.remote_ip6)
+
+ @property
def local_ip6_ll(self):
"""Local IPv6 linnk-local address on VPP interface (string)."""
return self._local_ip6_ll
@@ -62,17 +72,19 @@ class VppInterface(object):
@property
def local_ip6n_ll(self):
"""Local IPv6 link-local address - raw, suitable as API parameter."""
- return self.local_ip6n_ll
+ return self._local_ip6n_ll
@property
- def remote_ip6(self):
- """IPv6 address of remote peer "connected" to this interface."""
- return self._remote_hosts[0].ip6
+ def remote_ip6_ll(self):
+ """Link-local IPv6 address of remote peer
+ "connected" to this interface."""
+ return self._remote_ip6_ll
@property
- def remote_ip6n(self):
- """IPv6 address of remote peer - raw, suitable as API parameter"""
- return socket.inet_pton(socket.AF_INET6, self.remote_ip6)
+ def remote_ip6n_ll(self):
+ """Link-local IPv6 address of remote peer
+ - raw, suitable as API parameter"""
+ return self._remote_ip6n_ll
@property
def name(self):
@@ -196,6 +208,9 @@ class VppInterface(object):
self._local_ip6_ll = mk_ll_addr(self.local_mac)
self._local_ip6n_ll = socket.inet_pton(socket.AF_INET6,
self.local_ip6_ll)
+ self._remote_ip6_ll = mk_ll_addr(self.remote_mac)
+ self._remote_ip6n_ll = socket.inet_pton(socket.AF_INET6,
+ self.remote_ip6_ll)
def config_ip4(self):
"""Configure IPv4 address on the VPP interface."""
diff --git a/test/vpp_papi_provider.py b/test/vpp_papi_provider.py
index 8fc0bfdae85..bd770efdd36 100644
--- a/test/vpp_papi_provider.py
+++ b/test/vpp_papi_provider.py
@@ -480,6 +480,13 @@ class VppPapiProvider(object):
'mrd': mrd,
'sw_if_index': sw_if_index})
+ def ip6_nd_address_autoconfig(self, sw_if_index, enable,
+ install_default_routes):
+ return self.api(self.papi.ip6_nd_address_autoconfig,
+ {'sw_if_index': sw_if_index,
+ 'enable': enable,
+ 'install_default_routes': install_default_routes})
+
def want_macs_learn_events(self, enable_disable=1, scan_delay=0,
max_macs_in_event=0, learn_limit=0):
return self.api(self.papi.want_l2_macs_events,