"""Script to Test the SFC scenarios in ONOS."""
# !/usr/bin/python
#
-# Copyright (c) 2015 All rights reserved
+# Copyright (c) CREATED5 All rights reserved
# This program and the accompanying materials
# are made available under the terms of the Apache License, Version 2.0
# which accompanies this distribution, and is available at
# Testcase 1 : Prerequisites configuration for SFC
# Testcase 2 : Creation of 3 VNF Nodes and Attaching Ports
-# TestCase 3 : Configure SFC [Port pair,Port Group ,Flow classifer
-# TestCase 4 : Configure Port Chain and verify the flows are added.
-# TestCase 5 : Verify traffic with VNF node.
-# TestCase 6 : Remove the Port Chain and Verify the traffic.
+# Testcase 3 : Configure SFC [Port pair,Port Group ,Flow classifer
+# Testcase 4 : Configure Port Chain and verify the flows are added.
+# Testcase 5 : Verify traffic with VNF node.
+# Testcase 6 : Remove the Port Chain and Verify the traffic.
# Testcase 7 : Cleanup
# ###########################################################################
#
+import time
import functest.utils.functest_logger as ft_logger
import functest.utils.functest_utils as functest_utils
-import time
from Sfc_fun import Sfc_fun
+logger = ft_logger.Logger("sfc").getLogger()
+Sfc_obj = Sfc_fun()
-class Sfc:
- """Script to Test the SFC scenarios in ONOS."""
- logger = ft_logger.Logger("sfc").getLogger()
- Sfc_obj = Sfc_fun()
- start_time = time.time()
- status = "PASS"
- print("################################################################")
- print(" OPNFV SFC Script ")
- print("################################################################")
+OK = 200
+CREATED = 201
+ACCEPTED = 202
+NO_CONTENT = 204
+
+
+def PreConfig():
logger.info("Testcase 1 : Prerequisites configuration for SFC")
- #########################################################################
- logger.info("\t1.1 Creation of Auth-Token")
- if (Sfc_obj.getToken() == 200):
- logger.info("\t\tCreation of Token is successfull")
- else:
- status = "FAIL"
- logger.error("\t\t : Creation of Token is NOT successfull")
- #########################################################################
- logger.info("\t1.2 Creation of Network")
- if (Sfc_obj.createNetworks() == 201):
- logger.info("\t\tCreation of network is successfull")
- else:
- status = "FAIL"
- logger.error("\t\t : Creation of network is NOT successfull")
- #########################################################################
- logger.info("\t1.3 Creation of Subnetwork")
- if (Sfc_obj.createSubnets() == 201):
- logger.info("\t\tCreation of Subnetwork is successfull")
- else:
- status = "FAIL"
- logger.error("\t\t : Creation of Subnetwork is NOT successfull")
- print ("\n###########################################################\n")
- ########################################################################
- logger.info("Testcase 2 : Creation of 3 VNF Nodes and Attaching Ports")
- #########################################################################
- logger.info("\t2.1 Creation of Ports")
- if (Sfc_obj.createPorts() == 201):
- logger.info("\t\tCreation of Port is successfull")
- else:
- status = "FAIL"
- logger.error("\t\t : Creation of Port is NOT successfull")
- #########################################################################
- logger.info("\t2.2 Creation of VM-Compute-Node")
- if (Sfc_obj.createVm() == 202):
- logger.info("\t\tCreation of VM is successfull")
- else:
- status = "FAIL"
- logger.error("\t\t : Creation of VM is NOT successfull")
- #########################################################################
- logger.info("\t2.3 Check VM Status")
- if (Sfc_obj.checkVmState() == 200):
- logger.info("\t\tVM are in active state")
- else:
- status = "FAIL"
- logger.error("\t\t : VM is NOT Active")
- #########################################################################
- logger.info("\t\t2.4 Router Creation")
- if (Sfc_obj.createRouter() == 201):
- logger.info("\t\t Router Creation is Successful")
- else:
- status = "FAIL"
- logger.error("\t\t : Router Creation is NOT Successful")
- #########################################################################
- logger.info("\t\t2.5 Attachement of Interface to VM")
- if (Sfc_obj.attachInterface() == 200):
- logger.info("\t\t Interface attached to VM")
- else:
- status = "FAIL"
- logger.error("\t\t : Interface NOT attached to VM")
- #########################################################################
- logger.info("\t\t2.6 Attachement of FLoating Ip to VM")
- if (Sfc_obj.addFloatingIp() == 202):
- logger.info("\t\t Floating Ip attached to VM SUccessful")
- else:
- status = "FAIL"
- logger.error("\t\t : Floating Ip NOT attached to VM ")
- print ("\n###########################################################\n")
- ########################################################################
- logger.info(
- "TestCase 3 : Configure SFC [Portair,PortGroup,Flow classifer]")
- #########################################################################
- logger.info("\t3.1 Creation of Port Pair")
- if (Sfc_obj.createPortPair() == 201):
- logger.info("\t\tCreation of Port pair is successful")
- else:
- status = "FAIL"
- logger.error("\t\t : Creation of Port pair is NOT successful")
+ logger.info("1.1 Creation of Auth-Token")
+ if (Sfc_obj.getToken() == OK):
+ logger.info("Creation of Token is successfull")
+ else:
+ logger.error(" : Creation of Token is NOT successfull")
+ fail()
+ logger.info("1.2 Creation of Network")
+ if (Sfc_obj.createNetworks() == CREATED):
+ logger.info("Creation of network is successfull")
+ else:
+ logger.error(" : Creation of network is NOT successfull")
+ fail()
+ logger.info("1.3 Creation of Subnetwork")
- #########################################################################
- logger.info("\t3.2 Getting the Port Pair ID")
- if (Sfc_obj.getPortPair() == 200):
- logger.info("\t\tSuccessfully got Port Pair ID")
+ if (Sfc_obj.createSubnets() == CREATED):
+ logger.info("Creation of Subnetwork is successfull")
else:
- status = "FAIL"
- logger.error("\t\t : UnSuccessfully got Port Pair ID")
+ logger.error(" : Creation of Subnetwork is NOT successfull")
+ fail()
- #########################################################################
- logger.info("\t3.3 Creation of Port Pair Group")
- if (Sfc_obj.createPortGroup() == 201):
- logger.info("\t\tPort Pair Group successfully Created")
- else:
- status = "FAIL"
- logger.error("\t\t : Port Pair Group NOT successfully Created")
- #########################################################################
- logger.info("\t3.4 Getting Port Pair Group ID ")
+def CreateNodes():
+ logger.info("Testcase 2 : Creation of 3 VNF Nodes and Attaching Ports")
+ logger.info("2.1 Creation of Ports")
+ if (Sfc_obj.createPorts() == CREATED):
+ logger.info("Creation of Port is successfull")
+ else:
+ logger.error(" : Creation of Port is NOT successfull")
+ fail()
+ logger.info("2.2 Creation of VM-Compute-Node")
+ if (Sfc_obj.createVm() == ACCEPTED):
+ logger.info("Creation of VM is successfull")
+ else:
+ logger.error(" : Creation of VM is NOT successfull")
+ fail()
+ logger.info("2.3 Check VM Status")
+ if (Sfc_obj.checkVmState() == OK):
+ logger.info("VM are in active state")
+ else:
+ logger.error(" : VM is NOT Active")
+ fail()
+ logger.info("2.4 Router Creation")
+ if (Sfc_obj.createRouter() == CREATED):
+ logger.info("Creation of Router is Successful")
+ else:
+ logger.error(" : Router Creation is NOT Successful")
+ fail()
+ logger.info("2.5 Attachement of Interface to VM")
+ if (Sfc_obj.attachInterface() == OK):
+ logger.info("Interface attached to VM")
+ else:
+ logger.error(" : Interface NOT attached to VM")
+ fail()
+ logger.info("2.6 Attachement of FLoating Ip to VM")
+ if (Sfc_obj.addFloatingIp() == ACCEPTED):
+ logger.info("Floating Ip attached to VM SUccessful")
+ else:
+ logger.error(" : Floating Ip NOT attached to VM ")
+ fail()
- if (Sfc_obj.getPortGroup() == 200):
- logger.info("\t\tPort Pair Group ID successfully received")
- else:
- status = "FAIL"
- logger.error("\t\t : Port Pair Group ID NOT successfully received")
- #########################################################################
- logger.info("\t3.5 Creation of Flow Classifier")
- if (Sfc_obj.createFlowClassifier() == 201):
- logger.info("\t\tFlow Classifier successfully Created")
- else:
- status = "FAIL"
- logger.error("\t\t : Flow Classifier NOT successfully Created")
- print ("\n###########################################################\n")
- ########################################################################
+def ConfigSfc():
logger.info(
- "TestCase 4 : Configure Port Chain and verify flows are added")
- #########################################################################
- logger.info("\t4.1 Creation of PortChain")
- if (Sfc_obj.createPortChain() == 201):
- logger.info("\t\tPortChain successfully Created")
+ "Testcase 3 : Configure SFC [Portair,PortGroup,Flow classifer]")
+ logger.info("3.1 Creation of Port Pair")
+ if (Sfc_obj.createPortPair() == CREATED):
+ logger.info("Creation of Port pair is successful")
+ else:
+ logger.error(" : Creation of Port pair is NOT successful")
+ fail()
+ logger.info("3.2 Getting the Port Pair ID")
+ if (Sfc_obj.getPortPair() == OK):
+ logger.info("Port Pair ID is successfully got")
+ else:
+ logger.error(" : UnSuccessfully got Port Pair ID")
+ fail()
+ logger.info("3.3 Creation of Port Pair Group")
+ if (Sfc_obj.createPortGroup() == CREATED):
+ logger.info("Creation of Port Pair Group is successful")
+ else:
+ logger.error(" : Creation of Port Pair Group is NOT successful")
+ fail()
+ logger.info("3.4 Getting Port Pair Group ID ")
+ if (Sfc_obj.getPortGroup() == OK):
+ logger.info("Port Pair Group ID is successfully received")
+ else:
+ logger.error(" : Port Pair Group ID is NOT successfully got")
+ fail()
+ logger.info("3.5 Creation of Flow Classifier")
+ if (Sfc_obj.createFlowClassifier() == CREATED):
+ logger.info("Creation of Flow Classifier is successful")
+ else:
+ logger.error(" : Creation of Flow Classifier is NOT successful")
+ fail()
+ logger.info(
+ "Testcase 4 : Configure Port Chain and verify flows are added")
+ logger.info("4.1 Creation of Port Chain")
+ if (Sfc_obj.createPortChain() == CREATED):
+ logger.info("Creation of Port Chain is successful")
else:
- status = "FAIL"
- logger.error("\t\tPortChain NOT successfully Created")
- print ("\n###########################################################\n")
- #########################################################################
- logger.info("\tTestCase 5 : Verify traffic with VNF node.")
+ logger.error("Creation of Port Chain is NOT successful")
+
+
+def VerifySfcTraffic():
+ status = "PASS"
+ logger.info("Testcase 5 : Verify traffic with VNF node.")
if (Sfc_obj.loginToVM() == "1"):
- logger.info("\t\tSFC function Working")
+ logger.info("SFC function Working")
else:
+ logger.error(" : SFC function not working")
status = "FAIL"
- logger.error("\t\t : SFC function not working")
- print ("\n###########################################################\n")
- #########################################################################
- logger.info("TestCase 6 : Remove the Port Chain and Verify the traffic")
- if (Sfc_obj.deletePortChain() == 204):
+
+ logger.info("Testcase 6 : Remove the Port Chain and Verify the traffic")
+ if (Sfc_obj.deletePortChain() == NO_CONTENT):
if (Sfc_obj.loginToVM() == "0"):
- logger.info("\t\tSFC function is removed Successfully")
+ logger.info("SFC function is removed Successfully")
else:
+ logger.error(":SFC function not Removed. Have some problem")
status = "FAIL"
- logger.error("\t\t:SFC function not Removed.Have some problem")
- if (Sfc_obj.deleteFlowClassifier() == 204):
- if (Sfc_obj.deletePortGroup() == 204):
- if (Sfc_obj.deletePortPair() == 204):
+ if (Sfc_obj.deleteFlowClassifier() == NO_CONTENT):
+ if (Sfc_obj.deletePortGroup() == NO_CONTENT):
+ if (Sfc_obj.deletePortPair() == NO_CONTENT):
logger.info(
- "\t\tSFC configuration is deleted successfully")
+ "SFC configuration is deleted successfully")
else:
- status = "FAIL"
- logger.error("\t\t : Port pair configuration is NOT\
+ logger.error(" : Port pair configuration is NOT\
deleted successfully")
+ status = "FAIL"
else:
- status = "FAIL"
- logger.error("\t\t : Port Group configuration is NOT \
+ logger.error(" : Port Group configuration is NOT \
deleted successfully")
- else:
status = "FAIL"
- logger.error("\t\t : Flow classifier configuration is NOT \
+ else:
+ logger.error(" : Flow classifier configuration is NOT \
deleted successfully")
+ status = "FAIL"
else:
- status = "FAIL"
- logger.error("\t\t:PortChain configuration is NOT deleted \
+ logger.error(":PortChain configuration is NOT deleted \
successfully")
- print ("\n###########################################################n")
- #######################################################################
+ status = "FAIL"
+ if (status == "FAIL"):
+ fail()
+
+
+def CleanUp():
logger.info("Testcase 7 : Cleanup")
- if (Sfc_obj.cleanup() == 204):
- logger.info("\t\tCleanUp is successfull")
+ if (Sfc_obj.cleanup() == NO_CONTENT):
+ logger.info("CleanUp is successfull")
else:
- status = "FAIL"
- logger.error("\t\t : CleanUp is NOT successfull")
- print ("###############################################################")
+ logger.error(" : CleanUp is NOT successfull")
+
+
+def fail():
+ CleanUp()
+ exit(-1)
+
+
+def main():
+ """Script to Test the SFC scenarios in ONOS."""
+ start_time = time.time()
+ PreConfig()
+ CreateNodes()
+ ConfigSfc()
+ VerifySfcTraffic()
+ CleanUp()
+ status = "PASS"
logger.info("Summary :")
try:
logger.debug("Push ONOS SFC results into DB")
except:
logger.error("Error pushing results into Database")
- if status == "FAIL":
- EXIT_CODE = -1
- exit(EXIT_CODE)
-
- print("############################END OF SCRIPT ######################")
+if __name__ == '__main__':
+ main()
-"""SFC functions."""
+import os
+import re
+import time
import json
+import requests
+
from multiprocessing import Process
from multiprocessing import Queue
-import os
-import re
-import time
+from pexpect import pxssh
import functest.utils.functest_logger as ft_logger
-from pexpect import pxssh
-import requests
+
+OK = 200
+CREATED = 201
+ACCEPTED = 202
+NO_CONTENT = 204
class Sfc_fun:
url = 'http://' + self.keystone_hostname + \
':5000/' + self.osver + '/tokens'
data = '{"auth": {"tenantName": "admin", "passwordCredentials":\
- { "username": "admin", "password": "console"}}}'
+ { "username": "admin", "password": "console"}}}'
headers = {"Accept": "application/json"}
response = requests.post(url, headers=headers, data=data)
- if (response.status_code == 200):
+ if (response.status_code == OK):
json1_data = json.loads(response.content)
self.logger.debug(response.status_code)
self.logger.debug(response.content)
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
response = requests.post(url, headers=headers, data=data)
- if (response.status_code == 201):
+ if (response.status_code == CREATED):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
"X-Auth-Token": self.token_id}
response = requests.post(url, headers=headers, data=data)
- if (response.status_code == 201):
+ if (response.status_code == CREATED):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
json1_data = json.loads(response.content)
"X-Auth-Token": self.token_id}
response = requests.post(url, headers=headers, data=data)
- if (response.status_code == 201):
+ if (response.status_code == CREATED):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
headers = {"Accept": "application/json", "Content-Type": "application/\
octet-stream", "X-Auth-Token": self.token_id}
response = requests.get(url, headers=headers)
- if (response.status_code == 200):
+ if (response.status_code == OK):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
- self.logger.info("\tFireWall Image is available")
+ self.logger.info("FireWall Image is available")
json1_data = json.loads(response.content)
self.logger.debug(json1_data)
self.image_id = json1_data['images'][0]['id']
"application/json", "X-Auth-Token": self.token_id}
response = requests.get(url, headers=headers)
- if (response.status_code == 200):
+ if (response.status_code == OK):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
- self.logger.info("\tFlavor is available")
+ self.logger.info("Flavor is available")
json1_data = json.loads(response.content)
self.logger.debug(json1_data)
self.flavorRef = json1_data['flavors'][0]['id']
headers = {"Accept": "application/json", "Content-Type":
"application/json", "X-Auth-Token": self.token_id}
response = requests.post(url, headers=headers, data=data)
- if (response.status_code == 202):
+ if (response.status_code == ACCEPTED):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
- self.logger.info("\tCreation of VM is successfull")
+ info = "Creation of VM" + str(y) + " is successfull"
+ self.logger.debug(info)
json1_data = json.loads(response.content)
self.logger.debug(json1_data)
headers = {"Accept": "application/json", "X-Auth-Token":
self.token_id}
response = requests.get(url, headers=headers)
- if (response.status_code == 200):
+ if (response.status_code == OK):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
json1_data = json.loads(response.content)
self.logger.debug(json1_data)
self.vm_active = json1_data['servers'][0]['status']
if (self.vm_active == "ACTIVE"):
- print ("\t\t\t\t\t\tVM" + str(y) + " is Active : " +
- self.vm_active)
+ info = "VM" + str(y) + \
+ " is Active : " + self.vm_active
else:
- print ("\t\t\t\t\t\tVM" + str(y) + " is NOT Active : " +
- self.vm_active)
+ info = "VM" + str(y) + " is NOT Active : " + \
+ self.vm_active
+ self.logger.debug(info)
else:
return(response.status_code)
return(response.status_code)
headers = {"Accept": "application/json", "X-Auth-Token":
self.token_id}
response = requests.post(url, headers=headers, data=data)
- if (response.status_code == 201):
- print ("\t\t\t\tCreation of Port Pair PP" + str(p) +
- " is successful")
+ if (response.status_code == CREATED):
+ info = "Creation of Port Pair PP" + str(p) + \
+ " is successful"
+ self.logger.debug(info)
else:
return(response.status_code)
self.token_id}
response = requests.get(url, headers=headers)
- if (response.status_code == 200):
+ if (response.status_code == OK):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
json1_data = json.loads(response.content)
headers = {"Accept": "application/json", "X-Auth-Token":
self.token_id}
response = requests.post(url, headers=headers, data=data)
- if (response.status_code == 201):
- print ("\t\t\t\tCreation of Port Group PG" + str(p) +
- "is successful")
+ if (response.status_code == CREATED):
+ info = "Creation of Port Group PG" + str(p) + \
+ "is successful"
+ self.logger.debug(info)
else:
return(response.status_code)
self.token_id}
response = requests.get(url, headers=headers)
- if (response.status_code == 200):
+ if (response.status_code == OK):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
json1_data = json.loads(response.content)
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
response = requests.post(url, headers=headers, data=data)
- if (response.status_code == 201):
+ if (response.status_code == CREATED):
json1_data = json.loads(response.content)
self.flow_class_if = json1_data['flow_classifier']['id']
- self.logger.info("\tCreation of Flow Classifier is successful")
+ self.logger.debug("Creation of Flow Classifier is successful")
return(response.status_code)
else:
return(response.status_code)
"Content-Type": "application/json",
"X-Auth-Token": self.token_id}
response = requests.post(url, headers=headers, data=data)
- if (response.status_code == 201):
- self.logger.info("\tCreation of PORT CHAIN is successful")
+ if (response.status_code == CREATED):
+ self.logger.debug("Creation of PORT CHAIN is successful")
json1_data = json.loads(response.content)
self.PC_id = json1_data['port_chain']['id']
return(response.status_code)
response = requests.get('http://' + self.onos_hostname +
':8181/onos/v1/flows',
auth=("karaf", "karaf"))
- if (response.status_code == 200):
- self.logger.info("\tFlow is successfully Queries")
+ if (response.status_code == OK):
+ self.logger.debug("Flow is successfully Queries")
json1_data = json.loads(response.content)
self.flowadd = json1_data['flows'][0]['state']
if (self.flowadd == "ADDED"):
- self.logger.info("\tFlow is successfully added to OVS")
+ self.logger.info("Flow is successfully added to OVS")
return(response.status_code)
else:
return(404)
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
response = requests.post(url, headers=headers, data=data)
- if (response.status_code == 201):
+ if (response.status_code == CREATED):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
- self.logger.info("\tCreation of Router is successfull")
+ self.logger.debug("Creation of Router is successfull")
json1_data = json.loads(response.content)
self.logger.debug(json1_data)
self.router_id = json1_data['router']['id']
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
response = requests.get(url, headers=headers)
- if (response.status_code == 200):
+ if (response.status_code == OK):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
json1_data = json.loads(response.content)
return(response.status_code)
############################################################
- self.logger.info("\tAttachment of Instance interface to Router")
+ self.logger.info("Attachment of Instance interface to Router")
Dicdata = {}
if self.subnetId != '':
Dicdata['subnet_id'] = self.subnetId
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
response = requests.put(url, headers=headers, data=data)
- if (response.status_code == 200):
+ if (response.status_code == OK):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
- self.logger.info("\tInterface attached successfull")
+ self.logger.info("Interface attached successfull")
else:
return(response.status_code)
############################################################
- self.logger.info("\tAttachment of Gateway to Router")
+ self.logger.info("Attachment of Gateway to Router")
Dicdata1 = {}
if self.pub_net_id != 0:
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
response = requests.put(url, headers=headers, data=data)
- if (response.status_code == 200):
+ if (response.status_code == OK):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
- self.logger.info("\tGateway Interface attached successfull")
+ self.logger.info("Gateway Interface attached successfull")
return(response.status_code)
else:
return(response.status_code)
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
response = requests.post(url, headers=headers, data=data)
- if (response.status_code == 200):
+ if (response.status_code == OK):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
- self.logger.info("\tFloating ip created successfully")
+ self.logger.info("Floating ip created successfully")
json1_data = json.loads(response.content)
self.logger.debug(json1_data)
self.vm_public_ip.append(json1_data['floating_ip']['ip'])
self.vm_public_id.append(json1_data['floating_ip']['id'])
else:
- self.logger.error("\tFloating ip NOT created successfully")
+ self.logger.error("Floating ip NOT created successfully")
Dicdata1 = {}
if self.address != '':
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
response = requests.post(url, headers=headers, data=data)
- if(response.status_code == 202):
+ if(response.status_code == ACCEPTED):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
- self.logger.info("\tPublic Ip successfully added to VM")
+ self.logger.info("Public Ip successfully added to VM")
else:
return(response.status_code)
return(response.status_code)
ping_re = re.search("transmitted.*received", s.before).group()
x = re.split('\s+', ping_re)
if (x[1] >= "1"):
- self.logger.info("\tPing is Successfull")
+ self.logger.info("Ping is Successfull")
else:
- self.logger.info("\tPing is NOT Successfull")
+ self.logger.info("Ping is NOT Successfull")
def vm1(queue1):
s = pxssh.pxssh()
output_pack = s.before
if(output_pack.find("nshc") != -1):
- self.logger.info("\tThe packet has reached VM2 Instance")
+ self.logger.info("The packet has reached VM2 Instance")
queue1.put("1")
else:
- self.logger.info("\tPacket not received in Instance")
+ self.logger.info("Packet not received in Instance")
queue1.put("0")
def ping(ip, timeout=300):
p1.join(10)
return (queue1.get())
else:
- print("Thread didnt run")
+ self.logger.error("Thread didnt run")
"""##################################################################"""
""" ######################## Stats Functions ################# #####"""
response = requests.get('http://' + self.onos_hostname +
':8181/onos/vtn/portChainDeviceMap/' +
self.PC_id, auth=("karaf", "karaf"))
- if (response.status_code == 200):
- self.logger.info("\tPortChainDeviceMap is successfully Queries")
+ if (response.status_code == OK):
+ self.logger.info("PortChainDeviceMap is successfully Queries")
return(response.status_code)
else:
return(response.status_code)
response = requests.get('http://' + self.onos_hostname +
':8181/onos/vtn/portChainSfMap/' +
self.PC_id, auth=("karaf", "karaf"))
- if (response.status_code == 200):
- self.logger.info("\tportChainSfMap is successfully Queries")
+ if (response.status_code == OK):
+ self.logger.info("portChainSfMap is successfully Queries")
return(response.status_code)
else:
return(response.status_code)
headers = {"Accept": "application/json", "Content-Type":
"application/json", "X-Auth-Token": self.token_id}
response = requests.delete(url, headers=headers)
- if (response.status_code == 200):
+ if (response.status_code == OK):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
return(response.status_code)
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
response = requests.delete(url, headers=headers)
- if (response.status_code == 200):
+ if (response.status_code == OK):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
return(response.status_code)
headers = {"Accept": "application/json", "X-Auth-Token":
self.token_id}
response = requests.delete(url, headers=headers)
- if (response.status_code == 204):
+ if (response.status_code == NO_CONTENT):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
- print ("\n\t\tPort " + self.port_grp_id[p] + "Deleted")
else:
return(response.status_code)
return(response.status_code)
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
response = requests.delete(url, headers=headers)
- if (response.status_code == 204):
+ if (response.status_code == NO_CONTENT):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
else:
def cleanup(self):
"""Cleanup."""
- print ("\n\t\tDeleting the VMs")
+ self.logger.info("Deleting VMs")
for y in range(0, 3):
url = 'http://' + self.nova_hostname + \
':8774/v2.1/servers/' + self.vm[y]
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
response = requests.delete(url, headers=headers)
- if (response.status_code == 204):
+ if (response.status_code == NO_CONTENT):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
- print ("\n\t\tVM" + str(y) + " is Deleted : ")
+ self.logger.debug("VM" + str(y) + " is Deleted : ")
time.sleep(10)
else:
return(response.status_code)
- print ("\n\t\tDeletion of Ports")
+ self.logger.info("Deleting Ports")
for x in range(self.i, self.numTerms):
url = 'http://' + self.neutron_hostname + ':9696/' + self.osver + \
'/ports/' + self.port_num[x]
self.token_id}
response = requests.delete(url, headers=headers)
- if (response.status_code == 204):
+ if (response.status_code == NO_CONTENT):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
- print ("\n\t\tPort" + str(x) + " Deleted")
+ self.logger.debug("Port" + str(x) + " Deleted")
else:
return(response.status_code)
- print ("\n\t\tDeleting Router")
+ self.logger.info("Deleting Router")
Dicdata = {}
Dicdata['external_gateway_info'] = {}
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
response = requests.put(url, headers=headers, data=data)
- if (response.status_code == 200):
+ if (response.status_code == OK):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
Dicdata1 = {}
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
response = requests.put(url, headers=headers, data=data)
- if (response.status_code == 200):
+ if (response.status_code == OK):
url = ('http://' + self.neutron_hostname + ':9696/' +
self.osver + '/routers/' + self.router_id)
headers = {"Accept": "application/json", "X-Auth-Token":
self.token_id}
response = requests.delete(url, headers=headers)
- if (response.status_code == 204):
+ if (response.status_code == NO_CONTENT):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
- print ("\n\t\tDeletion of Router is successfull")
else:
return(response.status_code)
else:
else:
return(response.status_code)
- print ("\n\t\tDeletion of Network")
+ self.logger.info("Deleting Network")
url = 'http://' + self.neutron_hostname + ':9696/' + self.osver + \
'/networks/' + self.net_id
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
response = requests.delete(url, headers=headers)
- if (response.status_code == 204):
+ if (response.status_code == NO_CONTENT):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
- print ("\n\t\tNetwork deleted Successfully")
else:
return(response.status_code)
- print ("\n\t\tDeletion of Floating ip")
+ self.logger.info("Deleting Floating ip")
for ip_num in range(0, 2):
url = 'http://' + self.neutron_hostname + ':9696/' + self.osver + \
'/floatingips/' + self.vm_public_id[ip_num]
headers = {"Accept": "application/json", "X-Auth-Token":
self.token_id}
response = requests.delete(url, headers=headers)
- if (response.status_code == 204):
+ if (response.status_code == NO_CONTENT):
self.logger.debug(response.status_code)
self.logger.debug(response.content)
else:
Code Review / functest-xtesting.git / commitdiff