X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=src%2Femuvim%2Ftest%2Funittests%2Ftest_openstack.py;h=3e9037a0a651d3c45ceb256b708e9a91cf3babe1;hb=3e16acbebd2c6669c1bbe5a7e3bb313aa1494e20;hp=707b999562d1e0d1fabef3e80c59aa1325b36271;hpb=e22979a58ea5ae049186cd14113fa78eab3c5860;p=osm%2Fvim-emu.git diff --git a/src/emuvim/test/unittests/test_openstack.py b/src/emuvim/test/unittests/test_openstack.py index 707b999..3e9037a 100755 --- a/src/emuvim/test/unittests/test_openstack.py +++ b/src/emuvim/test/unittests/test_openstack.py @@ -34,6 +34,8 @@ import os import unittest import requests import simplejson as json +import yaml +import time from emuvim.test.api_base_openstack import ApiBaseOpenStack @@ -58,326 +60,15 @@ class testRestApi(ApiBaseOpenStack): # start Mininet network self.startNet() - @unittest.skip("temporarily disabled") - def testChainingDummy(self): - print('->>>>>>> test Chaining Class->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - print(" ") - - headers = {'Content-type': 'application/json'} - test_heatapi_template_create_stack = open(os.path.join(os.path.dirname(__file__), "test_heatapi_template_chaining.json")).read() - url = "http://0.0.0.0:8004/v1/tenantabc123/stacks" - requests.post(url, data=json.dumps(json.loads(test_heatapi_template_create_stack)), headers=headers) - - - print('->>>>>>> test Chaining Versions ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/" - listapiversionstackresponse = requests.get(url, headers=headers) - self.assertEqual(listapiversionstackresponse.status_code, 200) - self.assertEqual(json.loads(listapiversionstackresponse.content)["versions"][0]["id"], "v1") - print(" ") - - print('->>>>>>> test Chaining List ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/chain/list" - chainlistresponse = requests.get(url, headers=headers) - self.assertEqual(chainlistresponse.status_code, 200) - self.assertEqual(json.loads(chainlistresponse.content)["chains"], []) - print(" ") - - print('->>>>>>> test Loadbalancing List ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/lb/list" - lblistresponse = requests.get(url, headers=headers) - self.assertEqual(lblistresponse.status_code, 200) - self.assertEqual(json.loads(lblistresponse.content)["loadbalancers"], []) - print(" ") - - testchain = "dc0_s1_firewall1/fire-out-0/dc0_s1_iperf1/iper-in-0" - print('->>>>>>> test Chain VNF Interfaces ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/chain/%s" %(testchain) - chainvnfresponse = requests.put(url) - self.assertEqual(chainvnfresponse.status_code, 200) - self.assertGreaterEqual(json.loads(chainvnfresponse.content)["cookie"], 0) - print(" ") - - print('->>>>>>> test Chaining List ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/chain/list" - chainlistresponse = requests.get(url, headers=headers) - self.assertEqual(chainlistresponse.status_code, 200) - self.assertEqual(json.loads(chainlistresponse.content)["chains"][0]["dst_vnf"], "dc0_s1_firewall1") - self.assertEqual(json.loads(chainlistresponse.content)["chains"][0]["dst_intf"], "fire-out-0") - self.assertEqual(json.loads(chainlistresponse.content)["chains"][0]["src_vnf"], "dc0_s1_iperf1") - self.assertEqual(json.loads(chainlistresponse.content)["chains"][0]["src_intf"], "iper-in-0") - print(" ") - - print('->>>>>>> test Chain VNF Delete Chain ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/chain/%s" % (testchain) - deletechainvnfresponse = requests.delete(url) - self.assertEqual(deletechainvnfresponse.status_code, 200) - self.assertEqual(deletechainvnfresponse.content, "true") - print(" ") - - print('->>>>>>> test Chaining List If Empty Again ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/chain/list" - chainlistresponse = requests.get(url, headers=headers) - self.assertEqual(chainlistresponse.status_code, 200) - self.assertEqual(json.loads(chainlistresponse.content)["chains"], []) - print(" ") - - testchain = "dc0_s1_firewall1/fire-out-0/dc0_s1_iperf1/iper-in-0" - print('->>>>>>> test Stack Chain VNF Interfaces ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/chain/%s" % (testchain) - stackchainvnfresponse = requests.post(url, data=json.dumps(json.loads('{"path":["dc1.s1", "s1","s2","s3","s1","dc1.s1"]}')), headers=headers) - self.assertEqual(stackchainvnfresponse.status_code, 200) - print (stackchainvnfresponse.content) - self.assertGreaterEqual(json.loads(stackchainvnfresponse.content)["cookie"], 0) - print(" ") - - - print('->>>>>>> test Stack Chaining List ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/chain/list" - chainlistresponse = requests.get(url, headers=headers) - self.assertEqual(chainlistresponse.status_code, 200) - print (chainlistresponse.content) - self.assertEqual(json.loads(chainlistresponse.content)["chains"][0]["dst_vnf"], "dc0_s1_firewall1") - self.assertEqual(json.loads(chainlistresponse.content)["chains"][0]["dst_intf"], "fire-out-0") - self.assertEqual(json.loads(chainlistresponse.content)["chains"][0]["src_vnf"], "dc0_s1_iperf1") - self.assertEqual(json.loads(chainlistresponse.content)["chains"][0]["src_intf"], "iper-in-0") - self.assertItemsEqual(json.loads(chainlistresponse.content)["chains"][0]["path"],['dc1.s1', 's1', 's2', 's3', 's1', 'dc1.s1']) - print(" ") - - print('->>>>>>> test Chain VNF Delete Chain ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/chain/%s" % (testchain) - deletechainvnfresponse = requests.delete(url) - self.assertEqual(deletechainvnfresponse.status_code, 200) - self.assertEqual(deletechainvnfresponse.content, "true") - print(" ") - - print('->>>>>>> test Chaining List If Empty Again ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/chain/list" - chainlistresponse = requests.get(url, headers=headers) - self.assertEqual(chainlistresponse.status_code, 200) - self.assertEqual(json.loads(chainlistresponse.content)["chains"], []) - print(" ") - - - testchain = "dc0_s1_firewall1/non-existing-interface/dc0_s1_iperf1/iper-in-0" - print('->>>>>>> test Chain VNF Interfaces ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/chain/%s" % (testchain) - chainvnfresponse = requests.put(url) - self.assertEqual(chainvnfresponse.status_code, 501) - print(" ") - - testchain = "dc0_s1_firewall1/fire-out-0/dc0_s1_iperf1/non-existing-interface" - print('->>>>>>> test Chain VNF Interfaces ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/chain/%s" % (testchain) - chainvnfresponse = requests.put(url) - self.assertEqual(chainvnfresponse.status_code, 501) - print(" ") - - testchain = "dc0_s1_firewall1/non-existing-interface/dc0_s1_iperf1/iper-in-0" - print('->>>>>>> test Chain VNF Delete Chain ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/chain/%s" % (testchain) - deletechainvnfresponse = requests.delete(url) - self.assertEqual(deletechainvnfresponse.status_code, 501) - print(" ") - - testchain = "dc0_s1_firewall1/fire-out-0/dc0_s1_iperf1/non-existing-interface" - print('->>>>>>> test Chain VNF Delete Chain ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/chain/%s" % (testchain) - deletechainvnfresponse = requests.delete(url) - self.assertEqual(deletechainvnfresponse.status_code, 501) - print(" ") - - testchain = "non-existent-dc/s1/firewall1/firewall1:cp03:output/dc0/s1/iperf1/iperf1:cp02:input" - print('->>>>>>> test Chain VNF Non-Existing DC ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/chain/%s" % (testchain) - chainvnfresponse = requests.put(url) - self.assertEqual(chainvnfresponse.status_code, 500) - print(" ") - - - testchain = "dc0/s1/firewall1/non-existent:cp03:output/dc0/s1/iperf1/iperf1:cp02:input" - print('->>>>>>> test Chain VNF Non-Existing Interfaces ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/chain/%s" % (testchain) - chainvnfresponse = requests.put(url) - self.assertEqual(chainvnfresponse.status_code, 500) - print(" ") - - testchain = "dc0/s1/firewall1/firewall1:cp03:output/dc0/s1/iperf1/non-existent:cp02:input" - print('->>>>>>> test Chain VNF Non-Existing Interfaces ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/chain/%s" % (testchain) - chainvnfresponse = requests.put(url) - self.assertEqual(chainvnfresponse.status_code, 500) - print(" ") - - testchain = "dc0/s1/firewall1/firewall1:cp03:output/dc0/s1/iperf1/iperf1:cp02:input" - print('->>>>>>> test Chain VNF Interfaces ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/chain/%s" % (testchain) - chainvnfresponse = requests.put(url) - print (chainvnfresponse.content) - self.assertEqual(chainvnfresponse.status_code, 200) - self.assertGreaterEqual(json.loads(chainvnfresponse.content)["cookie"], 0) - print(" ") - - print('->>>>>>> test Chain VNF Delete Chain ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/chain/%s" % (testchain) - deletechainvnfresponse = requests.delete(url) - self.assertEqual(deletechainvnfresponse.status_code, 200) - self.assertEqual(deletechainvnfresponse.content, "true") - print(" ") - - print('->>>>>>> test Chaining List If Empty Again ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/chain/list" - chainlistresponse = requests.get(url, headers=headers) - self.assertEqual(chainlistresponse.status_code, 200) - self.assertEqual(json.loads(chainlistresponse.content)["chains"], []) - print(" ") - - testchain = "dc0/s1/firewall1/firewall1:cp03:output/dc0/s1/iperf1/iperf1:cp02:input" - print('->>>>>>> test Stack Chain VNF Interfaces ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/chain/%s" % (testchain) - stackchainvnfresponse = requests.post(url, data=json.dumps( - json.loads('{"path":["dc1.s1", "s1","s2","s3","s1","dc1.s1"]}')), headers=headers) - self.assertEqual(stackchainvnfresponse.status_code, 200) - print (stackchainvnfresponse.content) - self.assertGreaterEqual(json.loads(stackchainvnfresponse.content)["cookie"], 0) - print(" ") - - testchain = "dc0/s1/firewall1/firewall1:cp03:output/dc0/s1/iperf1/iperf1:cp02:input" - print('->>>>>>> test Stack Chain VNF Interfaces ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/chain/%s" % (testchain) - stackchainvnfresponse = requests.delete(url, headers=headers) - self.assertEqual(stackchainvnfresponse.status_code, 200) - print(" ") - - - print('->>>>>>> test Loadbalancing ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/lb/dc0/s1/firewall1/firewall1:cp03:output" - lblistresponse = requests.post(url, data=json.dumps( - {"dst_vnf_interfaces":[{"pop":"dc0","stack":"s1","server":"iperf1","port":"iperf1:cp02:input"}]}) - , headers=headers) - print (lblistresponse.content) - self.assertEqual(lblistresponse.status_code, 200) - self.assertIn("dc0_s1_firewall1:fire-out-0", lblistresponse.content) - print(" ") - - print('->>>>>>> test Loadbalancing List ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/lb/list" - lblistresponse = requests.get(url, headers=headers) - self.assertEqual(lblistresponse.status_code, 200) - print (lblistresponse.content ) - self.assertEqual(json.loads(lblistresponse.content)["loadbalancers"][0]["paths"][0]["dst_vnf"], "dc0_s1_iperf1") - self.assertEqual(json.loads(lblistresponse.content)["loadbalancers"][0]["paths"][0]["dst_intf"], "iper-in-0") - self.assertEqual(json.loads(lblistresponse.content)["loadbalancers"][0]["src_vnf"], "dc0_s1_firewall1") - self.assertEqual(json.loads(lblistresponse.content)["loadbalancers"][0]["src_intf"],"fire-out-0") - print(" ") - - print('->>>>>>> test delete Loadbalancing ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/lb/dc0/s1/firewall1/firewall1:cp03:output" - lbdeleteresponse = requests.delete(url, headers=headers) - print (lbdeleteresponse.content) - self.assertEqual(lbdeleteresponse.status_code, 200) - print(" ") - - print('->>>>>>> testFloatingLoadbalancer ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/lb/dc0/floating/bla/blubb" - lblistresponse = requests.post(url, data=json.dumps( - {"dst_vnf_interfaces":[{"pop":"dc0","stack":"s1","server":"iperf1","port":"iperf1:cp02:input"}]}) - , headers=headers) - print (lblistresponse.content) - self.assertEqual(lblistresponse.status_code, 200) - resp = json.loads(lblistresponse.content) - self.assertIsNotNone(resp.get('cookie')) - self.assertIsNotNone(resp.get('floating_ip')) - cookie = resp.get('cookie') - print(" ") - - print('->>>>>>> testDeleteFloatingLoadbalancer ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/lb/dc0/floating/%s/blubb" % cookie - lblistresponse = requests.delete(url, headers=headers) - print (lblistresponse.content) - self.assertEqual(lblistresponse.status_code, 200) - print(" ") - - print('->>>>>>> testLoadbalancingCustomPath ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/lb/dc0_s1_firewall1/fire-out-0" - lblistresponse = requests.post(url, data=json.dumps( - {"dst_vnf_interfaces":{"dc0_s1_iperf1":"iper-in-0"}, - "path": {"dc0_s1_iperf1": {"iper-in-0": ["dc1.s1", "s1","s2","s3","s1","dc1.s1"]}}}), headers=headers) - print (lblistresponse.content) - self.assertEqual(lblistresponse.status_code, 200) - self.assertIn("dc0_s1_firewall1:fire-out-0", lblistresponse.content) - print(" ") - - print('->>>>>>> testLoadbalancingListCustomPath ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/lb/list" - lblistresponse = requests.get(url, headers=headers) - self.assertEqual(lblistresponse.status_code, 200) - print (lblistresponse.content ) - self.assertEqual(json.loads(lblistresponse.content)["loadbalancers"][0]["paths"][0]["dst_vnf"], "dc0_s1_iperf1") - self.assertEqual(json.loads(lblistresponse.content)["loadbalancers"][0]["paths"][0]["dst_intf"], "iper-in-0") - self.assertEqual(json.loads(lblistresponse.content)["loadbalancers"][0]["paths"][0]["path"], - ["dc1.s1", "s1","s2","s3","s1","dc1.s1"] ) - self.assertEqual(json.loads(lblistresponse.content)["loadbalancers"][0]["src_vnf"], "dc0_s1_firewall1") - self.assertEqual(json.loads(lblistresponse.content)["loadbalancers"][0]["src_intf"],"fire-out-0") - print(" ") - - - print('->>>>>>> test Delete Loadbalancing ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/lb/dc0_s1_firewall1/fire-out-0" - lblistresponse = requests.delete(url, headers=headers) - self.assertEqual(lblistresponse.status_code, 200) - print(" ") - - print('->>>>>>> test Query Topology ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:4000/v1/topo" - topolistresponse = requests.get(url, headers=headers) - print(topolistresponse.content) - self.assertEqual(topolistresponse.status_code, 200) - print(" ") - - def testNovaDummy(self): print('->>>>>>> test Nova Dummy Class->>>>>>>>>>>>>>>') print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') print(" ") headers = {'Content-type': 'application/json'} - test_heatapi_template_create_stack = open(os.path.join(os.path.dirname(__file__), "test_heatapi_template_create_stack.json")).read() + test_heatapi_template_create_stack = open(os.path.join(os.path.dirname(__file__), "templates/test_heatapi_template_create_stack.yml")).read() url = "http://0.0.0.0:18004/v1/tenantabc123/stacks" - requests.post(url, data=json.dumps(json.loads(test_heatapi_template_create_stack)), + requests.post(url, data=json.dumps(yaml.load(test_heatapi_template_create_stack)), headers=headers) print('->>>>>>> test Nova List Versions ->>>>>>>>>>>>>>>') @@ -530,8 +221,6 @@ class testRestApi(ApiBaseOpenStack): if image["name"] == "ubuntu:trusty": ubuntu_image_id = image["id"] - - print('->>>>>>> test Nova Create Server Instance ->>>>>>>>>>>>>>>') print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') url = "http://0.0.0.0:18774/v2.1/id_bla/servers" @@ -572,17 +261,15 @@ class testRestApi(ApiBaseOpenStack): self.assertEqual(listnonexistingserverdetailsresponse.status_code, 404) print(" ") - - def testNeutronDummy(self): print('->>>>>>> test Neutron Dummy Class->>>>>>>>>>>>>>>') print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') print(" ") headers = {'Content-type': 'application/json'} - test_heatapi_template_create_stack = open(os.path.join(os.path.dirname(__file__), "test_heatapi_template_create_stack.json")).read() + test_heatapi_template_create_stack = open(os.path.join(os.path.dirname(__file__), "templates/test_heatapi_template_create_stack.yml")).read() url = "http://0.0.0.0:18004/v1/tenantabc123/stacks" - requests.post(url, data=json.dumps(json.loads(test_heatapi_template_create_stack)), headers=headers) + requests.post(url, data=json.dumps(yaml.load(test_heatapi_template_create_stack)), headers=headers) # test_heatapi_keystone_get_token = open("test_heatapi_keystone_get_token.json").read() print('->>>>>>> test Neutron List Versions ->>>>>>>>>>>>>>>') @@ -779,8 +466,6 @@ class testRestApi(ApiBaseOpenStack): self.assertEqual(updatenonexistingsubnetresponse.status_code, 404) print(" ") - - print('->>>>>>> test Neutron List Ports ->>>>>>>>>>>>>>>') print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') url = "http://0.0.0.0:19696/v2.0/ports" @@ -936,7 +621,7 @@ class testRestApi(ApiBaseOpenStack): print(" ") headers = {'Content-type': 'application/json'} - test_heatapi_keystone_get_token = open(os.path.join(os.path.dirname(__file__), "test_heatapi_keystone_get_token.json")).read() + test_heatapi_keystone_get_token = open(os.path.join(os.path.dirname(__file__), "templates/test_heatapi_keystone_get_token.yml")).read() print('->>>>>>> test Keystone List Versions ->>>>>>>>>>>>>>>') print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') @@ -957,20 +642,19 @@ class testRestApi(ApiBaseOpenStack): print('->>>>>>> test Keystone Get Token ->>>>>>>>>>>>>>>') print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') url = "http://0.0.0.0:15000/v2.0/tokens" - gettokenstackresponse = requests.post(url, data=json.dumps(json.loads(test_heatapi_keystone_get_token)), headers=headers) + gettokenstackresponse = requests.post(url, data=json.dumps(yaml.load(test_heatapi_keystone_get_token)), headers=headers) self.assertEqual(gettokenstackresponse.status_code, 200) self.assertEqual(json.loads(gettokenstackresponse.content)["access"]["user"]["name"], "tenantName") print(" ") - def testHeatDummy(self): print('->>>>>>> test Heat Dummy Class->>>>>>>>>>>>>>>') print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') print(" ") headers = {'Content-type': 'application/json'} - test_heatapi_template_create_stack = open(os.path.join(os.path.dirname(__file__), "test_heatapi_template_create_stack.json")).read() - test_heatapi_template_update_stack = open(os.path.join(os.path.dirname(__file__), "test_heatapi_template_update_stack.json")).read() + test_heatapi_template_create_stack = open(os.path.join(os.path.dirname(__file__), "templates/test_heatapi_template_create_stack.yml")).read() + test_heatapi_template_update_stack = open(os.path.join(os.path.dirname(__file__), "templates/test_heatapi_template_update_stack.yml")).read() print('->>>>>>> test Heat List API Versions Stack ->>>>>>>>>>>>>>>') print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') @@ -983,7 +667,7 @@ class testRestApi(ApiBaseOpenStack): print('->>>>>>> test Create Stack ->>>>>>>>>>>>>>>') print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') url = "http://0.0.0.0:18004/v1/tenantabc123/stacks" - createstackresponse = requests.post(url, data=json.dumps(json.loads(test_heatapi_template_create_stack)), headers=headers) + createstackresponse = requests.post(url, data=json.dumps(yaml.load(test_heatapi_template_create_stack)), headers=headers) self.assertEqual(createstackresponse.status_code, 201) self.assertNotEqual(json.loads(createstackresponse.content)["stack"]["id"], "") print(" ") @@ -1030,7 +714,7 @@ class testRestApi(ApiBaseOpenStack): print('->>>>>>> test Update Stack ->>>>>>>>>>>>>>>') print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') url = "http://0.0.0.0:18004/v1/tenantabc123updateStack/stacks/%s"% json.loads(createstackresponse.content)['stack']['id'] - updatestackresponse = requests.put(url, data=json.dumps(json.loads(test_heatapi_template_update_stack)), + updatestackresponse = requests.put(url, data=json.dumps(yaml.load(test_heatapi_template_update_stack)), headers=headers) self.assertEqual(updatestackresponse.status_code, 202) liststackdetailsresponse = requests.get(url, headers=headers) @@ -1052,119 +736,202 @@ class testRestApi(ApiBaseOpenStack): self.assertEqual(deletestackdetailsresponse.status_code, 204) print(" ") + def testNeutronSFC(self): + """ + Tests the Neutron Service Function Chaining implementation. As Some functions build up on others, a + complete environment is created here: - def test_CombinedTesting(self): - print('->>>>>>> test Combinded tests->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - print(" ") + Ports: p1, p2, p3, p4 + Port Pairs: pp1(p1, p2), pp2(p3, p4) + Port Pair Groups: ppg1(pp1, pp2) + Flow Classifiers: fc1 + Port Chain: pc1(ppg1, fc1) + """ headers = {'Content-type': 'application/json'} - test_heatapi_template_create_stack = open(os.path.join(os.path.dirname(__file__), - "test_heatapi_template_create_stack.json")).read() - test_heatapi_template_update_stack = open(os.path.join(os.path.dirname(__file__), - "test_heatapi_template_update_stack.json")).read() - print('->>>>>>> test Combined Create Stack ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:18004/v1/tenantabc123/stacks" - createstackresponse = requests.post(url, - data=json.dumps(json.loads(test_heatapi_template_create_stack)), - headers=headers) - self.assertEqual(createstackresponse.status_code, 201) - self.assertNotEqual(json.loads(createstackresponse.content)["stack"]["id"], "") - print(" ") + print('->>>>>>> Create ports p1 - p4 ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + # Get network id + network_resp = requests.get("http://0.0.0.0:19696/v2.0/networks?name=default", headers=headers) + self.assertEqual(network_resp.status_code, 200) + network_id = json.loads(network_resp.content)["networks"][0]["id"] - print('->>>>>>> test Combined Neutron List Ports ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') url = "http://0.0.0.0:19696/v2.0/ports" - listportsesponse = requests.get(url, headers=headers) - self.assertEqual(listportsesponse.status_code, 200) - self.assertEqual(len(json.loads(listportsesponse.content)["ports"]), 9) - for port in json.loads(listportsesponse.content)["ports"]: - self.assertEqual(len(str(port['fixed_ips'][0]['subnet_id'])), 36) - print(" ") - - print('->>>>>>> test Combined Neutron List Networks ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:19696/v2.0/networks" - listnetworksesponse = requests.get(url, headers=headers) - self.assertEqual(listnetworksesponse.status_code, 200) - self.assertEqual(len(json.loads(listnetworksesponse.content)["networks"]), 10) - for net in json.loads(listnetworksesponse.content)["networks"]: - self.assertEqual(len(str(net['subnets'][0])), 36) - print(" ") - - print('->>>>>>> test Combined Update Stack ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:18004/v1/tenantabc123updateStack/stacks/%s"% \ - json.loads(createstackresponse.content)['stack']['id'] - updatestackresponse = requests.put(url, - data=json.dumps(json.loads(test_heatapi_template_update_stack)), - headers=headers) - self.assertEqual(updatestackresponse.status_code, 202) - liststackdetailsresponse = requests.get(url, headers=headers) - self.assertEqual(json.loads(liststackdetailsresponse.content)["stack"]["stack_status"], "UPDATE_COMPLETE") - print(" ") - - print('->>>>>>> test Combined Neutron List Ports ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:19696/v2.0/ports" - listportsesponse = requests.get(url, headers=headers) - self.assertEqual(listportsesponse.status_code, 200) - self.assertEqual(len(json.loads(listportsesponse.content)["ports"]), 18) - for port in json.loads(listportsesponse.content)["ports"]: - self.assertEqual(len(str(port['fixed_ips'][0]['subnet_id'])), 36) - print(" ") - - print('->>>>>>> test Combined Neutron List Networks ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:19696/v2.0/networks" - listnetworksesponse = requests.get(url, headers=headers) - self.assertEqual(listnetworksesponse.status_code, 200) - self.assertEqual(len(json.loads(listnetworksesponse.content)["networks"]), 14) - for net in json.loads(listnetworksesponse.content)["networks"]: - self.assertEqual(len(str(net['subnets'][0])), 36) - print(" ") - - - # workflow create floating ip and assign it to a server - - print('->>>>>>> CombinedNeutronCreateFloatingIP ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:19696/v2.0/floatingips" - createflip = requests.post(url, headers=headers, - data='{"floatingip":{"floating_network_id":"default"}}') - self.assertEqual(createflip.status_code, 200) - self.assertIsNotNone(json.loads(createflip.content)["floatingip"].get("port_id")) - port_id = json.loads(createflip.content)["floatingip"].get("port_id") - print(" ") - - print('->>>>>>> CombinedNovaGetServer ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:18774/v2.1/id_bla/servers/detail" - listserverapisdetailedresponse = requests.get(url, headers=headers) - self.assertEqual(listserverapisdetailedresponse.status_code, 200) - self.assertEqual(json.loads(listserverapisdetailedresponse.content)["servers"][0]["status"], "ACTIVE") - server_id = json.loads(listserverapisdetailedresponse.content)["servers"][0]["id"] - print(" ") - - print('->>>>>>> CombinedNovaAssignInterface ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:18774/v2.1/id_bla/servers/%s/os-interface" % server_id - assign = requests.post(url, headers=headers, - data='{"interfaceAttachment":{"net_id": "default"}}') - self.assertEqual(assign.status_code, 202) - self.assertIsNotNone(json.loads(assign.content)["interfaceAttachment"].get("port_id")) - port_id = json.loads(assign.content)["interfaceAttachment"].get("port_id") - print(" ") - - print('->>>>>>> CombinedNovaDeleteInterface ->>>>>>>>>>>>>>>') - print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') - url = "http://0.0.0.0:18774/v2.1/id_bla/servers/%s/os-interface/%s" % (server_id, port_id) - getintfs = requests.delete(url, headers=headers) - self.assertEqual(getintfs.status_code, 202) - print(" ") - + port_request = '{"port": {"name": "%s", "network_id": "%s"}}' + p1_resp = requests.post(url, data=port_request % ("p1", network_id), headers=headers) + self.assertEqual(p1_resp.status_code, 201) + p2_resp = requests.post(url, data=port_request % ("p2", network_id), headers=headers) + self.assertEqual(p2_resp.status_code, 201) + p3_resp = requests.post(url, data=port_request % ("p3", network_id), headers=headers) + self.assertEqual(p3_resp.status_code, 201) + p4_resp = requests.post(url, data=port_request % ("p4", network_id), headers=headers) + self.assertEqual(p4_resp.status_code, 201) + + p1_id = json.loads(p1_resp.content)["port"]["id"] + p2_id = json.loads(p2_resp.content)["port"]["id"] + p3_id = json.loads(p3_resp.content)["port"]["id"] + p4_id = json.loads(p4_resp.content)["port"]["id"] + + print('->>>>>>> test Neutron SFC Port Pair Create ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + url = "http://0.0.0.0:19696/v2.0/sfc/port_pairs" + pp1_resp = requests.post(url, data='{"port_pair": {"name": "pp1", "ingress": "%s", "egress": "%s"}}' % (p1_id, p2_id), headers=headers) + self.assertEqual(pp1_resp.status_code, 201) + pp2_resp = requests.post(url, data='{"port_pair": {"name": "pp2", "ingress": "%s", "egress": "%s"}}' % (p3_id, p4_id), headers=headers) + self.assertEqual(pp2_resp.status_code, 201) + pp3_resp = requests.post(url, data='{"port_pair": {"name": "pp3", "ingress": "%s", "egress": "%s"}}' % (p3_id, p4_id), headers=headers) + self.assertEqual(pp3_resp.status_code, 201) + + pp1_id = json.loads(pp1_resp.content)["port_pair"]["id"] + pp2_id = json.loads(pp2_resp.content)["port_pair"]["id"] + pp3_id = json.loads(pp3_resp.content)["port_pair"]["id"] + + print('->>>>>>> test Neutron SFC Port Pair Update ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + url = "http://0.0.0.0:19696/v2.0/sfc/port_pairs/%s" % pp3_id + pp3_update_resp = requests.put(url, data='{"port_pair": {"description": "port_pair_update"}}', headers=headers) + self.assertEqual(pp3_update_resp.status_code, 200) + self.assertEqual(json.loads(pp3_update_resp.content)["port_pair"]["description"], "port_pair_update") + + print('->>>>>>> test Neutron SFC Port Pair Delete ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + url = "http://0.0.0.0:19696/v2.0/sfc/port_pairs/%s" % pp3_id + pp3_delete_resp = requests.delete(url, headers=headers) + self.assertEqual(pp3_delete_resp.status_code, 204) + + print('->>>>>>> test Neutron SFC Port Pair List ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + url = "http://0.0.0.0:19696/v2.0/sfc/port_pairs" + pp_list_resp = requests.get(url, headers=headers) + self.assertEqual(pp_list_resp.status_code, 200) + self.assertEqual(len(json.loads(pp_list_resp.content)["port_pairs"]), 2) # only pp1 and pp2 should be left + + print('->>>>>>> test Neutron SFC Port Pair Show ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + url = "http://0.0.0.0:19696/v2.0/sfc/port_pairs/%s" % pp2_id + pp2_show_resp = requests.get(url, headers=headers) + self.assertEqual(pp2_show_resp.status_code, 200) + self.assertEqual(json.loads(pp2_show_resp.content)["port_pair"]["name"], "pp2") + + + print('->>>>>>> test Neutron SFC Port Pair Group Create ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + url = "http://0.0.0.0:19696/v2.0/sfc/port_pair_groups" + ppg1_resp = requests.post(url, data='{"port_pair_group": {"name": "ppg1", "port_pairs": ["%s"]}}' % (pp1_id), headers=headers) + self.assertEqual(ppg1_resp.status_code, 201) + ppg2_resp = requests.post(url, data='{"port_pair_group": {"name": "ppg2", "port_pairs": ["%s"]}}' % (pp2_id), headers=headers) + self.assertEqual(ppg2_resp.status_code, 201) + ppg3_resp = requests.post(url, data='{"port_pair_group": {"name": "ppg3", "port_pairs": ["%s"]}}' % (pp2_id), headers=headers) + self.assertEqual(ppg3_resp.status_code, 201) + + ppg1_id = json.loads(ppg1_resp.content)["port_pair_group"]["id"] + ppg2_id = json.loads(ppg2_resp.content)["port_pair_group"]["id"] + ppg3_id = json.loads(ppg3_resp.content)["port_pair_group"]["id"] + + print('->>>>>>> test Neutron SFC Port Pair Group Update ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + url = "http://0.0.0.0:19696/v2.0/sfc/port_pair_groups/%s" % ppg3_id + ppg3_update_resp = requests.put(url, data='{"port_pair_group": {"description": "port_pair_group_update"}}', headers=headers) + self.assertEqual(ppg3_update_resp.status_code, 200) + self.assertEqual(json.loads(ppg3_update_resp.content)["port_pair_group"]["description"], "port_pair_group_update") + + print('->>>>>>> test Neutron SFC Port Pair Group Delete ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + url = "http://0.0.0.0:19696/v2.0/sfc/port_pair_groups/%s" % ppg3_id + ppg3_delete_resp = requests.delete(url, headers=headers) + self.assertEqual(ppg3_delete_resp.status_code, 204) + + print('->>>>>>> test Neutron SFC Port Pair Group List ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + url = "http://0.0.0.0:19696/v2.0/sfc/port_pair_groups" + ppg_list_resp = requests.get(url, headers=headers) + self.assertEqual(ppg_list_resp.status_code, 200) + self.assertEqual(len(json.loads(ppg_list_resp.content)["port_pair_groups"]), 2) # only ppg1 and ppg2 should be left + + print('->>>>>>> test Neutron SFC Port Pair Group Show ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + url = "http://0.0.0.0:19696/v2.0/sfc/port_pair_groups/%s" % ppg2_id + ppg2_show_resp = requests.get(url, headers=headers) + self.assertEqual(ppg2_show_resp.status_code, 200) + self.assertEqual(json.loads(ppg2_show_resp.content)["port_pair_group"]["name"], "ppg2") + + print('->>>>>>> test Neutron SFC Flow Classifier Create ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + url = "http://0.0.0.0:19696/v2.0/sfc/flow_classifiers" + fc1_resp = requests.post(url, data='{"flow_classifier": {"name": "fc1", "source_port_range_min": 22, "source_port_range_max": 4000}}', headers=headers) + self.assertEqual(fc1_resp.status_code, 201) + fc2_resp = requests.post(url, data='{"flow_classifier": {"name": "fc2", "source_port_range_min": 22, "source_port_range_max": 4000}}', headers=headers) + self.assertEqual(fc2_resp.status_code, 201) + + fc1_id = json.loads(fc1_resp.content)["flow_classifier"]["id"] + fc2_id = json.loads(fc2_resp.content)["flow_classifier"]["id"] + + print('->>>>>>> test Neutron SFC Flow Classifier Update ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + url = "http://0.0.0.0:19696/v2.0/sfc/flow_classifiers/%s" % fc2_id + fc2_update_resp = requests.put(url, data='{"flow_classifier": {"description": "flow_classifier_update"}}', headers=headers) + self.assertEqual(fc2_update_resp.status_code, 200) + self.assertEqual(json.loads(fc2_update_resp.content)["flow_classifier"]["description"], "flow_classifier_update") + + print('->>>>>>> test Neutron SFC Flow Classifier Delete ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + url = "http://0.0.0.0:19696/v2.0/sfc/flow_classifiers/%s" % fc2_id + fc2_delete_resp = requests.delete(url, headers=headers) + self.assertEqual(fc2_delete_resp.status_code, 204) + + print('->>>>>>> test Neutron SFC Flow Classifier List ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + url = "http://0.0.0.0:19696/v2.0/sfc/flow_classifiers" + fc_list_resp = requests.get(url, headers=headers) + self.assertEqual(fc_list_resp.status_code, 200) + self.assertEqual(len(json.loads(fc_list_resp.content)["flow_classifiers"]), 1) # only fc1 + + print('->>>>>>> test Neutron SFC Flow Classifier Show ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + url = "http://0.0.0.0:19696/v2.0/sfc/flow_classifiers/%s" % fc1_id + fc1_show_resp = requests.get(url, headers=headers) + self.assertEqual(fc1_show_resp.status_code, 200) + self.assertEqual(json.loads(fc1_show_resp.content)["flow_classifier"]["name"], "fc1") + + + print('->>>>>>> test Neutron SFC Port Chain Create ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + url = "http://0.0.0.0:19696/v2.0/sfc/port_chains" + pc1_resp = requests.post(url, data='{"port_chain": {"name": "pc1", "port_pair_groups": ["%s"], "flow_classifiers": ["%s"]}}' % (ppg1_id, fc1_id), headers=headers) + self.assertEqual(pc1_resp.status_code, 201) + pc2_resp = requests.post(url, data='{"port_chain": {"name": "pc2", "port_pair_groups": ["%s"], "flow_classifiers": ["%s"]}}' % (ppg1_id, fc1_id), headers=headers) + self.assertEqual(pc2_resp.status_code, 201) + + pc1_id = json.loads(pc1_resp.content)["port_chain"]["id"] + pc2_id = json.loads(pc2_resp.content)["port_chain"]["id"] + + print('->>>>>>> test Neutron SFC Port Chain Update ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + url = "http://0.0.0.0:19696/v2.0/sfc/port_chains/%s" % pc2_id + pc2_update_resp = requests.put(url, data='{"port_chain": {"description": "port_chain_update"}}', headers=headers) + self.assertEqual(pc2_update_resp.status_code, 200) + self.assertEqual(json.loads(pc2_update_resp.content)["port_chain"]["description"], "port_chain_update") + + print('->>>>>>> test Neutron SFC Port Chain Delete ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + url = "http://0.0.0.0:19696/v2.0/sfc/port_chains/%s" % pc2_id + pc2_delete_resp = requests.delete(url, headers=headers) + self.assertEqual(pc2_delete_resp.status_code, 204) + + print('->>>>>>> test Neutron SFC Port Chain List ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + url = "http://0.0.0.0:19696/v2.0/sfc/port_chains" + pc_list_resp = requests.get(url, headers=headers) + self.assertEqual(pc_list_resp.status_code, 200) + self.assertEqual(len(json.loads(pc_list_resp.content)["port_chains"]), 1) # only pc1 + + print('->>>>>>> test Neutron SFC Port Chain Show ->>>>>>>>>>>>>>>') + print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>') + url = "http://0.0.0.0:19696/v2.0/sfc/port_chains/%s" % pc1_id + pc1_show_resp = requests.get(url, headers=headers) + self.assertEqual(pc1_show_resp.status_code, 200) + self.assertEqual(json.loads(pc1_show_resp.content)["port_chain"]["name"], "pc1") if __name__ == '__main__': unittest.main()