@keyword('From node "${node}" interface "${port}" ARP-ping '
'IPv4 address "${ip_address}"')
def arp_ping(node, interface, ip_address):
+ """Send an ARP ping from the specified node.
+
+ :param node: Node in topology.
+ :param ip_address: Destination IP address for the ARP packet.
+ :param interface: Name of an interface to send the ARP packet from.
+ :type node: dict
+ :type ip_address: str
+ :type interface: str
+ """
log.debug('From node {} interface {} ARP-ping IPv4 address {}'.
format(Topology.get_node_hostname(node),
interface, ip_address))
:rtype: int
"""
for net in nodes_addr.values():
- for p in net['ports'].values():
- if p['node'] == node['host'] and p['if'] == port:
+ for net_port in net['ports'].values():
+ if net_port['node'] == node['host'] and net_port['if'] == port:
return net['prefix']
raise Exception('Subnet not found for node {n} port {p}'.
:rtype: str
"""
for net in nodes_addr.values():
- for p in net['ports'].values():
- if p['node'] == node['host'] and p['if'] == port:
+ for net_port in net['ports'].values():
+ if net_port['node'] == node['host'] and net_port['if'] == port:
return net['net_addr']
raise Exception('Subnet not found for node {n} port {p}'.
@staticmethod
def send_ping_from_node_to_dst(node, destination, namespace=None,
- ping_count=3):
+ ping_count=3, interface=None):
"""Send a ping from node to destination. Optionally, you can define a
- namespace from where to send a ping.
+ namespace and interface from where to send a ping.
:param node: Node to start ping on.
:param destination: IPv4 address where to send ping.
- :param namespace: Namespace to send ping from.
- :param ping_count: Number of pings to send.
+ :param namespace: Namespace to send ping from. Optional
+ :param ping_count: Number of pings to send. Default 3
+ :param interface: Interface from where to send ping. Optional
:type node: dict
:type destination: str
:type namespace: str
:type ping_count: int
+ :type interface: str
:raises RuntimeError: If no response for ping, raise error
"""
cmd = ''
if namespace is not None:
cmd = 'ip netns exec {0} ping -c{1} {2}'.format(
namespace, ping_count, destination)
+ elif interface is not None:
+ cmd = 'ping -I {0} -c{1} {2}'.format(
+ interface, ping_count, destination)
else:
cmd = 'ping -c{0} {1}'.format(ping_count, destination)
- rc, stdout, stderr = exec_cmd(node, cmd, sudo=True)
- if rc != 0:
+ ret_code, _, _ = exec_cmd(node, cmd, sudo=True)
+ if ret_code != 0:
raise RuntimeError("Ping Not Successful")
+
+ @staticmethod
+ def set_linux_interface_arp(node, interface, ip_addr, mac, namespace=None):
+ """Set arp on interface in linux.
+
+ :param node: Node where to execute command.
+ :param interface: Interface in namespace.
+ :param ip_addr: IP address for ARP entry.
+ :param mac: MAC address.
+ :param namespace: Execute command in namespace. Optional
+ :type node: dict
+ :type interface: str
+ :type ip_addr: str
+ :type mac: str
+ :type namespace: str
+ :raises RuntimeError: Could not set ARP properly.
+ """
+ if namespace is not None:
+ cmd = 'ip netns exec {} arp -i {} -s {} {}'.format(
+ namespace, interface, ip_addr, mac)
+ else:
+ cmd = 'arp -i {} -s {} {}'.format(interface, ip_addr, mac)
+ ret_code, _, stderr = exec_cmd(node, cmd, sudo=True)
+ if ret_code != 0:
+ raise RuntimeError("Arp set not successful, reason:{}".
+ format(stderr))