Code Review
/
csit.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
review
|
tree
raw
|
inline
| side by side
Trending: new daily set
[csit.git]
/
resources
/
libraries
/
python
/
IPv6Util.py
diff --git
a/resources/libraries/python/IPv6Util.py
b/resources/libraries/python/IPv6Util.py
index
28e5f7d
..
683f892
100644
(file)
--- a/
resources/libraries/python/IPv6Util.py
+++ b/
resources/libraries/python/IPv6Util.py
@@
-15,11
+15,11
@@
from resources.libraries.python.InterfaceUtil import InterfaceUtil
from resources.libraries.python.IPUtil import IPUtil
from resources.libraries.python.InterfaceUtil import InterfaceUtil
from resources.libraries.python.IPUtil import IPUtil
-from resources.libraries.python.PapiExecutor import PapiExecutor
+from resources.libraries.python.PapiExecutor import Papi
Socket
Executor
from resources.libraries.python.topology import NodeType
from resources.libraries.python.topology import NodeType
-class IPv6Util
(object)
:
+class IPv6Util:
"""IPv6 utilities"""
@staticmethod
"""IPv6 utilities"""
@staticmethod
@@
-31,16
+31,16
@@
class IPv6Util(object):
:type node: dict
:type interface: str
"""
:type node: dict
:type interface: str
"""
- cmd =
'sw_interface_ip6nd_ra_config'
+ cmd =
u"sw_interface_ip6nd_ra_config"
args = dict(
sw_if_index=InterfaceUtil.get_interface_index(node, interface),
args = dict(
sw_if_index=InterfaceUtil.get_interface_index(node, interface),
- suppress=1)
- err_msg = 'Failed to suppress ICMPv6 router advertisement message on ' \
- 'interface {ifc}'.format(ifc=interface)
+ suppress=1
+ )
+ err_msg = f"Failed to suppress ICMPv6 router advertisement message " \
+ f"on interface {interface}"
- with PapiExecutor(node) as papi_exec:
- papi_exec.add(cmd, **args).get_replies(err_msg). \
- verify_reply(err_msg=err_msg)
+ with PapiSocketExecutor(node) as papi_exec:
+ papi_exec.add(cmd, **args).get_reply(err_msg)
@staticmethod
def vpp_ra_send_after_interval(node, interface, interval=2):
@staticmethod
def vpp_ra_send_after_interval(node, interface, interval=2):
@@
-54,16
+54,16
@@
class IPv6Util(object):
:type interface: str
:type interval: int
"""
:type interface: str
:type interval: int
"""
- cmd =
'sw_interface_ip6nd_ra_config'
+ cmd =
u"sw_interface_ip6nd_ra_config"
args = dict(
sw_if_index=InterfaceUtil.get_interface_index(node, interface),
args = dict(
sw_if_index=InterfaceUtil.get_interface_index(node, interface),
- initial_interval=int(interval))
- err_msg = 'Failed to set router advertisement interval on ' \
- 'interface {ifc}'.format(ifc=interface)
+ initial_interval=int(interval)
+ )
+ err_msg = f"Failed to set router advertisement interval " \
+ f"on interface {interface}"
- with PapiExecutor(node) as papi_exec:
- papi_exec.add(cmd, **args).get_replies(err_msg). \
- verify_reply(err_msg=err_msg)
+ with PapiSocketExecutor(node) as papi_exec:
+ papi_exec.add(cmd, **args).get_reply(err_msg)
@staticmethod
def vpp_all_ra_suppress_link_layer(nodes):
@staticmethod
def vpp_all_ra_suppress_link_layer(nodes):
@@
-74,10
+74,11
@@
class IPv6Util(object):
:type nodes: dict
"""
for node in nodes.values():
:type nodes: dict
"""
for node in nodes.values():
- if node[
'type'
] == NodeType.TG:
+ if node[
u"type"
] == NodeType.TG:
continue
continue
- for port_k in node[
'interfaces'
].keys():
+ for port_k in node[
u"interfaces"
].keys():
ip6_addr_list = IPUtil.vpp_get_interface_ip_addresses(
ip6_addr_list = IPUtil.vpp_get_interface_ip_addresses(
- node, port_k, 'ipv6')
+ node, port_k, u"ipv6"
+ )
if ip6_addr_list:
IPv6Util.vpp_ra_suppress_link_layer(node, port_k)
if ip6_addr_list:
IPv6Util.vpp_ra_suppress_link_layer(node, port_k)