+ SchedUtils.py: Do not use `len(SEQUENCE)` to determine if a sequence is empty
+ VatHistory.py: Do not use `len(SEQUENCE)` to determine if a sequence is empty
+ VppCounters.py: Do not use `len(SEQUENCE)` to determine if a sequence is empty
+ Memif.py:
++ Do not use `len(SEQUENCE)` to determine if a sequence is empty
++ Either all return statements in a function should return an expression,
or none of them should.
++ Update :return: on possible None.
+ Classify.py: Unnecessary "else" after "return"
+ ContainerUtils.py: Useless super delegation in method '__init__'
+ CpuUtils.py: Do not use `len(SEQUENCE)` to determine if a sequence is empty
+ DropRateSearch.py: Either all return statements in a function
should return an expression, or none of them should.
+ IPv4NodeAddress.py: Do not use `len(SEQUENCE)` to determine if a sequence is empty
++ Also improve docstrings.
+ IPv4Setup.py: Useless super delegation in method '__init__'
+ IPv6Setup.py: Do not use `len(SEQUENCE)` to determine if a sequence is empty
++ Also improve docstrings.
+ IPv6Setup.py: standard import "from ipaddress import IPv6Network"
should be placed before "from robot.api import logger"
+ MacSwap.py: Trailing newlines
+ NATUtil.py: Do not use `len(SEQUENCE)` to determine if a sequence is empty
+ NodePath.py: Unnecessary "else" after "return"
+ Tap.py: Do not use `len(SEQUENCE)` to determine if a sequence is empty
+ topology.py: Either all return statements in a function
should return an expression, or none of them should.
+ topology.py: Unnecessary "else" after "return"
++ Do not use `len(SEQUENCE)` to determine if a sequence is empty
++ Improve docstrings
+ DUTSetup.py: Do not use `len(SEQUENCE)` to determine if a sequence is empty
++ Also do not compare int(ret_code) just to access zero-ness.
+ ssh.py: Do not use `len(SEQUENCE)` to determine if a sequence is empty
+ InterfaceUtil.py: Unnecessary "else" after "return"
Change-Id: Iba4244aa79661ee7df15fed5c7c6dbf04dfa88b2
Signed-off-by: Vratko Polak <vrpolak@cisco.com>
19 files changed:
)
if session_index is not None:
return data[0][session_index]
)
if session_index is not None:
return data[0][session_index]
@staticmethod
def vpp_log_plugin_acl_settings(node):
@staticmethod
def vpp_log_plugin_acl_settings(node):
class LXC(ContainerEngine):
"""LXC implementation."""
class LXC(ContainerEngine):
"""LXC implementation."""
- def __init__(self):
- """Initialize LXC object."""
- super(LXC, self).__init__()
+ # Implicit constructor is inherited.
def acquire(self, force=True):
"""Acquire a privileged system object where configuration is stored.
def acquire(self, force=True):
"""Acquire a privileged system object where configuration is stored.
class Docker(ContainerEngine):
"""Docker implementation."""
class Docker(ContainerEngine):
"""Docker implementation."""
- def __init__(self):
- """Initialize Docker object."""
- super(Docker, self).__init__()
+ # Implicit constructor is inherited.
def acquire(self, force=True):
"""Pull an image or a repository from a registry.
def acquire(self, force=True):
"""Pull an image or a repository from a registry.
ret, stderr))
node['cpuinfo'] = list()
for line in stdout.split("\n"):
ret, stderr))
node['cpuinfo'] = list()
for line in stdout.split("\n"):
- if len(line) > 0 and line[0] != "#":
+ if line and line[0] != "#":
node['cpuinfo'].append([CpuUtils.__str2int(x) for x in
line.split(",")])
node['cpuinfo'].append([CpuUtils.__str2int(x) for x in
line.split(",")])
'ActiveEnterTimestamp {name}` | '
'awk \'{{print $2 $3}}\')"'.
format(name=service))
'ActiveEnterTimestamp {name}` | '
'awk \'{{print $2 $3}}\')"'.
format(name=service))
raise RuntimeError('DUT {host} failed to get logs from unit {name}'.
format(host=node['host'], name=service))
raise RuntimeError('DUT {host} failed to get logs from unit {name}'.
format(host=node['host'], name=service))
ret_code, _, _ = \
ssh.exec_command_sudo('service {name} restart'.
format(name=service), timeout=120)
ret_code, _, _ = \
ssh.exec_command_sudo('service {name} restart'.
format(name=service), timeout=120)
raise RuntimeError('DUT {host} failed to start service {name}'.
format(host=node['host'], name=service))
raise RuntimeError('DUT {host} failed to start service {name}'.
format(host=node['host'], name=service))
ssh.exec_command('sudo -Sn bash {0}/{1}/dut_setup.sh'.
format(Constants.REMOTE_FW_DIR,
Constants.RESOURCES_LIB_SH), timeout=120)
ssh.exec_command('sudo -Sn bash {0}/{1}/dut_setup.sh'.
format(Constants.REMOTE_FW_DIR,
Constants.RESOURCES_LIB_SH), timeout=120)
raise RuntimeError('DUT test setup script failed at node {name}'.
format(name=node['host']))
raise RuntimeError('DUT test setup script failed at node {name}'.
format(name=node['host']))
logger.trace('Try {}: Get VPP PID'.format(i))
ret_code, stdout, stderr = ssh.exec_command('pidof vpp')
logger.trace('Try {}: Get VPP PID'.format(i))
ret_code, stdout, stderr = ssh.exec_command('pidof vpp')
raise RuntimeError('Not possible to get PID of VPP process '
'on node: {0}\n {1}'.
format(node['host'], stdout + stderr))
if len(stdout.splitlines()) == 1:
return int(stdout)
raise RuntimeError('Not possible to get PID of VPP process '
'on node: {0}\n {1}'.
format(node['host'], stdout + stderr))
if len(stdout.splitlines()) == 1:
return int(stdout)
- elif len(stdout.splitlines()) == 0:
+ elif not stdout.splitlines():
logger.debug("No VPP PID found on node {0}".
format(node['host']))
continue
logger.debug("No VPP PID found on node {0}".
format(node['host']))
continue
# Try to read number of VFs from PCI address of QAT device
for _ in range(3):
ret_code, stdout, _ = ssh.exec_command(cmd)
# Try to read number of VFs from PCI address of QAT device
for _ in range(3):
ret_code, stdout, _ = ssh.exec_command(cmd)
try:
sriov_numvfs = int(stdout)
except ValueError:
try:
sriov_numvfs = int(stdout)
except ValueError:
format(numvfs, cryptodev.replace(':', r'\:'), timeout=180)
ret_code, _, _ = ssh.exec_command_sudo("sh -c '{0}'".format(cmd))
format(numvfs, cryptodev.replace(':', r'\:'), timeout=180)
ret_code, _, _ = ssh.exec_command_sudo("sh -c '{0}'".format(cmd))
raise RuntimeError('Failed to initialize {0} VFs on QAT device '
' on host {1}'.format(numvfs, node['host']))
raise RuntimeError('Failed to initialize {0} VFs on QAT device '
' on host {1}'.format(numvfs, node['host']))
"sh -c 'echo {0} | tee /sys/bus/pci/devices/{1}/driver/unbind'"
.format(pci_addr, pci_addr.replace(':', r'\:')), timeout=180)
"sh -c 'echo {0} | tee /sys/bus/pci/devices/{1}/driver/unbind'"
.format(pci_addr, pci_addr.replace(':', r'\:')), timeout=180)
raise RuntimeError('Failed to unbind PCI device {0} from driver on '
'host {1}'.format(pci_addr, node['host']))
raise RuntimeError('Failed to unbind PCI device {0} from driver on '
'host {1}'.format(pci_addr, node['host']))
"sh -c 'echo {0} | tee /sys/bus/pci/drivers/{1}/bind'".format(
pci_addr, driver), timeout=180)
"sh -c 'echo {0} | tee /sys/bus/pci/drivers/{1}/bind'".format(
pci_addr, driver), timeout=180)
raise RuntimeError('Failed to bind PCI device {0} to {1} driver on '
'host {2}'.format(pci_addr, driver,
node['host']))
raise RuntimeError('Failed to bind PCI device {0} to {1} driver on '
'host {2}'.format(pci_addr, driver,
node['host']))
for i in range(3):
logger.trace('Try number {0}: Get PCI device driver'.format(i))
for i in range(3):
logger.trace('Try number {0}: Get PCI device driver'.format(i))
cmd = 'lspci -vmmks {0}'.format(pci_addr)
ret_code, stdout, _ = ssh.exec_command(cmd)
cmd = 'lspci -vmmks {0}'.format(pci_addr)
ret_code, stdout, _ = ssh.exec_command(cmd)
raise RuntimeError("'{0}' failed on '{1}'"
.format(cmd, node['host']))
for line in stdout.splitlines():
raise RuntimeError("'{0}' failed on '{1}'"
.format(cmd, node['host']))
for line in stdout.splitlines():
continue
name = None
value = None
continue
name = None
value = None
cmd = 'grep -w {0} /proc/modules'.format(module)
ret_code, _, _ = ssh.exec_command(cmd)
cmd = 'grep -w {0} /proc/modules'.format(module)
ret_code, _, _ = ssh.exec_command(cmd)
if force_load:
# Module is not loaded and we want to load it
DUTSetup.kernel_module_load(node, module)
if force_load:
# Module is not loaded and we want to load it
DUTSetup.kernel_module_load(node, module)
ret_code, _, _ = ssh.exec_command_sudo("modprobe {0}".format(module))
ret_code, _, _ = ssh.exec_command_sudo("modprobe {0}".format(module))
raise RuntimeError('Failed to load {0} kernel module on host {1}'.
format(module, node['host']))
raise RuntimeError('Failed to load {0} kernel module on host {1}'.
format(module, node['host']))
cmd = "[[ -f /etc/redhat-release ]]"
return_code, _, _ = ssh.exec_command(cmd)
cmd = "[[ -f /etc/redhat-release ]]"
return_code, _, _ = ssh.exec_command(cmd)
- if int(return_code) == 0:
+ if not int(return_code):
# workaroud - uninstall existing vpp installation until
# start-testcase script is updated on all virl servers
rpm_pkgs_remove = "vpp*"
cmd_u = 'yum -y remove "{0}"'.format(rpm_pkgs_remove)
r_rcode, _, r_err = ssh.exec_command_sudo(cmd_u, timeout=90)
# workaroud - uninstall existing vpp installation until
# start-testcase script is updated on all virl servers
rpm_pkgs_remove = "vpp*"
cmd_u = 'yum -y remove "{0}"'.format(rpm_pkgs_remove)
r_rcode, _, r_err = ssh.exec_command_sudo(cmd_u, timeout=90)
raise RuntimeError('Failed to remove previous VPP'
'installation on host {0}:\n{1}'
.format(node['host'], r_err))
raise RuntimeError('Failed to remove previous VPP'
'installation on host {0}:\n{1}'
.format(node['host'], r_err))
for pkg in vpp_rpm_pkgs) + "*.rpm"
cmd_i = "rpm -ivh {0}".format(rpm_pkgs)
ret_code, _, err = ssh.exec_command_sudo(cmd_i, timeout=90)
for pkg in vpp_rpm_pkgs) + "*.rpm"
cmd_i = "rpm -ivh {0}".format(rpm_pkgs)
ret_code, _, err = ssh.exec_command_sudo(cmd_i, timeout=90)
raise RuntimeError('Failed to install VPP on host {0}:'
'\n{1}'.format(node['host'], err))
else:
raise RuntimeError('Failed to install VPP on host {0}:'
'\n{1}'.format(node['host'], err))
else:
deb_pkgs_remove = "vpp*"
cmd_u = 'apt-get purge -y "{0}"'.format(deb_pkgs_remove)
r_rcode, _, r_err = ssh.exec_command_sudo(cmd_u, timeout=90)
deb_pkgs_remove = "vpp*"
cmd_u = 'apt-get purge -y "{0}"'.format(deb_pkgs_remove)
r_rcode, _, r_err = ssh.exec_command_sudo(cmd_u, timeout=90)
raise RuntimeError('Failed to remove previous VPP'
'installation on host {0}:\n{1}'
.format(node['host'], r_err))
raise RuntimeError('Failed to remove previous VPP'
'installation on host {0}:\n{1}'
.format(node['host'], r_err))
for pkg in vpp_deb_pkgs) + "*.deb"
cmd_i = "dpkg -i --force-all {0}".format(deb_pkgs)
ret_code, _, err = ssh.exec_command_sudo(cmd_i, timeout=90)
for pkg in vpp_deb_pkgs) + "*.deb"
cmd_i = "dpkg -i --force-all {0}".format(deb_pkgs)
ret_code, _, err = ssh.exec_command_sudo(cmd_i, timeout=90)
raise RuntimeError('Failed to install VPP on host {0}:'
'\n{1}'.format(node['host'], err))
else:
raise RuntimeError('Failed to install VPP on host {0}:'
'\n{1}'.format(node['host'], err))
else:
while True:
res = []
for dummy in range(self._max_attempts):
while True:
res = []
for dummy in range(self._max_attempts):
- res.append(self.measure_loss(rate, self._frame_size,
- self._loss_acceptance,
- self._loss_acceptance_type,
- traffic_type))
+ res.append(self.measure_loss(
+ rate, self._frame_size, self._loss_acceptance,
+ self._loss_acceptance_type, traffic_type))
res = self._get_res_based_on_search_type(res)
res = self._get_res_based_on_search_type(res)
:rtype: tuple
:raises Exception: If search failed.
"""
:rtype: tuple
:raises Exception: If search failed.
"""
- if self._search_result == SearchResults.FAILURE:
- raise Exception('Search FAILED')
- elif self._search_result in [SearchResults.SUCCESS,
- SearchResults.SUSPICIOUS]:
+ if self._search_result in [
+ SearchResults.SUCCESS, SearchResults.SUSPICIOUS]:
return self._search_result_rate, self.get_latency()
return self._search_result_rate, self.get_latency()
+ raise Exception('Search FAILED')
def binary_search(self, b_min, b_max, traffic_type, skip_max_rate=False,
skip_warmup=False):
def binary_search(self, b_min, b_max, traffic_type, skip_max_rate=False,
skip_warmup=False):
class IPv4NetworkGenerator(object):
class IPv4NetworkGenerator(object):
- """IPv4 network generator."""
+ """IPv4 network generator.
+
+ TODO: Conform to https://docs.python.org/2/library/stdtypes.html#typeiter
+ """
+
def __init__(self, networks):
def __init__(self, networks):
+ """Populate internal list of valid networks.
+
:param networks: List of strings containing IPv4 subnet
:param networks: List of strings containing IPv4 subnet
+ :raise RuntimeError: If no IPv4 networks are added.
- self._networks = list()
for network in networks:
net = IPv4Network(unicode(network))
self._networks.append(net)
for network in networks:
net = IPv4Network(unicode(network))
self._networks.append(net)
- if len(self._networks) == 0:
- raise Exception('No IPv4 networks')
+ if not self._networks:
+ raise RuntimeError("No IPv4 networks")
+ """Pop and return network from internal list.
+
:returns: Next network in form (IPv4Network, subnet).
:raises StopIteration: If there are no more elements.
"""
:returns: Next network in form (IPv4Network, subnet).
:raises StopIteration: If there are no more elements.
"""
- if len(self._networks):
return self._networks.pop()
else:
raise StopIteration()
return self._networks.pop()
else:
raise StopIteration()
class Tg(IPv4Node):
"""Traffic generator node"""
class Tg(IPv4Node):
"""Traffic generator node"""
- def __init__(self, node_info):
- super(Tg, self).__init__(node_info)
+
+ # Implicit constructor is inherited.
def _execute(self, cmd):
"""Executes the specified command on TG using SSH.
def _execute(self, cmd):
"""Executes the specified command on TG using SSH.
class Dut(IPv4Node):
"""Device under test"""
class Dut(IPv4Node):
"""Device under test"""
- def __init__(self, node_info):
- super(Dut, self).__init__(node_info)
+
+ # Implicit contructor is inherited.
def get_sw_if_index(self, interface):
"""Get sw_if_index of specified interface from current node.
def get_sw_if_index(self, interface):
"""Get sw_if_index of specified interface from current node.
"""Library to set up IPv6 in topology."""
"""Library to set up IPv6 in topology."""
-from robot.api import logger
from ipaddress import IPv6Network
from ipaddress import IPv6Network
+from robot.api import logger
from resources.libraries.python.ssh import SSH
from resources.libraries.python.topology import NodeType, Topology
from resources.libraries.python.ssh import SSH
from resources.libraries.python.topology import NodeType, Topology
class IPv6Networks(object):
"""IPv6 network iterator.
class IPv6Networks(object):
"""IPv6 network iterator.
- :param networks: List of the available IPv6 networks.
- :type networks: list
+ TODO: Conform to https://docs.python.org/2/library/stdtypes.html#typeiter
def __init__(self, networks):
def __init__(self, networks):
- self._networks = list()
+ """Initialize internal list of valid networks.
+
+ :param networks: List of the available IPv6 networks.
+ :type networks: list
+ :raise RuntimeError: If no networks were added.
+ """
+ self._networks = []
for network in networks:
net = IPv6Network(unicode(network))
self._networks.append(net)
for network in networks:
net = IPv6Network(unicode(network))
self._networks.append(net)
- num = len(self._networks)
- if num == 0:
- raise Exception('No IPv6 networks')
+ if not self._networks:
+ raise RuntimeError('No IPv6 networks')
def next_network(self):
"""Get the next element of the iterator.
def next_network(self):
"""Get the next element of the iterator.
:rtype: IPv6Network object
:raises StopIteration: If there is no more elements.
"""
:rtype: IPv6Network object
:raises StopIteration: If there is no more elements.
"""
- if len(self._networks):
return self._networks.pop()
else:
raise StopIteration()
return self._networks.pop()
else:
raise StopIteration()
"tap_dump.vat")
if name is None:
return response[0]
"tap_dump.vat")
if name is None:
return response[0]
- else:
- for item in response[0]:
- if name == item['dev_name']:
- return item
- return {}
+ for item in response[0]:
+ if name == item['dev_name']:
+ return item
+ return {}
@staticmethod
def create_subinterface(node, interface, sub_id, outer_vlan_id=None,
@staticmethod
def create_subinterface(node, interface, sub_id, outer_vlan_id=None,
else:
raise ValueError('Node {} has not DUT NodeType: "{}"'.
format(node['host'], node['type']))
else:
raise ValueError('Node {} has not DUT NodeType: "{}"'.
format(node['host'], node['type']))
memif_data = memif_data.replace(garbage, '')
for line in memif_data.splitlines():
memif_data = memif_data.replace(garbage, '')
for line in memif_data.splitlines():
- if line.startswith('Sending') or len(line) == 0:
+ if not line or line.startswith('Sending'):
continue
elif line.startswith('memif'):
if memif_name:
continue
elif line.startswith('memif'):
if memif_name:
:param sw_if_idx: DUT node.
:type node: dict
:type sw_if_idx: int
:param sw_if_idx: DUT node.
:type node: dict
:type sw_if_idx: int
- :returns: Memif interface name.
+ :returns: Memif interface name, or None if not found.
:rtype: str
"""
with VatTerminal(node, json_param=False) as vat:
:rtype: str
"""
with VatTerminal(node, json_param=False) as vat:
:param sw_if_idx: DUT node.
:type node: dict
:type sw_if_idx: int
:param sw_if_idx: DUT node.
:type node: dict
:type sw_if_idx: int
- :returns: Memif interface MAC address.
+ :returns: Memif interface MAC address, or None if not found.
:rtype: str
"""
with VatTerminal(node, json_param=False) as vat:
:rtype: str
"""
with VatTerminal(node, json_param=False) as vat:
for item in memif_data:
if memif_data[item]['sw_if_index'] == str(sw_if_idx):
return memif_data[item].get('mac', None)
for item in memif_data:
if memif_data[item]['sw_if_index'] == str(sw_if_idx):
return memif_data[item].get('mac', None)
@staticmethod
def vpp_get_memif_interface_socket(node, sw_if_idx):
@staticmethod
def vpp_get_memif_interface_socket(node, sw_if_idx):
:param sw_if_idx: DUT node.
:type node: dict
:type sw_if_idx: int
:param sw_if_idx: DUT node.
:type node: dict
:type sw_if_idx: int
- :returns: Memif interface socket path.
+ :returns: Memif interface socket path, or None if not found.
:rtype: str
"""
with VatTerminal(node, json_param=False) as vat:
:rtype: str
"""
with VatTerminal(node, json_param=False) as vat:
for item in memif_data:
if memif_data[item]['sw_if_index'] == str(sw_if_idx):
return memif_data[item].get('socket', None)
for item in memif_data:
if memif_data[item]['sw_if_index'] == str(sw_if_idx):
return memif_data[item].get('socket', None)
@staticmethod
def vpp_get_memif_interface_id(node, sw_if_idx):
@staticmethod
def vpp_get_memif_interface_id(node, sw_if_idx):
:param sw_if_idx: DUT node.
:type node: dict
:type sw_if_idx: int
:param sw_if_idx: DUT node.
:type node: dict
:type sw_if_idx: int
- :returns: Memif interface ID.
+ :returns: Memif interface ID, or None if not found.
:rtype: int
"""
with VatTerminal(node, json_param=False) as vat:
:rtype: int
"""
with VatTerminal(node, json_param=False) as vat:
for item in memif_data:
if memif_data[item]['sw_if_index'] == str(sw_if_idx):
return int(memif_data[item].get('id', None))
for item in memif_data:
if memif_data[item]['sw_if_index'] == str(sw_if_idx):
return int(memif_data[item].get('id', None))
@staticmethod
def vpp_get_memif_interface_role(node, sw_if_idx):
@staticmethod
def vpp_get_memif_interface_role(node, sw_if_idx):
:param sw_if_idx: DUT node.
:type node: dict
:type sw_if_idx: int
:param sw_if_idx: DUT node.
:type node: dict
:type sw_if_idx: int
- :returns: Memif interface role.
+ :returns: Memif interface role, or None if not found.
:rtype: int
"""
with VatTerminal(node, json_param=False) as vat:
:rtype: int
"""
with VatTerminal(node, json_param=False) as vat:
for item in memif_data:
if memif_data[item]['sw_if_index'] == str(sw_if_idx):
return memif_data[item].get('role', None)
for item in memif_data:
if memif_data[item]['sw_if_index'] == str(sw_if_idx):
return memif_data[item].get('role', None)
items = line.split(" ")
while "" in items:
items.remove("")
items = line.split(" ")
while "" in items:
items.remove("")
# no ports were returned
data.append({
"local_address": items[0],
# no ports were returned
data.append({
"local_address": items[0],
for trash in ("", "vat#"):
while trash in items:
items.remove(trash)
for trash in ("", "vat#"):
while trash in items:
items.remove(trash)
data.append({
# items[0] is the table header - "sw_if_index"
"sw_if_index": items[1],
data.append({
# items[0] is the table header - "sw_if_index"
"sw_if_index": items[1],
"""
if not self._path_iter:
return None, None
"""
if not self._path_iter:
return None, None
- else:
- return self._path_iter.pop()
+ return self._path_iter.pop()
def first_interface(self):
"""Return first interface on the path.
def first_interface(self):
"""Return first interface on the path.
print 'Reading VPP worker thread PID failed.'
else:
for pid in out.split("\n"):
print 'Reading VPP worker thread PID failed.'
else:
for pid in out.split("\n"):
- if len(pid) > 0 and pid[0] != "#":
+ if pid and pid[0] != '#':
SchedUtils.set_proc_scheduling_rr(node, int(pid))
break
else:
SchedUtils.set_proc_scheduling_rr(node, int(pid))
break
else:
:raises RuntimeError: Specified interface was not found.
"""
tap_if = InterfaceUtil.tap_dump(node, tap_name)
:raises RuntimeError: Specified interface was not found.
"""
tap_if = InterfaceUtil.tap_dump(node, tap_name)
raise RuntimeError(
'Tap interface :{} does not exist'.format(tap_name))
raise RuntimeError(
'Tap interface :{} does not exist'.format(tap_name))
:type node: dict
"""
if node['type'] == NodeType.DUT:
:type node: dict
"""
if node['type'] == NodeType.DUT:
- sequence = "\nno VAT command executed"\
- if len(DICT__DUTS_VAT_HISTORY[node['host']]) == 0\
- else "".join("\n{}".format(cmd)
- for cmd in DICT__DUTS_VAT_HISTORY[node['host']])
- logger.trace("{0} VAT command history:{1}\n".
- format(node['host'], sequence))
+ sequence = "\nno VAT command executed"
+ if DICT__DUTS_VAT_HISTORY[node['host']]:
+ sequence = "".join(["\n{}".format(
+ cmd) for cmd in DICT__DUTS_VAT_HISTORY[node['host']]])
+ logger.trace(
+ "{0} VAT command history:{1}\n".format(node['host'], sequence))
@staticmethod
def show_vat_history_on_all_duts(nodes):
@staticmethod
def show_vat_history_on_all_duts(nodes):
vat.vat_terminal_exec_cmd('want_stats enable')
for _ in range(0, 12):
stats_table = vat.vat_terminal_exec_cmd('dump_stats_table')
vat.vat_terminal_exec_cmd('want_stats enable')
for _ in range(0, 12):
stats_table = vat.vat_terminal_exec_cmd('dump_stats_table')
- if_counters = stats_table['interface_counters']
- if len(if_counters) > 0:
+ if stats_table['interface_counters']:
self._stats_table = stats_table
return stats_table
time.sleep(1)
self._stats_table = stats_table
return stats_table
time.sleep(1)
return 0
if_counters = self._stats_table.get('interface_counters')
return 0
if_counters = self._stats_table.get('interface_counters')
- if if_counters is None or len(if_counters) == 0:
logger.trace('No interface counters.')
return 0
for counter in if_counters:
if counter['vnet_counter_type'] == version:
data = counter['data']
return data[if_index]
logger.trace('No interface counters.')
return 0
for counter in if_counters:
if counter['vnet_counter_type'] == version:
data = counter['data']
return data[if_index]
- logger.trace('{i} {v} counter not found.'.format(i=interface,
- v=version))
+ logger.trace('{i} {v} counter not found.'.format(
+ i=interface, v=version))
raise TypeError('Node parameter is None')
if cmd is None:
raise TypeError('Command parameter is None')
raise TypeError('Node parameter is None')
if cmd is None:
raise TypeError('Command parameter is None')
raise ValueError('Empty command parameter')
ssh = SSH()
raise ValueError('Empty command parameter')
ssh = SSH()
:type node: dict
:type iface_key: str/int
:returns: Return sw_if_index or None if not found.
:type node: dict
:type iface_key: str/int
:returns: Return sw_if_index or None if not found.
"""
try:
if isinstance(iface_key, basestring):
return node['interfaces'][iface_key].get('vpp_sw_index')
# TODO: use only iface_key, do not use integer
"""
try:
if isinstance(iface_key, basestring):
return node['interfaces'][iface_key].get('vpp_sw_index')
# TODO: use only iface_key, do not use integer
- else:
- return int(iface_key)
except (KeyError, ValueError):
return None
except (KeyError, ValueError):
return None
:raises TypeError: If provided interface name is not a string.
"""
try:
:raises TypeError: If provided interface name is not a string.
"""
try:
- if isinstance(iface_name, basestring):
- iface_key = Topology.get_interface_by_name(node, iface_name)
- return node['interfaces'][iface_key].get('vpp_sw_index')
- else:
+ if not isinstance(iface_name, basestring):
raise TypeError("Interface name must be a string.")
raise TypeError("Interface name must be a string.")
+ iface_key = Topology.get_interface_by_name(node, iface_name)
+ return node['interfaces'][iface_key].get('vpp_sw_index')
except (KeyError, ValueError):
return None
except (KeyError, ValueError):
return None
:param iface_keys: Interface keys for lookup.
:type node: dict
:type iface_keys: strings
:param iface_keys: Interface keys for lookup.
:type node: dict
:type iface_keys: strings
+ :returns: Numa node of most given interfaces or 0.
+ :rtype: int
"""
numa_list = []
for if_key in iface_keys:
"""
numa_list = []
for if_key in iface_keys:
numa_cnt_mc = Counter(numa_list).most_common()
numa_cnt_mc = Counter(numa_list).most_common()
- if len(numa_cnt_mc) > 0 and numa_cnt_mc[0][0] != -1:
+ if numa_cnt_mc and numa_cnt_mc[0][0] != -1:
- elif len(numa_cnt_mc) > 1 and numa_cnt_mc[0][0] == -1:
+ if len(numa_cnt_mc) > 1 and numa_cnt_mc[0][0] == -1:
@staticmethod
def get_interface_mac(node, iface_key):
@staticmethod
def get_interface_mac(node, iface_key):
continue
if if_val['link'] == link_name:
return node_data, if_key
continue
if if_val['link'] == link_name:
return node_data, if_key
@staticmethod
def get_interface_pci_addr(node, iface_key):
@staticmethod
def get_interface_pci_addr(node, iface_key):
:param filter_list: Link filter criteria.
:type node: dict
:type filter_list: list of strings
:param filter_list: Link filter criteria.
:type node: dict
:type filter_list: list of strings
- :returns: List of strings representing link names occupied by the node.
- :rtype: list
+ :returns: List of link names occupied by the node.
+ :rtype: None or list of string
"""
interfaces = node['interfaces']
link_names = []
"""
interfaces = node['interfaces']
link_names = []
.format(str(interface)))
else:
link_names.append(interface['link'])
.format(str(interface)))
else:
link_names.append(interface['link'])
- if len(link_names) == 0:
link_names = None
return link_names
link_names = None
return link_names
:param node2: Connected node.
:type node1: dict
:type node2: dict
:param node2: Connected node.
:type node1: dict
:type node2: dict
- :returns: Name of link connecting the two nodes together.
+ :returns: Name of a link connecting the two nodes together.
:rtype: str
:raises RuntimeError: If no links are found.
"""
connecting_links = self.get_active_connecting_links(node1, node2)
:rtype: str
:raises RuntimeError: If no links are found.
"""
connecting_links = self.get_active_connecting_links(node1, node2)
- if len(connecting_links) == 0:
+ if not connecting_links:
raise RuntimeError("No links connecting the nodes were found")
raise RuntimeError("No links connecting the nodes were found")
- else:
- return connecting_links[0]
+ return connecting_links[0]
@keyword('Get egress interfaces name on "${node1}" for link with '
'"${node2}"')
@keyword('Get egress interfaces name on "${node1}" for link with '
'"${node2}"')
"""
interfaces = []
links = self.get_active_connecting_links(node1, node2)
"""
interfaces = []
links = self.get_active_connecting_links(node1, node2)
raise RuntimeError('No link between nodes')
for interface in node1['interfaces'].values():
link = interface.get('link')
raise RuntimeError('No link between nodes')
for interface in node1['interfaces'].values():
link = interface.get('link')
Code Review / csit.git / commitdiff