"""
Classes and utilities to manage IXN application.
@author yoram@ignissoft.com
"""
import sys
from os import path
import time
from builtins import range
from trafficgenerator.tgn_utils import TgnError, is_true
from trafficgenerator.tgn_tcl import build_obj_ref_list
from trafficgenerator.trafficgenerator import TrafficGenerator
from ixnetwork.ixn_object import IxnObject
from ixnetwork.ixn_port import IxnPort
from ixnetwork.ixn_traffic import IxnTrafficItem, IxnL23TrafficItem
from ixnetwork.ixn_protocol import (IxnBgpRouter, IxnOspfRouter, IxnOspfV3Router, IxnBgpRouteRange, IxnOspfRouteRange,
IxnPimsmRouter, IxnIgmpHost, IxnIgmpQuerier, IxnPimsmSource, IxnStpBridge,
IxnOpenFlowDevice, IxnLacp, IxnIsisRouter, IxnIsisRouteRange, IxnLdpRouter,
IxnLdpRouteRange)
from ixnetwork.ixn_interface import IxnInterface, IxnInterfaceL3
from ixnetwork.ixn_topology import IxnTopology, IxnDeviceGroup, IxnNgpfEthernet, IxnNgpfIpv4
from ixnetwork.ixn_statistics_view import IxnStatisticsView
from ixnetwork.ixn_root import IxnRoot
from ixnetwork.ixn_protocol_stack import IxnRange
from ixnetwork.ixn_hw import IxnHw, IxnChassis, IxnCard, IxnPhyPort
[docs]class IxnApp(TrafficGenerator):
""" IxNetwork driver. Equivalent to IxNetwork Application. """
root = None
def __init__(self, logger, api_wrapper=None):
""" Set all kinds of application level objects - logger, api, etc.
:param logger: python logger (e.g. logging.getLogger('log'))
:param api_wrapper: api wrapper object inheriting and implementing IxnApi base class.
"""
super(self.__class__, self).__init__()
self.logger = logger
self.api = api_wrapper
IxnObject.logger = self.logger
IxnObject.api = self.api
IxnObject.str_2_class = TYPE_2_OBJECT
[docs] def connect(self, tcl_server='localhost', tcl_port=8009):
self.api.connect(tcl_server, tcl_port)
self.root = IxnRoot(objRef=self.api.getRoot(), objType='root')
IxnObject.root = self.root
IxnObject.root.hw = self.root.get_child('availableHardware')
[docs] def disconnect(self):
""" Disconnect from chassis and server. """
pass
#
# IxNetwork operation commands.
#
[docs] def commit(self):
self.api.commit()
[docs] def load_config(self, config_file_name):
self.api.loadConfig(config_file_name)
self.commit()
[docs] def new_config(self):
self.api.newConfig()
self.commit()
[docs] def save_config(self, config_file_name):
self.commit()
self.api.saveConfig(path.abspath(config_file_name))
#
# IxNetwork GUI commands.
#
[docs] def send_arp_ns(self):
self.api.execute('sendArpAll')
self.api.execute('sendNsAll')
[docs] def send_rs(self):
self.api.execute('sendRsAll')
[docs] def protocols_start(self):
""" Start all protocols.
It is the calling function responsibility to wait for all protocols to start.
"""
self.api.execute('startAllProtocols')
[docs] def protocols_stop(self):
""" Stop all protocols.
It is the calling function responsibility to wait for all protocols to stop.
"""
self.api.execute('stopAllProtocols')
[docs] def traffic_apply(self):
self.root.get_child_static('traffic').execute('apply')
[docs] def l23_traffic_start(self, blocking=False):
self.root.get_child_static('traffic').execute('startStatelessTraffic')
self.wait_traffic_state("started", timeout=16)
if blocking:
self.wait_traffic_state("stopped", timeout=sys.maxsize)
else:
time.sleep(2)
[docs] def l23_traffic_stop(self):
self.root.get_child_static('traffic').execute('stopStatelessTraffic')
self.wait_traffic_state("stopped", timeout=8)
[docs] def wait_traffic_state(self, state, timeout):
for _ in range(timeout):
if self.root.get_child_static('traffic').get_attribute('state') == state:
return
time.sleep(1)
raise TgnError('Traffic failed, traffic is {} after {} seconds'.
format(self.root.get_child_static('traffic').get_attribute('isTrafficRunning'), timeout))
[docs] def protocol_start(self, protocol):
""" Start a protocol and wait for all protocols to start.
:raise TgnError: if some protocol failed to start.
"""
self.protocol_action(protocol, 'start')
[docs] def protocol_stop(self, protocol):
""" Stop all protocols and wait for all protocols to stop.
Raises:
TgnError: if some protocol failed to stop.
"""
self.protocol_action(protocol, 'stop')
[docs] def protocol_abort(self, protocol):
""" Abort all protocols and wait for all protocols to stop.
Raises:
TgnError: if some protocol failed to stop.
"""
self.protocol_action(protocol, 'abort')
[docs] def protocol_action(self, protocol, action):
action_state = {'start': 'started', 'stop': 'stopped', 'stop': 'stopped'}
protocol_objs = []
for port in self.root.get_objects_by_type('vport'):
protocols = port.get_child_static('protocols')
protocol_obj = protocols.get_child_static(protocol)
if is_true(protocol_obj.get_attribute('enabled')):
protocol_obj.execute(action)
protocol_objs.append(protocol_obj)
# Must wait before reading state
time.sleep(2)
for protocol_obj in protocol_objs:
runningState = protocol_obj.get_attribute('runningState')
timer = 16
while timer and runningState != action_state[action]:
time.sleep(1)
runningState = protocol_obj.get_attribute('runningState')
timer -= 1
if runningState != action_state[action]:
raise TgnError('Failed to {} port {} protocol {}'.
format(action, port.obj_name(), protocol))
#
# Not tested beyond this point.
#
[docs] def l23TrafficOperation(self, action):
tis = build_obj_ref_list(self.get_objects(IxnL23TrafficItem))
self.api.execute(action + 'StatelessTraffic', tis)
[docs] def l47TrafficOperation(self, action):
self.api.execute(action + 'ApplicationTraffic', self.root + '/traffic')
[docs] def l47TrafficApply(self):
self.l47TrafficOperation('apply')
[docs] def l47TrafficStart(self):
self.l47TrafficOperation('start')
[docs] def l47TrafficStop(self):
self.l47TrafficOperation('stop')
#
# IxNetwork results commands.
#
[docs] def getStatistics(self, view):
if (view not in self.objects):
self.objects[view] = IxnStatisticsView(self.api, view)
return self.objects[view].getStatistics()
TYPE_2_OBJECT = {'availableHardware': IxnHw,
'bridge': IxnStpBridge,
'card': IxnCard,
'chassis': IxnChassis,
'device': IxnOpenFlowDevice,
'deviceGroup': IxnDeviceGroup,
'ethernet': IxnNgpfEthernet,
'host': IxnIgmpHost,
'interface': {'vport': IxnInterface},
'ipv4': {'interface': IxnInterfaceL3,
'etherenet': IxnNgpfIpv4},
'ipv6': IxnInterfaceL3,
'lacp': IxnLacp,
'neighborRange': IxnBgpRouter,
'port': IxnPhyPort,
'querier': IxnIgmpQuerier,
'range': IxnRange,
'router': {'ospf': IxnOspfRouter,
'ospfV3': IxnOspfV3Router,
'pimsm': IxnPimsmRouter,
'isis': IxnIsisRouter,
'ldp': IxnLdpRouter},
'routeRange': {'bgp': IxnBgpRouteRange,
'ospf': IxnOspfRouteRange,
'isis': IxnIsisRouteRange,
'ldp': IxnLdpRouteRange},
'source': {'interface': IxnPimsmSource},
'topology': IxnTopology,
'trafficItem': IxnTrafficItem,
'vport': IxnPort,
}