[ixia/Keysight] sample test cases and fixtures

tests/common/ixia/ixia_fixtures.py

@@ -0,0 +1,75 @@
+import pytest
+import pprint
+# from common.devices import SonicHost, Localhost, PTFHost, EosHost, FanoutHost
+from common.devices import FanoutHost
+""" 
+In an IXIA testbed, there is no PTF docker. 
+Hence, we use ptf_ip field to store IXIA API server. 
+This fixture returns the IP address of the IXIA API server.
+"""
+@pytest.fixture(scope = "module")
+def ixia_api_serv_ip(testbed):
+    return testbed['ptf_ip']
+
+"""
+Return the username of IXIA API server
+"""
+@pytest.fixture(scope = "module")
+def ixia_api_serv_user(duthost):
+    return duthost.host.options['variable_manager']._hostvars[duthost.hostname]['secret_group_vars']['ixia_api_server']['user']
+
+"""
+Return the password of IXIA API server
+"""
+@pytest.fixture(scope = "module")
+def ixia_api_serv_passwd(duthost):
+    return duthost.host.options['variable_manager']._hostvars[duthost.hostname]['secret_group_vars']['ixia_api_server']['password']
+
+"""
+Return REST port. 
+"""
+@pytest.fixture(scope = "module")
+def ixia_api_serv_port(duthost):
+    return duthost.host.options['variable_manager']._hostvars[duthost.hostname]['secret_group_vars']['ixia_api_server']['rest_port']
+
+"""
+IXIA PTF can spawn multiple session on the same REST port. Optional for LINUX, Rewuired for windows 
+Return the session ID. 
+"""
+@pytest.fixture(scope = "module")
+def ixia_api_serv_session_id(duthost):
+    return duthost.host.options['variable_manager']._hostvars[duthost.hostname]['secret_group_vars']['ixia_api_server']['session_id']
+
+"""
+IXIA chassis are leaf fanout switches in the testbed.
+This fixture returns the hostnames and IP addresses of the IXIA chassis in the dictionary format.
+"""  
+@pytest.fixture(scope = "module")
+def ixia_dev(duthost, ansible_adhoc, conn_graph_facts, creds):
+    dev_conn     = conn_graph_facts['device_conn'] if 'device_conn' in conn_graph_facts else {}
+    fanout_hosts = {}
+    result       = dict()
+    # WA for virtual testbed which has no fanout
+    try:
+        for dut_port in dev_conn.keys():
+            fanout_rec  = dev_conn[dut_port]
+            fanout_host = fanout_rec['peerdevice']
+            fanout_port = fanout_rec['peerport']
+            if fanout_host in fanout_hosts.keys():
+                fanout  = fanout_hosts[fanout_host]
+            else:
+                user = pswd = None
+                host_vars = ansible_adhoc().options['inventory_manager'].get_host(fanout_host).vars
+                os_type = 'eos' if 'os' not in host_vars else host_vars['os']
+                if os_type == "ixia":
+                    fanout  = FanoutHost(ansible_adhoc, os_type, fanout_host, 'FanoutLeaf', user, pswd)
+                    fanout_hosts[fanout_host] = fanout
+                    fanout.add_port_map(dut_port, fanout_port)
+    except:
+        pass
+
+    ixia_dev_hostnames = fanout_hosts.keys()
+    for hostname in ixia_dev_hostnames:
+        result[hostname] = duthost.host.options['inventory_manager'].get_host(hostname).get_vars()['ansible_host']
+
+    return result

tests/common/ixia/ixia_helpers.py

