Create flow table entries for classifiers
[osm/vim-emu.git] / src / emuvim / test / unittests / test_openstack.py
1 # Copyright (c) 2015 SONATA-NFV and Paderborn University
2 # ALL RIGHTS RESERVED.
3 #
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
7 #
8 # http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15 #
16 # Neither the name of the SONATA-NFV, Paderborn University
17 # nor the names of its contributors may be used to endorse or promote
18 # products derived from this software without specific prior written
19 # permission.
20 #
21 # This work has been performed in the framework of the SONATA project,
22 # funded by the European Commission under Grant number 671517 through
23 # the Horizon 2020 and 5G-PPP programmes. The authors would like to
24 # acknowledge the contributions of their colleagues of the SONATA
25 # partner consortium (www.sonata-nfv.eu).
26 import os
27 import unittest
28 import requests
29 import simplejson as json
30 import yaml
31
32 from emuvim.test.api_base_openstack import ApiBaseOpenStack
33
34
35 class testRestApi(ApiBaseOpenStack):
36 """
37 Tests to check the REST API endpoints of the emulator.
38 """
39
40 def setUp(self):
41 # create network
42 self.createNet(nswitches=3, ndatacenter=2, nhosts=2,
43 ndockers=0, autolinkswitches=True)
44
45 # setup links
46 self.net.addLink(self.dc[0], self.h[0])
47 self.net.addLink(self.h[1], self.dc[1])
48 self.net.addLink(self.dc[0], self.dc[1])
49
50 # start api
51 self.startApi()
52
53 # start Mininet network
54 self.startNet()
55
56 def testNovaDummy(self):
57 print('->>>>>>> test Nova Dummy Class->>>>>>>>>>>>>>>')
58 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
59 print(" ")
60
61 headers = {'Content-type': 'application/json'}
62 test_heatapi_template_create_stack = open(os.path.join(os.path.dirname(
63 __file__), "templates/test_heatapi_template_create_stack.yml")).read()
64 url = "http://0.0.0.0:18004/v1/tenantabc123/stacks"
65 requests.post(url, data=json.dumps(yaml.load(test_heatapi_template_create_stack)),
66 headers=headers)
67
68 print('->>>>>>> test Nova List Versions ->>>>>>>>>>>>>>>')
69 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
70 url = "http://0.0.0.0:18774/"
71 listapiversionnovaresponse = requests.get(url, headers=headers)
72 self.assertEqual(listapiversionnovaresponse.status_code, 200)
73 self.assertEqual(json.loads(listapiversionnovaresponse.content)[
74 "versions"][0]["id"], "v2.1")
75 self.assertEqual(json.loads(listapiversionnovaresponse.content)[
76 "versions"][0]["status"], "CURRENT")
77 self.assertEqual(json.loads(listapiversionnovaresponse.content)[
78 "versions"][0]["version"], "2.38")
79 self.assertEqual(json.loads(listapiversionnovaresponse.content)[
80 "versions"][0]["min_version"], "2.1")
81 self.assertEqual(json.loads(listapiversionnovaresponse.content)[
82 "versions"][0]["updated"], "2013-07-23T11:33:21Z")
83 print(" ")
84
85 print('->>>>>>> test Nova Version Show ->>>>>>>>>>>>>>>')
86 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
87 url = "http://0.0.0.0:18774/v2.1/id_bla"
88 listapiversion21novaresponse = requests.get(url, headers=headers)
89 self.assertEqual(listapiversion21novaresponse.status_code, 200)
90 self.assertEqual(json.loads(listapiversion21novaresponse.content)[
91 "version"]["id"], "v2.1")
92 self.assertEqual(json.loads(listapiversion21novaresponse.content)[
93 "version"]["status"], "CURRENT")
94 self.assertEqual(json.loads(listapiversion21novaresponse.content)[
95 "version"]["version"], "2.38")
96 self.assertEqual(json.loads(listapiversion21novaresponse.content)[
97 "version"]["min_version"], "2.1")
98 self.assertEqual(json.loads(listapiversion21novaresponse.content)[
99 "version"]["updated"], "2013-07-23T11:33:21Z")
100 print(" ")
101
102 print('->>>>>>> test Nova Version List Server APIs ->>>>>>>>>>>>>>>')
103 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
104 url = "http://0.0.0.0:18774/v2.1/id_bla/servers"
105 listserverapisnovaresponse = requests.get(url, headers=headers)
106 self.assertEqual(listserverapisnovaresponse.status_code, 200)
107 self.assertNotEqual(json.loads(listserverapisnovaresponse.content)[
108 "servers"][0]["name"], "")
109 print(" ")
110
111 print('->>>>>>> test Nova Delete Server APIs ->>>>>>>>>>>>>>>')
112 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
113 url = "http://0.0.0.0:18774/v2.1/id_bla/servers/%s" % (
114 json.loads(listserverapisnovaresponse.content)["servers"][0]["id"])
115 deleteserverapisnovaresponse = requests.delete(url, headers=headers)
116 self.assertEqual(deleteserverapisnovaresponse.status_code, 204)
117 print(" ")
118
119 print('->>>>>>> test Nova Delete Non-Existing Server APIs ->>>>>>>>>>>>>>>')
120 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
121 url = "http://0.0.0.0:18774/v2.1/id_bla/servers/non-existing-ix"
122 deleteserverapisnovaresponse = requests.delete(url, headers=headers)
123 self.assertEqual(deleteserverapisnovaresponse.status_code, 404)
124 print(" ")
125
126 print('->>>>>>> testNovaVersionListServerAPIs_withPortInformation ->>>>>>>>>>>>>>>')
127 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
128 url = "http://0.0.0.0:18774/v2.1/id_bla/servers/andPorts"
129 listserverapisnovaresponse = requests.get(url, headers=headers)
130 self.assertEqual(listserverapisnovaresponse.status_code, 200)
131 self.assertNotEqual(json.loads(listserverapisnovaresponse.content)[
132 "servers"][0]["name"], "")
133 print(" ")
134
135 print('->>>>>>> test Nova List Flavors ->>>>>>>>>>>>>>>')
136 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
137 url = "http://0.0.0.0:18774/v2.1/id_bla/flavors"
138 listflavorsresponse = requests.get(url, headers=headers)
139 self.assertEqual(listflavorsresponse.status_code, 200)
140 self.assertIn(json.loads(listflavorsresponse.content)["flavors"][0]["name"], [
141 "m1.nano", "m1.tiny", "m1.micro", "m1.small"])
142 self.assertIn(json.loads(listflavorsresponse.content)["flavors"][1]["name"], [
143 "m1.nano", "m1.tiny", "m1.micro", "m1.small"])
144 self.assertIn(json.loads(listflavorsresponse.content)["flavors"][2]["name"], [
145 "m1.nano", "m1.tiny", "m1.micro", "m1.small"])
146 print(" ")
147
148 print('->>>>>>> testNovaAddFlavors ->>>>>>>>>>>>>>>')
149 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
150 url = "http://0.0.0.0:18774/v2.1/id_bla/flavors"
151 addflavorsresponse = requests.post(url,
152 data='{"flavor":{"name": "testFlavor", "vcpus": "test_vcpus", "ram": 1024, "disk": 10}}',
153 headers=headers)
154 self.assertEqual(addflavorsresponse.status_code, 200)
155 self.assertIsNotNone(json.loads(
156 addflavorsresponse.content)["flavor"]["id"])
157 self.assertIsNotNone(json.loads(addflavorsresponse.content)[
158 "flavor"]["links"][0]['href'])
159 print(" ")
160
161 print('->>>>>>> test Nova List Flavors Detail ->>>>>>>>>>>>>>>')
162 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
163 url = "http://0.0.0.0:18774/v2.1/id_bla/flavors/detail"
164 listflavorsdetailresponse = requests.get(url, headers=headers)
165 self.assertEqual(listflavorsdetailresponse.status_code, 200)
166 self.assertIn(json.loads(listflavorsdetailresponse.content)[
167 "flavors"][0]["name"], ["m1.nano", "m1.tiny", "m1.micro", "m1.small"])
168 self.assertIn(json.loads(listflavorsdetailresponse.content)[
169 "flavors"][1]["name"], ["m1.nano", "m1.tiny", "m1.micro", "m1.small"])
170 self.assertIn(json.loads(listflavorsdetailresponse.content)[
171 "flavors"][2]["name"], ["m1.nano", "m1.tiny", "m1.micro", "m1.small"])
172 print(" ")
173
174 print('->>>>>>> testNovaAddFlavors ->>>>>>>>>>>>>>>')
175 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
176 url = "http://0.0.0.0:18774/v2.1/id_bla/flavors/detail"
177 addflavorsresponse = requests.post(url,
178 data='{"flavor":{"name": "testFlavor", "vcpus": "test_vcpus", "ram": 1024, "disk": 10}}',
179 headers=headers)
180 self.assertEqual(addflavorsresponse.status_code, 200)
181 self.assertIsNotNone(json.loads(
182 addflavorsresponse.content)["flavor"]["id"])
183 self.assertIsNotNone(json.loads(addflavorsresponse.content)[
184 "flavor"]["links"][0]['href'])
185 print(" ")
186
187 print('->>>>>>> test Nova List Flavor By Id ->>>>>>>>>>>>>>>')
188
189 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
190 url = "http://0.0.0.0:18774/v2.1/id_bla/flavors/%s" % (
191 json.loads(listflavorsdetailresponse.content)["flavors"][0]["name"])
192 listflavorsbyidresponse = requests.get(url, headers=headers)
193 self.assertEqual(listflavorsbyidresponse.status_code, 200)
194 self.assertEqual(json.loads(listflavorsbyidresponse.content)[
195 "flavor"]["id"], json.loads(listflavorsdetailresponse.content)["flavors"][0]["id"])
196 print(" ")
197
198 print('->>>>>>> test Nova List Images ->>>>>>>>>>>>>>>')
199 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
200 url = "http://0.0.0.0:18774/v2.1/id_bla/images"
201 listimagesresponse = requests.get(url, headers=headers)
202 self.assertEqual(listimagesresponse.status_code, 200)
203 print(listimagesresponse.content)
204 print(" ")
205
206 print('->>>>>>> test Nova List Images Details ->>>>>>>>>>>>>>>')
207 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
208 url = "http://0.0.0.0:18774/v2.1/id_bla/images/detail"
209 listimagesdetailsresponse = requests.get(url, headers=headers)
210 self.assertEqual(listimagesdetailsresponse.status_code, 200)
211 self.assertEqual(json.loads(listimagesdetailsresponse.content)[
212 "images"][0]["metadata"]["architecture"], "x86_64")
213 print(" ")
214
215 print('->>>>>>> test Nova List Image By Id ->>>>>>>>>>>>>>>')
216 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
217 url = "http://0.0.0.0:18774/v2.1/id_bla/images/%s" % (
218 json.loads(listimagesdetailsresponse.content)["images"][0]["id"])
219 listimagebyidresponse = requests.get(url, headers=headers)
220 self.assertEqual(listimagebyidresponse.status_code, 200)
221 self.assertEqual(json.loads(listimagebyidresponse.content)[
222 "image"]["id"], json.loads(listimagesdetailsresponse.content)["images"][0]["id"])
223 print(" ")
224
225 print('->>>>>>> test Nova List Image By Non-Existend Id ->>>>>>>>>>>>>>>')
226 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
227 url = "http://0.0.0.0:18774/v2.1/id_bla/images/non_existing_id"
228 listimagebynonexistingidresponse = requests.get(url, headers=headers)
229 self.assertEqual(listimagebynonexistingidresponse.status_code, 404)
230 print(" ")
231
232 # find ubuntu id
233 for image in json.loads(listimagesresponse.content)["images"]:
234 if image["name"] == "ubuntu:trusty":
235 ubuntu_image_id = image["id"]
236
237 print('->>>>>>> test Nova Create Server Instance ->>>>>>>>>>>>>>>')
238 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
239 url = "http://0.0.0.0:18774/v2.1/id_bla/servers"
240 data = '{"server": {"name": "X", "flavorRef": "%s", "imageRef":"%s"}}' % (
241 json.loads(listflavorsresponse.content)["flavors"][0]["id"], ubuntu_image_id)
242 createserverinstance = requests.post(url, data=data, headers=headers)
243 self.assertEqual(createserverinstance.status_code, 200)
244 self.assertEqual(json.loads(createserverinstance.content)[
245 "server"]["image"]["id"], ubuntu_image_id)
246 print(" ")
247
248 print('->>>>>>> test Nova Create Server Instance With Already Existing Name ->>>>>>>>>>>>>>>')
249 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
250 url = "http://0.0.0.0:18774/v2.1/id_bla/servers"
251 data = '{"server": {"name": "X", "flavorRef": "%s", "imageRef":"%s"}}' % (
252 json.loads(listflavorsresponse.content)["flavors"][0]["id"], ubuntu_image_id)
253 createserverinstance = requests.post(url, data=data, headers=headers)
254 self.assertEqual(createserverinstance.status_code, 409)
255 print(" ")
256
257 print('->>>>>>> test Nova Version List Server APIs Detailed ->>>>>>>>>>>>>>>')
258 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
259 url = "http://0.0.0.0:18774/v2.1/id_bla/servers/detail"
260 listserverapisdetailedresponse = requests.get(url, headers=headers)
261 self.assertEqual(listserverapisdetailedresponse.status_code, 200)
262 self.assertEqual(json.loads(listserverapisdetailedresponse.content)[
263 "servers"][0]["status"], "ACTIVE")
264 print(" ")
265
266 print('->>>>>>> test Nova Show Server Details ->>>>>>>>>>>>>>>')
267 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
268 url = "http://0.0.0.0:18774/v2.1/id_bla/servers/%s" % (
269 json.loads(listserverapisdetailedresponse.content)["servers"][0]["id"])
270 listserverdetailsresponse = requests.get(url, headers=headers)
271 self.assertEqual(listserverdetailsresponse.status_code, 200)
272 self.assertEqual(json.loads(listserverdetailsresponse.content)[
273 "server"]["flavor"]["links"][0]["rel"], "bookmark")
274 print(" ")
275
276 print('->>>>>>> test Nova Show Non-Existing Server Details ->>>>>>>>>>>>>>>')
277 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
278 url = "http://0.0.0.0:18774/v2.1/id_bla/servers/non_existing_server_id"
279 listnonexistingserverdetailsresponse = requests.get(
280 url, headers=headers)
281 self.assertEqual(listnonexistingserverdetailsresponse.status_code, 404)
282 print(" ")
283
284 def testNeutronDummy(self):
285 print('->>>>>>> test Neutron Dummy Class->>>>>>>>>>>>>>>')
286 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
287 print(" ")
288
289 headers = {'Content-type': 'application/json'}
290 test_heatapi_template_create_stack = open(os.path.join(os.path.dirname(
291 __file__), "templates/test_heatapi_template_create_stack.yml")).read()
292 url = "http://0.0.0.0:18004/v1/tenantabc123/stacks"
293 requests.post(url, data=json.dumps(
294 yaml.load(test_heatapi_template_create_stack)), headers=headers)
295 # test_heatapi_keystone_get_token = open("test_heatapi_keystone_get_token.json").read()
296
297 print('->>>>>>> test Neutron List Versions ->>>>>>>>>>>>>>>')
298 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
299 url = "http://0.0.0.0:19696/"
300 listapiversionstackresponse = requests.get(url, headers=headers)
301 self.assertEqual(listapiversionstackresponse.status_code, 200)
302 self.assertEqual(json.loads(listapiversionstackresponse.content)[
303 "versions"][0]["id"], "v2.0")
304 print(" ")
305
306 print('->>>>>>> test Neutron Show API v2.0 ->>>>>>>>>>>>>>>')
307 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
308 url = "http://0.0.0.0:19696/v2.0"
309 listapiversionv20response = requests.get(url, headers=headers)
310 self.assertEqual(listapiversionv20response.status_code, 200)
311 self.assertEqual(json.loads(listapiversionv20response.content)[
312 "resources"][0]["name"], "subnet")
313 self.assertEqual(json.loads(listapiversionv20response.content)[
314 "resources"][1]["name"], "network")
315 self.assertEqual(json.loads(listapiversionv20response.content)[
316 "resources"][2]["name"], "ports")
317 print(" ")
318
319 print('->>>>>>> test Neutron List Networks ->>>>>>>>>>>>>>>')
320 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
321 url = "http://0.0.0.0:19696/v2.0/networks"
322 listnetworksesponse1 = requests.get(url, headers=headers)
323 self.assertEqual(listnetworksesponse1.status_code, 200)
324 self.assertEqual(json.loads(listnetworksesponse1.content)[
325 "networks"][0]["status"], "ACTIVE")
326 listNetworksId = json.loads(listnetworksesponse1.content)[
327 "networks"][0]["id"]
328 listNetworksName = json.loads(listnetworksesponse1.content)[
329 "networks"][0]["name"]
330 listNetworksId2 = json.loads(listnetworksesponse1.content)[
331 "networks"][1]["id"]
332 print(" ")
333
334 print('->>>>>>> test Neutron List Non-Existing Networks ->>>>>>>>>>>>>>>')
335 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
336 url = "http://0.0.0.0:19696/v2.0/networks?name=non_existent_network_name"
337 listnetworksesponse2 = requests.get(url, headers=headers)
338 self.assertEqual(listnetworksesponse2.status_code, 404)
339 print(" ")
340
341 print('->>>>>>> test Neutron List Networks By Name ->>>>>>>>>>>>>>>')
342 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
343 # tcpdump-vnf:input:net:9df6a98f-9e11-4cb7-b3c0-InAdUnitTest
344 url = "http://0.0.0.0:19696/v2.0/networks?name=" + listNetworksName
345 listnetworksesponse3 = requests.get(url, headers=headers)
346 self.assertEqual(listnetworksesponse3.status_code, 200)
347 self.assertEqual(json.loads(listnetworksesponse3.content)[
348 "networks"][0]["name"], listNetworksName)
349 print(" ")
350
351 print('->>>>>>> test Neutron List Networks By Id ->>>>>>>>>>>>>>>')
352 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
353 # tcpdump-vnf:input:net:9df6a98f-9e11-4cb7-b3c0-InAdUnitTest
354 url = "http://0.0.0.0:19696/v2.0/networks?id=" + listNetworksId
355 listnetworksesponse4 = requests.get(url, headers=headers)
356 self.assertEqual(listnetworksesponse4.status_code, 200)
357 self.assertEqual(json.loads(listnetworksesponse4.content)[
358 "networks"][0]["id"], listNetworksId)
359 print(" ")
360
361 print('->>>>>>> test Neutron List Networks By Multiple Ids ->>>>>>>>>>>>>>>')
362 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
363 url = "http://0.0.0.0:19696/v2.0/networks?id=" + listNetworksId + "&id=" + \
364 listNetworksId2 # tcpdump-vnf:input:net:9df6a98f-9e11-4cb7-b3c0-InAdUnitTest
365 listnetworksesponse5 = requests.get(url, headers=headers)
366 self.assertEqual(listnetworksesponse5.status_code, 200)
367 self.assertEqual(json.loads(listnetworksesponse5.content)[
368 "networks"][0]["id"], listNetworksId)
369 self.assertEqual(json.loads(listnetworksesponse5.content)[
370 "networks"][1]["id"], listNetworksId2)
371 print(" ")
372
373 print('->>>>>>> test Neutron Show Network ->>>>>>>>>>>>>>>')
374 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
375 url = "http://0.0.0.0:19696/v2.0/networks/" + listNetworksId
376 shownetworksesponse = requests.get(url, headers=headers)
377 self.assertEqual(shownetworksesponse.status_code, 200)
378 self.assertEqual(json.loads(shownetworksesponse.content)[
379 "network"]["status"], "ACTIVE")
380 print(" ")
381
382 print('->>>>>>> test Neutron Show Network Non-ExistendNetwork ->>>>>>>>>>>>>>>')
383 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
384 url = "http://0.0.0.0:19696/v2.0/networks/non_existent_network_id"
385 shownetworksesponse2 = requests.get(url, headers=headers)
386 self.assertEqual(shownetworksesponse2.status_code, 404)
387 print(" ")
388
389 print('->>>>>>> test Neutron Create Network ->>>>>>>>>>>>>>>')
390 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
391 url = "http://0.0.0.0:19696/v2.0/networks"
392 createnetworkresponse = requests.post(
393 url, data='{"network": {"name": "sample_network","admin_state_up": true}}', headers=headers)
394 self.assertEqual(createnetworkresponse.status_code, 201)
395 self.assertEqual(json.loads(createnetworkresponse.content)[
396 "network"]["status"], "ACTIVE")
397 print(" ")
398
399 print('->>>>>>> test Neutron Create Network With Existing Name ->>>>>>>>>>>>>>>')
400 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
401 url = "http://0.0.0.0:19696/v2.0/networks"
402 createnetworkresponsefailure = requests.post(
403 url, data='{"network": {"name": "sample_network","admin_state_up": true}}', headers=headers)
404 self.assertEqual(createnetworkresponsefailure.status_code, 400)
405 print(" ")
406
407 print('->>>>>>> test Neutron Update Network ->>>>>>>>>>>>>>>')
408 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
409 url = "http://0.0.0.0:19696/v2.0/networks/%s" % (
410 json.loads(createnetworkresponse.content)["network"]["id"])
411 updatenetworkresponse = requests.put(
412 url, data='{"network": {"status": "ACTIVE", "admin_state_up":true, "tenant_id":"abcd123", "name": "sample_network_new_name", "shared":false}}', headers=headers)
413 self.assertEqual(updatenetworkresponse.status_code, 200)
414 self.assertEqual(json.loads(updatenetworkresponse.content)[
415 "network"]["name"], "sample_network_new_name")
416 self.assertEqual(json.loads(updatenetworkresponse.content)[
417 "network"]["tenant_id"], "abcd123")
418 print(" ")
419
420 print('->>>>>>> test Neutron Update Non-Existing Network ->>>>>>>>>>>>>>>')
421 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
422 url = "http://0.0.0.0:19696/v2.0/networks/non-existing-name123"
423 updatenetworkresponse = requests.put(
424 url, data='{"network": {"name": "sample_network_new_name"}}', headers=headers)
425 self.assertEqual(updatenetworkresponse.status_code, 404)
426 print(" ")
427
428 print('->>>>>>> test Neutron List Subnets ->>>>>>>>>>>>>>>')
429 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
430 url = "http://0.0.0.0:19696/v2.0/subnets"
431 listsubnetsresponse = requests.get(url, headers=headers)
432 listSubnetName = json.loads(listsubnetsresponse.content)[
433 "subnets"][0]["name"]
434 listSubnetId = json.loads(listsubnetsresponse.content)[
435 "subnets"][0]["id"]
436 listSubnetId2 = json.loads(listsubnetsresponse.content)[
437 "subnets"][1]["id"]
438 self.assertEqual(listsubnetsresponse.status_code, 200)
439 self.assertNotIn('None', listSubnetName)
440 print(" ")
441
442 print('->>>>>>> test Neutron List Subnets By Name ->>>>>>>>>>>>>>>')
443 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
444 url = "http://0.0.0.0:19696/v2.0/subnets?name=" + listSubnetName
445 listsubnetByNameresponse = requests.get(url, headers=headers)
446 self.assertEqual(listsubnetByNameresponse.status_code, 200)
447 self.assertNotIn('None', json.loads(
448 listsubnetByNameresponse.content)["subnets"][0]["name"])
449 print(" ")
450
451 print('->>>>>>> test Neutron List Subnets By Id ->>>>>>>>>>>>>>>')
452 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
453 url = "http://0.0.0.0:19696/v2.0/subnets?id=" + listSubnetId
454 listsubnetsbyidresponse = requests.get(url, headers=headers)
455 self.assertEqual(listsubnetsbyidresponse.status_code, 200)
456 self.assertNotIn("None", json.loads(
457 listsubnetsbyidresponse.content)["subnets"][0]["name"])
458 print(" ")
459
460 print('->>>>>>> test Neutron List Subnets By Multiple Id ->>>>>>>>>>>>>>>')
461 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
462 url = "http://0.0.0.0:19696/v2.0/subnets?id=" + \
463 listSubnetId + "&id=" + listSubnetId2
464 listsubnetsbymultipleidsresponse = requests.get(url, headers=headers)
465 self.assertEqual(listsubnetsbymultipleidsresponse.status_code, 200)
466 self.assertNotIn("None", json.loads(
467 listsubnetsbymultipleidsresponse.content)["subnets"][0]["name"])
468 print(" ")
469
470 print('->>>>>>> test Neutron Show Subnet->>>>>>>>>>>>>>>')
471 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
472 url = "http://0.0.0.0:19696/v2.0/subnets/%s" % (
473 json.loads(listsubnetsresponse.content)["subnets"][0]["id"])
474 showsubnetsresponse = requests.get(url, headers=headers)
475 self.assertEqual(showsubnetsresponse.status_code, 200)
476 self.assertNotIn("None", json.loads(
477 showsubnetsresponse.content)["subnet"]["name"])
478 print(" ")
479
480 print('->>>>>>> test Neutron Show Non-Existing Subnet->>>>>>>>>>>>>>>')
481 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
482 url = "http://0.0.0.0:19696/v2.0/subnets/non-existing-id123"
483 showsubnetsresponse = requests.get(url, headers=headers)
484 self.assertEqual(showsubnetsresponse.status_code, 404)
485 print(" ")
486
487 print('->>>>>>> test Neutron Create Subnet ->>>>>>>>>>>>>>>')
488 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
489 url = "http://0.0.0.0:19696/v2.0/subnets"
490 createsubnetdata = '{"subnet": {"name": "new_subnet", "network_id": "%s","ip_version": 4,"cidr": "10.0.0.1/24"} }' % (
491 json.loads(createnetworkresponse.content)["network"]["id"])
492 createsubnetresponse = requests.post(
493 url, data=createsubnetdata, headers=headers)
494 self.assertEqual(createsubnetresponse.status_code, 201)
495 self.assertEqual(json.loads(createsubnetresponse.content)[
496 "subnet"]["name"], "new_subnet")
497 print(" ")
498
499 print('->>>>>>> test Neutron Create Second Subnet ->>>>>>>>>>>>>>>')
500 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
501 url = "http://0.0.0.0:19696/v2.0/subnets"
502 createsubnetdata = '{"subnet": {"name": "new_subnet", "network_id": "%s","ip_version": 4,"cidr": "10.0.0.1/24"} }' % (
503 json.loads(createnetworkresponse.content)["network"]["id"])
504 createsubnetfailureresponse = requests.post(
505 url, data=createsubnetdata, headers=headers)
506 self.assertEqual(createsubnetfailureresponse.status_code, 409)
507 print(" ")
508
509 print('->>>>>>> test Neutron Update Subnet ->>>>>>>>>>>>>>>')
510 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
511 url = "http://0.0.0.0:19696/v2.0/subnets/%s" % (
512 json.loads(createsubnetresponse.content)["subnet"]["id"])
513 updatesubnetdata = '{"subnet": {"name": "new_subnet_new_name", "network_id":"some_id", "tenant_id":"new_tenant_id", "allocation_pools":"change_me", "gateway_ip":"192.168.1.120", "ip_version":4, "cidr":"10.0.0.1/24", "id":"some_new_id", "enable_dhcp":true} }'
514 updatesubnetresponse = requests.put(
515 url, data=updatesubnetdata, headers=headers)
516 self.assertEqual(updatesubnetresponse.status_code, 200)
517 self.assertEqual(json.loads(updatesubnetresponse.content)[
518 "subnet"]["name"], "new_subnet_new_name")
519 print(" ")
520
521 print('->>>>>>> test Neutron Update Non-Existing Subnet ->>>>>>>>>>>>>>>')
522 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
523 url = "http://0.0.0.0:19696/v2.0/subnets/non-existing-subnet-12345"
524 updatenonexistingsubnetdata = '{"subnet": {"name": "new_subnet_new_name"} }'
525 updatenonexistingsubnetresponse = requests.put(
526 url, data=updatenonexistingsubnetdata, headers=headers)
527 self.assertEqual(updatenonexistingsubnetresponse.status_code, 404)
528 print(" ")
529
530 print('->>>>>>> test Neutron List Ports ->>>>>>>>>>>>>>>')
531 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
532 url = "http://0.0.0.0:19696/v2.0/ports"
533 listportsesponse = requests.get(url, headers=headers)
534 self.assertEqual(listportsesponse.status_code, 200)
535 self.assertEqual(json.loads(listportsesponse.content)
536 ["ports"][0]["status"], "ACTIVE")
537 listPortsName = json.loads(listportsesponse.content)[
538 "ports"][0]["name"]
539 listPortsId1 = json.loads(listportsesponse.content)["ports"][0]["id"]
540 listPortsId2 = json.loads(listportsesponse.content)["ports"][1]["id"]
541 print(" ")
542
543 print('->>>>>>> test Neutron List Ports By Name ->>>>>>>>>>>>>>>')
544 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
545 url = "http://0.0.0.0:19696/v2.0/ports?name=" + listPortsName
546 listportsbynameesponse = requests.get(url, headers=headers)
547 self.assertEqual(listportsbynameesponse.status_code, 200)
548 self.assertEqual(json.loads(listportsbynameesponse.content)[
549 "ports"][0]["name"], listPortsName)
550 print(" ")
551
552 print('->>>>>>> test Neutron List Ports By Id ->>>>>>>>>>>>>>>')
553 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
554 url = "http://0.0.0.0:19696/v2.0/ports?id=" + listPortsId1
555 listportsbyidesponse = requests.get(url, headers=headers)
556 self.assertEqual(listportsbyidesponse.status_code, 200)
557 self.assertEqual(json.loads(listportsbyidesponse.content)[
558 "ports"][0]["id"], listPortsId1)
559 print(" ")
560
561 print('->>>>>>> test Neutron List Ports By Multiple Ids ->>>>>>>>>>>>>>>')
562 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
563 url = "http://0.0.0.0:19696/v2.0/ports?id=" + \
564 listPortsId1 + "&id=" + listPortsId2
565 listportsbymultipleidsesponse = requests.get(url, headers=headers)
566 self.assertEqual(listportsbymultipleidsesponse.status_code, 200)
567 self.assertEqual(json.loads(listportsbymultipleidsesponse.content)[
568 "ports"][0]["id"], listPortsId1)
569 print(" ")
570
571 print('->>>>>>> test Neutron List Ports By Device ->>>>>>>>>>>>>>>')
572 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
573 server_url = "http://0.0.0.0:18774/v2.1/id_bla/servers/firewall1:9df6a98f-9e11-4cb7-b3c0-InAdUnitTest"
574 server_response = requests.get(server_url, headers=headers)
575 firewall1_server = json.loads(server_response.content)["server"]
576 device_id = firewall1_server["id"]
577 url = "http://0.0.0.0:19696/v2.0/ports?device_id=%s" % device_id
578 list_ports_by_device_id_response = requests.get(url, headers=headers)
579 self.assertEqual(list_ports_by_device_id_response.status_code, 200)
580 list_ports_by_device_id_ports = json.loads(list_ports_by_device_id_response.content)["ports"]
581
582 self.assertTrue(any(list_ports_by_device_id_ports), "Expected at least one port for device")
583 for port in list_ports_by_device_id_ports:
584 self.assertTrue(port["name"].startswith("firewall1:"), "Expected all ports to belong to firewall1")
585 print(" ")
586
587 print('->>>>>>> test Neutron List Non-Existing Ports ->>>>>>>>>>>>>>>')
588 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
589 url = "http://0.0.0.0:19696/v2.0/ports?id=non-existing-port-id"
590 listportsbynonexistingidsesponse = requests.get(url, headers=headers)
591 self.assertEqual(listportsbynonexistingidsesponse.status_code, 404)
592 print(" ")
593
594 print('->>>>>>> test Neutron Show Port ->>>>>>>>>>>>>>>')
595 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
596 url = "http://0.0.0.0:19696/v2.0/ports/%s" % (
597 json.loads(listportsesponse.content)["ports"][0]["id"])
598 showportresponse = requests.get(url, headers=headers)
599 self.assertEqual(showportresponse.status_code, 200)
600 self.assertEqual(json.loads(showportresponse.content)
601 ["port"]["status"], "ACTIVE")
602 print(" ")
603
604 print('->>>>>>> test Neutron Show Non-Existing Port ->>>>>>>>>>>>>>>')
605 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
606 url = "http://0.0.0.0:19696/v2.0/ports/non-existing-portid123"
607 shownonexistingportresponse = requests.get(url, headers=headers)
608 self.assertEqual(shownonexistingportresponse.status_code, 404)
609 print(" ")
610
611 print('->>>>>>> test Neutron Create Port In Non-Existing Network ->>>>>>>>>>>>>>>')
612 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
613 url = "http://0.0.0.0:19696/v2.0/ports"
614 createnonexistingportdata = '{"port": {"name": "new_port", "network_id": "non-existing-id"} }'
615 createnonexistingnetworkportresponse = requests.post(
616 url, data=createnonexistingportdata, headers=headers)
617 self.assertEqual(createnonexistingnetworkportresponse.status_code, 404)
618 print(" ")
619
620 print('->>>>>>> test Neutron Create Port ->>>>>>>>>>>>>>>')
621 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
622 url = "http://0.0.0.0:19696/v2.0/ports"
623 createportdata = '{"port": {"name": "new_port", "network_id": "%s", "admin_state_up":true, "device_id":"device_id123", "device_owner":"device_owner123", "fixed_ips":"change_me","id":"new_id1234", "mac_address":"12:34:56:78:90", "status":"change_me", "tenant_id":"tenant_id123"} }' % (json.loads(createnetworkresponse.content)[
624 "network"]["id"])
625 createportresponse = requests.post(
626 url, data=createportdata, headers=headers)
627 self.assertEqual(createportresponse.status_code, 201)
628 print(createportresponse.content)
629 createport = json.loads(createportresponse.content)["port"]
630 self.assertEqual(createport["name"], "new_port")
631 print(" ")
632
633 print('->>>>>>> test Neutron Create Port With Existing Name ->>>>>>>>>>>>>>>')
634 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
635 url = "http://0.0.0.0:19696/v2.0/ports"
636 network_id = json.loads(createnetworkresponse.content)["network"]["id"]
637 createportwithexistingnamedata = '{"port": {"name": "duplicate_port_name", "network_id": "%s"} }' % network_id
638 createportwithexistingnameresponse1 = requests.post(
639 url, data=createportwithexistingnamedata, headers=headers)
640 createportwithexistingnameresponse2 = requests.post(
641 url, data=createportwithexistingnamedata, headers=headers)
642 createportwithexistingname1 = json.loads(createportwithexistingnameresponse1.content)["port"]
643 createportwithexistingname2 = json.loads(createportwithexistingnameresponse2.content)["port"]
644 self.assertEqual(createportwithexistingnameresponse1.status_code, 201)
645 self.assertEqual(createportwithexistingnameresponse2.status_code, 201)
646 self.assertEqual(createportwithexistingname1["name"], "duplicate_port_name")
647 self.assertEqual(createportwithexistingname2["name"], "duplicate_port_name")
648 self.assertNotEqual(createportwithexistingname1["id"], createportwithexistingname2["id"], "Duplicate port should have different id")
649 print(" ")
650
651 print('->>>>>>> test Neutron Create Port Without Name ->>>>>>>>>>>>>>>')
652 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
653 url = "http://0.0.0.0:19696/v2.0/ports"
654 createportdatawithoutname = '{"port": {"network_id": "%s"} }' % (
655 json.loads(createnetworkresponse.content)["network"]["id"])
656 createportwithoutnameresponse = requests.post(
657 url, data=createportdatawithoutname, headers=headers)
658 self.assertEqual(createportwithoutnameresponse.status_code, 201)
659 self.assertIn("port:cp", json.loads(
660 createportwithoutnameresponse.content)["port"]["name"])
661 print(" ")
662
663 print('->>>>>>> test Neutron Update Port ->>>>>>>>>>>>>>>')
664 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
665 print(json.loads(createportresponse.content)["port"]["name"])
666 url = "http://0.0.0.0:19696/v2.0/ports/%s" % (
667 json.loads(createportresponse.content)["port"]["name"])
668 updateportdata = '{"port": {"name": "new_port_new_name", "admin_state_up":true, "device_id":"device_id123", "device_owner":"device_owner123", "fixed_ips":"change_me","mac_address":"12:34:56:78:90", "status":"change_me", "tenant_id":"tenant_id123", "network_id":"network_id123"} }'
669 updateportresponse = requests.put(
670 url, data=updateportdata, headers=headers)
671 self.assertEqual(updateportresponse.status_code, 200)
672 self.assertEqual(json.loads(updateportresponse.content)[
673 "port"]["name"], "new_port_new_name")
674 print(" ")
675
676 print('->>>>>>> test Neutron Update Non-Existing Port ->>>>>>>>>>>>>>>')
677 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
678 url = "http://0.0.0.0:19696/v2.0/ports/non-existing-port-ip"
679 updatenonexistingportdata = '{"port": {"name": "new_port_new_name"} }'
680 updatenonexistingportresponse = requests.put(
681 url, data=updatenonexistingportdata, headers=headers)
682 self.assertEqual(updatenonexistingportresponse.status_code, 404)
683 print(" ")
684
685 print('->>>>>>> test Neutron Delete Port ->>>>>>>>>>>>>>>')
686 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
687 righturl = "http://0.0.0.0:19696/v2.0/ports/%s" % (
688 json.loads(createportresponse.content)["port"]["id"])
689 deleterightportresponse = requests.delete(righturl, headers=headers)
690 self.assertEqual(deleterightportresponse.status_code, 204)
691 print(" ")
692
693 print('->>>>>>> test Neutron Delete Non-Existing Port ->>>>>>>>>>>>>>>')
694 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
695 wrongurl = "http://0.0.0.0:19696/v2.0/ports/unknownid"
696 deletewrongportresponse = requests.delete(wrongurl, headers=headers)
697 self.assertEqual(deletewrongportresponse.status_code, 404)
698 print(" ")
699
700 print('->>>>>>> test Neutron Delete Subnet ->>>>>>>>>>>>>>>')
701 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
702 wrongurl = "http://0.0.0.0:19696/v2.0/subnets/unknownid"
703 righturl = "http://0.0.0.0:19696/v2.0/subnets/%s" % (
704 json.loads(updatesubnetresponse.content)["subnet"]["id"])
705 deletewrongsubnetresponse = requests.delete(wrongurl, headers=headers)
706 deleterightsubnetresponse = requests.delete(righturl, headers=headers)
707 self.assertEqual(deletewrongsubnetresponse.status_code, 404)
708 self.assertEqual(deleterightsubnetresponse.status_code, 204)
709 print(" ")
710
711 print('->>>>>>> test Neutron Delete Network ->>>>>>>>>>>>>>>')
712 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
713 righturl = "http://0.0.0.0:19696/v2.0/networks/%s" % (
714 json.loads(createnetworkresponse.content)["network"]["id"])
715 deleterightnetworkresponse = requests.delete(righturl, headers=headers)
716 self.assertEqual(deleterightnetworkresponse.status_code, 204)
717 print(" ")
718
719 print('->>>>>>> test Neutron Delete Non-Existing Network ->>>>>>>>>>>>>>>')
720 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
721 wrongurl = "http://0.0.0.0:19696/v2.0/networks/unknownid"
722 deletewrongnetworkresponse = requests.delete(wrongurl, headers=headers)
723 self.assertEqual(deletewrongnetworkresponse.status_code, 404)
724 print(" ")
725
726 def testKeystomeDummy(self):
727 print('->>>>>>> test Keystone Dummy Class->>>>>>>>>>>>>>>')
728 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
729 print(" ")
730
731 headers = {'Content-type': 'application/json'}
732 test_heatapi_keystone_get_token = open(os.path.join(os.path.dirname(
733 __file__), "templates/test_heatapi_keystone_get_token.yml")).read()
734
735 print('->>>>>>> test Keystone List Versions ->>>>>>>>>>>>>>>')
736 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
737 url = "http://0.0.0.0:15000/"
738 listapiversionstackresponse = requests.get(url, headers=headers)
739 self.assertEqual(listapiversionstackresponse.status_code, 200)
740 self.assertEqual(json.loads(listapiversionstackresponse.content)[
741 "versions"]["values"][0]["id"], "v2.0")
742 print(" ")
743
744 print('->>>>>>> test Keystone Show ApiV2 ->>>>>>>>>>>>>>>')
745 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
746 url = "http://0.0.0.0:15000/v2.0"
747 showapiversionstackresponse = requests.get(url, headers=headers)
748 self.assertEqual(showapiversionstackresponse.status_code, 200)
749 self.assertEqual(json.loads(showapiversionstackresponse.content)[
750 "version"]["id"], "v2.0")
751 print(" ")
752
753 print('->>>>>>> test Keystone Get Token ->>>>>>>>>>>>>>>')
754 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
755 url = "http://0.0.0.0:15000/v2.0/tokens"
756 gettokenstackresponse = requests.post(url, data=json.dumps(
757 yaml.load(test_heatapi_keystone_get_token)), headers=headers)
758 self.assertEqual(gettokenstackresponse.status_code, 200)
759 self.assertEqual(json.loads(gettokenstackresponse.content)[
760 "access"]["user"]["name"], "tenantName")
761 print(" ")
762
763 def testHeatDummy(self):
764 print('->>>>>>> test Heat Dummy Class->>>>>>>>>>>>>>>')
765 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
766 print(" ")
767
768 headers = {'Content-type': 'application/json'}
769 test_heatapi_template_create_stack = open(os.path.join(os.path.dirname(
770 __file__), "templates/test_heatapi_template_create_stack.yml")).read()
771 test_heatapi_template_update_stack = open(os.path.join(os.path.dirname(
772 __file__), "templates/test_heatapi_template_update_stack.yml")).read()
773
774 print('->>>>>>> test Heat List API Versions Stack ->>>>>>>>>>>>>>>')
775 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
776 url = "http://0.0.0.0:18004/"
777 listapiversionstackresponse = requests.get(url, headers=headers)
778 self.assertEqual(listapiversionstackresponse.status_code, 200)
779 self.assertEqual(json.loads(listapiversionstackresponse.content)[
780 "versions"][0]["id"], "v1.0")
781 print(" ")
782
783 print('->>>>>>> test Create Stack ->>>>>>>>>>>>>>>')
784 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
785 url = "http://0.0.0.0:18004/v1/tenantabc123/stacks"
786 createstackresponse = requests.post(url, data=json.dumps(
787 yaml.load(test_heatapi_template_create_stack)), headers=headers)
788 self.assertEqual(createstackresponse.status_code, 201)
789 self.assertNotEqual(json.loads(
790 createstackresponse.content)["stack"]["id"], "")
791 print(" ")
792
793 print('->>>>>>> test Create Stack With Existing Name ->>>>>>>>>>>>>>>')
794 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
795 url = "http://0.0.0.0:18004/v1/tenantabc123/stacks"
796 createstackwithexistingnameresponse = requests.post(
797 url, data='{"stack_name" : "s1"}', headers=headers)
798 self.assertEqual(createstackwithexistingnameresponse.status_code, 409)
799 print(" ")
800
801 print('->>>>>>> test Create Stack With Unsupported Version ->>>>>>>>>>>>>>>')
802 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
803 url = "http://0.0.0.0:18004/v1/tenantabc123/stacks"
804 createstackwitheunsupportedversionresponse = requests.post(
805 url, data='{"stack_name" : "stackname123", "template" : {"heat_template_version": "2015-04-29"}}', headers=headers)
806 self.assertEqual(
807 createstackwitheunsupportedversionresponse.status_code, 400)
808 print(" ")
809
810 print('->>>>>>> test List Stack ->>>>>>>>>>>>>>>')
811 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
812 url = "http://0.0.0.0:18004/v1/tenantabc123/stacks"
813 liststackresponse = requests.get(url, headers=headers)
814 self.assertEqual(liststackresponse.status_code, 200)
815 self.assertEqual(json.loads(liststackresponse.content)[
816 "stacks"][0]["stack_status"], "CREATE_COMPLETE")
817 print(" ")
818
819 print('->>>>>>> test Show Stack ->>>>>>>>>>>>>>>')
820 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
821 url = "http://0.0.0.0:18004/v1/tenantabc123showStack/stacks/%s" % json.loads(
822 createstackresponse.content)['stack']['id']
823 liststackdetailsresponse = requests.get(url, headers=headers)
824 self.assertEqual(liststackdetailsresponse.status_code, 200)
825 self.assertEqual(json.loads(liststackdetailsresponse.content)[
826 "stack"]["stack_status"], "CREATE_COMPLETE")
827 print(" ")
828
829 print('->>>>>>> test Show Non-Exisitng Stack ->>>>>>>>>>>>>>>')
830 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
831 url = "http://0.0.0.0:18004/v1/tenantabc123showStack/stacks/non_exisitng_id123"
832 listnonexistingstackdetailsresponse = requests.get(
833 url, headers=headers)
834 self.assertEqual(listnonexistingstackdetailsresponse.status_code, 404)
835 print(" ")
836
837 print('->>>>>>> test Update Stack ->>>>>>>>>>>>>>>')
838 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
839 url = "http://0.0.0.0:18004/v1/tenantabc123updateStack/stacks/%s" % json.loads(
840 createstackresponse.content)['stack']['id']
841 updatestackresponse = requests.put(url, data=json.dumps(yaml.load(test_heatapi_template_update_stack)),
842 headers=headers)
843 self.assertEqual(updatestackresponse.status_code, 202)
844 liststackdetailsresponse = requests.get(url, headers=headers)
845 self.assertEqual(json.loads(liststackdetailsresponse.content)[
846 "stack"]["stack_status"], "UPDATE_COMPLETE")
847 print(" ")
848
849 print('->>>>>>> test Update Non-Existing Stack ->>>>>>>>>>>>>>>')
850 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
851 url = "http://0.0.0.0:18004/v1/tenantabc123updateStack/stacks/non_existing_id_1234"
852 updatenonexistingstackresponse = requests.put(
853 url, data={"non": "sense"}, headers=headers)
854 self.assertEqual(updatenonexistingstackresponse.status_code, 404)
855 print(" ")
856
857 print('->>>>>>> test Delete Stack ->>>>>>>>>>>>>>>')
858 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
859 url = "http://0.0.0.0:18004/v1/tenantabc123showStack/stacks/%s" % \
860 json.loads(createstackresponse.content)['stack']['id']
861 deletestackdetailsresponse = requests.delete(url, headers=headers)
862 self.assertEqual(deletestackdetailsresponse.status_code, 204)
863 print(" ")
864
865 def testNeutronSFC(self):
866 """
867 Tests the Neutron Service Function Chaining implementation. As Some functions build up on others, a
868 complete environment is created here:
869
870 Ports: p1, p2, p3, p4
871 Port Pairs: pp1(p1, p2), pp2(p3, p4)
872 Port Pair Groups: ppg1(pp1, pp2)
873 Flow Classifiers: fc1
874 Port Chain: pc1(ppg1, fc1)
875 """
876
877 headers = {'Content-type': 'application/json'}
878
879 print('->>>>>>> Create ports p1 - p6 ->>>>>>>>>>>>>>>')
880 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
881 # Get network id
882 network_resp = requests.get(
883 "http://0.0.0.0:19696/v2.0/networks?name=default", headers=headers)
884 self.assertEqual(network_resp.status_code, 200)
885 network_id = json.loads(network_resp.content)["networks"][0]["id"]
886
887 port_responses = map(lambda name: requests.post("http://0.0.0.0:19696/v2.0/ports",
888 data='{"port": {"name": "%s", "network_id": "%s"}}' %
889 (name, network_id),
890 headers=headers),
891 ["p1", "p2", "p3", "p4", "p5", "p6"])
892
893 for port in port_responses:
894 self.assertEqual(port.status_code, 201)
895
896 port_ids = map(lambda response: json.loads(response.content)["port"]["id"], port_responses)
897
898 listflavorsresponse = requests.get("http://0.0.0.0:18774/v2.1/id_bla/flavors", headers=headers)
899 self.assertEqual(listflavorsresponse.status_code, 200)
900 flavors = json.loads(listflavorsresponse.content)["flavors"]
901 m1_tiny_flavor = filter(lambda flavor: flavor["name"] == "m1.tiny", flavors)[0]
902
903 listimagesdetailsresponse = requests.get("http://0.0.0.0:18774/v2.1/id_bla/images/detail", headers=headers)
904 self.assertEqual(listimagesdetailsresponse.status_code, 200)
905 images = json.loads(listimagesdetailsresponse.content)["images"]
906 ubuntu_image = filter(lambda image: image["name"] == "ubuntu:trusty", images)[0]
907
908 server_url = "http://0.0.0.0:18774/v2.1/id_bla/servers"
909 server_template = \
910 '{"server": {' \
911 '"name": "%s",' \
912 '"networks": [{"port": "%s"}, {"port": "%s"}],' \
913 '"flavorRef": "%s",' \
914 '"imageRef": "%s"' \
915 '}}'
916 server_responses = map(lambda spec: (
917 requests.post(server_url,
918 data=server_template % (
919 spec["name"],
920 spec["ingress"],
921 spec["egress"],
922 m1_tiny_flavor["id"],
923 ubuntu_image["id"]
924 ),
925 headers=headers)
926 ), [
927 {"name": "s1", "ingress": "p1", "egress": "p2"},
928 {"name": "s2", "ingress": "p3", "egress": "p4"},
929 {"name": "s3", "ingress": "p5", "egress": "p6"},
930 ])
931 for response in server_responses:
932 self.assertEqual(response.status_code, 200)
933
934 print('->>>>>>> test Neutron SFC Port Pair Create ->>>>>>>>>>>>>>>')
935 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
936 url = "http://0.0.0.0:19696/v2.0/sfc/port_pairs"
937 pp1_resp = requests.post(url, data='{"port_pair": {"name": "pp1", "ingress": "%s", "egress": "%s"}}' % (
938 port_ids[0], port_ids[1]), headers=headers)
939 self.assertEqual(pp1_resp.status_code, 201)
940 pp2_resp = requests.post(url, data='{"port_pair": {"name": "pp2", "ingress": "%s", "egress": "%s"}}' % (
941 port_ids[2], port_ids[3]), headers=headers)
942 self.assertEqual(pp2_resp.status_code, 201)
943 pp3_resp = requests.post(url, data='{"port_pair": {"name": "pp3", "ingress": "%s", "egress": "%s"}}' % (
944 port_ids[4], port_ids[5]), headers=headers)
945 self.assertEqual(pp3_resp.status_code, 201)
946
947 pp1_id = json.loads(pp1_resp.content)["port_pair"]["id"]
948 pp2_id = json.loads(pp2_resp.content)["port_pair"]["id"]
949 pp3_id = json.loads(pp3_resp.content)["port_pair"]["id"]
950
951 print('->>>>>>> test Neutron SFC Port Pair Update ->>>>>>>>>>>>>>>')
952 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
953 url = "http://0.0.0.0:19696/v2.0/sfc/port_pairs/%s" % pp3_id
954 pp3_update_resp = requests.put(
955 url, data='{"port_pair": {"description": "port_pair_update"}}', headers=headers)
956 self.assertEqual(pp3_update_resp.status_code, 200)
957 self.assertEqual(json.loads(pp3_update_resp.content)[
958 "port_pair"]["description"], "port_pair_update")
959
960 print('->>>>>>> test Neutron SFC Port Pair Delete ->>>>>>>>>>>>>>>')
961 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
962 url = "http://0.0.0.0:19696/v2.0/sfc/port_pairs/%s" % pp3_id
963 pp3_delete_resp = requests.delete(url, headers=headers)
964 self.assertEqual(pp3_delete_resp.status_code, 204)
965
966 print('->>>>>>> test Neutron SFC Port Pair List ->>>>>>>>>>>>>>>')
967 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
968 url = "http://0.0.0.0:19696/v2.0/sfc/port_pairs"
969 pp_list_resp = requests.get(url, headers=headers)
970 self.assertEqual(pp_list_resp.status_code, 200)
971 pp_list = json.loads(pp_list_resp.content)["port_pairs"]
972 # only pp1 and pp2 should be left
973 self.assertEqual(len(pp_list), 2)
974
975 print('->>>>>>> test Neutron SFC Port Pair List filtered by id ->>>>>>>>>>>>>>>')
976 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
977 url = "http://0.0.0.0:19696/v2.0/sfc/port_pairs?id=%s" % pp_list[0]["id"]
978 pp_list_filtered_by_id_resp = requests.get(url, headers=headers)
979 pp_list_filtered_by_id = json.loads(pp_list_filtered_by_id_resp.content)["port_pairs"]
980 self.assertEqual(pp_list_filtered_by_id_resp.status_code, 200)
981 self.assertEqual(len(pp_list_filtered_by_id), 1)
982 self.assertEqual(pp_list_filtered_by_id[0], pp_list[0])
983
984 print('->>>>>>> test Neutron SFC Port Pair Show ->>>>>>>>>>>>>>>')
985 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
986 url = "http://0.0.0.0:19696/v2.0/sfc/port_pairs/%s" % pp2_id
987 pp2_show_resp = requests.get(url, headers=headers)
988 self.assertEqual(pp2_show_resp.status_code, 200)
989 self.assertEqual(json.loads(pp2_show_resp.content)
990 ["port_pair"]["name"], "pp2")
991
992 print('->>>>>>> test Neutron SFC Port Pair Group Create ->>>>>>>>>>>>>>>')
993 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
994 url = "http://0.0.0.0:19696/v2.0/sfc/port_pair_groups"
995 ppg1_resp = requests.post(
996 url, data='{"port_pair_group": {"name": "ppg1", "port_pairs": ["%s"]}}' % (pp1_id), headers=headers)
997 self.assertEqual(ppg1_resp.status_code, 201)
998 ppg2_resp = requests.post(
999 url, data='{"port_pair_group": {"name": "ppg2", "port_pairs": ["%s"]}}' % (pp2_id), headers=headers)
1000 self.assertEqual(ppg2_resp.status_code, 201)
1001 ppg3_resp = requests.post(
1002 url, data='{"port_pair_group": {"name": "ppg3", "port_pairs": ["%s"]}}' % (pp2_id), headers=headers)
1003 self.assertEqual(ppg3_resp.status_code, 201)
1004
1005 ppg1_id = json.loads(ppg1_resp.content)["port_pair_group"]["id"]
1006 ppg2_id = json.loads(ppg2_resp.content)["port_pair_group"]["id"]
1007 ppg3_id = json.loads(ppg3_resp.content)["port_pair_group"]["id"]
1008
1009 print('->>>>>>> test Neutron SFC Port Pair Group Update ->>>>>>>>>>>>>>>')
1010 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
1011 url = "http://0.0.0.0:19696/v2.0/sfc/port_pair_groups/%s" % ppg3_id
1012 ppg3_update_resp = requests.put(
1013 url, data='{"port_pair_group": {"description": "port_pair_group_update"}}', headers=headers)
1014 self.assertEqual(ppg3_update_resp.status_code, 200)
1015 self.assertEqual(json.loads(ppg3_update_resp.content)[
1016 "port_pair_group"]["description"], "port_pair_group_update")
1017
1018 print('->>>>>>> test Neutron SFC Port Pair Group Delete ->>>>>>>>>>>>>>>')
1019 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
1020 url = "http://0.0.0.0:19696/v2.0/sfc/port_pair_groups/%s" % ppg3_id
1021 ppg3_delete_resp = requests.delete(url, headers=headers)
1022 self.assertEqual(ppg3_delete_resp.status_code, 204)
1023
1024 print('->>>>>>> test Neutron SFC Port Pair Group List ->>>>>>>>>>>>>>>')
1025 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
1026 url = "http://0.0.0.0:19696/v2.0/sfc/port_pair_groups"
1027 ppg_list_resp = requests.get(url, headers=headers)
1028 self.assertEqual(ppg_list_resp.status_code, 200)
1029 # only ppg1 and ppg2 should be left
1030 self.assertEqual(
1031 len(json.loads(ppg_list_resp.content)["port_pair_groups"]), 2)
1032
1033 print('->>>>>>> test Neutron SFC Port Pair Group Show ->>>>>>>>>>>>>>>')
1034 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
1035 url = "http://0.0.0.0:19696/v2.0/sfc/port_pair_groups/%s" % ppg2_id
1036 ppg2_show_resp = requests.get(url, headers=headers)
1037 self.assertEqual(ppg2_show_resp.status_code, 200)
1038 self.assertEqual(json.loads(ppg2_show_resp.content)[
1039 "port_pair_group"]["name"], "ppg2")
1040
1041 print('->>>>>>> test Neutron SFC Flow Classifier Create ->>>>>>>>>>>>>>>')
1042 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
1043 url = "http://0.0.0.0:19696/v2.0/sfc/flow_classifiers"
1044 fc1_resp = requests.post(
1045 url, data='{"flow_classifier": {"name": "fc1", "logical_source_port": "p1", "source_port_range_min": 22, "source_port_range_max": 4000}}', headers=headers)
1046 self.assertEqual(fc1_resp.status_code, 201)
1047 fc2_resp = requests.post(
1048 url, data='{"flow_classifier": {"name": "fc2", "logical_source_port": "p2", "source_port_range_min": 22, "source_port_range_max": 4000}}', headers=headers)
1049 self.assertEqual(fc2_resp.status_code, 201)
1050
1051 fc1_id = json.loads(fc1_resp.content)["flow_classifier"]["id"]
1052 fc2_id = json.loads(fc2_resp.content)["flow_classifier"]["id"]
1053
1054 print('->>>>>>> test Neutron SFC Flow Classifier Update ->>>>>>>>>>>>>>>')
1055 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
1056 url = "http://0.0.0.0:19696/v2.0/sfc/flow_classifiers/%s" % fc2_id
1057 fc2_update_resp = requests.put(
1058 url, data='{"flow_classifier": {"description": "flow_classifier_update"}}', headers=headers)
1059 self.assertEqual(fc2_update_resp.status_code, 200)
1060 self.assertEqual(json.loads(fc2_update_resp.content)[
1061 "flow_classifier"]["description"], "flow_classifier_update")
1062
1063 print('->>>>>>> test Neutron SFC Flow Classifier Delete ->>>>>>>>>>>>>>>')
1064 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
1065 url = "http://0.0.0.0:19696/v2.0/sfc/flow_classifiers/%s" % fc2_id
1066 fc2_delete_resp = requests.delete(url, headers=headers)
1067 self.assertEqual(fc2_delete_resp.status_code, 204)
1068
1069 print('->>>>>>> test Neutron SFC Flow Classifier List ->>>>>>>>>>>>>>>')
1070 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
1071 url = "http://0.0.0.0:19696/v2.0/sfc/flow_classifiers"
1072 fc_list_resp = requests.get(url, headers=headers)
1073 self.assertEqual(fc_list_resp.status_code, 200)
1074 self.assertEqual(len(json.loads(fc_list_resp.content)
1075 ["flow_classifiers"]), 1) # only fc1
1076
1077 print('->>>>>>> test Neutron SFC Flow Classifier Show ->>>>>>>>>>>>>>>')
1078 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
1079 url = "http://0.0.0.0:19696/v2.0/sfc/flow_classifiers/%s" % fc1_id
1080 fc1_show_resp = requests.get(url, headers=headers)
1081 self.assertEqual(fc1_show_resp.status_code, 200)
1082 self.assertEqual(json.loads(fc1_show_resp.content)[
1083 "flow_classifier"]["name"], "fc1")
1084
1085 print('->>>>>>> test Neutron SFC Port Chain Create ->>>>>>>>>>>>>>>')
1086 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
1087 url = "http://0.0.0.0:19696/v2.0/sfc/port_chains"
1088 pc1_resp = requests.post(url, data='{"port_chain": {"name": "pc1", "port_pair_groups": ["%s"], "flow_classifiers": ["%s"]}}' % (
1089 ppg1_id, fc1_id), headers=headers)
1090 self.assertEqual(pc1_resp.status_code, 201)
1091 pc2_resp = requests.post(url, data='{"port_chain": {"name": "pc2", "port_pair_groups": ["%s"], "flow_classifiers": ["%s"]}}' % (
1092 ppg1_id, fc1_id), headers=headers)
1093 self.assertEqual(pc2_resp.status_code, 201)
1094
1095 pc1_id = json.loads(pc1_resp.content)["port_chain"]["id"]
1096 pc2_id = json.loads(pc2_resp.content)["port_chain"]["id"]
1097
1098 print('->>>>>>> test Neutron SFC Port Chain Update ->>>>>>>>>>>>>>>')
1099 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
1100 url = "http://0.0.0.0:19696/v2.0/sfc/port_chains/%s" % pc2_id
1101 pc2_update_resp = requests.put(
1102 url, data='{"port_chain": {"description": "port_chain_update"}}', headers=headers)
1103 self.assertEqual(pc2_update_resp.status_code, 200)
1104 self.assertEqual(json.loads(pc2_update_resp.content)[
1105 "port_chain"]["description"], "port_chain_update")
1106
1107 print('->>>>>>> test Neutron SFC Port Chain Delete ->>>>>>>>>>>>>>>')
1108 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
1109 url = "http://0.0.0.0:19696/v2.0/sfc/port_chains/%s" % pc2_id
1110 pc2_delete_resp = requests.delete(url, headers=headers)
1111 self.assertEqual(pc2_delete_resp.status_code, 204)
1112
1113 print('->>>>>>> test Neutron SFC Port Chain List ->>>>>>>>>>>>>>>')
1114 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
1115 url = "http://0.0.0.0:19696/v2.0/sfc/port_chains"
1116 pc_list_resp = requests.get(url, headers=headers)
1117 self.assertEqual(pc_list_resp.status_code, 200)
1118 self.assertEqual(len(json.loads(pc_list_resp.content)
1119 ["port_chains"]), 1) # only pc1
1120
1121 print('->>>>>>> test Neutron SFC Port Chain Show ->>>>>>>>>>>>>>>')
1122 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
1123 url = "http://0.0.0.0:19696/v2.0/sfc/port_chains/%s" % pc1_id
1124 pc1_show_resp = requests.get(url, headers=headers)
1125 self.assertEqual(pc1_show_resp.status_code, 200)
1126 self.assertEqual(json.loads(pc1_show_resp.content)
1127 ["port_chain"]["name"], "pc1")
1128
1129
1130 if __name__ == '__main__':
1131 unittest.main()