diff options
Diffstat (limited to 'functest/opnfv_tests/sdn/onos/sfc/sfc_onos.py')
| -rw-r--r-- | functest/opnfv_tests/sdn/onos/sfc/sfc_onos.py | 186 |
1 files changed, 105 insertions, 81 deletions
diff --git a/functest/opnfv_tests/sdn/onos/sfc/sfc_onos.py b/functest/opnfv_tests/sdn/onos/sfc/sfc_onos.py index 78bf6f4b..8ca32e9b 100644 --- a/functest/opnfv_tests/sdn/onos/sfc/sfc_onos.py +++ b/functest/opnfv_tests/sdn/onos/sfc/sfc_onos.py @@ -57,8 +57,8 @@ class SfcOnos: self.container_format = "bare" self.disk_format = "qcow2" self.imagename = "TestSfcVm" - self.createImage = "/home/root1/devstack/files/images/\ - firewall_block_image.img" + self.createImage = ("/home/root1/devstack/files/images/" + "firewall_block_image.img") self.vm_name = "vm" self.imageRef = "test" @@ -107,10 +107,10 @@ class SfcOnos: def getToken(self): """Get the keystone token value from Openstack .""" - url = 'http://' + self.keystone_hostname + \ - ':5000/' + self.osver + '/tokens' - data = '{"auth": {"tenantName": "admin", "passwordCredentials":\ - { "username": "admin", "password": "console"}}}' + url = 'http://%s:5000/%s/tokens' % (self.keystone_hostname, + self.osver) + data = ('{"auth": {"tenantName": "admin", "passwordCredentials":' + '{ "username": "admin", "password": "console"}}}') headers = {"Accept": "application/json"} response = requests.post(url, headers=headers, data=data) if (response.status_code == OK): @@ -133,8 +133,8 @@ class SfcOnos: Dicdata['admin_state_up'] = self.admin_state_up Dicdata = {'network': Dicdata} data = json.dumps(Dicdata, indent=4) - url = 'http://' + self.neutron_hostname + \ - ':9696/' + self.osver + '/networks' + url = 'http://%s:9696/%s/networks' % (self.neutron_hostname, + self.osver) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.post(url, headers=headers, data=data) @@ -163,8 +163,8 @@ class SfcOnos: Dicdata = {'subnet': Dicdata} data = json.dumps(Dicdata, indent=4) - url = 'http://' + self.neutron_hostname + \ - ':9696/' + self.osver + '/subnets' + url = 'http://%s:9696/%s/subnets' % (self.neutron_hostname, + self.osver) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.post(url, headers=headers, data=data) @@ -196,8 +196,8 @@ class SfcOnos: Dicdata = {'port': Dicdata} data = json.dumps(Dicdata, indent=4) - url = 'http://' + self.neutron_hostname + \ - ':9696/' + self.osver + '/ports' + url = 'http://%s:9696/%s/ports' % (self.neutron_hostname, + self.osver) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.post(url, headers=headers, data=data) @@ -217,8 +217,8 @@ class SfcOnos: def createVm(self): """Creation of Instance, using firewall image.""" - url = 'http://' + self.glance_hostname + \ - ':9292/v2/images?name=TestSfcVm' + url = ("http://%s:9292/v2/images?" + "name=TestSfcVm" % (self.glance_hostname)) headers = {"Accept": "application/json", "Content-Type": "application/\ octet-stream", "X-Auth-Token": self.token_id} response = requests.get(url, headers=headers) @@ -232,8 +232,10 @@ class SfcOnos: else: return(response.status_code) - url = 'http://' + self.nova_hostname + \ - ':8774/v2.1/' + self.tenant_id + '/flavors?name=m1.tiny' + url = ("http://%s:8774//v2.1/%s/ports/" + "%s/flavors?name=m1.tiny" % (self.nova_hostname, + self.tenant_id)) + headers = {"Accept": "application/json", "Content-Type": "application/json", "X-Auth-Token": self.token_id} response = requests.get(url, headers=headers) @@ -266,9 +268,8 @@ class SfcOnos: Dicdata['networks'] = org_nw_port Dicdata = {'server': Dicdata} data = json.dumps(Dicdata, indent=4) - - url = ('http://' + self.nova_hostname + ':8774/v2.1/' + - self.tenant_id + '/servers') + url = 'http://%s:8774/v2.1/%s/servers' % (self.nova_hostname, + self.tenant_id) headers = {"Accept": "application/json", "Content-Type": "application/json", "X-Auth-Token": self.token_id} response = requests.post(url, headers=headers, data=data) @@ -291,9 +292,8 @@ class SfcOnos: """Checking the Status of the Instance.""" time.sleep(10) for y in range(0, 3): - url = 'http://' + \ - self.nova_hostname + \ - ':8774/v2.1/servers/detail?name=vm' + str(y) + url = ("http://%s:8774/v2.1/servers/" + "detail?name=vm" + str(y)) % (self.neutron_hostname) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.get(url, headers=headers) @@ -304,11 +304,9 @@ class SfcOnos: self.logger.debug(json1_data) self.vm_active = json1_data['servers'][0]['status'] if (self.vm_active == "ACTIVE"): - info = "VM" + str(y) + \ - " is Active : " + self.vm_active + info = "VM" + str(y) + " is Active : " + self.vm_active else: - info = "VM" + 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) @@ -329,15 +327,14 @@ class SfcOnos: Dicdata = {'port_pair': Dicdata} data = json.dumps(Dicdata, indent=4) - - url = 'http://' + self.neutron_hostname + ':9696/' + self.osver + \ - '/sfc/port_pairs' + url = 'http://%s:9696/%s/sfc/port_pairs' % (self.neutron_hostname, + self.osver) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.post(url, headers=headers, data=data) if (response.status_code == CREATED): - info = "Creation of Port Pair PP" + str(p) + \ - " is successful" + info = ("Creation of Port Pair PP" + str(p) + + " is successful") self.logger.debug(info) else: return(response.status_code) @@ -347,10 +344,11 @@ class SfcOnos: def getPortPair(self): """Query the Portpair id value.""" for p in range(0, 1): - url = 'http://' + self.neutron_hostname + ':9696/' + self.osver + \ - '/sfc/port_pairs?name=PP1' - headers = {"Accept": "application/json", "X-Auth-Token": - self.token_id} + url = ("http://%s:9696/%s/ports/" + "sfc/port_pairs?name=PP1" % (self.neutron_hostname, + self.osver)) + headers = {"Accept": "application/json", + "X-Auth-Token": self.token_id} response = requests.get(url, headers=headers) if (response.status_code == OK): @@ -376,14 +374,15 @@ class SfcOnos: Dicdata = {'port_pair_group': Dicdata} data = json.dumps(Dicdata, indent=4) - url = 'http://' + self.neutron_hostname + ':9696/' + self.osver + \ - '/sfc/port_pair_groups' + url = ("http://%s:9696/%s/" + "sfc/port_pair_groups" % (self.neutron_hostname, + self.osver)) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.post(url, headers=headers, data=data) if (response.status_code == CREATED): - info = "Creation of Port Group PG" + str(p) + \ - "is successful" + info = ("Creation of Port Group PG" + str(p) + + "is successful") self.logger.debug(info) else: return(response.status_code) @@ -393,8 +392,9 @@ class SfcOnos: def getPortGroup(self): """Query the PortGroup id.""" for p in range(0, 1): - url = 'http://' + self.neutron_hostname + ':9696/' + self.osver + \ - '/sfc/port_pair_groups?name=PG' + str(p) + url = ("http://%s:9696/%s/sfc/port_pair_groups" + "?name=PG" + str(p)) % (self.neutron_hostname, + self.osver) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.get(url, headers=headers) @@ -425,8 +425,9 @@ class SfcOnos: Dicdata = {'flow_classifier': Dicdata} data = json.dumps(Dicdata, indent=4) - url = 'http://' + self.neutron_hostname + ':9696/' + self.osver + \ - '/sfc/flow_classifiers' + url = ("http://%s:9696/%s/" + "sfc/flow_classifiers" % (self.neutron_hostname, + self.osver)) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.post(url, headers=headers, data=data) @@ -455,8 +456,8 @@ class SfcOnos: Dicdata = {'port_chain': Dicdata} data = json.dumps(Dicdata, indent=4) - url = 'http://' + self.neutron_hostname + ':9696/' + self.osver + \ - '/sfc/port_chains' + url = 'http://%s:9696/%s/sfc/port_chains' % (self.neutron_hostname, + self.osver) headers = {"Accept": "application/json", "Content-Type": "application/json", "X-Auth-Token": self.token_id} @@ -499,8 +500,8 @@ class SfcOnos: Dicdata = {'router': Dicdata} data = json.dumps(Dicdata, indent=4) - url = 'http://' + self.neutron_hostname + ':9696/' + \ - self.osver + '/routers.json' + url = 'http://%s:9696/%s/routers.json' % (self.neutron_hostname, + self.osver) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.post(url, headers=headers, data=data) @@ -517,10 +518,10 @@ class SfcOnos: def attachInterface(self): """Attachment of instance ports to the Router.""" - url = 'http://' + self.neutron_hostname + ':9696/' + self.osver + \ - '/networks?name=admin_floating_net' - headers = {"Accept": "application/json", - "X-Auth-Token": self.token_id} + url = ("http://%s:9696/%s/networks" + "?name=admin_floating_net" % (self.neutron_hostname, + self.osver)) + headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.get(url, headers=headers) if (response.status_code == OK): self.logger.debug(response.status_code) @@ -540,8 +541,10 @@ class SfcOnos: Dicdata['subnet_id'] = self.subnetId data = json.dumps(Dicdata, indent=4) - url = 'http://' + self.neutron_hostname + ':9696/' + self.osver + \ - '/routers/' + self.router_id + '/add_router_interface' + url = ("http://%s:9696/%s/routers" + "/%s/add_router_interface" % (self.neutron_hostname, + self.osver, + self.router_id)) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.put(url, headers=headers, data=data) @@ -561,8 +564,9 @@ class SfcOnos: Dicdata1 = {'external_gateway_info': Dicdata1} Dicdata1 = {'router': Dicdata1} data = json.dumps(Dicdata1, indent=4) - url = 'http://' + self.neutron_hostname + ':9696/' + self.osver + \ - '/routers/' + self.router_id + url = 'http://%s:9696/%s/routers/%s' % (self.neutron_hostname, + self.osver, + self.router_id) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.put(url, headers=headers, data=data) @@ -581,7 +585,8 @@ class SfcOnos: Dicdata['pool'] = "admin_floating_net" data = json.dumps(Dicdata, indent=4) - url = 'http://' + self.nova_hostname + ':8774/v2.1/os-floating-ips' + url = ("http://%s:8774/v2.1/" + "os-floating-ips" % (self.nova_hostname)) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.post(url, headers=headers, data=data) @@ -602,8 +607,10 @@ class SfcOnos: Dicdata1 = {'addFloatingIp': Dicdata1} data = json.dumps(Dicdata1, indent=4) - url = 'http://' + self.nova_hostname + ':8774/v2.1/servers/' + \ - self.vm[ip_num] + '/action' + url = ("http://%s:8774/v2.1/" + "servers/%s/action" % (self.nova_hostname, + self.vm[ip_num])) + headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.post(url, headers=headers, data=data) @@ -710,8 +717,8 @@ class SfcOnos: def deletePortChain(self): """Deletion of PortChain.""" - url = 'http://' + self.neutron_hostname + ':9696/' + self.osver + \ - '/sfc/port_chains/' + self.PC_id + 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} response = requests.delete(url, headers=headers) @@ -724,8 +731,10 @@ class SfcOnos: def deleteFlowClassifier(self): """Deletion of Flow Classifier.""" - url = 'http://' + self.neutron_hostname + ':9696/' + self.osver + \ - '/sfc/flow_classifiers/' + self.flow_class_if + url = ("http://%s:9696/%s/sfc/" + "flow_classifiers/%s" % (self.neutron_hostname, + self.osver, + self.flow_class_if)) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.delete(url, headers=headers) @@ -739,8 +748,10 @@ class SfcOnos: def deletePortGroup(self): """Deletion of PortGroup.""" for p in range(0, 1): - url = 'http://' + self.neutron_hostname + ':9696/' + self.osver + \ - '/sfc/port_pair_groups/' + self.port_grp_id[p] + url = ("http://%s:9696/%s/sfc/" + "port_pair_groups/%s" % (self.neutron_hostname, + self.osver, + self.port_grp_id[p])) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.delete(url, headers=headers) @@ -754,8 +765,10 @@ class SfcOnos: def deletePortPair(self): """Deletion of Portpair.""" for p in range(1, 2): - url = 'http://' + self.neutron_hostname + ':9696/' + self.osver + \ - '/sfc/port_pairs/' + self.port_pair_id[0] + url = ("http://%s:9696/%s/sfc/" + "port_pairs/%s" % (self.neutron_hostname, + self.osver, + self.port_pair_id[0])) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.delete(url, headers=headers) @@ -770,8 +783,9 @@ class SfcOnos: """Cleanup.""" self.logger.info("Deleting VMs") for y in range(0, 3): - url = 'http://' + self.nova_hostname + \ - ':8774/v2.1/servers/' + self.vm[y] + url = ("http://%s:8774/v2.1/" + "/servers/%s" % (self.nova_hostname, + self.vm[y])) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.delete(url, headers=headers) @@ -784,8 +798,8 @@ class SfcOnos: return(response.status_code) 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] + url = ('http://' + self.neutron_hostname + ':9696/' + + self.osver + '/ports/' + self.port_num[x]) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.delete(url, headers=headers) @@ -802,8 +816,10 @@ class SfcOnos: Dicdata['external_gateway_info'] = {} Dicdata = {'router': Dicdata} data = json.dumps(Dicdata, indent=4) - url = 'http://' + self.neutron_hostname + ':9696/' + self.osver + \ - '/routers/' + self.router_id + url = ("http://%s:9696/%s/" + "/routers/%s" % (self.neutron_hostname, + self.osver, + self.router_id)) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.put(url, headers=headers, data=data) @@ -814,15 +830,18 @@ class SfcOnos: if self.subnetId != '': Dicdata1['subnet_id'] = self.subnetId data = json.dumps(Dicdata1, indent=4) - url = 'http://' + self.neutron_hostname + ':9696/' + self.osver + \ - '/routers/' + self.router_id + \ - '/remove_router_interface.json' + url = ("http://%s:9696/%s/routers/%s" + "/remove_router_interface.json" % (self.neutron_hostname, + self.osver, + self.router_id)) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.put(url, headers=headers, data=data) if (response.status_code == OK): - url = ('http://' + self.neutron_hostname + ':9696/' + - self.osver + '/routers/' + self.router_id) + url = ("http://%s:9696/%s/" + "routers/%s" % (self.neutron_hostname, + self.osver, + self.router_id)) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.delete(url, headers=headers) @@ -837,8 +856,10 @@ class SfcOnos: return(response.status_code) self.logger.info("Deleting Network") - url = 'http://' + self.neutron_hostname + ':9696/' + self.osver + \ - '/networks/' + self.net_id + url = "http://%s:9696/%s/networks/%s" % (self.neutron_hostname, + self.osver, + self.net_id) + headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.delete(url, headers=headers) @@ -850,8 +871,11 @@ class SfcOnos: 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] + url = ("http://%s:9696/%s/floatingips" + "/%s" % (self.neutron_hostname, + self.osver, + self.vm_public_id[ip_num])) + headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} response = requests.delete(url, headers=headers) |