@@ -0,0 +1,59 @@
+import re 
+
+"""
+@summary: given a DUT interface, return the management IP address of its neighbor IXIA device
+@param intf: DUT interface
+@param conn_graph_facts: testbed connectivity graph
+@param ixia_dev: the mapping of hostname to IP address of IXIA devices
+@return the management IP address of its neighbor IXIA device or None if we cannot find it
+"""
+def get_neigh_ixia_mgmt_ip(intf, conn_graph_facts, ixia_dev):
+    device_conn = conn_graph_facts['device_conn']
+    if intf not in device_conn:
+        return None 
+
+    ixia_dev_hostname = device_conn[intf]['peerdevice']
+    if ixia_dev_hostname not in ixia_dev:
+        return None 
+
+    return ixia_dev[ixia_dev_hostname]
+
+"""
+@summary: given a DUT interface, return the card of its neighbor IXIA device
+@param intf: DUT interface
+@param conn_graph_facts: testbed connectivity graph
+@return the card of its neighbor IXIA device or None if we cannot find it
+"""
+def get_neigh_ixia_card(intf, conn_graph_facts):
+    device_conn = conn_graph_facts['device_conn']
+    if intf not in device_conn:
+        return None 
+
+    ixia_intf = device_conn[intf]['peerport']
+    pattern = r'Card(\d+)/Port(\d+)'
+    m = re.match(pattern, ixia_intf)
+
+    if m is None:
+        return None 
+    else:
+        return m.group(1)    
+
+"""
+@summary: given a DUT interface, return the port of its neighbor IXIA device
+@param intf: DUT interface
+@param conn_graph_facts: testbed connectivity graph
+@return the port of its neighbor IXIA device or None if we cannot find it
+"""
+def get_neigh_ixia_port(intf, conn_graph_facts):
+    device_conn = conn_graph_facts['device_conn']
+    if intf not in device_conn:
+        return None
+
+    ixia_intf = device_conn[intf]['peerport']
+    pattern = r'Card(\d+)/Port(\d+)'
+    m = re.match(pattern, ixia_intf)
+
+    if m is None:
+        return None 
+    else:
+        return m.group(2)  

tests/ixia/test_ixia_traffic_restpy.py

