Refactoring: Made complete codebase PEP8 compatible.
Only PEP8 rule E501 (line too long) is not yet reflected
by this change.
The patch also adds automated code style checks to the
CI test stage using flake8.
It will let the tests fail if there is a code style
violation.
Change-Id: I90956dd424a46691546ef720351757d3c43451a7
Signed-off-by: peusterm <manuel.peuster@uni-paderborn.de>
diff --git a/src/emuvim/test/__init__.py b/src/emuvim/test/__init__.py
index 7e60065..d888119 100755
--- a/src/emuvim/test/__init__.py
+++ b/src/emuvim/test/__init__.py
@@ -1,27 +1,25 @@
-"""
-Copyright (c) 2015 SONATA-NFV
-ALL RIGHTS RESERVED.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-
-Neither the name of the SONATA-NFV [, ANY ADDITIONAL AFFILIATION]
-nor the names of its contributors may be used to endorse or promote
-products derived from this software without specific prior written
-permission.
-
-This work has been performed in the framework of the SONATA project,
-funded by the European Commission under Grant number 671517 through
-the Horizon 2020 and 5G-PPP programmes. The authors would like to
-acknowledge the contributions of their colleagues of the SONATA
-partner consortium (www.sonata-nfv.eu).
-"""
+# Copyright (c) 2015 SONATA-NFV and Paderborn University
+# ALL RIGHTS RESERVED.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Neither the name of the SONATA-NFV, Paderborn University
+# nor the names of its contributors may be used to endorse or promote
+# products derived from this software without specific prior written
+# permission.
+#
+# This work has been performed in the framework of the SONATA project,
+# funded by the European Commission under Grant number 671517 through
+# the Horizon 2020 and 5G-PPP programmes. The authors would like to
+# acknowledge the contributions of their colleagues of the SONATA
+# partner consortium (www.sonata-nfv.eu).
diff --git a/src/emuvim/test/api_base.py b/src/emuvim/test/api_base.py
index 0d994ca..bf0f4c0 100755
--- a/src/emuvim/test/api_base.py
+++ b/src/emuvim/test/api_base.py
@@ -1,35 +1,28 @@
-"""
-Copyright (c) 2015 SONATA-NFV
-ALL RIGHTS RESERVED.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-
-Neither the name of the SONATA-NFV [, ANY ADDITIONAL AFFILIATION]
-nor the names of its contributors may be used to endorse or promote
-products derived from this software without specific prior written
-permission.
-
-This work has been performed in the framework of the SONATA project,
-funded by the European Commission under Grant number 671517 through
-the Horizon 2020 and 5G-PPP programmes. The authors would like to
-acknowledge the contributions of their colleagues of the SONATA
-partner consortium (www.sonata-nfv.eu).
-"""
-
-"""
-Helper module that implements helpers for test implementations.
-"""
-
+# Copyright (c) 2015 SONATA-NFV and Paderborn University
+# ALL RIGHTS RESERVED.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Neither the name of the SONATA-NFV, Paderborn University
+# nor the names of its contributors may be used to endorse or promote
+# products derived from this software without specific prior written
+# permission.
+#
+# This work has been performed in the framework of the SONATA project,
+# funded by the European Commission under Grant number 671517 through
+# the Horizon 2020 and 5G-PPP programmes. The authors would like to
+# acknowledge the contributions of their colleagues of the SONATA
+# partner consortium (www.sonata-nfv.eu).
import unittest
import os
import subprocess
@@ -39,6 +32,7 @@
from mininet.clean import cleanup
from mininet.node import Controller
+
class SimpleTestTopology(unittest.TestCase):
"""
Helper class to do basic test setups.
@@ -72,7 +66,7 @@
# add some switches
# start from s1 because ovs does not like to have dpid = 0
# and switch name-number is being used by mininet to set the dpid
- for i in range(1, nswitches+1):
+ for i in range(1, nswitches + 1):
self.s.append(self.net.addSwitch('s%d' % i))
# if specified, chain all switches
if autolinkswitches:
@@ -92,14 +86,15 @@
self.h.append(self.net.addHost('h%d' % i))
# add some dockers
for i in range(0, ndockers):
- self.d.append(self.net.addDocker('d%d' % i, dimage="ubuntu:trusty"))
+ self.d.append(self.net.addDocker('d%d' %
+ i, dimage="ubuntu:trusty"))
def startApi(self):
self.api.start()
def stopApi(self):
self.api.stop()
-
+
def startNet(self):
self.net.start()
@@ -119,7 +114,8 @@
"""
List the containers managed by containernet
"""
- return self.getDockerCli().containers(filters={"label": "com.containernet"})
+ return self.getDockerCli().containers(
+ filters={"label": "com.containernet"})
@staticmethod
def setUp():
diff --git a/src/emuvim/test/api_base_openstack.py b/src/emuvim/test/api_base_openstack.py
index c951665..3909e1a 100755
--- a/src/emuvim/test/api_base_openstack.py
+++ b/src/emuvim/test/api_base_openstack.py
@@ -1,35 +1,28 @@
-"""
-Copyright (c) 2015 SONATA-NFV
-ALL RIGHTS RESERVED.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-
-Neither the name of the SONATA-NFV [, ANY ADDITIONAL AFFILIATION]
-nor the names of its contributors may be used to endorse or promote
-products derived from this software without specific prior written
-permission.
-
-This work has been performed in the framework of the SONATA project,
-funded by the European Commission under Grant number 671517 through
-the Horizon 2020 and 5G-PPP programmes. The authors would like to
-acknowledge the contributions of their colleagues of the SONATA
-partner consortium (www.sonata-nfv.eu).
-"""
-
-"""
-Helper module that implements helpers for test implementations.
-"""
-
+# Copyright (c) 2015 SONATA-NFV and Paderborn University
+# ALL RIGHTS RESERVED.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Neither the name of the SONATA-NFV, Paderborn University
+# nor the names of its contributors may be used to endorse or promote
+# products derived from this software without specific prior written
+# permission.
+#
+# This work has been performed in the framework of the SONATA project,
+# funded by the European Commission under Grant number 671517 through
+# the Horizon 2020 and 5G-PPP programmes. The authors would like to
+# acknowledge the contributions of their colleagues of the SONATA
+# partner consortium (www.sonata-nfv.eu).
import unittest
import os
import subprocess
@@ -40,6 +33,7 @@
from mininet.clean import cleanup
from mininet.node import Controller
+
class ApiBaseOpenStack(unittest.TestCase):
"""
Helper class to do basic test setups.
@@ -70,18 +64,19 @@
"""
self.net = DCNetwork(controller=controller, **kwargs)
for i in range(0, ndatacenter):
- self.api.append(OpenstackApiEndpoint("0.0.0.0", 15000+i))
+ self.api.append(OpenstackApiEndpoint("0.0.0.0", 15000 + i))
# add some switches
# start from s1 because ovs does not like to have dpid = 0
# and switch name-number is being used by mininet to set the dpid
- for i in range(1, nswitches+1):
+ for i in range(1, nswitches + 1):
self.s.append(self.net.addSwitch('s%d' % i))
# if specified, chain all switches
if autolinkswitches:
for i in range(0, len(self.s) - 1):
self.net.addLink(self.s[i], self.s[i + 1])
- self.net.addLink(self.s[2], self.s[0]) # link switches s1, s2 and s3
+ # link switches s1, s2 and s3
+ self.net.addLink(self.s[2], self.s[0])
# add some data centers
for i in range(0, ndatacenter):
@@ -89,7 +84,8 @@
self.net.addDatacenter(
'dc%d' % i,
metadata={"unittest_dc": i}))
- self.net.addLink(self.dc[0].switch, self.s[0]) # link switches dc0.s1 with s1
+ # link switches dc0.s1 with s1
+ self.net.addLink(self.dc[0].switch, self.s[0])
# connect data centers to the endpoint
for i in range(0, ndatacenter):
self.api[i].connect_datacenter(self.dc[i])
@@ -99,7 +95,8 @@
self.h.append(self.net.addHost('h%d' % i))
# add some dockers
for i in range(0, ndockers):
- self.d.append(self.net.addDocker('d%d' % i, dimage="ubuntu:trusty"))
+ self.d.append(self.net.addDocker('d%d' %
+ i, dimage="ubuntu:trusty"))
def startApi(self):
for i in self.api:
@@ -129,25 +126,26 @@
"""
List the containers managed by containernet
"""
- return self.getDockerCli().containers(filters={"label": "com.containernet"})
+ return self.getDockerCli().containers(
+ filters={"label": "com.containernet"})
@staticmethod
def setUp():
pass
-
def tearDown(self):
time.sleep(2)
print('->>>>>>> tear everything down ->>>>>>>>>>>>>>>')
- self.stopApi() # stop all flask threads
- self.stopNet() # stop some mininet and containernet stuff
+ self.stopApi() # stop all flask threads
+ self.stopNet() # stop some mininet and containernet stuff
cleanup()
# make sure that all pending docker containers are killed
- with open(os.devnull, 'w') as devnull: # kill a possibly running docker process that blocks the open ports
+ # kill a possibly running docker process that blocks the open ports
+ with open(os.devnull, 'w') as devnull:
subprocess.call("kill $(netstat -npl | grep '15000' | grep -o -e'[0-9]\+/docker' | grep -o -e '[0-9]\+')",
- stdout=devnull,
- stderr=devnull,
- shell=True)
+ stdout=devnull,
+ stderr=devnull,
+ shell=True)
with open(os.devnull, 'w') as devnull:
subprocess.call(
@@ -156,8 +154,3 @@
stderr=devnull,
shell=True)
time.sleep(2)
-
-
-
-
-
diff --git a/src/emuvim/test/base.py b/src/emuvim/test/base.py
index 397e2a1..6221765 100755
--- a/src/emuvim/test/base.py
+++ b/src/emuvim/test/base.py
@@ -1,35 +1,28 @@
-"""
-Copyright (c) 2015 SONATA-NFV
-ALL RIGHTS RESERVED.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-
-Neither the name of the SONATA-NFV [, ANY ADDITIONAL AFFILIATION]
-nor the names of its contributors may be used to endorse or promote
-products derived from this software without specific prior written
-permission.
-
-This work has been performed in the framework of the SONATA project,
-funded by the European Commission under Grant number 671517 through
-the Horizon 2020 and 5G-PPP programmes. The authors would like to
-acknowledge the contributions of their colleagues of the SONATA
-partner consortium (www.sonata-nfv.eu).
-"""
-
-"""
-Helper module that implements helpers for test implementations.
-"""
-
+# Copyright (c) 2015 SONATA-NFV and Paderborn University
+# ALL RIGHTS RESERVED.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Neither the name of the SONATA-NFV, Paderborn University
+# nor the names of its contributors may be used to endorse or promote
+# products derived from this software without specific prior written
+# permission.
+#
+# This work has been performed in the framework of the SONATA project,
+# funded by the European Commission under Grant number 671517 through
+# the Horizon 2020 and 5G-PPP programmes. The authors would like to
+# acknowledge the contributions of their colleagues of the SONATA
+# partner consortium (www.sonata-nfv.eu).
import unittest
import os
import subprocess
@@ -38,6 +31,7 @@
from mininet.clean import cleanup
from mininet.node import Controller
+
class SimpleTestTopology(unittest.TestCase):
"""
Helper class to do basic test setups.
@@ -70,7 +64,7 @@
# add some switches
# start from s1 because ovs does not like to have dpid = 0
# and switch name-number is being used by mininet to set the dpid
- for i in range(1, nswitches+1):
+ for i in range(1, nswitches + 1):
self.s.append(self.net.addSwitch('s%d' % i))
# if specified, chain all switches
if autolinkswitches:
@@ -87,7 +81,8 @@
self.h.append(self.net.addHost('h%d' % i))
# add some dockers
for i in range(0, ndockers):
- self.d.append(self.net.addDocker('d%d' % i, dimage="ubuntu:trusty"))
+ self.d.append(self.net.addDocker('d%d' %
+ i, dimage="ubuntu:trusty"))
def startNet(self):
self.net.start()
@@ -108,7 +103,8 @@
"""
List the containers managed by containernet
"""
- return self.getDockerCli().containers(filters={"label": "com.containernet"})
+ return self.getDockerCli().containers(
+ filters={"label": "com.containernet"})
@staticmethod
def setUp():
@@ -123,4 +119,4 @@
"sudo docker rm -f $(sudo docker ps --filter 'label=com.containernet' -a -q)",
stdout=devnull,
stderr=devnull,
- shell=True)
\ No newline at end of file
+ shell=True)
diff --git a/src/emuvim/test/integrationtests/__init__.py b/src/emuvim/test/integrationtests/__init__.py
index 7e60065..d888119 100755
--- a/src/emuvim/test/integrationtests/__init__.py
+++ b/src/emuvim/test/integrationtests/__init__.py
@@ -1,27 +1,25 @@
-"""
-Copyright (c) 2015 SONATA-NFV
-ALL RIGHTS RESERVED.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-
-Neither the name of the SONATA-NFV [, ANY ADDITIONAL AFFILIATION]
-nor the names of its contributors may be used to endorse or promote
-products derived from this software without specific prior written
-permission.
-
-This work has been performed in the framework of the SONATA project,
-funded by the European Commission under Grant number 671517 through
-the Horizon 2020 and 5G-PPP programmes. The authors would like to
-acknowledge the contributions of their colleagues of the SONATA
-partner consortium (www.sonata-nfv.eu).
-"""
+# Copyright (c) 2015 SONATA-NFV and Paderborn University
+# ALL RIGHTS RESERVED.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Neither the name of the SONATA-NFV, Paderborn University
+# nor the names of its contributors may be used to endorse or promote
+# products derived from this software without specific prior written
+# permission.
+#
+# This work has been performed in the framework of the SONATA project,
+# funded by the European Commission under Grant number 671517 through
+# the Horizon 2020 and 5G-PPP programmes. The authors would like to
+# acknowledge the contributions of their colleagues of the SONATA
+# partner consortium (www.sonata-nfv.eu).
diff --git a/src/emuvim/test/unittests/__init__.py b/src/emuvim/test/unittests/__init__.py
index 7e60065..d888119 100755
--- a/src/emuvim/test/unittests/__init__.py
+++ b/src/emuvim/test/unittests/__init__.py
@@ -1,27 +1,25 @@
-"""
-Copyright (c) 2015 SONATA-NFV
-ALL RIGHTS RESERVED.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-
-Neither the name of the SONATA-NFV [, ANY ADDITIONAL AFFILIATION]
-nor the names of its contributors may be used to endorse or promote
-products derived from this software without specific prior written
-permission.
-
-This work has been performed in the framework of the SONATA project,
-funded by the European Commission under Grant number 671517 through
-the Horizon 2020 and 5G-PPP programmes. The authors would like to
-acknowledge the contributions of their colleagues of the SONATA
-partner consortium (www.sonata-nfv.eu).
-"""
+# Copyright (c) 2015 SONATA-NFV and Paderborn University
+# ALL RIGHTS RESERVED.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Neither the name of the SONATA-NFV, Paderborn University
+# nor the names of its contributors may be used to endorse or promote
+# products derived from this software without specific prior written
+# permission.
+#
+# This work has been performed in the framework of the SONATA project,
+# funded by the European Commission under Grant number 671517 through
+# the Horizon 2020 and 5G-PPP programmes. The authors would like to
+# acknowledge the contributions of their colleagues of the SONATA
+# partner consortium (www.sonata-nfv.eu).
diff --git a/src/emuvim/test/unittests/test_emulator.py b/src/emuvim/test/unittests/test_emulator.py
index 94db55f..fc06a69 100755
--- a/src/emuvim/test/unittests/test_emulator.py
+++ b/src/emuvim/test/unittests/test_emulator.py
@@ -1,47 +1,36 @@
-"""
-Copyright (c) 2015 SONATA-NFV
-ALL RIGHTS RESERVED.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-
-Neither the name of the SONATA-NFV [, ANY ADDITIONAL AFFILIATION]
-nor the names of its contributors may be used to endorse or promote
-products derived from this software without specific prior written
-permission.
-
-This work has been performed in the framework of the SONATA project,
-funded by the European Commission under Grant number 671517 through
-the Horizon 2020 and 5G-PPP programmes. The authors would like to
-acknowledge the contributions of their colleagues of the SONATA
-partner consortium (www.sonata-nfv.eu).
-"""
-"""
-Test suite to automatically test emulator functionalities.
-Directly interacts with the emulator through the Mininet-like
-Python API.
-
-Does not test API endpoints. This is done in separated test suites.
-"""
-
-import time
+# Copyright (c) 2015 SONATA-NFV and Paderborn University
+# ALL RIGHTS RESERVED.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Neither the name of the SONATA-NFV, Paderborn University
+# nor the names of its contributors may be used to endorse or promote
+# products derived from this software without specific prior written
+# permission.
+#
+# This work has been performed in the framework of the SONATA project,
+# funded by the European Commission under Grant number 671517 through
+# the Horizon 2020 and 5G-PPP programmes. The authors would like to
+# acknowledge the contributions of their colleagues of the SONATA
+# partner consortium (www.sonata-nfv.eu).
import unittest
from emuvim.dcemulator.node import EmulatorCompute
from emuvim.test.base import SimpleTestTopology
from mininet.node import RemoteController
-#@unittest.skip("disabled topology tests for development")
-class testEmulatorTopology( SimpleTestTopology ):
+# @unittest.skip("disabled topology tests for development")
+class testEmulatorTopology(SimpleTestTopology):
"""
Tests to check the topology API of the emulator.
"""
@@ -68,7 +57,7 @@
# stop Mininet network
self.stopNet()
- #@unittest.skip("disabled to test if CI fails because this is the first test.")
+ # @unittest.skip("disabled to test if CI fails because this is the first test.")
def testMultipleDatacenterDirect(self):
"""
Create a two data centers and interconnect them.
@@ -115,7 +104,8 @@
# stop Mininet network
self.stopNet()
-class testEmulatorNetworking( SimpleTestTopology ):
+
+class testEmulatorNetworking(SimpleTestTopology):
def testSDNChainingSingleService_withLearning(self):
"""
@@ -137,8 +127,10 @@
self.startNet()
# add compute resources
- vnf1 = self.dc[0].startCompute("vnf1", network=[{'id':'intf1', 'ip':'10.0.10.1/24'}])
- vnf2 = self.dc[1].startCompute("vnf2", network=[{'id':'intf2', 'ip':'10.0.10.2/24'}])
+ vnf1 = self.dc[0].startCompute(
+ "vnf1", network=[{'id': 'intf1', 'ip': '10.0.10.1/24'}])
+ vnf2 = self.dc[1].startCompute(
+ "vnf2", network=[{'id': 'intf2', 'ip': '10.0.10.2/24'}])
# check number of running nodes
self.assertTrue(len(self.getContainernetContainers()) == 2)
self.assertTrue(len(self.net.hosts) == 2)
@@ -162,7 +154,8 @@
# should be connected because learning = True
self.assertTrue(self.net.ping([vnf1, vnf2]) <= 0.0)
# setup links
- self.net.setChain('vnf1', 'vnf2', 'intf1', 'intf2', bidirectional=True, cmd='add-flow')
+ self.net.setChain('vnf1', 'vnf2', 'intf1', 'intf2',
+ bidirectional=True, cmd='add-flow')
# should still be connected
self.assertTrue(self.net.ping([vnf1, vnf2]) <= 0.0)
# stop Mininet network
@@ -188,8 +181,10 @@
self.startNet()
# add compute resources
- vnf1 = self.dc[0].startCompute("vnf1", network=[{'id':'intf1', 'ip':'10.0.10.1/24'}])
- vnf2 = self.dc[1].startCompute("vnf2", network=[{'id':'intf2', 'ip':'10.0.10.2/24'}])
+ vnf1 = self.dc[0].startCompute(
+ "vnf1", network=[{'id': 'intf1', 'ip': '10.0.10.1/24'}])
+ vnf2 = self.dc[1].startCompute(
+ "vnf2", network=[{'id': 'intf2', 'ip': '10.0.10.2/24'}])
# check number of running nodes
self.assertTrue(len(self.getContainernetContainers()) == 2)
self.assertTrue(len(self.net.hosts) == 2)
@@ -213,7 +208,8 @@
# should be not not yet connected
self.assertTrue(self.net.ping([vnf1, vnf2]) > 0.0)
# setup links
- self.net.setChain('vnf1', 'vnf2', 'intf1', 'intf2', bidirectional=True, cmd='add-flow')
+ self.net.setChain('vnf1', 'vnf2', 'intf1', 'intf2',
+ bidirectional=True, cmd='add-flow')
# check connectivity by using ping
self.assertTrue(self.net.ping([vnf1, vnf2]) <= 0.0)
# stop Mininet network
@@ -239,19 +235,24 @@
# start Mininet network
self.startNet()
- ## First Service
+ # First Service
# add compute resources
- vnf1 = self.dc[0].startCompute("vnf1", network=[{'id': 'intf1', 'ip': '10.0.10.1/24'}])
- vnf2 = self.dc[1].startCompute("vnf2", network=[{'id': 'intf2', 'ip': '10.0.10.2/24'}])
+ vnf1 = self.dc[0].startCompute(
+ "vnf1", network=[{'id': 'intf1', 'ip': '10.0.10.1/24'}])
+ vnf2 = self.dc[1].startCompute(
+ "vnf2", network=[{'id': 'intf2', 'ip': '10.0.10.2/24'}])
# setup links
- self.net.setChain('vnf1', 'vnf2', 'intf1', 'intf2', bidirectional=True, cmd='add-flow', cookie=1)
+ self.net.setChain('vnf1', 'vnf2', 'intf1', 'intf2',
+ bidirectional=True, cmd='add-flow', cookie=1)
# check connectivity by using ping
self.assertTrue(self.net.ping([vnf1, vnf2]) <= 0.0)
- ## Second Service
+ # Second Service
# add compute resources
- vnf11 = self.dc[0].startCompute("vnf11", network=[{'id': 'intf1', 'ip': '10.0.20.1/24'}])
- vnf22 = self.dc[1].startCompute("vnf22", network=[{'id': 'intf2', 'ip': '10.0.20.2/24'}])
+ vnf11 = self.dc[0].startCompute(
+ "vnf11", network=[{'id': 'intf1', 'ip': '10.0.20.1/24'}])
+ vnf22 = self.dc[1].startCompute(
+ "vnf22", network=[{'id': 'intf2', 'ip': '10.0.20.2/24'}])
# check number of running nodes
self.assertTrue(len(self.getContainernetContainers()) == 4)
@@ -259,7 +260,8 @@
self.assertTrue(len(self.net.switches) == 5)
# setup links
- self.net.setChain('vnf11', 'vnf22', 'intf1', 'intf2', bidirectional=True, cmd='add-flow', cookie=2)
+ self.net.setChain('vnf11', 'vnf22', 'intf1', 'intf2',
+ bidirectional=True, cmd='add-flow', cookie=2)
# check connectivity by using ping
self.assertTrue(self.net.ping([vnf11, vnf22]) <= 0.0)
# check first service cannot ping second service
@@ -267,18 +269,21 @@
self.assertTrue(self.net.ping([vnf2, vnf11]) > 0.0)
# delete the first service chain
- self.net.setChain('vnf1', 'vnf2', 'intf1', 'intf2', bidirectional=True, cmd='del-flows', cookie=1)
+ self.net.setChain('vnf1', 'vnf2', 'intf1', 'intf2',
+ bidirectional=True, cmd='del-flows', cookie=1)
# check connectivity of first service is down
self.assertTrue(self.net.ping([vnf1, vnf2]) > 0.0)
- #time.sleep(100)
+ # time.sleep(100)
# check connectivity of second service is still up
self.assertTrue(self.net.ping([vnf11, vnf22]) <= 0.0)
# stop Mininet network
self.stopNet()
-#@unittest.skip("disabled compute tests for development")
-class testEmulatorCompute( SimpleTestTopology ):
+# @unittest.skip("disabled compute tests for development")
+
+
+class testEmulatorCompute(SimpleTestTopology):
"""
Tests to check the emulator's API to add and remove
compute resources at runtime.
@@ -304,7 +309,8 @@
self.assertTrue(len(self.net.switches) == 1)
# check compute list result
self.assertTrue(len(self.dc[0].listCompute()) == 1)
- self.assertTrue(isinstance(self.dc[0].listCompute()[0], EmulatorCompute))
+ self.assertTrue(isinstance(
+ self.dc[0].listCompute()[0], EmulatorCompute))
self.assertTrue(self.dc[0].listCompute()[0].name == "vnf1")
# check connectivity by using ping
self.assertTrue(self.net.ping([self.h[0], vnf1]) <= 0.0)
@@ -362,7 +368,8 @@
self.assertTrue(len(self.net.switches) == 1)
# check compute list result
self.assertTrue(len(self.dc[0].listCompute()) == 1)
- self.assertTrue(isinstance(self.dc[0].listCompute()[0], EmulatorCompute))
+ self.assertTrue(isinstance(
+ self.dc[0].listCompute()[0], EmulatorCompute))
self.assertTrue(self.dc[0].listCompute()[0].name == "vnf1")
# check connectivity by using ping
self.assertTrue(self.net.ping([self.h[0], vnf1]) <= 0.0)
@@ -407,7 +414,7 @@
Test multiple, interleaved add and remove operations and ensure
that always all expected compute instances are reachable.
"""
- # create network
+ # create network
self.createNet(
nswitches=3, ndatacenter=2, nhosts=0, ndockers=0,
autolinkswitches=True)
@@ -455,5 +462,6 @@
# stop Mininet network
self.stopNet()
+
if __name__ == '__main__':
unittest.main()
diff --git a/src/emuvim/test/unittests/test_openstack.py b/src/emuvim/test/unittests/test_openstack.py
index 3e9037a..a9273fb 100755
--- a/src/emuvim/test/unittests/test_openstack.py
+++ b/src/emuvim/test/unittests/test_openstack.py
@@ -1,41 +1,33 @@
-"""
-Copyright (c) 2015 SONATA-NFV
-ALL RIGHTS RESERVED.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-
-Neither the name of the SONATA-NFV [, ANY ADDITIONAL AFFILIATION]
-nor the names of its contributors may be used to endorse or promote
-products derived from this software without specific prior written
-permission.
-
-This work has been performed in the framework of the SONATA project,
-funded by the European Commission under Grant number 671517 through
-the Horizon 2020 and 5G-PPP programmes. The authors would like to
-acknowledge the contributions of their colleagues of the SONATA
-partner consortium (www.sonata-nfv.eu).
-"""
-
-"""
-Test suite to automatically test emulator REST API endpoints.
-"""
-
+# Copyright (c) 2015 SONATA-NFV and Paderborn University
+# ALL RIGHTS RESERVED.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Neither the name of the SONATA-NFV, Paderborn University
+# nor the names of its contributors may be used to endorse or promote
+# products derived from this software without specific prior written
+# permission.
+#
+# This work has been performed in the framework of the SONATA project,
+# funded by the European Commission under Grant number 671517 through
+# the Horizon 2020 and 5G-PPP programmes. The authors would like to
+# acknowledge the contributions of their colleagues of the SONATA
+# partner consortium (www.sonata-nfv.eu).
import os
import unittest
import requests
import simplejson as json
import yaml
-import time
from emuvim.test.api_base_openstack import ApiBaseOpenStack
@@ -47,7 +39,8 @@
def setUp(self):
# create network
- self.createNet(nswitches=3, ndatacenter=2, nhosts=2, ndockers=0, autolinkswitches=True)
+ self.createNet(nswitches=3, ndatacenter=2, nhosts=2,
+ ndockers=0, autolinkswitches=True)
# setup links
self.net.addLink(self.dc[0], self.h[0])
@@ -66,7 +59,8 @@
print(" ")
headers = {'Content-type': 'application/json'}
- test_heatapi_template_create_stack = open(os.path.join(os.path.dirname(__file__), "templates/test_heatapi_template_create_stack.yml")).read()
+ test_heatapi_template_create_stack = open(os.path.join(os.path.dirname(
+ __file__), "templates/test_heatapi_template_create_stack.yml")).read()
url = "http://0.0.0.0:18004/v1/tenantabc123/stacks"
requests.post(url, data=json.dumps(yaml.load(test_heatapi_template_create_stack)),
headers=headers)
@@ -76,11 +70,16 @@
url = "http://0.0.0.0:18774/"
listapiversionnovaresponse = requests.get(url, headers=headers)
self.assertEqual(listapiversionnovaresponse.status_code, 200)
- self.assertEqual(json.loads(listapiversionnovaresponse.content)["versions"][0]["id"], "v2.1")
- self.assertEqual(json.loads(listapiversionnovaresponse.content)["versions"][0]["status"], "CURRENT")
- self.assertEqual(json.loads(listapiversionnovaresponse.content)["versions"][0]["version"], "2.38")
- self.assertEqual(json.loads(listapiversionnovaresponse.content)["versions"][0]["min_version"], "2.1")
- self.assertEqual(json.loads(listapiversionnovaresponse.content)["versions"][0]["updated"], "2013-07-23T11:33:21Z")
+ self.assertEqual(json.loads(listapiversionnovaresponse.content)[
+ "versions"][0]["id"], "v2.1")
+ self.assertEqual(json.loads(listapiversionnovaresponse.content)[
+ "versions"][0]["status"], "CURRENT")
+ self.assertEqual(json.loads(listapiversionnovaresponse.content)[
+ "versions"][0]["version"], "2.38")
+ self.assertEqual(json.loads(listapiversionnovaresponse.content)[
+ "versions"][0]["min_version"], "2.1")
+ self.assertEqual(json.loads(listapiversionnovaresponse.content)[
+ "versions"][0]["updated"], "2013-07-23T11:33:21Z")
print(" ")
print('->>>>>>> test Nova Version Show ->>>>>>>>>>>>>>>')
@@ -88,11 +87,16 @@
url = "http://0.0.0.0:18774/v2.1/id_bla"
listapiversion21novaresponse = requests.get(url, headers=headers)
self.assertEqual(listapiversion21novaresponse.status_code, 200)
- self.assertEqual(json.loads(listapiversion21novaresponse.content)["version"]["id"], "v2.1")
- self.assertEqual(json.loads(listapiversion21novaresponse.content)["version"]["status"], "CURRENT")
- self.assertEqual(json.loads(listapiversion21novaresponse.content)["version"]["version"], "2.38")
- self.assertEqual(json.loads(listapiversion21novaresponse.content)["version"]["min_version"], "2.1")
- self.assertEqual(json.loads(listapiversion21novaresponse.content)["version"]["updated"], "2013-07-23T11:33:21Z")
+ self.assertEqual(json.loads(listapiversion21novaresponse.content)[
+ "version"]["id"], "v2.1")
+ self.assertEqual(json.loads(listapiversion21novaresponse.content)[
+ "version"]["status"], "CURRENT")
+ self.assertEqual(json.loads(listapiversion21novaresponse.content)[
+ "version"]["version"], "2.38")
+ self.assertEqual(json.loads(listapiversion21novaresponse.content)[
+ "version"]["min_version"], "2.1")
+ self.assertEqual(json.loads(listapiversion21novaresponse.content)[
+ "version"]["updated"], "2013-07-23T11:33:21Z")
print(" ")
print('->>>>>>> test Nova Version List Server APIs ->>>>>>>>>>>>>>>')
@@ -100,12 +104,14 @@
url = "http://0.0.0.0:18774/v2.1/id_bla/servers"
listserverapisnovaresponse = requests.get(url, headers=headers)
self.assertEqual(listserverapisnovaresponse.status_code, 200)
- self.assertNotEqual(json.loads(listserverapisnovaresponse.content)["servers"][0]["name"], "")
+ self.assertNotEqual(json.loads(listserverapisnovaresponse.content)[
+ "servers"][0]["name"], "")
print(" ")
print('->>>>>>> test Nova Delete Server APIs ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- url = "http://0.0.0.0:18774/v2.1/id_bla/servers/%s" % (json.loads(listserverapisnovaresponse.content)["servers"][0]["id"])
+ url = "http://0.0.0.0:18774/v2.1/id_bla/servers/%s" % (
+ json.loads(listserverapisnovaresponse.content)["servers"][0]["id"])
deleteserverapisnovaresponse = requests.delete(url, headers=headers)
self.assertEqual(deleteserverapisnovaresponse.status_code, 204)
print(" ")
@@ -117,13 +123,13 @@
self.assertEqual(deleteserverapisnovaresponse.status_code, 404)
print(" ")
-
print('->>>>>>> testNovaVersionListServerAPIs_withPortInformation ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:18774/v2.1/id_bla/servers/andPorts"
listserverapisnovaresponse = requests.get(url, headers=headers)
self.assertEqual(listserverapisnovaresponse.status_code, 200)
- self.assertNotEqual(json.loads(listserverapisnovaresponse.content)["servers"][0]["name"], "")
+ self.assertNotEqual(json.loads(listserverapisnovaresponse.content)[
+ "servers"][0]["name"], "")
print(" ")
print('->>>>>>> test Nova List Flavors ->>>>>>>>>>>>>>>')
@@ -131,9 +137,12 @@
url = "http://0.0.0.0:18774/v2.1/id_bla/flavors"
listflavorsresponse = requests.get(url, headers=headers)
self.assertEqual(listflavorsresponse.status_code, 200)
- self.assertIn(json.loads(listflavorsresponse.content)["flavors"][0]["name"], ["m1.nano", "m1.tiny", "m1.micro", "m1.small"])
- self.assertIn(json.loads(listflavorsresponse.content)["flavors"][1]["name"], ["m1.nano", "m1.tiny", "m1.micro", "m1.small"])
- self.assertIn(json.loads(listflavorsresponse.content)["flavors"][2]["name"], ["m1.nano", "m1.tiny", "m1.micro", "m1.small"])
+ self.assertIn(json.loads(listflavorsresponse.content)["flavors"][0]["name"], [
+ "m1.nano", "m1.tiny", "m1.micro", "m1.small"])
+ self.assertIn(json.loads(listflavorsresponse.content)["flavors"][1]["name"], [
+ "m1.nano", "m1.tiny", "m1.micro", "m1.small"])
+ self.assertIn(json.loads(listflavorsresponse.content)["flavors"][2]["name"], [
+ "m1.nano", "m1.tiny", "m1.micro", "m1.small"])
print(" ")
print('->>>>>>> testNovaAddFlavors ->>>>>>>>>>>>>>>')
@@ -143,8 +152,10 @@
data='{"flavor":{"name": "testFlavor", "vcpus": "test_vcpus", "ram": 1024, "disk": 10}}',
headers=headers)
self.assertEqual(addflavorsresponse.status_code, 200)
- self.assertIsNotNone(json.loads(addflavorsresponse.content)["flavor"]["id"])
- self.assertIsNotNone(json.loads(addflavorsresponse.content)["flavor"]["links"][0]['href'])
+ self.assertIsNotNone(json.loads(
+ addflavorsresponse.content)["flavor"]["id"])
+ self.assertIsNotNone(json.loads(addflavorsresponse.content)[
+ "flavor"]["links"][0]['href'])
print(" ")
print('->>>>>>> test Nova List Flavors Detail ->>>>>>>>>>>>>>>')
@@ -152,9 +163,12 @@
url = "http://0.0.0.0:18774/v2.1/id_bla/flavors/detail"
listflavorsdetailresponse = requests.get(url, headers=headers)
self.assertEqual(listflavorsdetailresponse.status_code, 200)
- self.assertIn(json.loads(listflavorsdetailresponse.content)["flavors"][0]["name"],["m1.nano", "m1.tiny", "m1.micro", "m1.small"])
- self.assertIn(json.loads(listflavorsdetailresponse.content)["flavors"][1]["name"],["m1.nano", "m1.tiny", "m1.micro", "m1.small"])
- self.assertIn(json.loads(listflavorsdetailresponse.content)["flavors"][2]["name"],["m1.nano", "m1.tiny", "m1.micro", "m1.small"])
+ self.assertIn(json.loads(listflavorsdetailresponse.content)[
+ "flavors"][0]["name"], ["m1.nano", "m1.tiny", "m1.micro", "m1.small"])
+ self.assertIn(json.loads(listflavorsdetailresponse.content)[
+ "flavors"][1]["name"], ["m1.nano", "m1.tiny", "m1.micro", "m1.small"])
+ self.assertIn(json.loads(listflavorsdetailresponse.content)[
+ "flavors"][2]["name"], ["m1.nano", "m1.tiny", "m1.micro", "m1.small"])
print(" ")
print('->>>>>>> testNovaAddFlavors ->>>>>>>>>>>>>>>')
@@ -164,17 +178,21 @@
data='{"flavor":{"name": "testFlavor", "vcpus": "test_vcpus", "ram": 1024, "disk": 10}}',
headers=headers)
self.assertEqual(addflavorsresponse.status_code, 200)
- self.assertIsNotNone(json.loads(addflavorsresponse.content)["flavor"]["id"])
- self.assertIsNotNone(json.loads(addflavorsresponse.content)["flavor"]["links"][0]['href'])
+ self.assertIsNotNone(json.loads(
+ addflavorsresponse.content)["flavor"]["id"])
+ self.assertIsNotNone(json.loads(addflavorsresponse.content)[
+ "flavor"]["links"][0]['href'])
print(" ")
print('->>>>>>> test Nova List Flavor By Id ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- url = "http://0.0.0.0:18774/v2.1/id_bla/flavors/%s" % (json.loads(listflavorsdetailresponse.content)["flavors"][0]["name"])
+ url = "http://0.0.0.0:18774/v2.1/id_bla/flavors/%s" % (
+ json.loads(listflavorsdetailresponse.content)["flavors"][0]["name"])
listflavorsbyidresponse = requests.get(url, headers=headers)
self.assertEqual(listflavorsbyidresponse.status_code, 200)
- self.assertEqual(json.loads(listflavorsbyidresponse.content)["flavor"]["id"], json.loads(listflavorsdetailresponse.content)["flavors"][0]["id"])
+ self.assertEqual(json.loads(listflavorsbyidresponse.content)[
+ "flavor"]["id"], json.loads(listflavorsdetailresponse.content)["flavors"][0]["id"])
print(" ")
print('->>>>>>> test Nova List Images ->>>>>>>>>>>>>>>')
@@ -183,10 +201,6 @@
listimagesresponse = requests.get(url, headers=headers)
self.assertEqual(listimagesresponse.status_code, 200)
print(listimagesresponse.content)
- # deactivated: highly depends on the environment in which the tests are executed. one cannot make such an assumption.
- #self.assertIn(json.loads(listimagesresponse.content)["images"][0]["name"],["google/cadvisor:latest", "ubuntu:trusty", "prom/pushgateway:latest"])
- #self.assertIn(json.loads(listimagesresponse.content)["images"][1]["name"],["google/cadvisor:latest", "ubuntu:trusty", "prom/pushgateway:latest"])
- #self.assertIn(json.loads(listimagesresponse.content)["images"][2]["name"],["google/cadvisor:latest", "ubuntu:trusty", "prom/pushgateway:latest"])
print(" ")
print('->>>>>>> test Nova List Images Details ->>>>>>>>>>>>>>>')
@@ -194,19 +208,18 @@
url = "http://0.0.0.0:18774/v2.1/id_bla/images/detail"
listimagesdetailsresponse = requests.get(url, headers=headers)
self.assertEqual(listimagesdetailsresponse.status_code, 200)
- # deactivated: highly depends on the environment in which the tests are executed. one cannot make such an assumption.
- #self.assertIn(json.loads(listimagesdetailsresponse.content)["images"][0]["name"],["google/cadvisor:latest", "ubuntu:trusty", "prom/pushgateway:latest"])
- #self.assertIn(json.loads(listimagesdetailsresponse.content)["images"][1]["name"],["google/cadvisor:latest", "ubuntu:trusty", "prom/pushgateway:latest"])
- #self.assertIn(json.loads(listimagesdetailsresponse.content)["images"][2]["name"],["google/cadvisor:latest", "ubuntu:trusty", "prom/pushgateway:latest"])
- self.assertEqual(json.loads(listimagesdetailsresponse.content)["images"][0]["metadata"]["architecture"],"x86_64")
+ self.assertEqual(json.loads(listimagesdetailsresponse.content)[
+ "images"][0]["metadata"]["architecture"], "x86_64")
print(" ")
print('->>>>>>> test Nova List Image By Id ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- url = "http://0.0.0.0:18774/v2.1/id_bla/images/%s" % (json.loads(listimagesdetailsresponse.content)["images"][0]["id"])
+ url = "http://0.0.0.0:18774/v2.1/id_bla/images/%s" % (
+ json.loads(listimagesdetailsresponse.content)["images"][0]["id"])
listimagebyidresponse = requests.get(url, headers=headers)
self.assertEqual(listimagebyidresponse.status_code, 200)
- self.assertEqual(json.loads(listimagebyidresponse.content)["image"]["id"],json.loads(listimagesdetailsresponse.content)["images"][0]["id"])
+ self.assertEqual(json.loads(listimagebyidresponse.content)[
+ "image"]["id"], json.loads(listimagesdetailsresponse.content)["images"][0]["id"])
print(" ")
print('->>>>>>> test Nova List Image By Non-Existend Id ->>>>>>>>>>>>>>>')
@@ -216,7 +229,7 @@
self.assertEqual(listimagebynonexistingidresponse.status_code, 404)
print(" ")
- #find ubuntu id
+ # find ubuntu id
for image in json.loads(listimagesresponse.content)["images"]:
if image["name"] == "ubuntu:trusty":
ubuntu_image_id = image["id"]
@@ -224,16 +237,19 @@
print('->>>>>>> test Nova Create Server Instance ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:18774/v2.1/id_bla/servers"
- data = '{"server": {"name": "X", "flavorRef": "%s", "imageRef":"%s"}}' % (json.loads(listflavorsresponse.content)["flavors"][0]["id"], ubuntu_image_id)
+ data = '{"server": {"name": "X", "flavorRef": "%s", "imageRef":"%s"}}' % (
+ json.loads(listflavorsresponse.content)["flavors"][0]["id"], ubuntu_image_id)
createserverinstance = requests.post(url, data=data, headers=headers)
self.assertEqual(createserverinstance.status_code, 200)
- self.assertEqual(json.loads(createserverinstance.content)["server"]["image"]["id"], ubuntu_image_id)
+ self.assertEqual(json.loads(createserverinstance.content)[
+ "server"]["image"]["id"], ubuntu_image_id)
print(" ")
print('->>>>>>> test Nova Create Server Instance With Already Existing Name ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:18774/v2.1/id_bla/servers"
- data = '{"server": {"name": "X", "flavorRef": "%s", "imageRef":"%s"}}' % (json.loads(listflavorsresponse.content)["flavors"][0]["id"], ubuntu_image_id)
+ data = '{"server": {"name": "X", "flavorRef": "%s", "imageRef":"%s"}}' % (
+ json.loads(listflavorsresponse.content)["flavors"][0]["id"], ubuntu_image_id)
createserverinstance = requests.post(url, data=data, headers=headers)
self.assertEqual(createserverinstance.status_code, 409)
print(" ")
@@ -243,21 +259,25 @@
url = "http://0.0.0.0:18774/v2.1/id_bla/servers/detail"
listserverapisdetailedresponse = requests.get(url, headers=headers)
self.assertEqual(listserverapisdetailedresponse.status_code, 200)
- self.assertEqual(json.loads(listserverapisdetailedresponse.content)["servers"][0]["status"], "ACTIVE")
+ self.assertEqual(json.loads(listserverapisdetailedresponse.content)[
+ "servers"][0]["status"], "ACTIVE")
print(" ")
print('->>>>>>> test Nova Show Server Details ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- url = "http://0.0.0.0:18774/v2.1/id_bla/servers/%s" % (json.loads(listserverapisdetailedresponse.content)["servers"][0]["id"])
+ url = "http://0.0.0.0:18774/v2.1/id_bla/servers/%s" % (
+ json.loads(listserverapisdetailedresponse.content)["servers"][0]["id"])
listserverdetailsresponse = requests.get(url, headers=headers)
self.assertEqual(listserverdetailsresponse.status_code, 200)
- self.assertEqual(json.loads(listserverdetailsresponse.content)["server"]["flavor"]["links"][0]["rel"], "bookmark")
+ self.assertEqual(json.loads(listserverdetailsresponse.content)[
+ "server"]["flavor"]["links"][0]["rel"], "bookmark")
print(" ")
print('->>>>>>> test Nova Show Non-Existing Server Details ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:18774/v2.1/id_bla/servers/non_existing_server_id"
- listnonexistingserverdetailsresponse = requests.get(url, headers=headers)
+ listnonexistingserverdetailsresponse = requests.get(
+ url, headers=headers)
self.assertEqual(listnonexistingserverdetailsresponse.status_code, 404)
print(" ")
@@ -267,9 +287,11 @@
print(" ")
headers = {'Content-type': 'application/json'}
- test_heatapi_template_create_stack = open(os.path.join(os.path.dirname(__file__), "templates/test_heatapi_template_create_stack.yml")).read()
+ test_heatapi_template_create_stack = open(os.path.join(os.path.dirname(
+ __file__), "templates/test_heatapi_template_create_stack.yml")).read()
url = "http://0.0.0.0:18004/v1/tenantabc123/stacks"
- requests.post(url, data=json.dumps(yaml.load(test_heatapi_template_create_stack)), headers=headers)
+ requests.post(url, data=json.dumps(
+ yaml.load(test_heatapi_template_create_stack)), headers=headers)
# test_heatapi_keystone_get_token = open("test_heatapi_keystone_get_token.json").read()
print('->>>>>>> test Neutron List Versions ->>>>>>>>>>>>>>>')
@@ -277,7 +299,8 @@
url = "http://0.0.0.0:19696/"
listapiversionstackresponse = requests.get(url, headers=headers)
self.assertEqual(listapiversionstackresponse.status_code, 200)
- self.assertEqual(json.loads(listapiversionstackresponse.content)["versions"][0]["id"], "v2.0")
+ self.assertEqual(json.loads(listapiversionstackresponse.content)[
+ "versions"][0]["id"], "v2.0")
print(" ")
print('->>>>>>> test Neutron Show API v2.0 ->>>>>>>>>>>>>>>')
@@ -285,9 +308,12 @@
url = "http://0.0.0.0:19696/v2.0"
listapiversionv20response = requests.get(url, headers=headers)
self.assertEqual(listapiversionv20response.status_code, 200)
- self.assertEqual(json.loads(listapiversionv20response.content)["resources"][0]["name"], "subnet")
- self.assertEqual(json.loads(listapiversionv20response.content)["resources"][1]["name"], "network")
- self.assertEqual(json.loads(listapiversionv20response.content)["resources"][2]["name"], "ports")
+ self.assertEqual(json.loads(listapiversionv20response.content)[
+ "resources"][0]["name"], "subnet")
+ self.assertEqual(json.loads(listapiversionv20response.content)[
+ "resources"][1]["name"], "network")
+ self.assertEqual(json.loads(listapiversionv20response.content)[
+ "resources"][2]["name"], "ports")
print(" ")
print('->>>>>>> test Neutron List Networks ->>>>>>>>>>>>>>>')
@@ -295,50 +321,62 @@
url = "http://0.0.0.0:19696/v2.0/networks"
listnetworksesponse1 = requests.get(url, headers=headers)
self.assertEqual(listnetworksesponse1.status_code, 200)
- self.assertEqual(json.loads(listnetworksesponse1.content)["networks"][0]["status"], "ACTIVE")
- listNetworksId = json.loads(listnetworksesponse1.content)["networks"][0]["id"]
- listNetworksName = json.loads(listnetworksesponse1.content)["networks"][0]["name"]
- listNetworksId2 = json.loads(listnetworksesponse1.content)["networks"][1]["id"]
+ self.assertEqual(json.loads(listnetworksesponse1.content)[
+ "networks"][0]["status"], "ACTIVE")
+ listNetworksId = json.loads(listnetworksesponse1.content)[
+ "networks"][0]["id"]
+ listNetworksName = json.loads(listnetworksesponse1.content)[
+ "networks"][0]["name"]
+ listNetworksId2 = json.loads(listnetworksesponse1.content)[
+ "networks"][1]["id"]
print(" ")
print('->>>>>>> test Neutron List Non-Existing Networks ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/networks?name=non_existent_network_name"
- listnetworksesponse2 = requests.get(url,headers=headers)
+ listnetworksesponse2 = requests.get(url, headers=headers)
self.assertEqual(listnetworksesponse2.status_code, 404)
print(" ")
print('->>>>>>> test Neutron List Networks By Name ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- url = "http://0.0.0.0:19696/v2.0/networks?name=" + listNetworksName #tcpdump-vnf:input:net:9df6a98f-9e11-4cb7-b3c0-InAdUnitTest
+ # tcpdump-vnf:input:net:9df6a98f-9e11-4cb7-b3c0-InAdUnitTest
+ url = "http://0.0.0.0:19696/v2.0/networks?name=" + listNetworksName
listnetworksesponse3 = requests.get(url, headers=headers)
self.assertEqual(listnetworksesponse3.status_code, 200)
- self.assertEqual(json.loads(listnetworksesponse3.content)["networks"][0]["name"], listNetworksName)
+ self.assertEqual(json.loads(listnetworksesponse3.content)[
+ "networks"][0]["name"], listNetworksName)
print(" ")
print('->>>>>>> test Neutron List Networks By Id ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- url = "http://0.0.0.0:19696/v2.0/networks?id=" + listNetworksId # tcpdump-vnf:input:net:9df6a98f-9e11-4cb7-b3c0-InAdUnitTest
+ # tcpdump-vnf:input:net:9df6a98f-9e11-4cb7-b3c0-InAdUnitTest
+ url = "http://0.0.0.0:19696/v2.0/networks?id=" + listNetworksId
listnetworksesponse4 = requests.get(url, headers=headers)
self.assertEqual(listnetworksesponse4.status_code, 200)
- self.assertEqual(json.loads(listnetworksesponse4.content)["networks"][0]["id"], listNetworksId)
+ self.assertEqual(json.loads(listnetworksesponse4.content)[
+ "networks"][0]["id"], listNetworksId)
print(" ")
print('->>>>>>> test Neutron List Networks By Multiple Ids ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- url = "http://0.0.0.0:19696/v2.0/networks?id=" + listNetworksId + "&id="+ listNetworksId2 # tcpdump-vnf:input:net:9df6a98f-9e11-4cb7-b3c0-InAdUnitTest
+ url = "http://0.0.0.0:19696/v2.0/networks?id=" + listNetworksId + "&id=" + \
+ listNetworksId2 # tcpdump-vnf:input:net:9df6a98f-9e11-4cb7-b3c0-InAdUnitTest
listnetworksesponse5 = requests.get(url, headers=headers)
self.assertEqual(listnetworksesponse5.status_code, 200)
- self.assertEqual(json.loads(listnetworksesponse5.content)["networks"][0]["id"], listNetworksId)
- self.assertEqual(json.loads(listnetworksesponse5.content)["networks"][1]["id"], listNetworksId2)
+ self.assertEqual(json.loads(listnetworksesponse5.content)[
+ "networks"][0]["id"], listNetworksId)
+ self.assertEqual(json.loads(listnetworksesponse5.content)[
+ "networks"][1]["id"], listNetworksId2)
print(" ")
print('->>>>>>> test Neutron Show Network ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- url = "http://0.0.0.0:19696/v2.0/networks/"+listNetworksId
+ url = "http://0.0.0.0:19696/v2.0/networks/" + listNetworksId
shownetworksesponse = requests.get(url, headers=headers)
self.assertEqual(shownetworksesponse.status_code, 200)
- self.assertEqual(json.loads(shownetworksesponse.content)["network"]["status"], "ACTIVE")
+ self.assertEqual(json.loads(shownetworksesponse.content)[
+ "network"]["status"], "ACTIVE")
print(" ")
print('->>>>>>> test Neutron Show Network Non-ExistendNetwork ->>>>>>>>>>>>>>>')
@@ -351,31 +389,39 @@
print('->>>>>>> test Neutron Create Network ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/networks"
- createnetworkresponse = requests.post(url, data='{"network": {"name": "sample_network","admin_state_up": true}}', headers=headers)
+ createnetworkresponse = requests.post(
+ url, data='{"network": {"name": "sample_network","admin_state_up": true}}', headers=headers)
self.assertEqual(createnetworkresponse.status_code, 201)
- self.assertEqual(json.loads(createnetworkresponse.content)["network"]["status"], "ACTIVE")
+ self.assertEqual(json.loads(createnetworkresponse.content)[
+ "network"]["status"], "ACTIVE")
print(" ")
print('->>>>>>> test Neutron Create Network With Existing Name ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/networks"
- createnetworkresponsefailure = requests.post(url,data='{"network": {"name": "sample_network","admin_state_up": true}}',headers=headers)
+ createnetworkresponsefailure = requests.post(
+ url, data='{"network": {"name": "sample_network","admin_state_up": true}}', headers=headers)
self.assertEqual(createnetworkresponsefailure.status_code, 400)
print(" ")
print('->>>>>>> test Neutron Update Network ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- url = "http://0.0.0.0:19696/v2.0/networks/%s" % (json.loads(createnetworkresponse.content)["network"]["id"])
- updatenetworkresponse = requests.put(url, data='{"network": {"status": "ACTIVE", "admin_state_up":true, "tenant_id":"abcd123", "name": "sample_network_new_name", "shared":false}}' , headers=headers)
+ url = "http://0.0.0.0:19696/v2.0/networks/%s" % (
+ json.loads(createnetworkresponse.content)["network"]["id"])
+ updatenetworkresponse = requests.put(
+ url, data='{"network": {"status": "ACTIVE", "admin_state_up":true, "tenant_id":"abcd123", "name": "sample_network_new_name", "shared":false}}', headers=headers)
self.assertEqual(updatenetworkresponse.status_code, 200)
- self.assertEqual(json.loads(updatenetworkresponse.content)["network"]["name"], "sample_network_new_name")
- self.assertEqual(json.loads(updatenetworkresponse.content)["network"]["tenant_id"], "abcd123")
+ self.assertEqual(json.loads(updatenetworkresponse.content)[
+ "network"]["name"], "sample_network_new_name")
+ self.assertEqual(json.loads(updatenetworkresponse.content)[
+ "network"]["tenant_id"], "abcd123")
print(" ")
print('->>>>>>> test Neutron Update Non-Existing Network ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/networks/non-existing-name123"
- updatenetworkresponse = requests.put(url, data='{"network": {"name": "sample_network_new_name"}}', headers=headers)
+ updatenetworkresponse = requests.put(
+ url, data='{"network": {"name": "sample_network_new_name"}}', headers=headers)
self.assertEqual(updatenetworkresponse.status_code, 404)
print(" ")
@@ -383,19 +429,23 @@
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/subnets"
listsubnetsresponse = requests.get(url, headers=headers)
- listSubnetName = json.loads(listsubnetsresponse.content)["subnets"][0]["name"]
- listSubnetId = json.loads(listsubnetsresponse.content)["subnets"][0]["id"]
- listSubnetId2 = json.loads(listsubnetsresponse.content)["subnets"][1]["id"]
+ listSubnetName = json.loads(listsubnetsresponse.content)[
+ "subnets"][0]["name"]
+ listSubnetId = json.loads(listsubnetsresponse.content)[
+ "subnets"][0]["id"]
+ listSubnetId2 = json.loads(listsubnetsresponse.content)[
+ "subnets"][1]["id"]
self.assertEqual(listsubnetsresponse.status_code, 200)
self.assertNotIn('None', listSubnetName)
print(" ")
print('->>>>>>> test Neutron List Subnets By Name ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- url = "http://0.0.0.0:19696/v2.0/subnets?name="+listSubnetName
+ url = "http://0.0.0.0:19696/v2.0/subnets?name=" + listSubnetName
listsubnetByNameresponse = requests.get(url, headers=headers)
self.assertEqual(listsubnetByNameresponse.status_code, 200)
- self.assertNotIn('None', json.loads(listsubnetByNameresponse.content)["subnets"][0]["name"])
+ self.assertNotIn('None', json.loads(
+ listsubnetByNameresponse.content)["subnets"][0]["name"])
print(" ")
print('->>>>>>> test Neutron List Subnets By Id ->>>>>>>>>>>>>>>')
@@ -403,25 +453,28 @@
url = "http://0.0.0.0:19696/v2.0/subnets?id=" + listSubnetId
listsubnetsbyidresponse = requests.get(url, headers=headers)
self.assertEqual(listsubnetsbyidresponse.status_code, 200)
- self.assertNotIn("None", json.loads(listsubnetsbyidresponse.content)["subnets"][0]["name"])
+ self.assertNotIn("None", json.loads(
+ listsubnetsbyidresponse.content)["subnets"][0]["name"])
print(" ")
print('->>>>>>> test Neutron List Subnets By Multiple Id ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- url = "http://0.0.0.0:19696/v2.0/subnets?id=" + listSubnetId +"&id="+listSubnetId2
+ url = "http://0.0.0.0:19696/v2.0/subnets?id=" + \
+ listSubnetId + "&id=" + listSubnetId2
listsubnetsbymultipleidsresponse = requests.get(url, headers=headers)
self.assertEqual(listsubnetsbymultipleidsresponse.status_code, 200)
- self.assertNotIn("None", json.loads(listsubnetsbymultipleidsresponse.content)["subnets"][0]["name"])
+ self.assertNotIn("None", json.loads(
+ listsubnetsbymultipleidsresponse.content)["subnets"][0]["name"])
print(" ")
-
-
print('->>>>>>> test Neutron Show Subnet->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- url = "http://0.0.0.0:19696/v2.0/subnets/%s" % (json.loads(listsubnetsresponse.content)["subnets"][0]["id"])
+ url = "http://0.0.0.0:19696/v2.0/subnets/%s" % (
+ json.loads(listsubnetsresponse.content)["subnets"][0]["id"])
showsubnetsresponse = requests.get(url, headers=headers)
self.assertEqual(showsubnetsresponse.status_code, 200)
- self.assertNotIn("None", json.loads(showsubnetsresponse.content)["subnet"]["name"])
+ self.assertNotIn("None", json.loads(
+ showsubnetsresponse.content)["subnet"]["name"])
print(" ")
print('->>>>>>> test Neutron Show Non-Existing Subnet->>>>>>>>>>>>>>>')
@@ -431,38 +484,46 @@
self.assertEqual(showsubnetsresponse.status_code, 404)
print(" ")
-
print('->>>>>>> test Neutron Create Subnet ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/subnets"
- createsubnetdata = '{"subnet": {"name": "new_subnet", "network_id": "%s","ip_version": 4,"cidr": "10.0.0.1/24"} }' % (json.loads(createnetworkresponse.content)["network"]["id"])
- createsubnetresponse = requests.post(url, data=createsubnetdata, headers=headers)
+ createsubnetdata = '{"subnet": {"name": "new_subnet", "network_id": "%s","ip_version": 4,"cidr": "10.0.0.1/24"} }' % (
+ json.loads(createnetworkresponse.content)["network"]["id"])
+ createsubnetresponse = requests.post(
+ url, data=createsubnetdata, headers=headers)
self.assertEqual(createsubnetresponse.status_code, 201)
- self.assertEqual(json.loads(createsubnetresponse.content)["subnet"]["name"], "new_subnet")
+ self.assertEqual(json.loads(createsubnetresponse.content)[
+ "subnet"]["name"], "new_subnet")
print(" ")
print('->>>>>>> test Neutron Create Second Subnet ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/subnets"
- createsubnetdata = '{"subnet": {"name": "new_subnet", "network_id": "%s","ip_version": 4,"cidr": "10.0.0.1/24"} }' % (json.loads(createnetworkresponse.content)["network"]["id"])
- createsubnetfailureresponse = requests.post(url, data=createsubnetdata, headers=headers)
+ createsubnetdata = '{"subnet": {"name": "new_subnet", "network_id": "%s","ip_version": 4,"cidr": "10.0.0.1/24"} }' % (
+ json.loads(createnetworkresponse.content)["network"]["id"])
+ createsubnetfailureresponse = requests.post(
+ url, data=createsubnetdata, headers=headers)
self.assertEqual(createsubnetfailureresponse.status_code, 409)
print(" ")
print('->>>>>>> test Neutron Update Subnet ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- url = "http://0.0.0.0:19696/v2.0/subnets/%s" % (json.loads(createsubnetresponse.content)["subnet"]["id"])
+ url = "http://0.0.0.0:19696/v2.0/subnets/%s" % (
+ json.loads(createsubnetresponse.content)["subnet"]["id"])
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} }'
- updatesubnetresponse = requests.put(url, data=updatesubnetdata, headers=headers)
+ updatesubnetresponse = requests.put(
+ url, data=updatesubnetdata, headers=headers)
self.assertEqual(updatesubnetresponse.status_code, 200)
- self.assertEqual(json.loads(updatesubnetresponse.content)["subnet"]["name"], "new_subnet_new_name")
+ self.assertEqual(json.loads(updatesubnetresponse.content)[
+ "subnet"]["name"], "new_subnet_new_name")
print(" ")
print('->>>>>>> test Neutron Update Non-Existing Subnet ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/subnets/non-existing-subnet-12345"
updatenonexistingsubnetdata = '{"subnet": {"name": "new_subnet_new_name"} }'
- updatenonexistingsubnetresponse = requests.put(url, data=updatenonexistingsubnetdata, headers=headers)
+ updatenonexistingsubnetresponse = requests.put(
+ url, data=updatenonexistingsubnetdata, headers=headers)
self.assertEqual(updatenonexistingsubnetresponse.status_code, 404)
print(" ")
@@ -471,8 +532,10 @@
url = "http://0.0.0.0:19696/v2.0/ports"
listportsesponse = requests.get(url, headers=headers)
self.assertEqual(listportsesponse.status_code, 200)
- self.assertEqual(json.loads(listportsesponse.content)["ports"][0]["status"], "ACTIVE")
- listPortsName = json.loads(listportsesponse.content)["ports"][0]["name"]
+ self.assertEqual(json.loads(listportsesponse.content)
+ ["ports"][0]["status"], "ACTIVE")
+ listPortsName = json.loads(listportsesponse.content)[
+ "ports"][0]["name"]
listPortsId1 = json.loads(listportsesponse.content)["ports"][0]["id"]
listPortsId2 = json.loads(listportsesponse.content)["ports"][1]["id"]
print(" ")
@@ -482,7 +545,8 @@
url = "http://0.0.0.0:19696/v2.0/ports?name=" + listPortsName
listportsbynameesponse = requests.get(url, headers=headers)
self.assertEqual(listportsbynameesponse.status_code, 200)
- self.assertEqual(json.loads(listportsbynameesponse.content)["ports"][0]["name"], listPortsName)
+ self.assertEqual(json.loads(listportsbynameesponse.content)[
+ "ports"][0]["name"], listPortsName)
print(" ")
print('->>>>>>> test Neutron List Ports By Id ->>>>>>>>>>>>>>>')
@@ -490,15 +554,18 @@
url = "http://0.0.0.0:19696/v2.0/ports?id=" + listPortsId1
listportsbyidesponse = requests.get(url, headers=headers)
self.assertEqual(listportsbyidesponse.status_code, 200)
- self.assertEqual(json.loads(listportsbyidesponse.content)["ports"][0]["id"], listPortsId1)
+ self.assertEqual(json.loads(listportsbyidesponse.content)[
+ "ports"][0]["id"], listPortsId1)
print(" ")
print('->>>>>>> test Neutron List Ports By Multiple Ids ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- url = "http://0.0.0.0:19696/v2.0/ports?id=" + listPortsId1 +"&id="+listPortsId2
+ url = "http://0.0.0.0:19696/v2.0/ports?id=" + \
+ listPortsId1 + "&id=" + listPortsId2
listportsbymultipleidsesponse = requests.get(url, headers=headers)
self.assertEqual(listportsbymultipleidsesponse.status_code, 200)
- self.assertEqual(json.loads(listportsbymultipleidsesponse.content)["ports"][0]["id"], listPortsId1)
+ self.assertEqual(json.loads(listportsbymultipleidsesponse.content)[
+ "ports"][0]["id"], listPortsId1)
print(" ")
print('->>>>>>> test Neutron List Non-Existing Ports ->>>>>>>>>>>>>>>')
@@ -510,10 +577,12 @@
print('->>>>>>> test Neutron Show Port ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- url = "http://0.0.0.0:19696/v2.0/ports/%s" % (json.loads(listportsesponse.content)["ports"][0]["id"])
+ url = "http://0.0.0.0:19696/v2.0/ports/%s" % (
+ json.loads(listportsesponse.content)["ports"][0]["id"])
showportresponse = requests.get(url, headers=headers)
self.assertEqual(showportresponse.status_code, 200)
- self.assertEqual(json.loads(showportresponse.content)["port"]["status"], "ACTIVE")
+ self.assertEqual(json.loads(showportresponse.content)
+ ["port"]["status"], "ACTIVE")
print(" ")
print('->>>>>>> test Neutron Show Non-Existing Port ->>>>>>>>>>>>>>>')
@@ -527,63 +596,76 @@
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/ports"
createnonexistingportdata = '{"port": {"name": "new_port", "network_id": "non-existing-id"} }'
- createnonexistingnetworkportresponse = requests.post(url, data=createnonexistingportdata, headers=headers)
+ createnonexistingnetworkportresponse = requests.post(
+ url, data=createnonexistingportdata, headers=headers)
self.assertEqual(createnonexistingnetworkportresponse.status_code, 404)
print(" ")
print('->>>>>>> test Neutron Create Port ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/ports"
- 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)["network"]["id"])
- createportresponse = requests.post(url, data=createportdata, headers=headers)
+ 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)[
+ "network"]["id"])
+ createportresponse = requests.post(
+ url, data=createportdata, headers=headers)
self.assertEqual(createportresponse.status_code, 201)
- print (createportresponse.content)
- self.assertEqual(json.loads(createportresponse.content)["port"]["name"], "new_port")
+ print(createportresponse.content)
+ self.assertEqual(json.loads(createportresponse.content)[
+ "port"]["name"], "new_port")
print(" ")
print('->>>>>>> test Neutron Create Port With Existing Name ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/ports"
- createportwithexistingnamedata = '{"port": {"name": "new_port", "network_id": "%s"} }' % (json.loads(createnetworkresponse.content)["network"]["id"])
- createportwithexistingnameresponse = requests.post(url, data=createportwithexistingnamedata, headers=headers)
+ createportwithexistingnamedata = '{"port": {"name": "new_port", "network_id": "%s"} }' % (
+ json.loads(createnetworkresponse.content)["network"]["id"])
+ createportwithexistingnameresponse = requests.post(
+ url, data=createportwithexistingnamedata, headers=headers)
self.assertEqual(createportwithexistingnameresponse.status_code, 500)
print(" ")
print('->>>>>>> test Neutron Create Port Without Name ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/ports"
- createportdatawithoutname = '{"port": {"network_id": "%s"} }' % (json.loads(createnetworkresponse.content)["network"]["id"])
- createportwithoutnameresponse = requests.post(url, data=createportdatawithoutname, headers=headers)
+ createportdatawithoutname = '{"port": {"network_id": "%s"} }' % (
+ json.loads(createnetworkresponse.content)["network"]["id"])
+ createportwithoutnameresponse = requests.post(
+ url, data=createportdatawithoutname, headers=headers)
self.assertEqual(createportwithoutnameresponse.status_code, 201)
- self.assertIn("port:cp", json.loads(createportwithoutnameresponse.content)["port"]["name"])
+ self.assertIn("port:cp", json.loads(
+ createportwithoutnameresponse.content)["port"]["name"])
print(" ")
print('->>>>>>> test Neutron Update Port ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
print(json.loads(createportresponse.content)["port"]["name"])
- url = "http://0.0.0.0:19696/v2.0/ports/%s" % (json.loads(createportresponse.content)["port"]["name"])
+ url = "http://0.0.0.0:19696/v2.0/ports/%s" % (
+ json.loads(createportresponse.content)["port"]["name"])
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"} }'
- updateportresponse = requests.put(url, data=updateportdata, headers=headers)
+ updateportresponse = requests.put(
+ url, data=updateportdata, headers=headers)
self.assertEqual(updateportresponse.status_code, 200)
- self.assertEqual(json.loads(updateportresponse.content)["port"]["name"], "new_port_new_name")
+ self.assertEqual(json.loads(updateportresponse.content)[
+ "port"]["name"], "new_port_new_name")
print(" ")
print('->>>>>>> test Neutron Update Non-Existing Port ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/ports/non-existing-port-ip"
updatenonexistingportdata = '{"port": {"name": "new_port_new_name"} }'
- updatenonexistingportresponse = requests.put(url, data=updatenonexistingportdata, headers=headers)
+ updatenonexistingportresponse = requests.put(
+ url, data=updatenonexistingportdata, headers=headers)
self.assertEqual(updatenonexistingportresponse.status_code, 404)
print(" ")
print('->>>>>>> test Neutron Delete Port ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- righturl = "http://0.0.0.0:19696/v2.0/ports/%s" % (json.loads(createportresponse.content)["port"]["id"])
+ righturl = "http://0.0.0.0:19696/v2.0/ports/%s" % (
+ json.loads(createportresponse.content)["port"]["id"])
deleterightportresponse = requests.delete(righturl, headers=headers)
self.assertEqual(deleterightportresponse.status_code, 204)
print(" ")
-
print('->>>>>>> test Neutron Delete Non-Existing Port ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
wrongurl = "http://0.0.0.0:19696/v2.0/ports/unknownid"
@@ -594,7 +676,8 @@
print('->>>>>>> test Neutron Delete Subnet ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
wrongurl = "http://0.0.0.0:19696/v2.0/subnets/unknownid"
- righturl = "http://0.0.0.0:19696/v2.0/subnets/%s" % (json.loads(updatesubnetresponse.content)["subnet"]["id"])
+ righturl = "http://0.0.0.0:19696/v2.0/subnets/%s" % (
+ json.loads(updatesubnetresponse.content)["subnet"]["id"])
deletewrongsubnetresponse = requests.delete(wrongurl, headers=headers)
deleterightsubnetresponse = requests.delete(righturl, headers=headers)
self.assertEqual(deletewrongsubnetresponse.status_code, 404)
@@ -603,7 +686,8 @@
print('->>>>>>> test Neutron Delete Network ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- righturl = "http://0.0.0.0:19696/v2.0/networks/%s" % (json.loads(createnetworkresponse.content)["network"]["id"])
+ righturl = "http://0.0.0.0:19696/v2.0/networks/%s" % (
+ json.loads(createnetworkresponse.content)["network"]["id"])
deleterightnetworkresponse = requests.delete(righturl, headers=headers)
self.assertEqual(deleterightnetworkresponse.status_code, 204)
print(" ")
@@ -621,14 +705,16 @@
print(" ")
headers = {'Content-type': 'application/json'}
- test_heatapi_keystone_get_token = open(os.path.join(os.path.dirname(__file__), "templates/test_heatapi_keystone_get_token.yml")).read()
+ test_heatapi_keystone_get_token = open(os.path.join(os.path.dirname(
+ __file__), "templates/test_heatapi_keystone_get_token.yml")).read()
print('->>>>>>> test Keystone List Versions ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:15000/"
listapiversionstackresponse = requests.get(url, headers=headers)
self.assertEqual(listapiversionstackresponse.status_code, 200)
- self.assertEqual(json.loads(listapiversionstackresponse.content)["versions"]["values"][0]["id"], "v2.0")
+ self.assertEqual(json.loads(listapiversionstackresponse.content)[
+ "versions"]["values"][0]["id"], "v2.0")
print(" ")
print('->>>>>>> test Keystone Show ApiV2 ->>>>>>>>>>>>>>>')
@@ -636,15 +722,18 @@
url = "http://0.0.0.0:15000/v2.0"
showapiversionstackresponse = requests.get(url, headers=headers)
self.assertEqual(showapiversionstackresponse.status_code, 200)
- self.assertEqual(json.loads(showapiversionstackresponse.content)["version"]["id"], "v2.0")
+ self.assertEqual(json.loads(showapiversionstackresponse.content)[
+ "version"]["id"], "v2.0")
print(" ")
print('->>>>>>> test Keystone Get Token ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:15000/v2.0/tokens"
- gettokenstackresponse = requests.post(url, data=json.dumps(yaml.load(test_heatapi_keystone_get_token)), headers=headers)
+ gettokenstackresponse = requests.post(url, data=json.dumps(
+ yaml.load(test_heatapi_keystone_get_token)), headers=headers)
self.assertEqual(gettokenstackresponse.status_code, 200)
- self.assertEqual(json.loads(gettokenstackresponse.content)["access"]["user"]["name"], "tenantName")
+ self.assertEqual(json.loads(gettokenstackresponse.content)[
+ "access"]["user"]["name"], "tenantName")
print(" ")
def testHeatDummy(self):
@@ -653,78 +742,91 @@
print(" ")
headers = {'Content-type': 'application/json'}
- test_heatapi_template_create_stack = open(os.path.join(os.path.dirname(__file__), "templates/test_heatapi_template_create_stack.yml")).read()
- test_heatapi_template_update_stack = open(os.path.join(os.path.dirname(__file__), "templates/test_heatapi_template_update_stack.yml")).read()
+ test_heatapi_template_create_stack = open(os.path.join(os.path.dirname(
+ __file__), "templates/test_heatapi_template_create_stack.yml")).read()
+ test_heatapi_template_update_stack = open(os.path.join(os.path.dirname(
+ __file__), "templates/test_heatapi_template_update_stack.yml")).read()
print('->>>>>>> test Heat List API Versions Stack ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:18004/"
listapiversionstackresponse = requests.get(url, headers=headers)
self.assertEqual(listapiversionstackresponse.status_code, 200)
- self.assertEqual(json.loads(listapiversionstackresponse.content)["versions"][0]["id"], "v1.0")
+ self.assertEqual(json.loads(listapiversionstackresponse.content)[
+ "versions"][0]["id"], "v1.0")
print(" ")
print('->>>>>>> test Create Stack ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:18004/v1/tenantabc123/stacks"
- createstackresponse = requests.post(url, data=json.dumps(yaml.load(test_heatapi_template_create_stack)), headers=headers)
+ createstackresponse = requests.post(url, data=json.dumps(
+ yaml.load(test_heatapi_template_create_stack)), headers=headers)
self.assertEqual(createstackresponse.status_code, 201)
- self.assertNotEqual(json.loads(createstackresponse.content)["stack"]["id"], "")
+ self.assertNotEqual(json.loads(
+ createstackresponse.content)["stack"]["id"], "")
print(" ")
print('->>>>>>> test Create Stack With Existing Name ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:18004/v1/tenantabc123/stacks"
- createstackwithexistingnameresponse = requests.post(url, data='{"stack_name" : "s1"}', headers=headers)
+ createstackwithexistingnameresponse = requests.post(
+ url, data='{"stack_name" : "s1"}', headers=headers)
self.assertEqual(createstackwithexistingnameresponse.status_code, 409)
print(" ")
print('->>>>>>> test Create Stack With Unsupported Version ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:18004/v1/tenantabc123/stacks"
- createstackwitheunsupportedversionresponse = requests.post(url, data='{"stack_name" : "stackname123", "template" : {"heat_template_version": "2015-04-29"}}', headers=headers)
- self.assertEqual(createstackwitheunsupportedversionresponse.status_code, 400)
+ createstackwitheunsupportedversionresponse = requests.post(
+ url, data='{"stack_name" : "stackname123", "template" : {"heat_template_version": "2015-04-29"}}', headers=headers)
+ self.assertEqual(
+ createstackwitheunsupportedversionresponse.status_code, 400)
print(" ")
-
print('->>>>>>> test List Stack ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:18004/v1/tenantabc123/stacks"
liststackresponse = requests.get(url, headers=headers)
self.assertEqual(liststackresponse.status_code, 200)
- self.assertEqual(json.loads(liststackresponse.content)["stacks"][0]["stack_status"], "CREATE_COMPLETE")
+ self.assertEqual(json.loads(liststackresponse.content)[
+ "stacks"][0]["stack_status"], "CREATE_COMPLETE")
print(" ")
-
print('->>>>>>> test Show Stack ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- url = "http://0.0.0.0:18004/v1/tenantabc123showStack/stacks/%s"% json.loads(createstackresponse.content)['stack']['id']
+ url = "http://0.0.0.0:18004/v1/tenantabc123showStack/stacks/%s" % json.loads(
+ createstackresponse.content)['stack']['id']
liststackdetailsresponse = requests.get(url, headers=headers)
self.assertEqual(liststackdetailsresponse.status_code, 200)
- self.assertEqual(json.loads(liststackdetailsresponse.content)["stack"]["stack_status"], "CREATE_COMPLETE")
+ self.assertEqual(json.loads(liststackdetailsresponse.content)[
+ "stack"]["stack_status"], "CREATE_COMPLETE")
print(" ")
print('->>>>>>> test Show Non-Exisitng Stack ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:18004/v1/tenantabc123showStack/stacks/non_exisitng_id123"
- listnonexistingstackdetailsresponse = requests.get(url, headers=headers)
+ listnonexistingstackdetailsresponse = requests.get(
+ url, headers=headers)
self.assertEqual(listnonexistingstackdetailsresponse.status_code, 404)
print(" ")
print('->>>>>>> test Update Stack ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- url = "http://0.0.0.0:18004/v1/tenantabc123updateStack/stacks/%s"% json.loads(createstackresponse.content)['stack']['id']
+ url = "http://0.0.0.0:18004/v1/tenantabc123updateStack/stacks/%s" % json.loads(
+ createstackresponse.content)['stack']['id']
updatestackresponse = requests.put(url, data=json.dumps(yaml.load(test_heatapi_template_update_stack)),
- headers=headers)
+ headers=headers)
self.assertEqual(updatestackresponse.status_code, 202)
liststackdetailsresponse = requests.get(url, headers=headers)
- self.assertEqual(json.loads(liststackdetailsresponse.content)["stack"]["stack_status"], "UPDATE_COMPLETE")
+ self.assertEqual(json.loads(liststackdetailsresponse.content)[
+ "stack"]["stack_status"], "UPDATE_COMPLETE")
print(" ")
print('->>>>>>> test Update Non-Existing Stack ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:18004/v1/tenantabc123updateStack/stacks/non_existing_id_1234"
- updatenonexistingstackresponse = requests.put(url, data={"non": "sense"}, headers=headers)
+ updatenonexistingstackresponse = requests.put(
+ url, data={"non": "sense"}, headers=headers)
self.assertEqual(updatenonexistingstackresponse.status_code, 404)
print(" ")
@@ -753,19 +855,24 @@
print('->>>>>>> Create ports p1 - p4 ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
# Get network id
- network_resp = requests.get("http://0.0.0.0:19696/v2.0/networks?name=default", headers=headers)
+ network_resp = requests.get(
+ "http://0.0.0.0:19696/v2.0/networks?name=default", headers=headers)
self.assertEqual(network_resp.status_code, 200)
network_id = json.loads(network_resp.content)["networks"][0]["id"]
url = "http://0.0.0.0:19696/v2.0/ports"
port_request = '{"port": {"name": "%s", "network_id": "%s"}}'
- p1_resp = requests.post(url, data=port_request % ("p1", network_id), headers=headers)
+ p1_resp = requests.post(url, data=port_request %
+ ("p1", network_id), headers=headers)
self.assertEqual(p1_resp.status_code, 201)
- p2_resp = requests.post(url, data=port_request % ("p2", network_id), headers=headers)
+ p2_resp = requests.post(url, data=port_request %
+ ("p2", network_id), headers=headers)
self.assertEqual(p2_resp.status_code, 201)
- p3_resp = requests.post(url, data=port_request % ("p3", network_id), headers=headers)
+ p3_resp = requests.post(url, data=port_request %
+ ("p3", network_id), headers=headers)
self.assertEqual(p3_resp.status_code, 201)
- p4_resp = requests.post(url, data=port_request % ("p4", network_id), headers=headers)
+ p4_resp = requests.post(url, data=port_request %
+ ("p4", network_id), headers=headers)
self.assertEqual(p4_resp.status_code, 201)
p1_id = json.loads(p1_resp.content)["port"]["id"]
@@ -776,11 +883,14 @@
print('->>>>>>> test Neutron SFC Port Pair Create ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/sfc/port_pairs"
- pp1_resp = requests.post(url, data='{"port_pair": {"name": "pp1", "ingress": "%s", "egress": "%s"}}' % (p1_id, p2_id), headers=headers)
+ pp1_resp = requests.post(url, data='{"port_pair": {"name": "pp1", "ingress": "%s", "egress": "%s"}}' % (
+ p1_id, p2_id), headers=headers)
self.assertEqual(pp1_resp.status_code, 201)
- pp2_resp = requests.post(url, data='{"port_pair": {"name": "pp2", "ingress": "%s", "egress": "%s"}}' % (p3_id, p4_id), headers=headers)
+ pp2_resp = requests.post(url, data='{"port_pair": {"name": "pp2", "ingress": "%s", "egress": "%s"}}' % (
+ p3_id, p4_id), headers=headers)
self.assertEqual(pp2_resp.status_code, 201)
- pp3_resp = requests.post(url, data='{"port_pair": {"name": "pp3", "ingress": "%s", "egress": "%s"}}' % (p3_id, p4_id), headers=headers)
+ pp3_resp = requests.post(url, data='{"port_pair": {"name": "pp3", "ingress": "%s", "egress": "%s"}}' % (
+ p3_id, p4_id), headers=headers)
self.assertEqual(pp3_resp.status_code, 201)
pp1_id = json.loads(pp1_resp.content)["port_pair"]["id"]
@@ -790,9 +900,11 @@
print('->>>>>>> test Neutron SFC Port Pair Update ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/sfc/port_pairs/%s" % pp3_id
- pp3_update_resp = requests.put(url, data='{"port_pair": {"description": "port_pair_update"}}', headers=headers)
+ pp3_update_resp = requests.put(
+ url, data='{"port_pair": {"description": "port_pair_update"}}', headers=headers)
self.assertEqual(pp3_update_resp.status_code, 200)
- self.assertEqual(json.loads(pp3_update_resp.content)["port_pair"]["description"], "port_pair_update")
+ self.assertEqual(json.loads(pp3_update_resp.content)[
+ "port_pair"]["description"], "port_pair_update")
print('->>>>>>> test Neutron SFC Port Pair Delete ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
@@ -805,24 +917,29 @@
url = "http://0.0.0.0:19696/v2.0/sfc/port_pairs"
pp_list_resp = requests.get(url, headers=headers)
self.assertEqual(pp_list_resp.status_code, 200)
- self.assertEqual(len(json.loads(pp_list_resp.content)["port_pairs"]), 2) # only pp1 and pp2 should be left
+ # only pp1 and pp2 should be left
+ self.assertEqual(
+ len(json.loads(pp_list_resp.content)["port_pairs"]), 2)
print('->>>>>>> test Neutron SFC Port Pair Show ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/sfc/port_pairs/%s" % pp2_id
pp2_show_resp = requests.get(url, headers=headers)
self.assertEqual(pp2_show_resp.status_code, 200)
- self.assertEqual(json.loads(pp2_show_resp.content)["port_pair"]["name"], "pp2")
-
+ self.assertEqual(json.loads(pp2_show_resp.content)
+ ["port_pair"]["name"], "pp2")
print('->>>>>>> test Neutron SFC Port Pair Group Create ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/sfc/port_pair_groups"
- ppg1_resp = requests.post(url, data='{"port_pair_group": {"name": "ppg1", "port_pairs": ["%s"]}}' % (pp1_id), headers=headers)
+ ppg1_resp = requests.post(
+ url, data='{"port_pair_group": {"name": "ppg1", "port_pairs": ["%s"]}}' % (pp1_id), headers=headers)
self.assertEqual(ppg1_resp.status_code, 201)
- ppg2_resp = requests.post(url, data='{"port_pair_group": {"name": "ppg2", "port_pairs": ["%s"]}}' % (pp2_id), headers=headers)
+ ppg2_resp = requests.post(
+ url, data='{"port_pair_group": {"name": "ppg2", "port_pairs": ["%s"]}}' % (pp2_id), headers=headers)
self.assertEqual(ppg2_resp.status_code, 201)
- ppg3_resp = requests.post(url, data='{"port_pair_group": {"name": "ppg3", "port_pairs": ["%s"]}}' % (pp2_id), headers=headers)
+ ppg3_resp = requests.post(
+ url, data='{"port_pair_group": {"name": "ppg3", "port_pairs": ["%s"]}}' % (pp2_id), headers=headers)
self.assertEqual(ppg3_resp.status_code, 201)
ppg1_id = json.loads(ppg1_resp.content)["port_pair_group"]["id"]
@@ -832,9 +949,11 @@
print('->>>>>>> test Neutron SFC Port Pair Group Update ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/sfc/port_pair_groups/%s" % ppg3_id
- ppg3_update_resp = requests.put(url, data='{"port_pair_group": {"description": "port_pair_group_update"}}', headers=headers)
+ ppg3_update_resp = requests.put(
+ url, data='{"port_pair_group": {"description": "port_pair_group_update"}}', headers=headers)
self.assertEqual(ppg3_update_resp.status_code, 200)
- self.assertEqual(json.loads(ppg3_update_resp.content)["port_pair_group"]["description"], "port_pair_group_update")
+ self.assertEqual(json.loads(ppg3_update_resp.content)[
+ "port_pair_group"]["description"], "port_pair_group_update")
print('->>>>>>> test Neutron SFC Port Pair Group Delete ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
@@ -847,21 +966,26 @@
url = "http://0.0.0.0:19696/v2.0/sfc/port_pair_groups"
ppg_list_resp = requests.get(url, headers=headers)
self.assertEqual(ppg_list_resp.status_code, 200)
- self.assertEqual(len(json.loads(ppg_list_resp.content)["port_pair_groups"]), 2) # only ppg1 and ppg2 should be left
+ # only ppg1 and ppg2 should be left
+ self.assertEqual(
+ len(json.loads(ppg_list_resp.content)["port_pair_groups"]), 2)
print('->>>>>>> test Neutron SFC Port Pair Group Show ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/sfc/port_pair_groups/%s" % ppg2_id
ppg2_show_resp = requests.get(url, headers=headers)
self.assertEqual(ppg2_show_resp.status_code, 200)
- self.assertEqual(json.loads(ppg2_show_resp.content)["port_pair_group"]["name"], "ppg2")
+ self.assertEqual(json.loads(ppg2_show_resp.content)[
+ "port_pair_group"]["name"], "ppg2")
print('->>>>>>> test Neutron SFC Flow Classifier Create ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/sfc/flow_classifiers"
- fc1_resp = requests.post(url, data='{"flow_classifier": {"name": "fc1", "source_port_range_min": 22, "source_port_range_max": 4000}}', headers=headers)
+ fc1_resp = requests.post(
+ url, data='{"flow_classifier": {"name": "fc1", "source_port_range_min": 22, "source_port_range_max": 4000}}', headers=headers)
self.assertEqual(fc1_resp.status_code, 201)
- fc2_resp = requests.post(url, data='{"flow_classifier": {"name": "fc2", "source_port_range_min": 22, "source_port_range_max": 4000}}', headers=headers)
+ fc2_resp = requests.post(
+ url, data='{"flow_classifier": {"name": "fc2", "source_port_range_min": 22, "source_port_range_max": 4000}}', headers=headers)
self.assertEqual(fc2_resp.status_code, 201)
fc1_id = json.loads(fc1_resp.content)["flow_classifier"]["id"]
@@ -870,9 +994,11 @@
print('->>>>>>> test Neutron SFC Flow Classifier Update ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/sfc/flow_classifiers/%s" % fc2_id
- fc2_update_resp = requests.put(url, data='{"flow_classifier": {"description": "flow_classifier_update"}}', headers=headers)
+ fc2_update_resp = requests.put(
+ url, data='{"flow_classifier": {"description": "flow_classifier_update"}}', headers=headers)
self.assertEqual(fc2_update_resp.status_code, 200)
- self.assertEqual(json.loads(fc2_update_resp.content)["flow_classifier"]["description"], "flow_classifier_update")
+ self.assertEqual(json.loads(fc2_update_resp.content)[
+ "flow_classifier"]["description"], "flow_classifier_update")
print('->>>>>>> test Neutron SFC Flow Classifier Delete ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
@@ -885,22 +1011,25 @@
url = "http://0.0.0.0:19696/v2.0/sfc/flow_classifiers"
fc_list_resp = requests.get(url, headers=headers)
self.assertEqual(fc_list_resp.status_code, 200)
- self.assertEqual(len(json.loads(fc_list_resp.content)["flow_classifiers"]), 1) # only fc1
+ self.assertEqual(len(json.loads(fc_list_resp.content)
+ ["flow_classifiers"]), 1) # only fc1
print('->>>>>>> test Neutron SFC Flow Classifier Show ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/sfc/flow_classifiers/%s" % fc1_id
fc1_show_resp = requests.get(url, headers=headers)
self.assertEqual(fc1_show_resp.status_code, 200)
- self.assertEqual(json.loads(fc1_show_resp.content)["flow_classifier"]["name"], "fc1")
-
+ self.assertEqual(json.loads(fc1_show_resp.content)[
+ "flow_classifier"]["name"], "fc1")
print('->>>>>>> test Neutron SFC Port Chain Create ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/sfc/port_chains"
- pc1_resp = requests.post(url, data='{"port_chain": {"name": "pc1", "port_pair_groups": ["%s"], "flow_classifiers": ["%s"]}}' % (ppg1_id, fc1_id), headers=headers)
+ pc1_resp = requests.post(url, data='{"port_chain": {"name": "pc1", "port_pair_groups": ["%s"], "flow_classifiers": ["%s"]}}' % (
+ ppg1_id, fc1_id), headers=headers)
self.assertEqual(pc1_resp.status_code, 201)
- pc2_resp = requests.post(url, data='{"port_chain": {"name": "pc2", "port_pair_groups": ["%s"], "flow_classifiers": ["%s"]}}' % (ppg1_id, fc1_id), headers=headers)
+ pc2_resp = requests.post(url, data='{"port_chain": {"name": "pc2", "port_pair_groups": ["%s"], "flow_classifiers": ["%s"]}}' % (
+ ppg1_id, fc1_id), headers=headers)
self.assertEqual(pc2_resp.status_code, 201)
pc1_id = json.loads(pc1_resp.content)["port_chain"]["id"]
@@ -909,9 +1038,11 @@
print('->>>>>>> test Neutron SFC Port Chain Update ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/sfc/port_chains/%s" % pc2_id
- pc2_update_resp = requests.put(url, data='{"port_chain": {"description": "port_chain_update"}}', headers=headers)
+ pc2_update_resp = requests.put(
+ url, data='{"port_chain": {"description": "port_chain_update"}}', headers=headers)
self.assertEqual(pc2_update_resp.status_code, 200)
- self.assertEqual(json.loads(pc2_update_resp.content)["port_chain"]["description"], "port_chain_update")
+ self.assertEqual(json.loads(pc2_update_resp.content)[
+ "port_chain"]["description"], "port_chain_update")
print('->>>>>>> test Neutron SFC Port Chain Delete ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
@@ -924,14 +1055,17 @@
url = "http://0.0.0.0:19696/v2.0/sfc/port_chains"
pc_list_resp = requests.get(url, headers=headers)
self.assertEqual(pc_list_resp.status_code, 200)
- self.assertEqual(len(json.loads(pc_list_resp.content)["port_chains"]), 1) # only pc1
+ self.assertEqual(len(json.loads(pc_list_resp.content)
+ ["port_chains"]), 1) # only pc1
print('->>>>>>> test Neutron SFC Port Chain Show ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
url = "http://0.0.0.0:19696/v2.0/sfc/port_chains/%s" % pc1_id
pc1_show_resp = requests.get(url, headers=headers)
self.assertEqual(pc1_show_resp.status_code, 200)
- self.assertEqual(json.loads(pc1_show_resp.content)["port_chain"]["name"], "pc1")
+ self.assertEqual(json.loads(pc1_show_resp.content)
+ ["port_chain"]["name"], "pc1")
+
if __name__ == '__main__':
unittest.main()
diff --git a/src/emuvim/test/unittests/test_resourcemodel.py b/src/emuvim/test/unittests/test_resourcemodel.py
index 4a21fee..c0cf0c4 100755
--- a/src/emuvim/test/unittests/test_resourcemodel.py
+++ b/src/emuvim/test/unittests/test_resourcemodel.py
@@ -1,31 +1,28 @@
-"""
-Copyright (c) 2015 SONATA-NFV
-ALL RIGHTS RESERVED.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-
-Neither the name of the SONATA-NFV [, ANY ADDITIONAL AFFILIATION]
-nor the names of its contributors may be used to endorse or promote
-products derived from this software without specific prior written
-permission.
-
-This work has been performed in the framework of the SONATA project,
-funded by the European Commission under Grant number 671517 through
-the Horizon 2020 and 5G-PPP programmes. The authors would like to
-acknowledge the contributions of their colleagues of the SONATA
-partner consortium (www.sonata-nfv.eu).
-"""
-
+# Copyright (c) 2015 SONATA-NFV and Paderborn University
+# ALL RIGHTS RESERVED.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Neither the name of the SONATA-NFV, Paderborn University
+# nor the names of its contributors may be used to endorse or promote
+# products derived from this software without specific prior written
+# permission.
+#
+# This work has been performed in the framework of the SONATA project,
+# funded by the European Commission under Grant number 671517 through
+# the Horizon 2020 and 5G-PPP programmes. The authors would like to
+# acknowledge the contributions of their colleagues of the SONATA
+# partner consortium (www.sonata-nfv.eu).
import time
import os
import unittest
@@ -34,7 +31,6 @@
from emuvim.dcemulator.resourcemodel.upb.simple import UpbSimpleCloudDcRM, UpbOverprovisioningCloudDcRM, UpbDummyRM
-
class testResourceModel(SimpleTestTopology):
"""
Test the general resource model API and functionality.
@@ -98,15 +94,15 @@
def __init__(self):
# take defaukt values from son-emu
self.resources = dict(
- cpu_period = -1,
- cpu_quota = -1,
- mem_limit = -1,
- memswap_limit = -1
+ cpu_period=-1,
+ cpu_quota=-1,
+ mem_limit=-1,
+ memswap_limit=-1
)
- #self.cpu_period = self.resources['cpu_period']
- #self.cpu_quota = self.resources['cpu_quota']
- #self.mem_limit = self.resources['mem_limit']
- #self.memswap_limit = self.resources['memswap_limit']
+ # self.cpu_period = self.resources['cpu_period']
+ # self.cpu_quota = self.resources['cpu_quota']
+ # self.mem_limit = self.resources['mem_limit']
+ # self.memswap_limit = self.resources['memswap_limit']
def updateCpuLimit(self, cpu_period, cpu_quota):
self.resources['cpu_period'] = cpu_period
@@ -121,8 +117,6 @@
return d
-
-
class testUpbSimpleCloudDcRM(SimpleTestTopology):
"""
Test the UpbSimpleCloudDc resource model.
@@ -139,35 +133,55 @@
E_MEM = 512
MAX_MU = 2048
# create dummy resource model environment
- reg = ResourceModelRegistrar(dc_emulation_max_cpu=E_CPU, dc_emulation_max_mem=E_MEM)
+ reg = ResourceModelRegistrar(
+ dc_emulation_max_cpu=E_CPU, dc_emulation_max_mem=E_MEM)
rm = UpbSimpleCloudDcRM(max_cu=MAX_CU, max_mu=MAX_MU)
reg.register("test_dc", rm)
c1 = createDummyContainerObject("c1", flavor="tiny")
rm.allocate(c1) # calculate allocation
- self.assertEqual(float(c1.resources['cpu_quota']) / c1.resources['cpu_period'], E_CPU / MAX_CU * 0.5) # validate compute result
- self.assertEqual(float(c1.resources['mem_limit']/1024/1024), float(E_MEM) / MAX_MU * 32) # validate memory result
+ # validate compute result
+ self.assertEqual(float(
+ c1.resources['cpu_quota']) / c1.resources['cpu_period'], E_CPU / MAX_CU * 0.5)
+ # validate memory result
+ self.assertEqual(
+ float(c1.resources['mem_limit'] / 1024 / 1024), float(E_MEM) / MAX_MU * 32)
c2 = createDummyContainerObject("c2", flavor="small")
rm.allocate(c2) # calculate allocation
- self.assertEqual(float(c2.resources['cpu_quota']) / c2.resources['cpu_period'], E_CPU / MAX_CU * 1) # validate compute result
- self.assertEqual(float(c2.resources['mem_limit']/1024/1024), float(E_MEM) / MAX_MU * 128) # validate memory result
+ # validate compute result
+ self.assertEqual(float(
+ c2.resources['cpu_quota']) / c2.resources['cpu_period'], E_CPU / MAX_CU * 1)
+ # validate memory result
+ self.assertEqual(
+ float(c2.resources['mem_limit'] / 1024 / 1024), float(E_MEM) / MAX_MU * 128)
c3 = createDummyContainerObject("c3", flavor="medium")
rm.allocate(c3) # calculate allocation
- self.assertEqual(float(c3.resources['cpu_quota']) / c3.resources['cpu_period'], E_CPU / MAX_CU * 4) # validate compute result
- self.assertEqual(float(c3.resources['mem_limit']/1024/1024), float(E_MEM) / MAX_MU * 256) # validate memory result
+ # validate compute result
+ self.assertEqual(float(
+ c3.resources['cpu_quota']) / c3.resources['cpu_period'], E_CPU / MAX_CU * 4)
+ # validate memory result
+ self.assertEqual(
+ float(c3.resources['mem_limit'] / 1024 / 1024), float(E_MEM) / MAX_MU * 256)
c4 = createDummyContainerObject("c4", flavor="large")
rm.allocate(c4) # calculate allocation
- self.assertEqual(float(c4.resources['cpu_quota']) / c4.resources['cpu_period'], E_CPU / MAX_CU * 8) # validate compute result
- self.assertEqual(float(c4.resources['mem_limit']/1024/1024), float(E_MEM) / MAX_MU * 512) # validate memory result
+ # validate compute result
+ self.assertEqual(float(
+ c4.resources['cpu_quota']) / c4.resources['cpu_period'], E_CPU / MAX_CU * 8)
+ # validate memory result
+ self.assertEqual(
+ float(c4.resources['mem_limit'] / 1024 / 1024), float(E_MEM) / MAX_MU * 512)
c5 = createDummyContainerObject("c5", flavor="xlarge")
rm.allocate(c5) # calculate allocation
- self.assertEqual(float(c5.resources['cpu_quota']) / c5.resources['cpu_period'], E_CPU / MAX_CU * 16) # validate compute result
- self.assertEqual(float(c5.resources['mem_limit']/1024/1024), float(E_MEM) / MAX_MU * 1024) # validate memory result
-
+ # validate compute result
+ self.assertEqual(float(
+ c5.resources['cpu_quota']) / c5.resources['cpu_period'], E_CPU / MAX_CU * 16)
+ # validate memory result
+ self.assertEqual(
+ float(c5.resources['mem_limit'] / 1024 / 1024), float(E_MEM) / MAX_MU * 1024)
def testAllocationCpuLimit(self):
"""
@@ -180,7 +194,8 @@
E_MEM = 512
MAX_MU = 4096
# create dummy resource model environment
- reg = ResourceModelRegistrar(dc_emulation_max_cpu=E_CPU, dc_emulation_max_mem=E_MEM)
+ reg = ResourceModelRegistrar(
+ dc_emulation_max_cpu=E_CPU, dc_emulation_max_mem=E_MEM)
rm = UpbSimpleCloudDcRM(max_cu=MAX_CU, max_mu=MAX_MU)
reg.register("test_dc", rm)
@@ -211,7 +226,8 @@
E_MEM = 512
MAX_MU = 2048
# create dummy resource model environment
- reg = ResourceModelRegistrar(dc_emulation_max_cpu=E_CPU, dc_emulation_max_mem=E_MEM)
+ reg = ResourceModelRegistrar(
+ dc_emulation_max_cpu=E_CPU, dc_emulation_max_mem=E_MEM)
rm = UpbSimpleCloudDcRM(max_cu=MAX_CU, max_mu=MAX_MU)
reg.register("test_dc", rm)
@@ -234,11 +250,9 @@
Test the free procedure.
:return:
"""
- # config
- E_CPU = 1.0
- MAX_CU = 100
# create dummy resource model environment
- reg = ResourceModelRegistrar(dc_emulation_max_cpu=1.0, dc_emulation_max_mem=512)
+ reg = ResourceModelRegistrar(
+ dc_emulation_max_cpu=1.0, dc_emulation_max_mem=512)
rm = UpbSimpleCloudDcRM(max_cu=100, max_mu=100)
reg.register("test_dc", rm)
c1 = createDummyContainerObject("c6", flavor="tiny")
@@ -279,8 +293,10 @@
self.assertTrue(len(r._allocated_compute_instances) == 1)
# check if there is a real limitation set for containers cgroup
- # deactivated for now, seems not to work in docker-in-docker setup used in CI
- self.assertEqual(float(tc1.resources['cpu_quota'])/tc1.resources['cpu_period'], 0.005)
+ # deactivated for now, seems not to work in docker-in-docker setup used
+ # in CI
+ self.assertEqual(
+ float(tc1.resources['cpu_quota']) / tc1.resources['cpu_period'], 0.005)
# check if free was called during stopCompute
self.dc[0].stopCompute("tc1")
@@ -307,39 +323,50 @@
E_MEM = 512
MAX_MU = 2048
# create dummy resource model environment
- reg = ResourceModelRegistrar(dc_emulation_max_cpu=E_CPU, dc_emulation_max_mem=E_MEM)
+ reg = ResourceModelRegistrar(
+ dc_emulation_max_cpu=E_CPU, dc_emulation_max_mem=E_MEM)
rm = UpbOverprovisioningCloudDcRM(max_cu=MAX_CU, max_mu=MAX_MU)
reg.register("test_dc", rm)
c1 = createDummyContainerObject("c1", flavor="small")
rm.allocate(c1) # calculate allocation
- self.assertAlmostEqual(float(c1.resources['cpu_quota']) / c1.resources['cpu_period'], E_CPU / MAX_CU * 1.0, places=5)
- self.assertAlmostEqual(float(c1.resources['mem_limit']/1024/1024), float(E_MEM) / MAX_MU * 128)
+ self.assertAlmostEqual(float(
+ c1.resources['cpu_quota']) / c1.resources['cpu_period'], E_CPU / MAX_CU * 1.0, places=5)
+ self.assertAlmostEqual(
+ float(c1.resources['mem_limit'] / 1024 / 1024), float(E_MEM) / MAX_MU * 128)
self.assertAlmostEqual(rm.cpu_op_factor, 1.0)
c2 = createDummyContainerObject("c2", flavor="small")
rm.allocate(c2) # calculate allocation
- self.assertAlmostEqual(float(c2.resources['cpu_quota']) / c2.resources['cpu_period'], E_CPU / MAX_CU * 1.0, places=5)
- self.assertAlmostEqual(float(c2.resources['mem_limit']/1024/1024), float(E_MEM) / MAX_MU * 128)
+ self.assertAlmostEqual(float(
+ c2.resources['cpu_quota']) / c2.resources['cpu_period'], E_CPU / MAX_CU * 1.0, places=5)
+ self.assertAlmostEqual(
+ float(c2.resources['mem_limit'] / 1024 / 1024), float(E_MEM) / MAX_MU * 128)
self.assertAlmostEqual(rm.cpu_op_factor, 1.0)
c3 = createDummyContainerObject("c3", flavor="small")
rm.allocate(c3) # calculate allocation
- self.assertAlmostEqual(float(c3.resources['cpu_quota']) / c3.resources['cpu_period'], E_CPU / MAX_CU * 1.0, places=5)
- self.assertAlmostEqual(float(c3.resources['mem_limit']/1024/1024), float(E_MEM) / MAX_MU * 128)
+ self.assertAlmostEqual(float(
+ c3.resources['cpu_quota']) / c3.resources['cpu_period'], E_CPU / MAX_CU * 1.0, places=5)
+ self.assertAlmostEqual(
+ float(c3.resources['mem_limit'] / 1024 / 1024), float(E_MEM) / MAX_MU * 128)
self.assertAlmostEqual(rm.cpu_op_factor, 1.0)
# from this container onwards, we should go to over provisioning mode:
c4 = createDummyContainerObject("c4", flavor="small")
rm.allocate(c4) # calculate allocation
- self.assertAlmostEqual(float(c4.resources['cpu_quota']) / c4.resources['cpu_period'], E_CPU / MAX_CU * (float(3) / 4), places=5)
- self.assertAlmostEqual(float(c4.resources['mem_limit']/1024/1024), float(E_MEM) / MAX_MU * 128, places=5)
+ self.assertAlmostEqual(float(
+ c4.resources['cpu_quota']) / c4.resources['cpu_period'], E_CPU / MAX_CU * (float(3) / 4), places=5)
+ self.assertAlmostEqual(float(
+ c4.resources['mem_limit'] / 1024 / 1024), float(E_MEM) / MAX_MU * 128, places=5)
self.assertAlmostEqual(rm.cpu_op_factor, 0.75)
c5 = createDummyContainerObject("c5", flavor="small")
rm.allocate(c5) # calculate allocation
- self.assertAlmostEqual(float(c5.resources['cpu_quota']) / c5.resources['cpu_period'], E_CPU / MAX_CU * (float(3) / 5), places=5)
- self.assertAlmostEqual(float(c5.resources['mem_limit']/1024/1024), float(E_MEM) / MAX_MU * 128)
+ self.assertAlmostEqual(float(
+ c5.resources['cpu_quota']) / c5.resources['cpu_period'], E_CPU / MAX_CU * (float(3) / 5), places=5)
+ self.assertAlmostEqual(
+ float(c5.resources['mem_limit'] / 1024 / 1024), float(E_MEM) / MAX_MU * 128)
self.assertAlmostEqual(rm.cpu_op_factor, 0.6)
@@ -359,7 +386,8 @@
E_MEM = 512
MAX_MU = 2048
# create dummy resource model environment
- reg = ResourceModelRegistrar(dc_emulation_max_cpu=E_CPU, dc_emulation_max_mem=E_MEM)
+ reg = ResourceModelRegistrar(
+ dc_emulation_max_cpu=E_CPU, dc_emulation_max_mem=E_MEM)
rm = UpbDummyRM(max_cu=MAX_CU, max_mu=MAX_MU)
reg.register("test_dc", rm)
@@ -370,4 +398,3 @@
c2 = createDummyContainerObject("c2", flavor="small")
rm.allocate(c2) # calculate allocation
self.assertEqual(len(rm._allocated_compute_instances), 2)
-
diff --git a/src/emuvim/test/unittests/test_restapi.py b/src/emuvim/test/unittests/test_restapi.py
index 4ea8008..ef157e5 100755
--- a/src/emuvim/test/unittests/test_restapi.py
+++ b/src/emuvim/test/unittests/test_restapi.py
@@ -1,36 +1,28 @@
-"""
-Copyright (c) 2015 SONATA-NFV
-ALL RIGHTS RESERVED.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-
-Neither the name of the SONATA-NFV [, ANY ADDITIONAL AFFILIATION]
-nor the names of its contributors may be used to endorse or promote
-products derived from this software without specific prior written
-permission.
-
-This work has been performed in the framework of the SONATA project,
-funded by the European Commission under Grant number 671517 through
-the Horizon 2020 and 5G-PPP programmes. The authors would like to
-acknowledge the contributions of their colleagues of the SONATA
-partner consortium (www.sonata-nfv.eu).
-"""
-
-"""
-Test suite to automatically test emulator REST API endpoints.
-"""
-
-import time
+# Copyright (c) 2015 SONATA-NFV and Paderborn University
+# ALL RIGHTS RESERVED.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Neither the name of the SONATA-NFV, Paderborn University
+# nor the names of its contributors may be used to endorse or promote
+# products derived from this software without specific prior written
+# permission.
+#
+# This work has been performed in the framework of the SONATA project,
+# funded by the European Commission under Grant number 671517 through
+# the Horizon 2020 and 5G-PPP programmes. The authors would like to
+# acknowledge the contributions of their colleagues of the SONATA
+# partner consortium (www.sonata-nfv.eu).
import unittest
from emuvim.test.api_base import SimpleTestTopology
import subprocess
@@ -60,13 +52,16 @@
print('->>>>>>> vim-emu compute start -d datacenter0 -n vnf1 ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- subprocess.call("vim-emu compute start -d datacenter0 -n vnf1", shell=True)
+ subprocess.call(
+ "vim-emu compute start -d datacenter0 -n vnf1", shell=True)
print('->>>>>>> vim-emu compute start -d datacenter0 -n vnf2 ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- subprocess.call("vim-emu compute start -d datacenter0 -n vnf2", shell=True)
+ subprocess.call(
+ "vim-emu compute start -d datacenter0 -n vnf2", shell=True)
print('->>>>>>> vim-emu compute start -d datacenter0 -n vnf3 ->>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- subprocess.call("vim-emu compute start -d datacenter1 -n vnf3", shell=True)
+ subprocess.call(
+ "vim-emu compute start -d datacenter1 -n vnf3", shell=True)
subprocess.call("vim-emu compute list", shell=True)
print('->>>>>>> checking running nodes, compute list, and connectivity >>>>>>>>>>')
@@ -78,33 +73,42 @@
# check compute list result
self.assertTrue(len(self.dc[0].listCompute()) == 2)
self.assertTrue(len(self.dc[1].listCompute()) == 1)
- self.assertTrue(isinstance(self.dc[0].listCompute()[0], EmulatorCompute))
- self.assertTrue(isinstance(self.dc[0].listCompute()[1], EmulatorCompute))
- self.assertTrue(isinstance(self.dc[1].listCompute()[0], EmulatorCompute))
+ self.assertTrue(isinstance(
+ self.dc[0].listCompute()[0], EmulatorCompute))
+ self.assertTrue(isinstance(
+ self.dc[0].listCompute()[1], EmulatorCompute))
+ self.assertTrue(isinstance(
+ self.dc[1].listCompute()[0], EmulatorCompute))
self.assertTrue(self.dc[0].listCompute()[1].name == "vnf1")
self.assertTrue(self.dc[0].listCompute()[0].name == "vnf2")
self.assertTrue(self.dc[1].listCompute()[0].name == "vnf3")
# check connectivity by using ping
- self.assertTrue(self.net.ping([self.dc[0].listCompute()[1], self.dc[0].listCompute()[0]]) <= 0.0)
- self.assertTrue(self.net.ping([self.dc[0].listCompute()[0], self.dc[1].listCompute()[0]]) <= 0.0)
- self.assertTrue(self.net.ping([self.dc[1].listCompute()[0], self.dc[0].listCompute()[1]]) <= 0.0)
+ self.assertTrue(self.net.ping(
+ [self.dc[0].listCompute()[1], self.dc[0].listCompute()[0]]) <= 0.0)
+ self.assertTrue(self.net.ping(
+ [self.dc[0].listCompute()[0], self.dc[1].listCompute()[0]]) <= 0.0)
+ self.assertTrue(self.net.ping(
+ [self.dc[1].listCompute()[0], self.dc[0].listCompute()[1]]) <= 0.0)
print('network add vnf1 vnf2->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- output = subprocess.check_output("vim-emu network add -src vnf1 -dst vnf2 -b -c 10", shell=True)
+ output = subprocess.check_output(
+ "vim-emu network add -src vnf1 -dst vnf2 -b -c 10", shell=True)
self.assertTrue("add-flow" in output)
self.assertTrue("success" in output)
print('network remove vnf1 vnf2->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- output = subprocess.check_output("vim-emu network remove -src vnf1 -dst vnf2 -b", shell=True)
+ output = subprocess.check_output(
+ "vim-emu network remove -src vnf1 -dst vnf2 -b", shell=True)
self.assertTrue("del-flows" in output)
self.assertTrue("success" in output)
print('>>>>> checking --> vim-emu compute stop -d datacenter0 -n vnf2 ->>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- output = subprocess.check_output("vim-emu compute stop -d datacenter0 -n vnf2", shell=True)
+ output = subprocess.check_output(
+ "vim-emu compute stop -d datacenter0 -n vnf2", shell=True)
# check number of running nodes
self.assertTrue(len(self.getContainernetContainers()) == 2)
@@ -123,7 +127,8 @@
print('>>>>> checking --> vim-emu compute status -d datacenter0 -n vnf1 ->>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- output = subprocess.check_output("vim-emu compute status -d datacenter0 -n vnf1", shell=True)
+ output = subprocess.check_output(
+ "vim-emu compute status -d datacenter0 -n vnf1", shell=True)
output = ast.literal_eval(output)
# check compute status result
@@ -138,7 +143,8 @@
print('->>>>> checking --> vim-emu datacenter status -d datacenter0 ->>>>>>>>')
print('->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
- output = subprocess.check_output("vim-emu datacenter status -d datacenter0", shell=True)
+ output = subprocess.check_output(
+ "vim-emu datacenter status -d datacenter0", shell=True)
# check datacenter status result
self.assertTrue("datacenter0" in output)
self.stopApi()
diff --git a/src/emuvim/test/unittests/test_sonata_dummy_gatekeeper.py b/src/emuvim/test/unittests/test_sonata_dummy_gatekeeper.py
index 3450ce9..cb18c57 100755
--- a/src/emuvim/test/unittests/test_sonata_dummy_gatekeeper.py
+++ b/src/emuvim/test/unittests/test_sonata_dummy_gatekeeper.py
@@ -1,40 +1,35 @@
-"""
-Copyright (c) 2015 SONATA-NFV
-ALL RIGHTS RESERVED.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-
-Neither the name of the SONATA-NFV [, ANY ADDITIONAL AFFILIATION]
-nor the names of its contributors may be used to endorse or promote
-products derived from this software without specific prior written
-permission.
-
-This work has been performed in the framework of the SONATA project,
-funded by the European Commission under Grant number 671517 through
-the Horizon 2020 and 5G-PPP programmes. The authors would like to
-acknowledge the contributions of their colleagues of the SONATA
-partner consortium (www.sonata-nfv.eu).
-"""
-
+# Copyright (c) 2015 SONATA-NFV and Paderborn University
+# ALL RIGHTS RESERVED.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Neither the name of the SONATA-NFV, Paderborn University
+# nor the names of its contributors may be used to endorse or promote
+# products derived from this software without specific prior written
+# permission.
+#
+# This work has been performed in the framework of the SONATA project,
+# funded by the European Commission under Grant number 671517 through
+# the Horizon 2020 and 5G-PPP programmes. The authors would like to
+# acknowledge the contributions of their colleagues of the SONATA
+# partner consortium (www.sonata-nfv.eu).
import time
import requests
import json
-import os
import unittest
from emuvim.test.base import SimpleTestTopology
from emuvim.api.sonata import SonataDummyGatekeeperEndpoint
from emuvim.api.sonata.dummygatekeeper import initialize_GK, parse_interface
-import mininet.clean
from ipaddress import ip_network
PACKAGE_PATH = "misc/sonata-demo-service.son"
@@ -42,10 +37,11 @@
class testSonataDummyGatekeeper(SimpleTestTopology):
-# @unittest.skip("disabled")
+ # @unittest.skip("disabled")
def test_GK_Api_start_service(self):
# create network
- self.createNet(nswitches=0, ndatacenter=2, nhosts=2, ndockers=0, enable_learning=True)
+ self.createNet(nswitches=0, ndatacenter=2, nhosts=2,
+ ndockers=0, enable_learning=True)
# setup links
self.net.addLink(self.dc[0], self.h[0])
self.net.addLink(self.dc[0], self.dc[1])
@@ -70,7 +66,8 @@
# instantiate service
self.service_uuid = json.loads(r.text).get("service_uuid")
- r2 = requests.post("http://127.0.0.1:55000/instantiations", data=json.dumps({"service_uuid": self.service_uuid}))
+ r2 = requests.post("http://127.0.0.1:55000/instantiations",
+ data=json.dumps({"service_uuid": self.service_uuid}))
self.assertEqual(r2.status_code, 201)
# give the emulator some time to instantiate everything
@@ -80,7 +77,8 @@
r3 = requests.get("http://127.0.0.1:55000/packages")
self.assertEqual(len(json.loads(r3.text).get("service_uuid_list")), 1)
r4 = requests.get("http://127.0.0.1:55000/instantiations")
- self.assertEqual(len(json.loads(r4.text).get("service_instantiations_list")), 1)
+ self.assertEqual(len(json.loads(r4.text).get(
+ "service_instantiations_list")), 1)
# check number of running nodes
self.assertTrue(len(self.getContainernetContainers()) == 3)
@@ -89,49 +87,59 @@
# check compute list result
self.assertEqual(len(self.dc[0].listCompute()), 2)
# check connectivity by using ping
- ELAN_list=[]
+ ELAN_list = []
# check E-Line connection, by checking the IP addresses
for link in self.net.deployed_elines:
- vnf_src, intf_src, vnf_sap_docker_name = parse_interface(link['connection_points_reference'][0])
+ vnf_src, intf_src, vnf_sap_docker_name = parse_interface(
+ link['connection_points_reference'][0])
print vnf_src, intf_src
src = self.net.getNodeByName(vnf_src)
if not src:
continue
network_list = src.getNetworkStatus()
- src_ip = [intf['ip'] for intf in network_list if intf['intf_name'] == intf_src][0]
- src_mask = [intf['netmask'] for intf in network_list if intf['intf_name'] == intf_src][0]
+ src_ip = [intf['ip']
+ for intf in network_list if intf['intf_name'] == intf_src][0]
+ src_mask = [intf['netmask']
+ for intf in network_list if intf['intf_name'] == intf_src][0]
- vnf_dst, intf_dst, vnf_sap_docker_name = parse_interface(link['connection_points_reference'][1])
+ vnf_dst, intf_dst, vnf_sap_docker_name = parse_interface(
+ link['connection_points_reference'][1])
dst = self.net.getNodeByName(vnf_dst)
if not dst:
continue
network_list = dst.getNetworkStatus()
- dst_ip = [intf['ip'] for intf in network_list if intf['intf_name'] == intf_dst][0]
- dst_mask = [intf['netmask'] for intf in network_list if intf['intf_name'] == intf_dst][0]
+ dst_ip = [intf['ip']
+ for intf in network_list if intf['intf_name'] == intf_dst][0]
+ dst_mask = [intf['netmask']
+ for intf in network_list if intf['intf_name'] == intf_dst][0]
- print "src = {0}:{1} ip={2} ".format(vnf_src, intf_src, src_ip, src_mask)
- print "dst = {0}:{1} ip={2} ".format(vnf_dst, intf_dst, dst_ip, dst_mask)
+ print "src = {0}:{1} ip={2} ".format(
+ vnf_src, intf_src, src_ip, src_mask)
+ print "dst = {0}:{1} ip={2} ".format(
+ vnf_dst, intf_dst, dst_ip, dst_mask)
# check if the E-Line IP's are in the same subnet
ret = ip_network(u'{0}'.format(src_ip, src_mask), strict=False)\
- .compare_networks(ip_network(u'{0}'.format(dst_ip, dst_mask),strict=False))
+ .compare_networks(ip_network(u'{0}'.format(dst_ip, dst_mask), strict=False))
self.assertTrue(ret == 0)
-
for vnf in self.dc[0].listCompute():
# check E LAN connection
network_list = vnf.getNetworkStatus()
- mgmt_ip = [intf['ip'] for intf in network_list if intf['intf_name'] == 'mgmt']
+ mgmt_ip = [intf['ip']
+ for intf in network_list if intf['intf_name'] == 'mgmt']
self.assertTrue(len(mgmt_ip) > 0)
ip_address = mgmt_ip[0]
ELAN_list.append(ip_address)
print ip_address
- # check ELAN connection by ping over the mgmt network (needs to be configured as ELAN in the test service)
+ # check ELAN connection by ping over the mgmt network (needs to be
+ # configured as ELAN in the test service)
for vnf in self.dc[0].listCompute():
network_list = vnf.getNetworkStatus()
- mgmt_ip = [intf['ip'] for intf in network_list if intf['intf_name'] == 'mgmt']
+ mgmt_ip = [intf['ip']
+ for intf in network_list if intf['intf_name'] == 'mgmt']
self.assertTrue(len(mgmt_ip) > 0)
ip_address = mgmt_ip[0]
print ELAN_list
@@ -140,7 +148,7 @@
test_ip_list.remove(ip_address)
for ip in test_ip_list:
# only take ip address, without netmask
- p = self.net.ping([vnf],manualdestip=ip.split('/')[0])
+ p = self.net.ping([vnf], manualdestip=ip.split('/')[0])
print p
self.assertTrue(p <= 0.0)
@@ -148,7 +156,7 @@
self.stopNet()
initialize_GK()
- #@unittest.skip("disabled")
+ # @unittest.skip("disabled")
def test_GK_Api_stop_service(self):
# create network
self.createNet(ndatacenter=2, nhosts=2)
@@ -176,7 +184,8 @@
# instantiate service
self.service_uuid = json.loads(r.text).get("service_uuid")
- r2 = requests.post("http://127.0.0.1:55001/instantiations", data=json.dumps({"service_uuid": self.service_uuid}))
+ r2 = requests.post("http://127.0.0.1:55001/instantiations",
+ data=json.dumps({"service_uuid": self.service_uuid}))
self.assertEqual(r2.status_code, 201)
# give the emulator some time to instantiate everything
@@ -186,7 +195,8 @@
r3 = requests.get("http://127.0.0.1:55001/packages")
self.assertEqual(len(json.loads(r3.text).get("service_uuid_list")), 1)
r4 = requests.get("http://127.0.0.1:55001/instantiations")
- self.assertEqual(len(json.loads(r4.text).get("service_instantiations_list")), 1)
+ self.assertEqual(len(json.loads(r4.text).get(
+ "service_instantiations_list")), 1)
# check number of running nodes
self.assertTrue(len(self.getContainernetContainers()) == 3)
@@ -196,18 +206,21 @@
self.assertEqual(len(self.dc[0].listCompute()), 2)
# stop the service
- service_instance_uuid = json.loads(r2.text).get("service_instance_uuid")
+ service_instance_uuid = json.loads(
+ r2.text).get("service_instance_uuid")
self.assertTrue(service_instance_uuid is not None)
- requests.delete("http://127.0.0.1:55001/instantiations", data=json.dumps({"service_uuid": self.service_uuid, "service_instance_uuid":service_instance_uuid}))
+ requests.delete("http://127.0.0.1:55001/instantiations", data=json.dumps(
+ {"service_uuid": self.service_uuid, "service_instance_uuid": service_instance_uuid}))
r5 = requests.get("http://127.0.0.1:55001/instantiations")
- self.assertTrue(len(json.loads(r5.text).get("service_instantiations_list")), 0) # note that there was 1 instance before
+ # note that there was 1 instance before
+ self.assertTrue(len(json.loads(r5.text).get(
+ "service_instantiations_list")), 0)
# stop Mininet network
self.stopNet()
initialize_GK()
-
@unittest.skip("disabled")
def test_GK_stress_service(self):
# create network
@@ -232,7 +245,8 @@
# instantiate service
self.service_uuid = json.loads(r.text).get("service_uuid")
- r2 = requests.post("http://127.0.0.1:55002/instantiations", data=json.dumps({"service_uuid": self.service_uuid}))
+ r2 = requests.post("http://127.0.0.1:55002/instantiations",
+ data=json.dumps({"service_uuid": self.service_uuid}))
self.assertEqual(r2.status_code, 201)
# give the emulator some time to instantiate everything
@@ -242,18 +256,21 @@
r3 = requests.get("http://127.0.0.1:55002/packages")
self.assertEqual(len(json.loads(r3.text).get("service_uuid_list")), 1)
r4 = requests.get("http://127.0.0.1:55002/instantiations")
- self.assertEqual(len(json.loads(r4.text).get("service_instantiations_list")), 1)
+ self.assertEqual(len(json.loads(r4.text).get(
+ "service_instantiations_list")), 1)
# stop the service
- service_instance_uuid = json.loads(r2.text).get("service_instance_uuid")
+ service_instance_uuid = json.loads(
+ r2.text).get("service_instance_uuid")
self.assertTrue(service_instance_uuid is not None)
- requests.delete("http://127.0.0.1:55002/instantiations", data=json.dumps({"service_uuid": self.service_uuid, "service_instance_uuid":service_instance_uuid}))
+ requests.delete("http://127.0.0.1:55002/instantiations", data=json.dumps(
+ {"service_uuid": self.service_uuid, "service_instance_uuid": service_instance_uuid}))
r5 = requests.get("http://127.0.0.1:55002/instantiations")
- self.assertTrue(len(json.loads(r5.text).get("service_instantiations_list")), 0) # note that there was 1 instance before
+ # note that there was 1 instance before
+ self.assertTrue(len(json.loads(r5.text).get(
+ "service_instantiations_list")), 0)
# stop Mininet network
self.stopNet()
initialize_GK()
-
-