1 # Copyright (c) 2015 SONATA-NFV and Paderborn University
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
8 # http://www.apache.org/licenses/LICENSE-2.0
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.
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
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).
29 import simplejson
as json
32 from emuvim
.test
.api_base_openstack
import ApiBaseOpenStack
35 class testRestApi(ApiBaseOpenStack
):
37 Tests to check the REST API endpoints of the emulator.
42 self
.createNet(nswitches
=3, ndatacenter
=2, nhosts
=2,
43 ndockers
=0, autolinkswitches
=True)
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])
53 # start Mininet network
56 def testNovaDummy(self
):
57 print('->>>>>>> test Nova Dummy Class->>>>>>>>>>>>>>>')
58 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
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
)),
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")
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")
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"], "")
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)
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)
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"], "")
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"])
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}}',
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'])
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"])
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}}',
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'])
187 print('->>>>>>> test Nova List Flavor By Id ->>>>>>>>>>>>>>>')
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"])
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
)
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")
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"])
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)
233 for image
in json
.loads(listimagesresponse
.content
)["images"]:
234 if image
["name"] == "ubuntu:trusty":
235 ubuntu_image_id
= image
["id"]
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
)
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)
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")
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")
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)
284 def testNeutronDummy(self
):
285 print('->>>>>>> test Neutron Dummy Class->>>>>>>>>>>>>>>')
286 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
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()
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")
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")
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
)[
328 listNetworksName
= json
.loads(listnetworksesponse1
.content
)[
329 "networks"][0]["name"]
330 listNetworksId2
= json
.loads(listnetworksesponse1
.content
)[
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)
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
)
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
)
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
)
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")
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)
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")
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)
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")
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)
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
)[
436 listSubnetId2
= json
.loads(listsubnetsresponse
.content
)[
438 self
.assertEqual(listsubnetsresponse
.status_code
, 200)
439 self
.assertNotIn('None', listSubnetName
)
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"])
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"])
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"])
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"])
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)
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")
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)
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")
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)
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
)[
539 listPortsId1
= json
.loads(listportsesponse
.content
)["ports"][0]["id"]
540 listPortsId2
= json
.loads(listportsesponse
.content
)["ports"][1]["id"]
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
)
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
)
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
)
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"]
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")
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)
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")
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)
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)
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
)[
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")
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")
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"])
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")
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)
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)
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)
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)
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)
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)
726 def testKeystomeDummy(self
):
727 print('->>>>>>> test Keystone Dummy Class->>>>>>>>>>>>>>>')
728 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
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()
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")
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")
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")
763 def testHeatDummy(self
):
764 print('->>>>>>> test Heat Dummy Class->>>>>>>>>>>>>>>')
765 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
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()
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")
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"], "")
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)
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
)
807 createstackwitheunsupportedversionresponse
.status_code
, 400)
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")
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")
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)
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
)),
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")
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)
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)
865 def testNeutronSFC(self
):
867 Tests the Neutron Service Function Chaining implementation. As Some functions build up on others, a
868 complete environment is created here:
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)
877 headers
= {'Content-type': 'application/json'}
879 print('->>>>>>> Create ports p1 - p6 ->>>>>>>>>>>>>>>')
880 print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
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"]
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"}}' %
891 ["p1", "p2", "p3", "p4", "p5", "p6"])
893 for port
in port_responses
:
894 self
.assertEqual(port
.status_code
, 201)
896 port_ids
= map(lambda response
: json
.loads(response
.content
)["port"]["id"], port_responses
)
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]
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]
908 server_url
= "http://0.0.0.0:18774/v2.1/id_bla/servers"
912 '"networks": [{"port": "%s"}, {"port": "%s"}],' \
913 '"flavorRef": "%s",' \
916 server_responses
= map(lambda spec
: (
917 requests
.post(server_url
,
918 data
=server_template
% (
922 m1_tiny_flavor
["id"],
927 {"name": "s1", "ingress": "p1", "egress": "p2"},
928 {"name": "s2", "ingress": "p3", "egress": "p4"},
929 {"name": "s3", "ingress": "p5", "egress": "p6"},
931 for response
in server_responses
:
932 self
.assertEqual(response
.status_code
, 200)
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)
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"]
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")
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)
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)
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])
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")
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)
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"]
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")
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)
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
1031 len(json
.loads(ppg_list_resp
.content
)["port_pair_groups"]), 2)
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")
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)
1051 fc1_id
= json
.loads(fc1_resp
.content
)["flow_classifier"]["id"]
1052 fc2_id
= json
.loads(fc2_resp
.content
)["flow_classifier"]["id"]
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")
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)
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
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")
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)
1095 pc1_id
= json
.loads(pc1_resp
.content
)["port_chain"]["id"]
1096 pc2_id
= json
.loads(pc2_resp
.content
)["port_chain"]["id"]
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")
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)
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
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")
1130 if __name__
== '__main__':