@@ -0,0 +1,183 @@
+###############################################################################
+# This test cases demonstrates:
+#   * All the fixtures required for running ixia script (please see the
+#     arguments of the test function)
+#   * How Ixia chassis card/ports are addressed
+#   * How you can configure/control ixia devices, start traffic and collect
+#     statistics using REST API
+#   * This simple sanity test cases can be used to check if testbed setup
+#     is correct or not - since it prints a lot of testbed data
+###############################################################################
+
+import logging
+import time
+import pytest
+import pprint
+from common.utilities import wait_until
+from common.fixtures.conn_graph_facts import conn_graph_facts
+from common.platform.interface_utils import check_interface_information
+from common.platform.daemon_utils import check_pmon_daemon_status
+from common.reboot import *
+from common.platform.device_utils import fanout_switch_port_lookup
+from common.helpers import assertions
+from ixnetwork_restpy import SessionAssistant, Files
+
+from common.ixia_fixtures import ixia_api_serv_ip, ixia_api_serv_user,\
+     ixia_api_serv_passwd, ixia_dev, ixia_api_serv_port, ixia_api_serv_session_id
+
+from common.ixia_helpers import get_neigh_ixia_mgmt_ip, get_neigh_ixia_card,\
+     get_neigh_ixia_port
+
+import time
+
+#pytestmark = [pytest.mark.disable_loganalyzer]
+
+def returnIxiaChassisIp (chassisDict, no) :
+    chList = []
+    for key in chassisDict.keys():
+        chList.append(chassisDict[key])
+
+    return (chList[no - 1])
+
+def parseDeviceConn (device_conn) :
+    retval = []
+    dive_con_dict = device_conn['device_conn']
+    for key in dive_con_dict.keys() :
+        pp =  dive_con_dict[key]['peerport']
+        string = pp + '/' + key
+        retval.append(string)
+    retval.sort()
+    return(retval)
+
+def getCard(ixiaCardPortList, num) :
+    card = ixiaCardPortList[num].split('/')[0]
+    cardNo = int(card.replace('Card', ''))
+    return(cardNo)
+
+def getPort(ixiaCardPortList, num) :
+    port = ixiaCardPortList[num].split('/')[1]
+    portNo = int(port.replace('Port', ''))
+    return(portNo)
+
+def test_testbed(testbed, conn_graph_facts, duthost, ixia_dev, ixia_api_serv_ip,
+                 ixia_api_serv_user, ixia_api_serv_passwd, ixia_api_serv_port,
+                 ixia_api_serv_session_id):
+    logger.info("Connection Graph Facts = %s " %(conn_graph_facts))
+    logger.info("DUT hostname = %s" %(duthost.hostname))
+    ixiaports = parseDeviceConn(conn_graph_facts)
+    logger.info("Ixia ports = %s" %(ixiaports))
+    logger.info("Ixia chassis IP = %s" %(ixia_dev))
+    logger.info("Ixia API server IP = %s" %(ixia_api_serv_ip))
+    logger.info("Ixia API server user = %s" %(ixia_api_serv_user))
+    logger.info("Ixia API server password = %s" %(ixia_api_serv_passwd))
+    logger.info("Ixia API server port = %s" %(ixia_api_serv_port))
+    logger.info("Ixia API server sessionId = %s" %(ixia_api_serv_session_id))
+
+    clientIp  = ixia_api_serv_ip
+    UserName  = ixia_api_serv_user
+    Password  = ixia_api_serv_passwd
+    RestPort  = ixia_api_serv_passwd
+    SessionId = ixia_api_serv_session_id
+    chassisIp = returnIxiaChassisIp(ixia_dev, 1)
+
+    cardId    = getCard(ixiaports, 0)
+    PortId1   = getPort(ixiaports, 0)
+    PortId2   = getPort(ixiaports, 1)
+    PortId3   = getPort(ixiaports, 2)
+    PortId4   = getPort(ixiaports, 3)
+    PortId5   = getPort(ixiaports, 4)
+    PortId6   = getPort(ixiaports, 5)
+
+    if (SessionId != "None") :
+        session = SessionAssistant(IpAddress = clientIp,
+                               UserName = UserName,
+                               Password = Password,
+                               RestPort = RestPort,
+                               SessionId = SessionId)
+    else :
+        session = SessionAssistant(IpAddress = clientIp,
+                               UserName = UserName,
+                               Password = Password,
+                               RestPort = RestPort)
+
+    sessionData = session.Session
+    ixNetwork   = session.Ixnetwork
+    ixNetwork.NewConfig()
+    portMap = session.PortMapAssistant()
+
+    vPort1 = portMap.Map(chassisIp, cardId, PortId1)
+    vPort2 = portMap.Map(chassisIp, cardId, PortId2)
+    vPort3 = portMap.Map(chassisIp, cardId, PortId3)
+    vPort4 = portMap.Map(chassisIp, cardId, PortId4)
+    vPort5 = portMap.Map(chassisIp, cardId, PortId5)
+    vPort6 = portMap.Map(chassisIp, cardId, PortId6)
+    #print ('connecting to chassis %s' %(chassisIp))
+
+    t1 = time.time()
+    portMap.Connect(ChassisTimeout=1200, ForceOwnership=True)
+    t2 = time.time()
+
+    time_taken = t2 - t1
+    logger.info("time-taken to connect = %s" %(time_taken))
+
+    vPort1.L1Config.NovusHundredGigLan.IeeeL1Defaults  =  False
+    vPort1.L1Config.NovusHundredGigLan.EnableAutoNegotiation =False
+    vPort1.L1Config.NovusHundredGigLan.EnableRsFec = True
+    vPort1.L1Config.NovusHundredGigLan.EnableRsFecStats = True
+
+    vPort2.L1Config.NovusHundredGigLan.IeeeL1Defaults  =  False
+    vPort2.L1Config.NovusHundredGigLan.EnableAutoNegotiation =False
+    vPort2.L1Config.NovusHundredGigLan.EnableRsFec = True
+    vPort2.L1Config.NovusHundredGigLan.EnableRsFecStats = True
+
+    vPort3.L1Config.NovusHundredGigLan.IeeeL1Defaults  =  False
+    vPort3.L1Config.NovusHundredGigLan.EnableAutoNegotiation =False
+    vPort3.L1Config.NovusHundredGigLan.EnableRsFec = True
+    vPort3.L1Config.NovusHundredGigLan.EnableRsFecStats = True
+
+    vPort4.L1Config.NovusHundredGigLan.IeeeL1Defaults  =  False
+    vPort4.L1Config.NovusHundredGigLan.EnableAutoNegotiation =False
+    vPort4.L1Config.NovusHundredGigLan.EnableRsFec = True
+    vPort4.L1Config.NovusHundredGigLan.EnableRsFecStats = True
+
+    vPort5.L1Config.NovusHundredGigLan.IeeeL1Defaults  =  False
+    vPort5.L1Config.NovusHundredGigLan.EnableAutoNegotiation =False
+    vPort5.L1Config.NovusHundredGigLan.EnableRsFec = True
+    vPort5.L1Config.NovusHundredGigLan.EnableRsFecStats = True
+
+    vPort6.L1Config.NovusHundredGigLan.IeeeL1Defaults  =  False
+    vPort6.L1Config.NovusHundredGigLan.EnableAutoNegotiation =False
+    vPort6.L1Config.NovusHundredGigLan.EnableRsFec = True
+    vPort6.L1Config.NovusHundredGigLan.EnableRsFecStats = True
+
+    vPort1.Name = 'Tx1'
+    vPort4.Name = 'Rx1'
+    state = vPort1.State
+
+    topology1    = ixNetwork.Topology.add(Name='Topo1', Ports=[vPort1, vPort2, vPort3, vPort4, vPort5, vPort6])
+    deviceGroup1 = topology1.DeviceGroup.add(Name='DG1', Multiplier='1')
+    ethernet1    = deviceGroup1.Ethernet.add(Name='Eth1')
+    ipv4         = ethernet1.Ipv4.add(Name='Ipv4')
+
+    ipv4.GatewayIp.ValueList(['192.168.1.1','192.168.1.1','192.168.1.1','192.168.1.1', '192.168.1.1', '192.168.1.1'])
+    ipv4.Address.ValueList(['192.168.1.2','192.168.1.3','192.168.1.4','192.168.1.5', '192.168.1.6', '192.168.1.7'])
+
+    ixNetwork.StartAllProtocols()
+    time.sleep(60)
+
+    # Traffic
+    traffic_item = ixNetwork.Traffic.TrafficItem.add(Name='Traffic Test', TrafficType='ipv4')
+    dest = [{'arg1': '/api/v1/sessions/1/ixnetwork/topology/1/deviceGroup/1/ethernet/1/ipv4/1', 'arg2': 1, 'arg3': 4, 'arg4': 1 ,'arg5': 1}]
+    src  = [{'arg1': '/api/v1/sessions/1/ixnetwork/topology/1/deviceGroup/1/ethernet/1/ipv4/1', 'arg2': 1, 'arg3': 1, 'arg4': 1, 'arg5': 1}]
+    endPoint = traffic_item.EndpointSet.add()
+    traffic_item.Tracking.find().TrackBy = ['trackingenabled0']
+    endPoint.ScalableSources = src
+    endPoint.ScalableDestinations = dest
+    traffic_item.Generate()
+    ixNetwork.Traffic.Apply()
+    ixNetwork.Traffic.Start()
+    time.sleep(10)
+    logger.info(session.StatViewAssistant('Traffic Item Statistics'))
+    ixNetwork.Traffic.Stop()
+
+    assert 1