49344b17c4
Adapting assertions in unit tests merged from trunk to reflect changes in the API due to RFE requested by Erik Carlin
841 lines
40 KiB
Python
841 lines
40 KiB
Python
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
|
|
|
# Copyright 2010-2011 ????
|
|
# 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.
|
|
# @author: Brad Hall, Nicira Networks
|
|
# @author: Salvatore Orlando, Citrix Systems
|
|
|
|
import logging
|
|
import unittest
|
|
|
|
import tests.unit.testlib_api as testlib
|
|
|
|
from quantum import api as server
|
|
from quantum.db import api as db
|
|
from quantum.common.wsgi import Serializer
|
|
|
|
|
|
LOG = logging.getLogger('quantum.tests.test_api')
|
|
|
|
|
|
class APITest(unittest.TestCase):
|
|
|
|
def _create_network(self, format, name=None, custom_req_body=None,
|
|
expected_res_status=200):
|
|
LOG.debug("Creating network")
|
|
content_type = "application/" + format
|
|
if name:
|
|
net_name = name
|
|
else:
|
|
net_name = self.network_name
|
|
network_req = testlib.new_network_request(self.tenant_id,
|
|
net_name, format,
|
|
custom_req_body)
|
|
network_res = network_req.get_response(self.api)
|
|
self.assertEqual(network_res.status_int, expected_res_status)
|
|
if expected_res_status == 200:
|
|
network_data = Serializer().deserialize(network_res.body,
|
|
content_type)
|
|
return network_data['networks']['network']['id']
|
|
|
|
def _create_port(self, network_id, port_state, format,
|
|
custom_req_body=None, expected_res_status=200):
|
|
LOG.debug("Creating port for network %s", network_id)
|
|
content_type = "application/%s" % format
|
|
port_req = testlib.new_port_request(self.tenant_id, network_id,
|
|
port_state, format,
|
|
custom_req_body)
|
|
port_res = port_req.get_response(self.api)
|
|
self.assertEqual(port_res.status_int, expected_res_status)
|
|
if expected_res_status == 200:
|
|
port_data = Serializer().deserialize(port_res.body, content_type)
|
|
return port_data['ports']['port']['id']
|
|
|
|
def _test_create_network(self, format):
|
|
LOG.debug("_test_create_network - format:%s - START", format)
|
|
content_type = "application/%s" % format
|
|
network_id = self._create_network(format)
|
|
show_network_req = testlib.show_network_request(self.tenant_id,
|
|
network_id,
|
|
format)
|
|
show_network_res = show_network_req.get_response(self.api)
|
|
self.assertEqual(show_network_res.status_int, 200)
|
|
network_data = Serializer().deserialize(show_network_res.body,
|
|
content_type)
|
|
self.assertEqual(network_id,
|
|
network_data['network']['id'])
|
|
LOG.debug("_test_create_network - format:%s - END", format)
|
|
|
|
def _test_create_network_badrequest(self, format):
|
|
LOG.debug("_test_create_network_badrequest - format:%s - START",
|
|
format)
|
|
bad_body = {'network': {'bad-attribute': 'very-bad'}}
|
|
self._create_network(format, custom_req_body=bad_body,
|
|
expected_res_status=400)
|
|
LOG.debug("_test_create_network_badrequest - format:%s - END",
|
|
format)
|
|
|
|
def _test_list_networks(self, format):
|
|
LOG.debug("_test_list_networks - format:%s - START", format)
|
|
content_type = "application/%s" % format
|
|
self._create_network(format, "net_1")
|
|
self._create_network(format, "net_2")
|
|
list_network_req = testlib.network_list_request(self.tenant_id,
|
|
format)
|
|
list_network_res = list_network_req.get_response(self.api)
|
|
self.assertEqual(list_network_res.status_int, 200)
|
|
network_data = self._net_serializer.deserialize(
|
|
list_network_res.body, content_type)
|
|
# Check network count: should return 2
|
|
self.assertEqual(len(network_data['networks']), 2)
|
|
LOG.debug("_test_list_networks - format:%s - END", format)
|
|
|
|
def _test_show_network(self, format):
|
|
LOG.debug("_test_show_network - format:%s - START", format)
|
|
content_type = "application/%s" % format
|
|
network_id = self._create_network(format)
|
|
show_network_req = testlib.show_network_request(self.tenant_id,
|
|
network_id,
|
|
format)
|
|
show_network_res = show_network_req.get_response(self.api)
|
|
self.assertEqual(show_network_res.status_int, 200)
|
|
network_data = self._net_serializer.deserialize(
|
|
show_network_res.body, content_type)
|
|
self.assertEqual({'id': network_id,
|
|
'name': self.network_name,
|
|
'PortCount': 0},
|
|
network_data['network'])
|
|
LOG.debug("_test_show_network - format:%s - END", format)
|
|
|
|
def _test_show_network_not_found(self, format):
|
|
LOG.debug("_test_show_network_not_found - format:%s - START", format)
|
|
show_network_req = testlib.show_network_request(self.tenant_id,
|
|
"A_BAD_ID",
|
|
format)
|
|
show_network_res = show_network_req.get_response(self.api)
|
|
self.assertEqual(show_network_res.status_int, 420)
|
|
LOG.debug("_test_show_network_not_found - format:%s - END", format)
|
|
|
|
def _test_rename_network(self, format):
|
|
LOG.debug("_test_rename_network - format:%s - START", format)
|
|
content_type = "application/%s" % format
|
|
new_name = 'new_network_name'
|
|
network_id = self._create_network(format)
|
|
update_network_req = testlib.update_network_request(self.tenant_id,
|
|
network_id,
|
|
new_name,
|
|
format)
|
|
update_network_res = update_network_req.get_response(self.api)
|
|
self.assertEqual(update_network_res.status_int, 202)
|
|
show_network_req = testlib.show_network_request(self.tenant_id,
|
|
network_id,
|
|
format)
|
|
show_network_res = show_network_req.get_response(self.api)
|
|
self.assertEqual(show_network_res.status_int, 200)
|
|
network_data = self._net_serializer.deserialize(
|
|
show_network_res.body, content_type)
|
|
self.assertEqual({'id': network_id,
|
|
'name': new_name,
|
|
'PortCount': 0},
|
|
network_data['network'])
|
|
LOG.debug("_test_rename_network - format:%s - END", format)
|
|
|
|
def _test_rename_network_badrequest(self, format):
|
|
LOG.debug("_test_rename_network_badrequest - format:%s - START",
|
|
format)
|
|
network_id = self._create_network(format)
|
|
bad_body = {'network': {'bad-attribute': 'very-bad'}}
|
|
update_network_req = testlib.\
|
|
update_network_request(self.tenant_id,
|
|
network_id, format,
|
|
custom_req_body=bad_body)
|
|
update_network_res = update_network_req.get_response(self.api)
|
|
self.assertEqual(update_network_res.status_int, 400)
|
|
LOG.debug("_test_rename_network_badrequest - format:%s - END",
|
|
format)
|
|
|
|
def _test_rename_network_not_found(self, format):
|
|
LOG.debug("_test_rename_network_not_found - format:%s - START",
|
|
format)
|
|
new_name = 'new_network_name'
|
|
update_network_req = testlib.update_network_request(self.tenant_id,
|
|
"A BAD ID",
|
|
new_name,
|
|
format)
|
|
update_network_res = update_network_req.get_response(self.api)
|
|
self.assertEqual(update_network_res.status_int, 420)
|
|
LOG.debug("_test_rename_network_not_found - format:%s - END",
|
|
format)
|
|
|
|
def _test_delete_network(self, format):
|
|
LOG.debug("_test_delete_network - format:%s - START", format)
|
|
content_type = "application/%s" % format
|
|
network_id = self._create_network(format)
|
|
LOG.debug("Deleting network %(network_id)s"\
|
|
" of tenant %(tenant_id)s", locals())
|
|
delete_network_req = testlib.network_delete_request(self.tenant_id,
|
|
network_id,
|
|
format)
|
|
delete_network_res = delete_network_req.get_response(self.api)
|
|
self.assertEqual(delete_network_res.status_int, 202)
|
|
list_network_req = testlib.network_list_request(self.tenant_id,
|
|
format)
|
|
list_network_res = list_network_req.get_response(self.api)
|
|
network_list_data = self._net_serializer.deserialize(
|
|
list_network_res.body, content_type)
|
|
network_count = len(network_list_data['networks'])
|
|
self.assertEqual(network_count, 0)
|
|
LOG.debug("_test_delete_network - format:%s - END", format)
|
|
|
|
def _test_delete_network_in_use(self, format):
|
|
LOG.debug("_test_delete_network_in_use - format:%s - START", format)
|
|
content_type = "application/%s" % format
|
|
port_state = "ACTIVE"
|
|
attachment_id = "test_attachment"
|
|
network_id = self._create_network(format)
|
|
LOG.debug("Deleting network %(network_id)s"\
|
|
" of tenant %(tenant_id)s", locals())
|
|
port_id = self._create_port(network_id, port_state, format)
|
|
#plug an attachment into the port
|
|
LOG.debug("Putting attachment into port %s", port_id)
|
|
attachment_req = testlib.put_attachment_request(self.tenant_id,
|
|
network_id,
|
|
port_id,
|
|
attachment_id)
|
|
attachment_res = attachment_req.get_response(self.api)
|
|
self.assertEquals(attachment_res.status_int, 202)
|
|
|
|
LOG.debug("Deleting network %(network_id)s"\
|
|
" of tenant %(tenant_id)s", locals())
|
|
delete_network_req = testlib.network_delete_request(self.tenant_id,
|
|
network_id,
|
|
format)
|
|
delete_network_res = delete_network_req.get_response(self.api)
|
|
self.assertEqual(delete_network_res.status_int, 421)
|
|
LOG.debug("_test_delete_network_in_use - format:%s - END", format)
|
|
|
|
def _test_list_ports(self, format):
|
|
LOG.debug("_test_list_ports - format:%s - START", format)
|
|
content_type = "application/%s" % format
|
|
port_state = "ACTIVE"
|
|
network_id = self._create_network(format)
|
|
self._create_port(network_id, port_state, format)
|
|
self._create_port(network_id, port_state, format)
|
|
list_port_req = testlib.port_list_request(self.tenant_id,
|
|
network_id, format)
|
|
list_port_res = list_port_req.get_response(self.api)
|
|
self.assertEqual(list_port_res.status_int, 200)
|
|
port_data = self._port_serializer.deserialize(
|
|
list_port_res.body, content_type)
|
|
# Check port count: should return 2
|
|
self.assertEqual(len(port_data['ports']), 2)
|
|
LOG.debug("_test_list_ports - format:%s - END", format)
|
|
|
|
def _test_show_port(self, format):
|
|
LOG.debug("_test_show_port - format:%s - START", format)
|
|
content_type = "application/%s" % format
|
|
port_state = "ACTIVE"
|
|
network_id = self._create_network(format)
|
|
port_id = self._create_port(network_id, port_state, format)
|
|
show_port_req = testlib.show_port_request(self.tenant_id,
|
|
network_id, port_id,
|
|
format)
|
|
show_port_res = show_port_req.get_response(self.api)
|
|
self.assertEqual(show_port_res.status_int, 200)
|
|
port_data = self._port_serializer.deserialize(
|
|
show_port_res.body, content_type)
|
|
self.assertEqual({'id': port_id, 'state': port_state},
|
|
port_data['port'])
|
|
LOG.debug("_test_show_port - format:%s - END", format)
|
|
|
|
def _test_show_port_networknotfound(self, format):
|
|
LOG.debug("_test_show_port_networknotfound - format:%s - START",
|
|
format)
|
|
port_state = "ACTIVE"
|
|
network_id = self._create_network(format)
|
|
port_id = self._create_port(network_id, port_state, format)
|
|
show_port_req = testlib.show_port_request(self.tenant_id,
|
|
"A_BAD_ID", port_id,
|
|
format)
|
|
show_port_res = show_port_req.get_response(self.api)
|
|
self.assertEqual(show_port_res.status_int, 420)
|
|
LOG.debug("_test_show_port_networknotfound - format:%s - END",
|
|
format)
|
|
|
|
def _test_show_port_portnotfound(self, format):
|
|
LOG.debug("_test_show_port_portnotfound - format:%s - START", format)
|
|
network_id = self._create_network(format)
|
|
show_port_req = testlib.show_port_request(self.tenant_id,
|
|
network_id,
|
|
"A_BAD_ID",
|
|
format)
|
|
show_port_res = show_port_req.get_response(self.api)
|
|
self.assertEqual(show_port_res.status_int, 430)
|
|
LOG.debug("_test_show_port_portnotfound - format:%s - END", format)
|
|
|
|
def _test_create_port(self, format):
|
|
LOG.debug("_test_create_port - format:%s - START", format)
|
|
content_type = "application/%s" % format
|
|
port_state = "ACTIVE"
|
|
network_id = self._create_network(format)
|
|
port_id = self._create_port(network_id, port_state, format)
|
|
show_port_req = testlib.show_port_request(self.tenant_id,
|
|
network_id, port_id, format)
|
|
show_port_res = show_port_req.get_response(self.api)
|
|
self.assertEqual(show_port_res.status_int, 200)
|
|
port_data = self._port_serializer.deserialize(
|
|
show_port_res.body, content_type)
|
|
self.assertEqual(port_id, port_data['port']['id'])
|
|
LOG.debug("_test_create_port - format:%s - END", format)
|
|
|
|
def _test_create_port_networknotfound(self, format):
|
|
LOG.debug("_test_create_port_networknotfound - format:%s - START",
|
|
format)
|
|
port_state = "ACTIVE"
|
|
self._create_port("A_BAD_ID", port_state, format,
|
|
expected_res_status=420)
|
|
LOG.debug("_test_create_port_networknotfound - format:%s - END",
|
|
format)
|
|
|
|
def _test_create_port_badrequest(self, format):
|
|
LOG.debug("_test_create_port_badrequest - format:%s - START", format)
|
|
bad_body = {'bad-resource': {'bad-attribute': 'bad-value'}}
|
|
network_id = self._create_network(format)
|
|
port_state = "ACTIVE"
|
|
self._create_port(network_id, port_state, format,
|
|
custom_req_body=bad_body, expected_res_status=400)
|
|
LOG.debug("_test_create_port_badrequest - format:%s - END", format)
|
|
|
|
def _test_delete_port(self, format):
|
|
LOG.debug("_test_delete_port - format:%s - START", format)
|
|
content_type = "application/%s" % format
|
|
port_state = "ACTIVE"
|
|
network_id = self._create_network(format)
|
|
port_id = self._create_port(network_id, port_state, format)
|
|
LOG.debug("Deleting port %(port_id)s for network %(network_id)s"\
|
|
" of tenant %(tenant_id)s", locals())
|
|
delete_port_req = testlib.port_delete_request(self.tenant_id,
|
|
network_id, port_id,
|
|
format)
|
|
delete_port_res = delete_port_req.get_response(self.api)
|
|
self.assertEqual(delete_port_res.status_int, 202)
|
|
list_port_req = testlib.port_list_request(self.tenant_id, network_id,
|
|
format)
|
|
list_port_res = list_port_req.get_response(self.api)
|
|
port_list_data = self._port_serializer.deserialize(
|
|
list_port_res.body, content_type)
|
|
port_count = len(port_list_data['ports'])
|
|
self.assertEqual(port_count, 0)
|
|
LOG.debug("_test_delete_port - format:%s - END", format)
|
|
|
|
def _test_delete_port_in_use(self, format):
|
|
LOG.debug("_test_delete_port_in_use - format:%s - START", format)
|
|
content_type = "application/" + format
|
|
port_state = "ACTIVE"
|
|
attachment_id = "test_attachment"
|
|
network_id = self._create_network(format)
|
|
port_id = self._create_port(network_id, port_state, format)
|
|
#plug an attachment into the port
|
|
LOG.debug("Putting attachment into port %s", port_id)
|
|
attachment_req = testlib.put_attachment_request(self.tenant_id,
|
|
network_id,
|
|
port_id,
|
|
attachment_id)
|
|
attachment_res = attachment_req.get_response(self.api)
|
|
self.assertEquals(attachment_res.status_int, 202)
|
|
LOG.debug("Deleting port %(port_id)s for network %(network_id)s"\
|
|
" of tenant %(tenant_id)s", locals())
|
|
delete_port_req = testlib.port_delete_request(self.tenant_id,
|
|
network_id, port_id,
|
|
format)
|
|
delete_port_res = delete_port_req.get_response(self.api)
|
|
self.assertEqual(delete_port_res.status_int, 432)
|
|
LOG.debug("_test_delete_port_in_use - format:%s - END", format)
|
|
|
|
def _test_delete_port_with_bad_id(self, format):
|
|
LOG.debug("_test_delete_port_with_bad_id - format:%s - START",
|
|
format)
|
|
port_state = "ACTIVE"
|
|
network_id = self._create_network(format)
|
|
self._create_port(network_id, port_state, format)
|
|
# Test for portnotfound
|
|
delete_port_req = testlib.port_delete_request(self.tenant_id,
|
|
network_id, "A_BAD_ID",
|
|
format)
|
|
delete_port_res = delete_port_req.get_response(self.api)
|
|
self.assertEqual(delete_port_res.status_int, 430)
|
|
LOG.debug("_test_delete_port_with_bad_id - format:%s - END", format)
|
|
|
|
def _test_delete_port_networknotfound(self, format):
|
|
LOG.debug("_test_delete_port_networknotfound - format:%s - START",
|
|
format)
|
|
port_state = "ACTIVE"
|
|
network_id = self._create_network(format)
|
|
port_id = self._create_port(network_id, port_state, format)
|
|
delete_port_req = testlib.port_delete_request(self.tenant_id,
|
|
"A_BAD_ID", port_id,
|
|
format)
|
|
delete_port_res = delete_port_req.get_response(self.api)
|
|
self.assertEqual(delete_port_res.status_int, 420)
|
|
LOG.debug("_test_delete_port_networknotfound - format:%s - END",
|
|
format)
|
|
|
|
def _test_set_port_state(self, format):
|
|
LOG.debug("_test_set_port_state - format:%s - START", format)
|
|
content_type = "application/%s" % format
|
|
port_state = 'DOWN'
|
|
new_port_state = 'ACTIVE'
|
|
network_id = self._create_network(format)
|
|
port_id = self._create_port(network_id, port_state, format)
|
|
update_port_req = testlib.update_port_request(self.tenant_id,
|
|
network_id, port_id,
|
|
new_port_state,
|
|
format)
|
|
update_port_res = update_port_req.get_response(self.api)
|
|
self.assertEqual(update_port_res.status_int, 200)
|
|
show_port_req = testlib.show_port_request(self.tenant_id,
|
|
network_id, port_id,
|
|
format)
|
|
show_port_res = show_port_req.get_response(self.api)
|
|
self.assertEqual(show_port_res.status_int, 200)
|
|
port_data = self._port_serializer.deserialize(
|
|
show_port_res.body, content_type)
|
|
self.assertEqual({'id': port_id, 'state': new_port_state},
|
|
port_data['port'])
|
|
LOG.debug("_test_set_port_state - format:%s - END", format)
|
|
|
|
def _test_set_port_state_networknotfound(self, format):
|
|
LOG.debug("_test_set_port_state_networknotfound - format:%s - START",
|
|
format)
|
|
port_state = 'DOWN'
|
|
new_port_state = 'ACTIVE'
|
|
network_id = self._create_network(format)
|
|
port_id = self._create_port(network_id, port_state, format)
|
|
update_port_req = testlib.update_port_request(self.tenant_id,
|
|
"A_BAD_ID", port_id,
|
|
new_port_state,
|
|
format)
|
|
update_port_res = update_port_req.get_response(self.api)
|
|
self.assertEqual(update_port_res.status_int, 420)
|
|
LOG.debug("_test_set_port_state_networknotfound - format:%s - END",
|
|
format)
|
|
|
|
def _test_set_port_state_portnotfound(self, format):
|
|
LOG.debug("_test_set_port_state_portnotfound - format:%s - START",
|
|
format)
|
|
port_state = 'DOWN'
|
|
new_port_state = 'ACTIVE'
|
|
network_id = self._create_network(format)
|
|
self._create_port(network_id, port_state, format)
|
|
update_port_req = testlib.update_port_request(self.tenant_id,
|
|
network_id,
|
|
"A_BAD_ID",
|
|
new_port_state,
|
|
format)
|
|
update_port_res = update_port_req.get_response(self.api)
|
|
self.assertEqual(update_port_res.status_int, 430)
|
|
LOG.debug("_test_set_port_state_portnotfound - format:%s - END",
|
|
format)
|
|
|
|
def _test_set_port_state_stateinvalid(self, format):
|
|
LOG.debug("_test_set_port_state_stateinvalid - format:%s - START",
|
|
format)
|
|
port_state = 'DOWN'
|
|
new_port_state = 'A_BAD_STATE'
|
|
network_id = self._create_network(format)
|
|
port_id = self._create_port(network_id, port_state, format)
|
|
update_port_req = testlib.update_port_request(self.tenant_id,
|
|
network_id, port_id,
|
|
new_port_state,
|
|
format)
|
|
update_port_res = update_port_req.get_response(self.api)
|
|
self.assertEqual(update_port_res.status_int, 431)
|
|
LOG.debug("_test_set_port_state_stateinvalid - format:%s - END",
|
|
format)
|
|
|
|
def _test_show_attachment(self, format):
|
|
LOG.debug("_test_show_attachment - format:%s - START", format)
|
|
content_type = "application/%s" % format
|
|
port_state = "ACTIVE"
|
|
network_id = self._create_network(format)
|
|
interface_id = "test_interface"
|
|
port_id = self._create_port(network_id, port_state, format)
|
|
put_attachment_req = testlib.put_attachment_request(self.tenant_id,
|
|
network_id,
|
|
port_id,
|
|
interface_id,
|
|
format)
|
|
put_attachment_res = put_attachment_req.get_response(self.api)
|
|
self.assertEqual(put_attachment_res.status_int, 202)
|
|
get_attachment_req = testlib.get_attachment_request(self.tenant_id,
|
|
network_id,
|
|
port_id,
|
|
format)
|
|
get_attachment_res = get_attachment_req.get_response(self.api)
|
|
attachment_data = Serializer().deserialize(get_attachment_res.body,
|
|
content_type)
|
|
self.assertEqual(attachment_data['attachment'], interface_id)
|
|
LOG.debug("_test_show_attachment - format:%s - END", format)
|
|
|
|
def _test_show_attachment_networknotfound(self, format):
|
|
LOG.debug("_test_show_attachment_networknotfound - format:%s - START",
|
|
format)
|
|
port_state = "ACTIVE"
|
|
network_id = self._create_network(format)
|
|
port_id = self._create_port(network_id, port_state, format)
|
|
get_attachment_req = testlib.get_attachment_request(self.tenant_id,
|
|
"A_BAD_ID",
|
|
port_id,
|
|
format)
|
|
get_attachment_res = get_attachment_req.get_response(self.api)
|
|
self.assertEqual(get_attachment_res.status_int, 420)
|
|
LOG.debug("_test_show_attachment_networknotfound - format:%s - END",
|
|
format)
|
|
|
|
def _test_show_attachment_portnotfound(self, format):
|
|
LOG.debug("_test_show_attachment_portnotfound - format:%s - START",
|
|
format)
|
|
port_state = "ACTIVE"
|
|
network_id = self._create_network(format)
|
|
self._create_port(network_id, port_state, format)
|
|
get_attachment_req = testlib.get_attachment_request(self.tenant_id,
|
|
network_id,
|
|
"A_BAD_ID",
|
|
format)
|
|
get_attachment_res = get_attachment_req.get_response(self.api)
|
|
self.assertEqual(get_attachment_res.status_int, 430)
|
|
LOG.debug("_test_show_attachment_portnotfound - format:%s - END",
|
|
format)
|
|
|
|
def _test_put_attachment(self, format):
|
|
LOG.debug("_test_put_attachment - format:%s - START", format)
|
|
port_state = "ACTIVE"
|
|
network_id = self._create_network(format)
|
|
interface_id = "test_interface"
|
|
port_id = self._create_port(network_id, port_state, format)
|
|
put_attachment_req = testlib.put_attachment_request(self.tenant_id,
|
|
network_id,
|
|
port_id,
|
|
interface_id,
|
|
format)
|
|
put_attachment_res = put_attachment_req.get_response(self.api)
|
|
self.assertEqual(put_attachment_res.status_int, 202)
|
|
LOG.debug("_test_put_attachment - format:%s - END", format)
|
|
|
|
def _test_put_attachment_networknotfound(self, format):
|
|
LOG.debug("_test_put_attachment_networknotfound - format:%s - START",
|
|
format)
|
|
port_state = 'DOWN'
|
|
interface_id = "test_interface"
|
|
network_id = self._create_network(format)
|
|
port_id = self._create_port(network_id, port_state, format)
|
|
put_attachment_req = testlib.put_attachment_request(self.tenant_id,
|
|
"A_BAD_ID",
|
|
port_id,
|
|
interface_id,
|
|
format)
|
|
put_attachment_res = put_attachment_req.get_response(self.api)
|
|
self.assertEqual(put_attachment_res.status_int, 420)
|
|
LOG.debug("_test_put_attachment_networknotfound - format:%s - END",
|
|
format)
|
|
|
|
def _test_put_attachment_portnotfound(self, format):
|
|
LOG.debug("_test_put_attachment_portnotfound - format:%s - START",
|
|
format)
|
|
port_state = 'DOWN'
|
|
interface_id = "test_interface"
|
|
network_id = self._create_network(format)
|
|
self._create_port(network_id, port_state, format)
|
|
put_attachment_req = testlib.put_attachment_request(self.tenant_id,
|
|
network_id,
|
|
"A_BAD_ID",
|
|
interface_id,
|
|
format)
|
|
put_attachment_res = put_attachment_req.get_response(self.api)
|
|
self.assertEqual(put_attachment_res.status_int, 430)
|
|
LOG.debug("_test_put_attachment_portnotfound - format:%s - END",
|
|
format)
|
|
|
|
def _test_delete_attachment(self, format):
|
|
LOG.debug("_test_delete_attachment - format:%s - START", format)
|
|
port_state = "ACTIVE"
|
|
network_id = self._create_network(format)
|
|
interface_id = "test_interface"
|
|
port_id = self._create_port(network_id, port_state, format)
|
|
put_attachment_req = testlib.put_attachment_request(self.tenant_id,
|
|
network_id,
|
|
port_id,
|
|
interface_id,
|
|
format)
|
|
put_attachment_res = put_attachment_req.get_response(self.api)
|
|
self.assertEqual(put_attachment_res.status_int, 202)
|
|
del_attachment_req = testlib.delete_attachment_request(self.tenant_id,
|
|
network_id,
|
|
port_id,
|
|
format)
|
|
del_attachment_res = del_attachment_req.get_response(self.api)
|
|
self.assertEqual(del_attachment_res.status_int, 202)
|
|
LOG.debug("_test_delete_attachment - format:%s - END", format)
|
|
|
|
def _test_delete_attachment_networknotfound(self, format):
|
|
LOG.debug("_test_delete_attachment_networknotfound -" \
|
|
" format:%s - START", format)
|
|
port_state = "ACTIVE"
|
|
network_id = self._create_network(format)
|
|
port_id = self._create_port(network_id, port_state, format)
|
|
del_attachment_req = testlib.delete_attachment_request(self.tenant_id,
|
|
"A_BAD_ID",
|
|
port_id,
|
|
format)
|
|
del_attachment_res = del_attachment_req.get_response(self.api)
|
|
self.assertEqual(del_attachment_res.status_int, 420)
|
|
LOG.debug("_test_delete_attachment_networknotfound -" \
|
|
" format:%s - END", format)
|
|
|
|
def _test_delete_attachment_portnotfound(self, format):
|
|
LOG.debug("_test_delete_attachment_portnotfound - " \
|
|
" format:%s - START", format)
|
|
port_state = "ACTIVE"
|
|
network_id = self._create_network(format)
|
|
self._create_port(network_id, port_state, format)
|
|
del_attachment_req = testlib.delete_attachment_request(self.tenant_id,
|
|
network_id,
|
|
"A_BAD_ID",
|
|
format)
|
|
del_attachment_res = del_attachment_req.get_response(self.api)
|
|
self.assertEqual(del_attachment_res.status_int, 430)
|
|
LOG.debug("_test_delete_attachment_portnotfound - " \
|
|
"format:%s - END", format)
|
|
|
|
def setUp(self):
|
|
options = {}
|
|
options['plugin_provider'] = 'quantum.plugins.SamplePlugin.FakePlugin'
|
|
self.api = server.APIRouterV01(options)
|
|
self.tenant_id = "test_tenant"
|
|
self.network_name = "test_network"
|
|
self._net_serializer = \
|
|
Serializer(server.networks.Controller._serialization_metadata)
|
|
self._port_serializer = \
|
|
Serializer(server.ports.Controller._serialization_metadata)
|
|
|
|
def tearDown(self):
|
|
"""Clear the test environment"""
|
|
# Remove database contents
|
|
db.clear_db()
|
|
|
|
def test_list_networks_json(self):
|
|
self._test_list_networks('json')
|
|
|
|
def test_list_networks_xml(self):
|
|
self._test_list_networks('xml')
|
|
|
|
def test_create_network_json(self):
|
|
self._test_create_network('json')
|
|
|
|
def test_create_network_xml(self):
|
|
self._test_create_network('xml')
|
|
|
|
def test_create_network_badrequest_json(self):
|
|
self._test_create_network_badrequest('json')
|
|
|
|
def test_create_network_badreqyest_xml(self):
|
|
self._test_create_network_badrequest('xml')
|
|
|
|
def test_show_network_not_found_json(self):
|
|
self._test_show_network_not_found('json')
|
|
|
|
def test_show_network_not_found_xml(self):
|
|
self._test_show_network_not_found('xml')
|
|
|
|
def test_show_network_json(self):
|
|
self._test_show_network('json')
|
|
|
|
def test_show_network_xml(self):
|
|
self._test_show_network('xml')
|
|
|
|
def test_delete_network_json(self):
|
|
self._test_delete_network('json')
|
|
|
|
def test_delete_network_xml(self):
|
|
self._test_delete_network('xml')
|
|
|
|
def test_rename_network_json(self):
|
|
self._test_rename_network('json')
|
|
|
|
def test_rename_network_xml(self):
|
|
self._test_rename_network('xml')
|
|
|
|
def test_rename_network_badrequest_json(self):
|
|
self._test_rename_network_badrequest('json')
|
|
|
|
def test_rename_network_badrequest_xml(self):
|
|
self._test_rename_network_badrequest('xml')
|
|
|
|
def test_rename_network_not_found_json(self):
|
|
self._test_rename_network_not_found('json')
|
|
|
|
def test_rename_network_not_found_xml(self):
|
|
self._test_rename_network_not_found('xml')
|
|
|
|
def test_delete_network_in_use_json(self):
|
|
self._test_delete_network_in_use('json')
|
|
|
|
def test_delete_network_in_use_xml(self):
|
|
self._test_delete_network_in_use('xml')
|
|
|
|
def test_list_ports_json(self):
|
|
self._test_list_ports('json')
|
|
|
|
def test_list_ports_xml(self):
|
|
self._test_list_ports('xml')
|
|
|
|
def test_show_port_json(self):
|
|
self._test_show_port('json')
|
|
|
|
def test_show_port_xml(self):
|
|
self._test_show_port('xml')
|
|
|
|
def test_show_port_networknotfound_json(self):
|
|
self._test_show_port_networknotfound('json')
|
|
|
|
def test_show_port_networknotfound_xml(self):
|
|
self._test_show_port_networknotfound('xml')
|
|
|
|
def test_show_port_portnotfound_json(self):
|
|
self._test_show_port_portnotfound('json')
|
|
|
|
def test_show_port_portnotfound_xml(self):
|
|
self._test_show_port_portnotfound('xml')
|
|
|
|
def test_create_port_json(self):
|
|
self._test_create_port('json')
|
|
|
|
def test_create_port_xml(self):
|
|
self._test_create_port('xml')
|
|
|
|
def test_create_port_networknotfound_json(self):
|
|
self._test_create_port_networknotfound('json')
|
|
|
|
def test_create_port_networknotfound_xml(self):
|
|
self._test_create_port_networknotfound('xml')
|
|
|
|
def test_create_port_badrequest_json(self):
|
|
self._test_create_port_badrequest('json')
|
|
|
|
def test_create_port_badrequest_xml(self):
|
|
self._test_create_port_badrequest('xml')
|
|
|
|
def test_delete_port_xml(self):
|
|
self._test_delete_port('xml')
|
|
|
|
def test_delete_port_json(self):
|
|
self._test_delete_port('json')
|
|
|
|
def test_delete_port_in_use_xml(self):
|
|
self._test_delete_port_in_use('xml')
|
|
|
|
def test_delete_port_in_use_json(self):
|
|
self._test_delete_port_in_use('json')
|
|
|
|
def test_delete_port_networknotfound_xml(self):
|
|
self._test_delete_port_networknotfound('xml')
|
|
|
|
def test_delete_port_networknotfound_json(self):
|
|
self._test_delete_port_networknotfound('json')
|
|
|
|
def test_delete_port_with_bad_id_xml(self):
|
|
self._test_delete_port_with_bad_id('xml')
|
|
|
|
def test_delete_port_with_bad_id_json(self):
|
|
self._test_delete_port_with_bad_id('json')
|
|
|
|
def test_set_port_state_xml(self):
|
|
self._test_set_port_state('xml')
|
|
|
|
def test_set_port_state_json(self):
|
|
self._test_set_port_state('json')
|
|
|
|
def test_set_port_state_networknotfound_xml(self):
|
|
self._test_set_port_state_networknotfound('xml')
|
|
|
|
def test_set_port_state_networknotfound_json(self):
|
|
self._test_set_port_state_networknotfound('json')
|
|
|
|
def test_set_port_state_portnotfound_xml(self):
|
|
self._test_set_port_state_portnotfound('xml')
|
|
|
|
def test_set_port_state_portnotfound_json(self):
|
|
self._test_set_port_state_portnotfound('json')
|
|
|
|
def test_set_port_state_stateinvalid_xml(self):
|
|
self._test_set_port_state_stateinvalid('xml')
|
|
|
|
def test_set_port_state_stateinvalid_json(self):
|
|
self._test_set_port_state_stateinvalid('json')
|
|
|
|
def test_show_attachment_xml(self):
|
|
self._test_show_attachment('xml')
|
|
|
|
def test_show_attachment_json(self):
|
|
self._test_show_attachment('json')
|
|
|
|
def test_show_attachment_networknotfound_xml(self):
|
|
self._test_show_attachment_networknotfound('xml')
|
|
|
|
def test_show_attachment_networknotfound_json(self):
|
|
self._test_show_attachment_networknotfound('json')
|
|
|
|
def test_show_attachment_portnotfound_xml(self):
|
|
self._test_show_attachment_portnotfound('xml')
|
|
|
|
def test_show_attachment_portnotfound_json(self):
|
|
self._test_show_attachment_portnotfound('json')
|
|
|
|
def test_put_attachment_xml(self):
|
|
self._test_put_attachment('xml')
|
|
|
|
def test_put_attachment_json(self):
|
|
self._test_put_attachment('json')
|
|
|
|
def test_put_attachment_networknotfound_xml(self):
|
|
self._test_put_attachment_networknotfound('xml')
|
|
|
|
def test_put_attachment_networknotfound_json(self):
|
|
self._test_put_attachment_networknotfound('json')
|
|
|
|
def test_put_attachment_portnotfound_xml(self):
|
|
self._test_put_attachment_portnotfound('xml')
|
|
|
|
def test_put_attachment_portnotfound_json(self):
|
|
self._test_put_attachment_portnotfound('json')
|
|
|
|
def test_delete_attachment_xml(self):
|
|
self._test_delete_attachment('xml')
|
|
|
|
def test_delete_attachment_json(self):
|
|
self._test_delete_attachment('json')
|
|
|
|
def test_delete_attachment_networknotfound_xml(self):
|
|
self._test_delete_attachment_networknotfound('xml')
|
|
|
|
def test_delete_attachment_networknotfound_json(self):
|
|
self._test_delete_attachment_networknotfound('json')
|
|
|
|
def test_delete_attachment_portnotfound_xml(self):
|
|
self._test_delete_attachment_portnotfound('xml')
|
|
|
|
def test_delete_attachment_portnotfound_json(self):
|
|
self._test_delete_attachment_portnotfound('json')
|