class ViewBuilder(object):
- def __init__(self, base_url):
+ def __init__(self, base_url=None):
"""
:param base_url: url of the root wsgi application
"""
class ViewBuilder(object):
- def __init__(self, base_url):
+ def __init__(self, base_url=None):
"""
:param base_url: url of the root wsgi application
"""
# @author: Somik Behera, Nicira Networks, Inc.
# @author: Brad Hall, Nicira Networks, Inc.
+import Cheetah.Template as cheetah_template
import httplib
import logging as LOG
import json
import sys
import urllib
+from client import Client
from manager import QuantumManager
from optparse import OptionParser
-from client import Client
+from quantum.api.views.networks import ViewBuilder as NetworkBuilder
+from quantum.api.views.ports import ViewBuilder as PortBuilder
FORMAT = "json"
+CLI_TEMPLATE = "../quantum/cli_output.template"
### -- Core CLI functions
+def prepare_output(cmd, tenant_id, response):
+ #add command and tenant to response for output generation
+ response['cmd'] = cmd
+ response['tenant_id'] = tenant_id
+ template_file = open(CLI_TEMPLATE).read()
+ output = str(cheetah_template.Template(template_file,
+ searchList=response))
+ return output
+
def list_nets(manager, *args):
tenant_id = args[0]
networks = manager.get_all_networks(tenant_id)
- print "Virtual Networks on Tenant:%s\n" % tenant_id
- for net in networks:
- id = net["net-id"]
- name = net["net-name"]
- print "\tNetwork ID:%s \n\tNetwork Name:%s \n" % (id, name)
+ builder=NetworkBuilder()
+ nw_list = [builder.build(network, net_detail=True, port_detail=False)['network']
+ for network in networks]
+ res = dict(networks=nw_list)
+ output = prepare_output("list_nets", tenant_id, res)
+ print output
def api_list_nets(client, *args):
tenant_id = args[0]
res = client.list_networks()
- LOG.debug(res)
- print "Virtual Networks on Tenant:%s\n" % tenant_id
- for n in res["networks"]:
- net_id = n["id"]
- print "\tNetwork ID:%s\n" % (net_id)
- # TODO(bgh): we should make this call pass back the name too
- # name = n["net-name"]
- # LOG.info("\tNetwork ID:%s \n\tNetwork Name:%s \n" % (id, name))
+ output = prepare_output("list_nets", tenant_id, res)
+ print output
def create_net(manager, *args):
"api_func": api_unplug_iface,
"args": ["tenant-id", "net-id", "port-id"]}, }
-
def help():
print "\nCommands:"
for k in commands.keys():
commands[cmd]["api_func"](client, *args)
else:
quantum = QuantumManager()
- manager = quantum.get_manager()
+ manager = quantum.get_plugin()
commands[cmd]["func"](manager, *args)
sys.exit(0)
""" Stubs for client tools unit tests """
+from quantum import api as server
from quantum import client
from tests.unit import testlib_api
"""Simulates a failure in fetch image_glance_disk."""
def fake_send_request(self, conn, method, action, body, headers):
- # ignore headers and connection
+ # ignore headers and connection
req = testlib_api.create_request(action, body,
"application/json", method)
res = req.get_response(api)
return res
stubs.Set(client.Client, '_send_request', fake_send_request)
+
+
+class FakeStdout:
+
+ def __init__(self):
+ self.content = []
+ def write(self, text):
+ self.content.append(text)
+
+ def make_string(self):
+ result = ''
+ for line in self.content:
+ result = result + line
+ return result
+
+
class FakeHTTPConnection:
""" stub HTTP connection class for CLI testing """
- pass
+ def __init__(self, _1, _2):
+ # Ignore host and port parameters
+ self._req = None
+ options = dict(plugin_provider = \
+ 'quantum.plugins.SamplePlugin.FakePlugin')
+ self._api = server.APIRouterV01(options)
+
+ def request(self, method, action, body, headers):
+ # TODO: remove version prefix from action!
+ parts = action.split('/', 2)
+ path = '/' + parts[2]
+ self._req = testlib_api.create_request(path, body, "application/json",
+ method)
+
+ def getresponse(self):
+ res = self._req.get_response(self._api)
+
+ def _fake_read():
+ """ Trick for macking a webob.Response look like a
+ httplib.Response
+
+ """
+ return res.body
+
+ setattr(res, 'read', _fake_read)
+ return res
+
import logging
-import stubout
import sys
import unittest
from quantum import cli
from quantum.client import Client
from quantum.db import api as db
+from quantum.manager import QuantumManager
from tests.unit.client_tools import stubs as client_stubs
LOG = logging.getLogger('quantum.tests.test_cli')
+FORMAT = 'json'
class CLITest(unittest.TestCase):
options = {}
options['plugin_provider'] = 'quantum.plugins.SamplePlugin.FakePlugin'
self.api = server.APIRouterV01(options)
- #self.client = Client("host", "port", False,
- # args[0], FORMAT)
self.tenant_id = "test_tenant"
self.network_name_1 = "test_network_1"
self.network_name_2 = "test_network_2"
- # Stubout do_request
- self.stubs = stubout.StubOutForTesting()
- client_stubs.stubout_send_request(self.stubs, self.api)
+ # Prepare client and plugin manager
+ self.client = Client(tenant = self.tenant_id, format = FORMAT,
+ testingStub = client_stubs.FakeHTTPConnection)
+ self.manager = QuantumManager(options).get_plugin()
# Redirect stdout
- # Pre-populate data
- pass
+ self.fake_stdout = client_stubs.FakeStdout()
+ sys.stdout = self.fake_stdout
+ # Pre-populate data for testing using db api
+ db.network_create(self.tenant_id, self.network_name_1)
+ db.network_create(self.tenant_id, self.network_name_2)
def tearDown(self):
"""Clear the test environment"""
db.clear_db()
+ sys.stdout = sys.__stdout__
+
+ def _verify_list_networks(self):
+ # Verification - get raw result from db
+ nw_list = db.network_list(self.tenant_id)
+ networks=[dict(id=nw.uuid, name=nw.name) for nw in nw_list]
+ # Fill CLI template
+ output = cli.prepare_output('list_nets', self.tenant_id,
+ dict(networks=networks))
+ # Verify!
+ # Must add newline at the end to match effect of print call
+ self.assertEquals(self.fake_stdout.make_string(), output + '\n')
+
+ def test_list_networks(self):
+ try:
+ cli.list_nets(self.manager, self.tenant_id)
+ LOG.debug("Operation completed. Verifying result")
+ LOG.debug(self.fake_stdout.content)
+ self._verify_list_networks()
+ except:
+ LOG.exception("Exception caught: %s", sys.exc_info())
+ self.fail("test_list_network_api failed due to an exception")
+
+
def test_list_networks_api(self):
- cli.api_list_nets(client)
- pass
+ try:
+ cli.api_list_nets(self.client, self.tenant_id)
+ LOG.debug("Operation completed. Verifying result")
+ LOG.debug(self.fake_stdout.content)
+ self._verify_list_networks()
+ except:
+ LOG.exception("Exception caught: %s", sys.exc_info())
+ self.fail("test_list_network_api failed due to an exception")
+
eventlet>=0.9.12
Routes>=1.12.3
+Cheetah>=2.0.1
mox==0.5.3
nose
Paste
Code Review / openstack-build / neutron-build.git / commitdiff