TestON/drivers/common/cli/emulator/lincoemininetdriver.py

#!/usr/bin/env python

"""
Copyright 2016 Open Networking Foundation (ONF)

Please refer questions to either the onos test mailing list at <onos-test@onosproject.org>,
the System Testing Plans and Results wiki page at <https://wiki.onosproject.org/x/voMg>,
or the System Testing Guide page at <https://wiki.onosproject.org/x/WYQg>

    TestON is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 2 of the License, or
    ( at your option ) any later version.

    TestON is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with TestON.  If not, see <http://www.gnu.org/licenses/>.


LincOEMininetDriver is an extension of the mininetclidriver to handle linc oe
"""

import pexpect
import re
import sys
import os
from drivers.common.cli.emulator.mininetclidriver import MininetCliDriver


class LincOEMininetDriver( MininetCliDriver ):

    def runOpticalMnScript( self, onosDirectory = 'onos', ctrllerIP = None, topology = 'opticalTest' ):
        import time
        import types
        """
            Description:
                This function is only meant for Packet Optical.
                It runs python script "opticalTest.py" to create the
                packet layer( mn ) and optical topology
            Optional:
                name - Name of onos directory. (ONOS | onos)
                topology - Name of optical topology to activate, defaults to opticalTest.py
            Required:
                ctrllerIP = Controller(s) IP address
            TODO: If no ctrllerIP is provided, a default
                $OC1 can be accepted
        """
        try:
            if ctrllerIP is None:
                main.log.error( "You need to specify the IP" )
                return main.FALSE
            else:
                controller = ''
                if isinstance( ctrllerIP, types.ListType ):
                    for i in xrange( len( ctrllerIP ) ):
                        controller += ctrllerIP[ i ] + ' '
                    main.log.info( "Mininet topology is being loaded with " +
                                   "controllers: " + controller )
                elif isinstance( ctrllerIP, types.StringType ):
                    controller = ctrllerIP
                    main.log.info( "Mininet topology is being loaded with " +
                                   "controller: " + controller )
                else:
                    main.log.info( "You need to specify a valid IP" )
                    return main.FALSE
                topoFile = "~/{0}/tools/test/topos/{1}.py".format( onosDirectory, topology )
                cmd = "sudo -E python {0} {1}".format( topoFile, controller )
                main.log.info( self.name + ": cmd = " + cmd )
                self.handle.sendline( cmd )
                lincStart = self.handle.expect( [ "mininet>", pexpect.TIMEOUT ], timeout=120 )
                if lincStart == 1:
                    self.handle.sendline( "\x03" )
                    self.handle.sendline( "sudo mn -c" )
                    self.handle.sendline( cmd )
                    lincStart = self.handle.expect( [ "mininet>", pexpect.TIMEOUT ], timeout=120 )
                if lincStart == 1:
                    main.log.error( "OpticalTest.py failed to start." )
                    return main.FALSE
                return main.TRUE
        except pexpect.EOF:
            main.log.error( self.name + ": EOF exception found" )
            main.log.error( self.name + ":     " + self.handle.before )
            return main.FALSE
        except Exception:
            main.log.exception( self.name + ": Uncaught exception!" )
            main.cleanAndExit()
            return main.FALSE

    def pingHostOptical( self, **pingParams ):
        """
        This function is only for Packet Optical related ping
        Use the next pingHost() function for all normal scenarios )
        Ping from one mininet host to another
        Currently the only supported Params: SRC and TARGET
        """
        args = utilities.parse_args( [ "SRC", "TARGET" ], **pingParams )
        command = args[ "SRC" ] + " ping " + \
            args[ "TARGET" ] + " -c 1 -i 1 -W 8"
        try:
            main.log.warn( "Sending: " + command )
            self.handle.sendline( command )
            i = self.handle.expect( [ command, pexpect.TIMEOUT ] )
            if i == 1:
                main.log.error(
                    self.name +
                    ": timeout when waiting for response from mininet" )
                main.log.error( "response: " + str( self.handle.before ) )
            i = self.handle.expect( [ "mininet>", pexpect.TIMEOUT ] )
            if i == 1:
                main.log.error(
                    self.name +
                    ": timeout when waiting for response from mininet" )
                main.log.error( "response: " + str( self.handle.before ) )
            response = self.handle.before
        except pexpect.EOF:
            main.log.error( self.name + ": EOF exception found" )
            main.log.error( self.name + ":     " + self.handle.before )
            main.cleanAndExit()
        except Exception:
            main.log.exception( self.name + ": Uncaught exception!" )
            main.cleanAndExit()
        main.log.info( self.name + ": Ping Response: " + response )
        if re.search( ',\s0\%\spacket\sloss', response ):
            main.log.info( self.name + ": no packets lost, host is reachable" )
            main.lastResult = main.TRUE
            return main.TRUE
        else:
            main.log.info(
                self.name +
                ": PACKET LOST, HOST IS NOT REACHABLE" )
            main.lastResult = main.FALSE
            return main.FALSE