from robot.api.deco import keyword
from resources.libraries.python.topology import Topology
-from resources.libraries.python.topology import NodeType
-from resources.libraries.python.TrafficScriptExecutor\
- import TrafficScriptExecutor
from resources.libraries.python.IPv4Setup import get_node
interface, ip_address))
get_node(node).arp_ping(ip_address, interface)
+ @staticmethod
+ def set_interface_address(node, interface, address, prefix_length):
+ """See IPv4Node.set_ip for more information.
+
+ :param node: Node where IP address should be set to.
+ :param interface: Interface name.
+ :param address: IP address.
+ :param prefix_length: Prefix length.
+ :type node: dict
+ :type interface: str
+ :type address: str
+ :type prefix_length: int
+ """
+ log.debug('Node {} interface {} has IPv4 address {} with prefix '
+ 'length {}'.format(Topology.get_node_hostname(node),
+ interface, address, prefix_length))
+ get_node(node).set_ip(interface, address, int(prefix_length))
+
@staticmethod
@keyword('Node "${node}" routes to IPv4 network "${network}" with prefix '
'length "${prefix_length}" using interface "${interface}" via '
def set_route(node, network, prefix_length, interface, gateway):
"""See IPv4Node.set_route for more information.
- :param node:
- :param network:
- :param prefix_length:
- :param interface:
- :param gateway:
- :return:
+ :param node: Node where IP address should be set to.
+ :param network: IP network.
+ :param prefix_length: Prefix length.
+ :param interface: Interface name.
+ :param gateway: Gateway.
+ :type node: dict
+ :type network: str
+ :type prefix_length: int
+ :type interface: str
+ :type gateway: str
"""
log.debug('Node {} routes to network {} with prefix length {} '
'via {} interface {}'.format(Topology.get_node_hostname(node),
get_node(node).set_route(network, int(prefix_length),
gateway, interface)
- @staticmethod
- @keyword('After ping is sent in topology "${nodes_info}" from node '
- '"${src_node}" interface "${src_port}" "${src_ip}" with '
- 'destination IPv4 address "${dst_ip}" of node "${dst_node}" '
- 'interface "${dst_port}" a ping response arrives and TTL is '
- 'decreased by "${hops}"')
- def send_ping(nodes_info, src_node, src_port, src_ip, dst_ip, dst_node,
- dst_port, hops):
- """Send IPv4 ping and wait for response.
-
- :param nodes_info: Dictionary containing information on all nodes
- in topology.
- :param src_node: Source node.
- :param src_port: Source interface.
- :param src_ip: Source ipv4 address.
- :param dst_ip: Destination ipv4 address.
- :param dst_node: Destination node.
- :param dst_port: Destination interface.
- :param hops: Number of hops between src_node and dst_node.
- """
- log.debug('After ping is sent from node "{}" interface "{}" '
- 'with destination IPv4 address of node "{}" interface "{}" '
- 'a ping response arrives and TTL is decreased by "${}"'.
- format(Topology.get_node_hostname(src_node), src_port,
- Topology.get_node_hostname(dst_node), dst_port, hops))
- dst_mac = None
- src_mac = Topology.get_interface_mac(src_node, src_port)
- if dst_node['type'] == NodeType.TG:
- dst_mac = Topology.get_interface_mac(src_node, src_port)
- _, adj_int = Topology.\
- get_adjacent_node_and_interface(nodes_info, src_node, src_port)
- first_hop_mac = adj_int['mac_address']
- args = '--src_if "{}" --src_mac "{}" --first_hop_mac "{}" ' \
- '--src_ip "{}" --dst_ip "{}" --hops "{}"'\
- .format(src_port, src_mac, first_hop_mac, src_ip, dst_ip, hops)
- if dst_node['type'] == NodeType.TG:
- args += ' --dst_if "{}" --dst_mac "{}"'.format(dst_port, dst_mac)
- TrafficScriptExecutor.run_traffic_script_on_node(
- "ipv4_ping_ttl_check.py", src_node, args)
-
@staticmethod
@keyword('Get IPv4 address prefix of node "${node}" interface "${port}" '
'from "${nodes_addr}"')
:param node: Node dictionary.
:param port: Interface name.
- :return: IPv4 prefix length
+ :param nodes_addr: Available nodes IPv4 addresses.
+ :type node: dict
+ :type port: str
+ :type nodes_addr: dict
+ :return: IPv4 prefix length.
+ :rtype: int
"""
for net in nodes_addr.values():
for p in net['ports'].values():
:param node: Node dictionary.
:param port: Interface name.
- :return: IPv4 subnet of 'str' type
+ :param nodes_addr: Available nodes IPv4 addresses.
+ :type node: dict
+ :type port: int
+ :type nodes_addr: dict
+ :return: IPv4 subnet.
+ :rtype: str
"""
for net in nodes_addr.values():
for p in net['ports'].values():
:return:
"""
get_node(node).flush_ip_addresses(port)
+
+ @staticmethod
+ def get_link_address(link, nodes_addr):
+ """Get link IPv4 address.
+
+ :param link: Link name.
+ :param nodes_addr: Available nodes IPv4 addresses.
+ :type link: str
+ :type nodes_addr: dict
+ :return: Link IPv4 address.
+ :rtype: str
+ """
+ net = nodes_addr.get(link)
+ if net is None:
+ raise ValueError('Link "{0}" not found'.format(link))
+ return net.get('net_addr')
+
+ @staticmethod
+ def get_link_prefix(link, nodes_addr):
+ """Get link IPv4 address prefix.
+
+ :param link: Link name.
+ :param nodes_addr: Available nodes IPv4 addresses.
+ :type link: str
+ :type nodes_addr: dict
+ :return: Link IPv4 address prefix.
+ :rtype: int
+ """
+ net = nodes_addr.get(link)
+ if net is None:
+ raise ValueError('Link "{0}" not found'.format(link))
+ return net.get('prefix')
Code Review / csit.git / blobdiff