TestON/tests/SdnIpTest/SdnIpTest.py

#Testing the basic functionality of SDN-IP

class SdnIpTest:
    def __init__(self):
        self.default = ''

    def CASE1(self, main):

        '''
        Test the SDN-IP functionality
        allRoutes_expected: all expected routes for all BGP peers
		routeIntents_expected: all expected MultiPointToSinglePointIntent intents
		bgpIntents_expected: expected PointToPointIntent intents
		allRoutes_actual: all routes from ONOS LCI
		routeIntents_actual: actual MultiPointToSinglePointIntent intents from ONOS CLI
		bgpIntents_actual: actual PointToPointIntent intents from ONOS CLI
        '''
        import time
        import json
        from operator import eq
        
        main.case("The test case is to help to setup the TestON environment and test new drivers")
        SDNIP_JSON_FILE_PATH = "../tests/SdnIpTest/sdnip.json"
        # all expected routes for all BGP peers
        allRoutes_expected = []
        main.step("Start to generate routes for BGP peer on host1")
        prefixes = main.Quaggacli.generate_prefixes(1, 3)
        main.log.info(prefixes)

        # TODO: delete the static prefix below after integration with Demo Mininet script
        prefixes = []
        prefixes.append("172.16.30.0/24")
        prefixes.append("1.0.0.0/24")
        prefixes.append("2.0.0.0/24")
        prefixes.append("3.0.0.0/24")


        for prefix in prefixes:
            # generate route with next hop
            allRoutes_expected.append(prefix + "/" + "1.1.1.1")

        routeIntents_expected = main.Quaggacli.generate_expected_onePeerRouteIntents(prefixes, "192.168.30.1", "00:00:00:00:03:01", SDNIP_JSON_FILE_PATH)

        # start to insert routes into the bgp peer
        main.step("Start Quagga-cli on host1")
        main.Quaggacli.loginQuagga("1.1.1.1")

        main.step("Start to configure Quagga on host1")
        main.Quaggacli.enter_config(64513)
    
        main.step("Start to generate and add routes for BGP peer on host1")
        routes = main.Quaggacli.generate_prefixes(8, 3)
        main.Quaggacli.add_routes(routes, 1)
             
        # add generates routes to allRoutes
        for route in routes:
            allRoutes_expected.append(route + "/" + "1.1.1.1")

        cell_name = main.params['ENV']['cellName']
        ONOS1_ip = main.params['CTRL']['ip1']

        main.step("Starting ONOS service")
        # TODO: start ONOS from Mininet Script
        # start_result = main.ONOSbench.onos_start("127.0.0.1")

        main.step("Set cell for ONOS-cli environment")
        main.ONOScli.set_cell(cell_name)
   
        main.step("Start ONOS-cli")
        # TODO: change the hardcode in start_onos_cli method in ONOS CLI driver
        time.sleep(5)
        main.ONOScli.start_onos_cli(ONOS1_ip)    

        main.step("Get devices in the network")
        list_result = main.ONOScli.devices(json_format=False)
        main.log.info(list_result)

        #get all routes inside SDN-IP
        get_routes_result = main.ONOScli.routes(json_format=True)  
                     
        # parse routes from ONOS CLI
        allRoutes_actual = main.Quaggacli.extract_actual_routes(get_routes_result)
      
        main.step("Check routes installed")
        main.log.info("Routes expected:")
        main.log.info(sorted(allRoutes_expected))
        main.log.info("Routes get from ONOS CLI:")
        main.log.info(allRoutes_actual)
        utilities.assert_equals(expect=sorted(allRoutes_expected), actual=allRoutes_actual,
                                onpass="***Routes in SDN-IP are correct!***",
                                onfail="***Routes in SDN-IP are wrong!***")

        time.sleep(2)
        get_intents_result = main.ONOScli.intents(json_format=True)


        main.step("Check MultiPointToSinglePointIntent intents installed")
        # route_intents_expected are generated when generating routes
        # get rpoute intents from ONOS CLI
        routeIntents_actual  = main.Quaggacli.extract_actual_routeIntents(get_intents_result)
        main.log.info("MultiPointToSinglePoint intents expected:")
        main.log.info(routeIntents_expected)
        main.log.info("MultiPointToSinglePoint intents get from ONOS CLI:")
        main.log.info(routeIntents_actual)
        utilities.assert_equals(expect=True, actual=eq(routeIntents_expected, routeIntents_actual),
                                onpass="***MultiPointToSinglePoint Intents in SDN-IP are correct!***",
                                onfail="***MultiPointToSinglePoint Intents in SDN-IP are wrong!***")


        main.step("Check BGP PointToPointIntent intents installed")
        # bgp intents expected
        bgpIntents_expected = main.Quaggacli.generate_expected_bgpIntents(SDNIP_JSON_FILE_PATH)
        # get BGP intents from ONOS CLI
        bgpIntents_actual = main.Quaggacli.extract_actual_bgpIntents(get_intents_result)

        bgpIntents_str_expected = str(bgpIntents_expected).replace('u',"")
        bgpIntents_str_actual = str(bgpIntents_actual)
        main.log.info("PointToPointIntent intents expected:")
        main.log.info(bgpIntents_str_expected)
        main.log.info("PointToPointIntent intents get from ONOS CLI:")
        main.log.info(bgpIntents_str_actual)

        utilities.assert_equals(expect=True, actual=eq(bgpIntents_str_expected, bgpIntents_str_actual),
                                onpass="***PointToPointIntent Intents in SDN-IP are correct!***",
                                onfail="***PointToPointIntent Intents in SDN-IP are wrong!***")

        #main.step("Test whether Mininet is started")
        #main.Mininet2.handle.sendline("xterm host1")
        #main.Mininet2.handle.expect("mininet>")