[ixia/Keysight] sample test cases and fixtures

tests/common/devices.py

@@ -852,6 +852,54 @@ def exec_template(self, ansible_root, ansible_playbook, inventory, **kwargs):
             raise Exception("Unable to execute template\n{}".format(res["localhost"]["stdout"]))
 
 
+class IxiaHost (AnsibleHostBase):
+    """ This class is a place-holder for running ansible module on Ixia
+    fanout devices in future (TBD).
+    """
+    def __init__ (self, ansible_adhoc, os, hostname, device_type) :
+        """ Initializing Ixia fanout host for using ansible modules.
+
+        Note: Right now, it is just a place holder.
+
+        Args:
+            ansible_adhoc :The pytest-ansible fixture
+            os (str): The os type of Ixia Fanout.
+            hostname (str): The Ixia fanout host-name
+            device_type (str): The Ixia fanout device type.
+        """ 
+
+        self.ansible_adhoc = ansible_adhoc
+        self.os            = os
+        self.hostname      = hostname
+        self.device_type   = device_type
+        super().__init__(IxiaHost, self)
+
+    def get_host_name (self):
+        """Returns the Ixia hostname
+
+        Args:
+            This function takes no argument.
+        """    
+        return self.hostname
+
+    def get_os (self) :
+        """Returns the os type of the ixia device.
+
+        Args:
+            This function takes no argument.
+        """    
+        return self.os
+
+    def execute (self, cmd) :
+        """Execute a given command on ixia fanout host.
+
+        Args: 
+           cmd (str): Command to be executed.
+        """ 
+        if (self.os == 'ixia') :
+            eval(cmd)
+
+
 class FanoutHost():
     """
     @summary: Class for Fanout switch
@@ -873,7 +921,7 @@ def __init__(self, ansible_adhoc, os, hostname, device_type, user, passwd, shell
         elif os == 'ixia':
             # TODO: add ixia chassis abstraction
             self.os = os
-            self.host = None
+            self.host = IxiaHost(ansible_adhoc, os, hostname, device_type)
         else:
             # Use eos host if the os type is unknown
             self.os = 'eos'

tests/common/ixia/__init__.py

@@ -0,0 +1 @@
+# Place for Ixia fixtures.

tests/common/ixia/common_helpers.py

@@ -0,0 +1,96 @@
+"""This module contains some auxiliary functions that are required 
+to support automation activities. These functions are used for various 
+secondary activities like convert the ansible Unicode STDOUT output 
+to string, get IP address in a subnet, increment an IP address, get 
+VLAN subnet etc.
+
+This file is also a placeholder for auxiliary function that are
+required for supporting automation with Ixia devices in future:
+like collecting diagnostics, uploading and downloading files 
+to/from API server, processing the statistics after obtaining them
+in .csv format etc. 
+"""
+
+import ipaddr
+from netaddr import IPNetwork
+
+def increment_ip_address (ip, incr=1) :
+    """
+    Increment IP address by an integer number.
+
+    Args: 
+       ip (str): IP address in string format.
+       incr (int): Increment by the specified number.
+
+    Return:
+       IP address in the argument incremented by the given integer.
+    """
+    ipaddress = ipaddr.IPv4Address(ip)
+    ipaddress = ipaddress + incr
+    return_value = ipaddress._string_from_ip_int(ipaddress._ip)
+    return(return_value)
+
+
+def ansible_stdout_to_str(ansible_stdout):
+    """
+    The stdout of Ansible host is essentially a list of unicode characters.
+    This function converts it to a string.
+
+    Args: 
+        ansible_stdout: stdout of Ansible
+
+    Returns:
+        Return a string
+    """
+    result = ""
+    for x in ansible_stdout:
+        result += x.encode('UTF8')
+    return result
+
+
+def get_vlan_subnet(host_ans):
+    """
+    Get VLAN subnet of a T0 device
+
+    Args:
+        host_ans: Ansible host instance of the device
+
+    Returns:
+        VLAN subnet, e.g., "192.168.1.1/24" where 192.168.1.1 is gateway 
+        and 24 is prefix length
+    """    
+    mg_facts = host_ans.minigraph_facts(host=host_ans.hostname)['ansible_facts']
+    mg_vlans = mg_facts['minigraph_vlans']
+
+    if len(mg_vlans) != 1:
+        print 'There should be only one Vlan at the DUT'
+        return None
+
+    mg_vlan_intfs = mg_facts['minigraph_vlan_interfaces']
+    prefix_len = mg_vlan_intfs[0]['prefixlen']
+    gw_addr = ansible_stdout_to_str(mg_vlan_intfs[0]['addr'])
+    return gw_addr + '/' + str(prefix_len)
+
+
+def get_addrs_in_subnet(subnet, number_of_ip):
+    """
+    Get N IP addresses in a subnet.
+
+    Args:
+        subnet (str): IPv4 subnet, e.g., '192.168.1.1/24'
+        number_of_ip (int): Number of IP addresses to get
+
+    Return:
+        Return n IPv4 addresses in this subnet in a list.
+    """
+    ip_addr = subnet.split('/')[0]
+    ip_addrs = [str(x) for x in list(IPNetwork(subnet))]
+    ip_addrs.remove(ip_addr)
+
+    """ Try to avoid network and broadcast addresses """
+    if len(ip_addrs) >= number_of_ip + 2:
+        del ip_addrs[0]
+        del ip_addrs[-1]
+
+    return ip_addrs[:number_of_ip]
+

tests/common/ixia/ixia_fixtures.py

@@ -0,0 +1,142 @@
+"""
+This module contains the necessary fixtures for running test cases with
+Ixia devices and IxNetwork. If more fixtures are required, they should be 
+included in this file.
+"""
+
+import pytest
+from ixnetwork_restpy import SessionAssistant
+
+@pytest.fixture(scope = "module")
+def ixia_api_serv_ip(testbed):
+    """ 
+    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.
+
+    Args: 
+       testbed (pytest fixture): The testbed fixture.
+
+    Returns:
+        Ixia API server IP
+    """
+    return testbed['ptf_ip']
+
+
+@pytest.fixture(scope = "module")
+def ixia_api_serv_user(duthost):
+    """
+    Return the username of Ixia API server.
+
+    Args:
+        duthost (pytest fixture): The duthost fixture.
+
+    Returns:
+        Ixia API server username.
+    """
+    return duthost.host.options['variable_manager']._hostvars[duthost.hostname]['secret_group_vars']['ixia_api_server']['user']
+
+
+@pytest.fixture(scope = "module")
+def ixia_api_serv_passwd(duthost):
+    """
+    Return the password of Ixia API server.
+
+    Args:
+        duthost (pytest fixture): The duthost fixture.
+
+    Returns:
+        Ixia API server password.
+    """
+    return duthost.host.options['variable_manager']._hostvars[duthost.hostname]['secret_group_vars']['ixia_api_server']['password']
+
+
+@pytest.fixture(scope = "module")
+def ixia_api_serv_port(duthost):
+    """
+    This fixture returns the TCP port for REST API of the ixia API server.
+
+    Args:
+        duthost (pytest fixture): The duthost fixture.
+
+    Returns:
+        Ixia API server REST port.
+    """
+    return duthost.host.options['variable_manager']._hostvars[duthost.hostname]['secret_group_vars']['ixia_api_server']['rest_port']
+
+
+@pytest.fixture(scope = "module")
+def ixia_api_serv_session_id(duthost):
+    """
+    Ixia API server can spawn multiple session on the same REST port.
+    Optional for LINUX, required for windows return the session ID.
+
+    Args:
+        duthost (pytest fixture): The duthost fixture.
+
+    Returns:
+        Ixia API server session id.
+    """
+    return duthost.host.options['variable_manager']._hostvars[duthost.hostname]['secret_group_vars']['ixia_api_server']['session_id']
+
+
+@pytest.fixture(scope = "module")
+def ixia_dev(duthost, fanouthosts):
+    """
+    Returns the Ixia chassis IP. This fixture can return multiple IPs if 
+    multiple Ixia chassis are present in the test topology.
+
+    Args:
+        duthost (pytest fixture): The duthost fixture. 
+        fanouthosts (pytest fixture): The fanouthosts fixture.
+
+    Returns:
+        Dictionary of Ixia Chassis IP/IPs.
+    """
+    result = dict()
+    ixia_dev_hostnames = fanouthosts.keys()
+    for hostname in ixia_dev_hostnames:
+        result[hostname] = duthost.host.options['inventory_manager'].get_host(hostname).get_vars()['ansible_host']
+    return result
+
+
+@pytest.fixture(scope = "function")
+def ixia_api_server_session(
+        ixia_api_serv_ip,
+        ixia_api_serv_user,
+        ixia_api_serv_passwd,
+        ixia_api_serv_port,
+        ixia_api_serv_session_id) :
+    """
+    Ixia session manager fixture.
+
+    Args:
+        ixia_api_serv_ip (pytest fixture): ixia_api_serv_ip fixture
+        ixia_api_serv_user (pytest fixture): ixia_api_serv_user fixture.
+        ixia_api_serv_passwd (pytest fixture): ixia_api_serv_passwd fixture.
+        ixia_api_serv_port (pytest fixture): ixia_api_serv_port fixture.
+        ixia_api_serv_session_id (pytest fixture): ixia_api_serv_session_id 
+            fixture.
+
+    Returns:
+        IxNetwork Session
+    """
+
+    if (ixia_api_serv_session_id.lower() != 'none') :
+        session = SessionAssistant(IpAddress=ixia_api_serv_ip,
+                                   UserName=ixia_api_serv_user,
+                                   Password=ixia_api_serv_passwd,
+                                   RestPort=ixia_api_serv_port,
+                                   SessionId=ixia_api_serv_session_id)
+    else :
+        session = SessionAssistant(IpAddress=ixia_api_serv_ip,
+                                   UserName=ixia_api_serv_user,
+                                   Password=ixia_api_serv_passwd,
+                                   RestPort=ixia_api_serv_port)
+    ixNetwork = session.Ixnetwork
+    ixNetwork.NewConfig()
+
+    yield session
+
+    ixNetwork.NewConfig()
+    session.Session.remove()