# 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 : Verify the Service Chain Statistics
-# TestCase 7 : Remove the Port Chain and Verify the traffic.
-# Testcase 8 : Cleanup
+# TestCase 6 : Remove the Port Chain and Verify the traffic.
+# Testcase 7 : Cleanup
# ###########################################################################
#
import functest.utils.functest_logger as ft_logger
logger.info("\t\tPortChain successfully Created")
else:
logger.error("\t\tPortChain NOT successfully Created")
+ print ("\n###########################################################\n")
#########################################################################
- logger.info("\t4.2 Verification of Flows Installed in OVS")
- if (Sfc_obj.checkFlowAdded() == 200):
- logger.info("\t\tFlows Installed successfully ")
- else:
- logger.error("\t\t : Flows NOT Installed successfully")
-
- if (Sfc_obj.portChainDeviceMap() == 200):
- logger.info("\t\t portChainDeviceMap successfully ")
- else:
- logger.error("\t\t:portChainDeviceMap NOT Installed successfully")
-
- if (Sfc_obj.portChainSfMap() == 200):
- logger.info("\t\tportChainSfMap successfully ")
- else:
- logger.error("\t\t : portChainSfMap NOT Installed successfully")
- print ("\n############################################################n")
- ########################################################################
logger.info("\tTestCase 5 : Verify traffic with VNF node.")
if (Sfc_obj.loginToVM() == "1"):
logger.info("\t\tSFC function Working")
else:
logger.error("\t\t : SFC function not working")
print ("\n###########################################################\n")
- logger.info("TestCase 6 : Verify the Service Chain Statistics")
- if (Sfc_obj.portChainDeviceMap() == 200):
- logger.info("\t\tportChainDeviceMap successfully ")
- else:
- logger.error("\t\t:portChainDeviceMap NOT Installed successfully")
-
- if (Sfc_obj.portChainSfMap() == 200):
- logger.info("\t\tportChainSfMap successfully ")
- else:
- logger.error("\t\t : portChainSfMap NOT Installed successfully")
- print ("\n##########################################################\n")
- logger.info("TestCase 7 : Remove the Port Chain and Verify the traffic")
+ #########################################################################
+ logger.info("TestCase 6 : Remove the Port Chain and Verify the traffic")
if (Sfc_obj.deletePortChain() == 204):
if (Sfc_obj.loginToVM() == "0"):
logger.info("\t\tSFC function is removed Successfully")
successfully")
print ("\n###########################################################n")
#######################################################################
- logger.info("Testcase 8 : Cleanup")
+ logger.info("Testcase 7 : Cleanup")
if (Sfc_obj.cleanup() == 204):
logger.info("\t\tCleanUp is successfull")
else:
self.token_id = 0
self.net_id = 0
self.image_id = 0
- self.os_hostname = 'openstack_ip'
+ self.keystone_hostname = 'keystone_ip'
+ self.neutron_hostname = 'neutron_ip'
+ self.nova_hostname = 'nova_ip'
+ self.glance_hostname = 'glance_ip'
self.onos_hostname = 'onos_ip'
# Network variables #######
self.netname = "test_nw"
def getToken(self):
"""Get the keystone token value from Openstack ."""
- url = 'http://'+self.os_hostname+':5000/'+self.osver+'/tokens'
+ url = 'http://'+self.keystone_hostname+':5000/'+self.osver+'/tokens'
data = '{"auth": {"tenantName": "admin", "passwordCredentials":\
- { "username": "admin", "password": "admin"}}}'
+ { "username": "admin", "password": "console"}}}'
headers = {"Accept": "application/json"}
response = requests.post(url, headers=headers, data=data)
if (response.status_code == 200):
Dicdata['admin_state_up'] = self.admin_state_up
Dicdata = {'network': Dicdata}
data = json.dumps(Dicdata, indent=4)
- url = 'http://'+self.os_hostname+':9696/'+self.osver+'/networks'
+ url = 'http://'+self.neutron_hostname+':9696/'+self.osver+'/networks'
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
response = requests.post(url, headers=headers, data=data)
Dicdata = {'subnet': Dicdata}
data = json.dumps(Dicdata, indent=4)
- url = 'http://'+self.os_hostname+':9696/'+self.osver+'/subnets'
+ url = 'http://'+self.neutron_hostname+':9696/'+self.osver+'/subnets'
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
response = requests.post(url, headers=headers, data=data)
Dicdata['admin_state_up'] = self.admin_state_up
if self.security_groups != '':
Dicdata['security_groups'] = self.security_groups
- if self.port_security_enabled != '':
- Dicdata['port_security_enabled'] = self.port_security_enabled
+ # if self.port_security_enabled != '':
+ # Dicdata['port_security_enabled'] = self.port_security_enabled
Dicdata = {'port': Dicdata}
data = json.dumps(Dicdata, indent=4)
- url = 'http://'+self.os_hostname+':9696/'+self.osver+'/ports'
+ url = 'http://'+self.neutron_hostname+':9696/'+self.osver+'/ports'
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
response = requests.post(url, headers=headers, data=data)
def createVm(self):
"""Creation of Instance, using firewall image."""
- url = 'http://'+self.os_hostname+':9292/v2/images?name=TestSfcVm'
+ url = 'http://'+self.glance_hostname+':9292/v2/images?name=TestSfcVm'
headers = {"Accept": "application/json", "Content-Type": "application/\
octet-stream", "X-Auth-Token": self.token_id}
response = requests.get(url, headers=headers)
Dicdata = {'server': Dicdata}
data = json.dumps(Dicdata, indent=4)
- url = 'http://'+self.os_hostname+':8774/v2.1/'+self.tenant_id + \
+ url = 'http://'+self.nova_hostname+':8774/v2.1/'+self.tenant_id + \
'/servers'
headers = {"Accept": "application/json", "Content-Type":
"application/json", "X-Auth-Token": self.token_id}
"""Checking the Status of the Instance."""
time.sleep(10)
for y in range(0, 3):
- url = 'http://'+self.os_hostname + \
- ':8774/v2.1/servers/detail?name=vm'+str(y)
+ url = 'http://' + \
+ self.nova_hostname+':8774/v2.1/servers/detail?name=vm'+str(y)
headers = {"Accept": "application/json", "X-Auth-Token":
self.token_id}
response = requests.get(url, headers=headers)
Dicdata = {'port_pair': Dicdata}
data = json.dumps(Dicdata, indent=4)
- url = 'http://'+self.os_hostname+':9696/'+self.osver + \
+ url = 'http://'+self.neutron_hostname+':9696/'+self.osver + \
'/sfc/port_pairs'
headers = {"Accept": "application/json", "X-Auth-Token":
self.token_id}
def getPortPair(self):
"""Query the Portpair id value."""
for p in range(0, 1):
- url = 'http://'+self.os_hostname+':9696/'+self.osver + \
+ url = 'http://'+self.neutron_hostname+':9696/'+self.osver + \
'/sfc/port_pairs?name=PP1'
headers = {"Accept": "application/json", "X-Auth-Token":
self.token_id}
Dicdata = {'port_pair_group': Dicdata}
data = json.dumps(Dicdata, indent=4)
- url = 'http://'+self.os_hostname+':9696/'+self.osver + \
+ url = 'http://'+self.neutron_hostname+':9696/'+self.osver + \
'/sfc/port_pair_groups'
headers = {"Accept": "application/json", "X-Auth-Token":
self.token_id}
def getPortGroup(self):
"""Query the PortGroup id."""
for p in range(0, 1):
- url = 'http://'+self.os_hostname+':9696/'+self.osver + \
+ url = 'http://'+self.neutron_hostname+':9696/'+self.osver + \
'/sfc/port_pair_groups?name=PG'+str(p)
headers = {"Accept": "application/json", "X-Auth-Token":
self.token_id}
Dicdata = {'flow_classifier': Dicdata}
data = json.dumps(Dicdata, indent=4)
- url = 'http://'+self.os_hostname+':9696/'+self.osver + \
+ url = 'http://'+self.neutron_hostname+':9696/'+self.osver + \
'/sfc/flow_classifiers'
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
Dicdata = {'port_chain': Dicdata}
data = json.dumps(Dicdata, indent=4)
- url = 'http://'+self.os_hostname+':9696/'+self.osver + \
+ url = 'http://'+self.neutron_hostname+':9696/'+self.osver + \
'/sfc/port_chains'
headers = {"Accept": "application/json",
"Content-Type": "application/json",
Dicdata = {'router': Dicdata}
data = json.dumps(Dicdata, indent=4)
- url = 'http://'+self.os_hostname+':9696/'+self.osver+'/routers.json'
+ url = 'http://'+self.neutron_hostname+':9696/' + \
+ self.osver+'/routers.json'
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
response = requests.post(url, headers=headers, data=data)
def attachInterface(self):
"""Attachment of instance ports to the Router."""
- url = 'http://'+self.os_hostname+':9696/'+self.osver + \
+ url = 'http://'+self.neutron_hostname+':9696/'+self.osver + \
'/networks?name=admin_floating_net'
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
Dicdata['subnet_id'] = self.subnetId
data = json.dumps(Dicdata, indent=4)
- url = 'http://'+self.os_hostname+':9696/'+self.osver + \
+ url = 'http://'+self.neutron_hostname+':9696/'+self.osver + \
'/routers/'+self.router_id+'/add_router_interface'
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
Dicdata1 = {'external_gateway_info': Dicdata1}
Dicdata1 = {'router': Dicdata1}
data = json.dumps(Dicdata1, indent=4)
- url = 'http://' + self.os_hostname+':9696/'+self.osver + \
+ url = 'http://' + self.neutron_hostname+':9696/'+self.osver + \
'/routers/' + self.router_id
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
"""Attachment of Floating Ip to the Router."""
for ip_num in range(0, 2):
Dicdata = {}
- if self.ip_pool != '':
- Dicdata['pool'] = ""
+ Dicdata['pool'] = "admin_floating_net"
data = json.dumps(Dicdata, indent=4)
- url = 'http://'+self.os_hostname+':8774/v2.1/os-floating-ips'
+ url = 'http://'+self.nova_hostname+':8774/v2.1/os-floating-ips'
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
response = requests.post(url, headers=headers, data=data)
Dicdata1 = {'addFloatingIp': Dicdata1}
data = json.dumps(Dicdata1, indent=4)
- url = 'http://'+self.os_hostname + \
- ':8774/v2.1/servers/'+self.vm[ip_num]+'/action'
+ url = 'http://' + self.nova_hostname + ':8774/v2.1/servers/' + \
+ self.vm[ip_num]+'/action'
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
response = requests.post(url, headers=headers, data=data)
self.logger.info("\tPacket not received in Instance")
queue1.put("0")
- def ping(ip, timeout=300):
+ def ping(ip, timeout=120):
while True:
time.sleep(1)
self.logger.debug("Pinging %s. Waiting for response..." % ip)
result0 = ping(self.vm_public_ip[0])
result1 = ping(self.vm_public_ip[1])
if result0 == 0 and result1 == 0:
- time.sleep(300)
+ time.sleep(120)
queue1 = Queue()
p1 = Process(target=vm1, args=(queue1, ))
p1.start()
def deletePortChain(self):
"""Deletion of PortChain."""
- url = 'http://'+self.os_hostname+':9696/'+self.osver + \
+ url = 'http://'+self.neutron_hostname+':9696/'+self.osver + \
'/sfc/port_chains/'+self.PC_id
headers = {"Accept": "application/json", "Content-Type":
"application/json", "X-Auth-Token": self.token_id}
def deleteFlowClassifier(self):
"""Deletion of Flow Classifier."""
- url = 'http://'+self.os_hostname+':9696/'+self.osver + \
+ url = 'http://'+self.neutron_hostname+':9696/'+self.osver + \
'/sfc/flow_classifiers/'+self.flow_class_if
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
def deletePortGroup(self):
"""Deletion of PortGroup."""
for p in range(0, 1):
- url = 'http://'+self.os_hostname+':9696/'+self.osver + \
+ url = 'http://'+self.neutron_hostname+':9696/'+self.osver + \
'/sfc/port_pair_groups/'+self.port_grp_id[p]
headers = {"Accept": "application/json", "X-Auth-Token":
self.token_id}
def deletePortPair(self):
"""Deletion of Portpair."""
for p in range(1, 2):
- url = 'http://'+self.os_hostname+':9696/'+self.osver + \
+ url = 'http://'+self.neutron_hostname+':9696/'+self.osver + \
'/sfc/port_pairs/'+self.port_pair_id[0]
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
"""Cleanup."""
print ("\n\t\tDeleting the VMs")
for y in range(0, 3):
- url = 'http://'+self.os_hostname+':8774/v2.1/servers/'+self.vm[y]
+ 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)
return(response.status_code)
print ("\n\t\tDeletion of Ports")
for x in range(self.i, self.numTerms):
- url = 'http://'+self.os_hostname+':9696/'+self.osver + \
+ url = 'http://'+self.neutron_hostname+':9696/'+self.osver + \
'/ports/'+self.port_num[x]
headers = {"Accept": "application/json", "X-Auth-Token":
self.token_id}
Dicdata['external_gateway_info'] = {}
Dicdata = {'router': Dicdata}
data = json.dumps(Dicdata, indent=4)
- url = 'http://'+self.os_hostname+':9696/'+self.osver + \
+ url = 'http://'+self.neutron_hostname+':9696/'+self.osver + \
'/routers/'+self.router_id
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
if self.subnetId != '':
Dicdata1['subnet_id'] = self.subnetId
data = json.dumps(Dicdata1, indent=4)
- url = 'http://'+self.os_hostname+':9696/'+self.osver + \
+ url = 'http://'+self.neutron_hostname+':9696/'+self.osver + \
'/routers/'+self.router_id + \
'/remove_router_interface.json'
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
response = requests.put(url, headers=headers, data=data)
if (response.status_code == 200):
- url = 'http://'+self.os_hostname+':9696/'+self.osver + \
+ url = 'http://'+self.neutron_hostname+':9696/'+self.osver + \
'/routers/'+self.router_id
headers = {"Accept": "application/json", "X-Auth-Token":
self.token_id}
return(response.status_code)
print ("\n\t\tDeletion of Network")
- url = 'http://'+self.os_hostname+':9696/'+self.osver + \
+ url = 'http://'+self.neutron_hostname+':9696/'+self.osver + \
'/networks/'+self.net_id
headers = {"Accept": "application/json",
"X-Auth-Token": self.token_id}
print ("\n\t\tDeletion of Floating ip")
for ip_num in range(0, 2):
- url = 'http://'+self.os_hostname+':9696/'+self.osver + \
+ 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}