return faults.Fault(e)
network = self.network_manager.\
create_network(tenant_id,
- request_params['network-name'])
+ request_params['net-name'])
builder = networks_view.get_view_builder(request)
result = builder.build(network)
return dict(networks=result)
def get_resource(self, request, tenant_id, network_id, id):
try:
- result = self.network_manager.get_interface_details(
- tenant_id, network_id, id)
+ result = self.network_manager.get_port_details(
+ tenant_id, network_id, id).get('attachment-id', None)
return dict(attachment=result)
except exception.NetworkNotFound as e:
return faults.Fault(faults.NetworkNotFound(e))
def _build_detail(self, port_data):
"""Return a simple model of a server."""
return dict(port=dict(id=port_data['port-id'],
- attachment=port_data['attachment'],
state=port_data['port-state']))
Route the incoming request to a controller based on self.map.
If no match, return a 404.
"""
- LOG.debug("HERE - wsgi.Router.__call__")
return self._router
@staticmethod
arg_dict = req.environ['wsgiorg.routing_args'][1]
action = arg_dict['action']
method = getattr(self, action)
- LOG.debug("ARG_DICT:%s", arg_dict)
- LOG.debug("Action:%s", action)
- LOG.debug("Method:%s", method)
- LOG.debug("%s %s" % (req.method, req.url))
del arg_dict['controller']
del arg_dict['action']
if 'format' in arg_dict:
are attached to the network
"""
LOG.debug("FakePlugin.get_network_details() called")
- return self._get_network(tenant_id, net_id)
+ net = self._get_network(tenant_id, net_id)
+ return {'net-id':str(net.uuid),
+ 'net-name':net.name}
def create_network(self, tenant_id, net_name):
"""
net = self._get_network(tenant_id, net_id)
# Verify that no attachments are plugged into the network
if net:
- if net['net-ports']:
- for port in db.port_list(net_id):
- if port['interface-id']:
- raise exc.NetworkInUse(net_id=net_id)
+ for port in db.port_list(net_id):
+ if port['interface-id']:
+ raise exc.NetworkInUse(net_id=net_id)
db.network_destroy(net_id)
return net
# Network not found
that is attached to this particular port.
"""
LOG.debug("FakePlugin.get_port_details() called")
- return self._get_port(tenant_id, net_id, port_id)
+ port = self._get_port(tenant_id, net_id, port_id)
+ return {'port-id':str(port.uuid),
+ 'attachment-id':port.interface_id,
+ 'port-state':port.state}
+
def create_port(self, tenant_id, net_id, port_state=None):
"""
"""
Updates the state of a port on the specified Virtual Network.
"""
- port=self._get_port(tenant_id, net_id, port_id)
LOG.debug("FakePlugin.update_port() called")
- self._validate_port_state(port_state)
- db.port_set_state(new_state)
+ self._validate_port_state(new_state)
+ db.port_set_state(port_id,new_state)
return
def delete_port(self, tenant_id, net_id, port_id):
LOG.debug("FakePlugin.delete_port() called")
net = self._get_network(tenant_id, net_id)
port = self._get_port(tenant_id, net_id, port_id)
- if port['attachment']:
+ if port['interface_id']:
raise exc.PortInUse(net_id=net_id, port_id=port_id,
- att_id=port['attachment'])
+ att_id=port['interface_id'])
try:
port = db.port_destroy(port_id)
except Exception, e:
self._validate_attachment(tenant_id, net_id, port_id,
remote_interface_id)
port = self._get_port(tenant_id, net_id, port_id)
- if port['attachment']:
+ if port['interface_id']:
raise exc.PortInUse(net_id=net_id, port_id=port_id,
- att_id=port['attachment'])
+ att_id=port['interface_id'])
db.port_set_attachment(port_id, remote_interface_id)
def unplug_interface(self, tenant_id, net_id, port_id):
# @author: Brad Hall, Nicira Networks
# @author: Salvatore Orlando, Citrix Systems
-import tests.unit.testlib as testlib
+import logging
import unittest
+import tests.unit.testlib as testlib
+
from quantum import api as server
+from quantum.common.wsgi import Serializer
+
+LOG = logging.getLogger('quantum.tests.test_api')
+
class APIPortsTest(unittest.TestCase):
def setUp(self):
- self.api = server.APIRouterv01()
+ self.api = server.APIRouterV01()
+ self.tenant_id = "test_tenant"
+ self.network_name = "test_network"
+ def tearDown(self):
+ """Clear the test environment"""
+ # Remove all the networks.
+ network_req = testlib.create_list_networks_request(self.tenant_id)
+ network_res = network_req.get_response(self.api)
+ network_data = Serializer().deserialize(network_res.body,"application/xml")
+ for network in network_data["networks"].values():
+ network_delete_req = testlib. \
+ create_network_delete_request(self.tenant_id,network['id'])
+ network_delete_req.get_response(self.api)
+
+
+
# Fault names copied here for reference
#
# _fault_names = {
# 470: "serviceUnavailable",
# 471: "pluginFault"}
- def test_deletePort(self):
- tenant = "tenant1"
- network = "test1"
- req = testlib.create_network_request(tenant, network)
- network_obj = self.network.create(req, tenant)
- network_id = network_obj["networks"]["network"]["id"]
- req = testlib.create_empty_request()
- rv = self.port.create(req, tenant, network_id)
- port_id = rv["ports"]["port"]["id"]
- self.assertTrue(port_id > 0)
- rv = self.port.delete("", tenant, network_id, port_id)
- self.assertEqual(rv.status_int, 202)
-
- def test_deletePortNegative(self):
- tenant = "tenant1"
- network = "test1"
+
+ def _test_delete_port(self, format):
+ content_type = "application/" + format
+ port_state = "ACTIVE"
+ network_req = testlib.create_new_network_request(self.tenant_id,
+ self.network_name,
+ format)
+ network_res = network_req.get_response(self.api)
+ self.assertEqual(network_res.status_int, 200)
+ network_data = Serializer().deserialize(network_res.body,
+ content_type)
+ network_id = network_data["networks"]["network"]["id"]
+ port_req = testlib.create_new_port_request(self.tenant_id,
+ network_id, port_state,
+ format)
+ port_res = port_req.get_response(self.api)
+ self.assertEqual(port_res.status_int, 200)
+ port_data = Serializer().deserialize(port_res.body, content_type)
+ port_id = port_data["ports"]["port"]["id"]
+ LOG.debug("Deleting port %(port_id)s for network %(network_id)s"\
+ " of tenant %(tenant_id)s", locals())
+ delete_port_req = testlib.create_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)
+
+ def test_deletePort_xml(self):
+ self._test_delete_port('xml')
+
+ def test_deletePort_json(self):
+ self._test_delete_port('json')
+
+
+ #def test_deletePortNegative(self):
+ # tenant = "tenant1"
+ # network = "test1"
# Check for network not found
- rv = self.port.delete("", tenant, network, 2)
- self.assertEqual(rv.wrapped_exc.status_int, 420)
+ #rv = self.port.delete("", tenant, network, 2)
+ #self.assertEqual(rv.wrapped_exc.status_int, 420)
# Create a network to put the port on
- req = testlib.create_network_request(tenant, network)
- network_obj = self.network.create(req, tenant)
- network_id = network_obj["networks"]["network"]["id"]
+ #req = testlib.create_network_request(tenant, network)
+ #network_obj = self.network.create(req, tenant)
+ #network_id = network_obj["networks"]["network"]["id"]
# Test for portnotfound
- rv = self.port.delete("", tenant, network_id, 2)
- self.assertEqual(rv.wrapped_exc.status_int, 430)
+ #rv = self.port.delete("", tenant, network_id, 2)
+ #self.assertEqual(rv.wrapped_exc.status_int, 430)
# Test for portinuse
- rv = self.port.create(req, tenant, network_id)
- port_id = rv["ports"]["port"]["id"]
- req = testlib.create_attachment_request(tenant, network_id,
- port_id, "fudd")
- rv = self.port.attach_resource(req, tenant, network_id, port_id)
- self.assertEqual(rv.status_int, 202)
- rv = self.port.delete("", tenant, network_id, port_id)
- self.assertEqual(rv.wrapped_exc.status_int, 432)
+ #rv = self.port.create(req, tenant, network_id)
+ #port_id = rv["ports"]["port"]["id"]
+ #req = testlib.create_attachment_request(tenant, network_id,
+ # port_id, "fudd")
+ #rv = self.port.attach_resource(req, tenant, network_id, port_id)
+ #self.assertEqual(rv.status_int, 202)
+ #rv = self.port.delete("", tenant, network_id, port_id)
+ #self.assertEqual(rv.wrapped_exc.status_int, 432)
+#
\ No newline at end of file
from quantum.common.wsgi import Serializer
-class Request(webob.Request):
-
- def best_match_content_type(self):
- return "application/json"
-
- def get_content_type(self):
- return "application/json"
-
-
-def create_request(path, body):
- req = Request.blank(path)
- req.method = "POST"
+def create_request(path, body, content_type, method= 'GET'):
+ req = webob.Request.blank(path)
+ req.method = method
req.headers = {}
- req.headers['Accept'] = "application/json"
+ req.headers['Accept'] = content_type
req.body = body
return req
+def create_list_networks_request(tenant_id, format='xml'):
+ method = 'GET'
+ path = "/tenants/%(tenant_id)s/networks.%(format)s" % locals()
+ content_type = "application/" + format
+ return create_request(path, None, content_type, method)
+
-def create_empty_request():
- return create_request("/v0.1/tenant.json", "")
+def create_new_network_request(tenant_id, network_name, format='xml'):
+ method = 'POST'
+ path = "/tenants/%(tenant_id)s/networks.%(format)s" % locals()
+ data = {'network': {'net-name': '%s' % network_name}}
+ content_type = "application/" + format
+ body = Serializer().serialize(data, content_type)
+ return create_request(path, body, content_type, method)
-def create_network_request(tenant_id, network_name):
- path = "/v0.1/tenants/%s/networks.json" % tenant_id
- data = {'network': {'network-name': '%s' % network_name}}
- content_type = "application/json"
+def create_network_delete_request(tenant_id, network_id, format='xml'):
+ method = 'DELETE'
+ path = "/tenants/%(tenant_id)s/networks/" \
+ "%(network_id)s.%(format)s" % locals()
+ content_type = "application/" + format
+ return create_request(path, None, content_type, method)
+
+
+def create_new_port_request(tenant_id, network_id, port_state, format='xml'):
+ method = 'POST'
+ path = "/tenants/%(tenant_id)s/networks/" \
+ "%(network_id)s/ports.%(format)s" % locals()
+ data = {'port': {'port-state': '%s' % port_state}}
+ content_type = "application/" + format
body = Serializer().serialize(data, content_type)
- return create_request(path, body)
+ return create_request(path, body, content_type, method)
+
+def create_port_delete_request(tenant_id, network_id, port_id, format='xml'):
+ method = 'DELETE'
+ path = "/tenants/%(tenant_id)s/networks/" \
+ "%(network_id)s/ports/%(port_id)s.%(format)s" % locals()
+ content_type = "application/" + format
+ return create_request(path, None, content_type, method)
def create_attachment_request(tid, nid, pid, attachment_id):
Code Review / openstack-build / neutron-build.git / commitdiff