openstack
/
python-neutronclient
Code Issues Proposed changes

Merge "Remove class-based import in the code repo"

This commit is contained in:
Jenkins 2013-06-15 23:33:59 +00:00 committed by Gerrit Code Review
parent 12bc22b07a b5a416ac34
commit 92d9698ceb
21 changed files with 337 additions and 405 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
quantumclient/common/clientmanager.py
View File

@ -20,7 +20,7 @@
import logging
from quantumclient.client import HTTPClient
from quantumclient import client
from quantumclient.quantum import client as quantum_client
@ -74,13 +74,13 @@ class ClientManager(object):
def initialize(self):
if not self._url:
httpclient = HTTPClient(username=self._username,
tenant_name=self._tenant_name,
password=self._password,
region_name=self._region_name,
auth_url=self._auth_url,
endpoint_type=self._endpoint_type,
insecure=self._insecure)
httpclient = client.HTTPClient(username=self._username,
tenant_name=self._tenant_name,
password=self._password,
region_name=self._region_name,
auth_url=self._auth_url,
endpoint_type=self._endpoint_type,
insecure=self._insecure)
httpclient.authenticate()
# Populate other password flow attributes
self._token = httpclient.auth_token

4
quantumclient/common/command.py
View File

@ -19,10 +19,10 @@
OpenStack base command
"""
from cliff.command import Command
from cliff import command
class OpenStackCommand(Command):
class OpenStackCommand(command.Command):
"""Base class for OpenStack commands
"""

17
quantumclient/quantum/v2_0/floatingip.py
View File

@ -19,14 +19,9 @@ import argparse
import logging
from quantumclient.quantum import v2_0 as quantumv20
from quantumclient.quantum.v2_0 import CreateCommand
from quantumclient.quantum.v2_0 import DeleteCommand
from quantumclient.quantum.v2_0 import ListCommand
from quantumclient.quantum.v2_0 import QuantumCommand
from quantumclient.quantum.v2_0 import ShowCommand
class ListFloatingIP(ListCommand):
class ListFloatingIP(quantumv20.ListCommand):
"""List floating ips that belong to a given tenant."""
resource = 'floatingip'
@ -37,7 +32,7 @@ class ListFloatingIP(ListCommand):
sorting_support = True
class ShowFloatingIP(ShowCommand):
class ShowFloatingIP(quantumv20.ShowCommand):
"""Show information of a given floating ip."""
resource = 'floatingip'
@ -45,7 +40,7 @@ class ShowFloatingIP(ShowCommand):
allow_names = False
class CreateFloatingIP(CreateCommand):
class CreateFloatingIP(quantumv20.CreateCommand):
"""Create a floating ip for a given tenant."""
resource = 'floatingip'
@ -83,7 +78,7 @@ class CreateFloatingIP(CreateCommand):
return body
class DeleteFloatingIP(DeleteCommand):
class DeleteFloatingIP(quantumv20.DeleteCommand):
"""Delete a given floating ip."""
log = logging.getLogger(__name__ + '.DeleteFloatingIP')
@ -91,7 +86,7 @@ class DeleteFloatingIP(DeleteCommand):
allow_names = False
class AssociateFloatingIP(QuantumCommand):
class AssociateFloatingIP(quantumv20.QuantumCommand):
"""Create a mapping between a floating ip and a fixed ip."""
api = 'network'
@ -130,7 +125,7 @@ class AssociateFloatingIP(QuantumCommand):
_('Associated floatingip %s') % parsed_args.floatingip_id)
class DisassociateFloatingIP(QuantumCommand):
class DisassociateFloatingIP(quantumv20.QuantumCommand):
"""Remove a mapping from a floating ip to a fixed ip.
"""

16
quantumclient/quantum/v2_0/network.py
View File

@ -19,11 +19,7 @@ import argparse
import logging
from quantumclient.common import exceptions
from quantumclient.quantum.v2_0 import CreateCommand
from quantumclient.quantum.v2_0 import DeleteCommand
from quantumclient.quantum.v2_0 import ListCommand
from quantumclient.quantum.v2_0 import ShowCommand
from quantumclient.quantum.v2_0 import UpdateCommand
from quantumclient.quantum import v2_0 as quantumv20
def _format_subnets(network):
@ -34,7 +30,7 @@ def _format_subnets(network):
return ''
class ListNetwork(ListCommand):
class ListNetwork(quantumv20.ListCommand):
"""List networks that belong to a given tenant."""
# Length of a query filter on subnet id
@ -101,14 +97,14 @@ class ListExternalNetwork(ListNetwork):
return super(ListExternalNetwork, self).retrieve_list(parsed_args)
class ShowNetwork(ShowCommand):
class ShowNetwork(quantumv20.ShowCommand):
"""Show information of a given network."""
resource = 'network'
log = logging.getLogger(__name__ + '.ShowNetwork')
class CreateNetwork(CreateCommand):
class CreateNetwork(quantumv20.CreateCommand):
"""Create a network for a given tenant."""
resource = 'network'
@ -142,14 +138,14 @@ class CreateNetwork(CreateCommand):
return body
class DeleteNetwork(DeleteCommand):
class DeleteNetwork(quantumv20.DeleteCommand):
"""Delete a given network."""
log = logging.getLogger(__name__ + '.DeleteNetwork')
resource = 'network'
class UpdateNetwork(UpdateCommand):
class UpdateNetwork(quantumv20.UpdateCommand):
"""Update network's information."""
log = logging.getLogger(__name__ + '.UpdateNetwork')

17
quantumclient/quantum/v2_0/port.py
View File

@ -20,11 +20,6 @@ import logging
from quantumclient.common import utils
from quantumclient.quantum import v2_0 as quantumv20
from quantumclient.quantum.v2_0 import CreateCommand
from quantumclient.quantum.v2_0 import DeleteCommand
from quantumclient.quantum.v2_0 import ListCommand
from quantumclient.quantum.v2_0 import ShowCommand
from quantumclient.quantum.v2_0 import UpdateCommand
def _format_fixed_ips(port):
@ -34,7 +29,7 @@ def _format_fixed_ips(port):
return ''
class ListPort(ListCommand):
class ListPort(quantumv20.ListCommand):
"""List ports that belong to a given tenant."""
resource = 'port'
@ -45,7 +40,7 @@ class ListPort(ListCommand):
sorting_support = True
class ListRouterPort(ListCommand):
class ListRouterPort(quantumv20.ListCommand):
"""List ports that belong to a given tenant, with specified router."""
resource = 'port'
@ -71,14 +66,14 @@ class ListRouterPort(ListCommand):
return super(ListRouterPort, self).get_data(parsed_args)
class ShowPort(ShowCommand):
class ShowPort(quantumv20.ShowCommand):
"""Show information of a given port."""
resource = 'port'
log = logging.getLogger(__name__ + '.ShowPort')
class CreatePort(CreateCommand):
class CreatePort(quantumv20.CreateCommand):
"""Create a port for a given tenant."""
resource = 'port'
@ -163,14 +158,14 @@ class CreatePort(CreateCommand):
return body
class DeletePort(DeleteCommand):
class DeletePort(quantumv20.DeleteCommand):
"""Delete a given port."""
resource = 'port'
log = logging.getLogger(__name__ + '.DeletePort')
class UpdatePort(UpdateCommand):
class UpdatePort(quantumv20.UpdateCommand):
"""Update port's information."""
resource = 'port'

9
quantumclient/quantum/v2_0/quota.py
View File

@ -24,7 +24,6 @@ from cliff import show
from quantumclient.common import exceptions
from quantumclient.common import utils
from quantumclient.quantum import v2_0 as quantumv20
from quantumclient.quantum.v2_0 import QuantumCommand
def get_tenant_id(tenant_id, client):
@ -32,7 +31,7 @@ def get_tenant_id(tenant_id, client):
client.get_quotas_tenant()['tenant']['tenant_id'])
class DeleteQuota(QuantumCommand):
class DeleteQuota(quantumv20.QuantumCommand):
"""Delete defined quotas of a given tenant."""
api = 'network'
@ -64,7 +63,7 @@ class DeleteQuota(QuantumCommand):
return
class ListQuota(QuantumCommand, lister.Lister):
class ListQuota(quantumv20.QuantumCommand, lister.Lister):
"""List defined quotas of all tenants."""
api = 'network'
@ -93,7 +92,7 @@ class ListQuota(QuantumCommand, lister.Lister):
for s in info))
class ShowQuota(QuantumCommand, show.ShowOne):
class ShowQuota(quantumv20.QuantumCommand, show.ShowOne):
"""Show quotas of a given tenant
"""
@ -140,7 +139,7 @@ class ShowQuota(QuantumCommand, show.ShowOne):
return None
class UpdateQuota(QuantumCommand, show.ShowOne):
class UpdateQuota(quantumv20.QuantumCommand, show.ShowOne):
"""Define tenant's quotas not to use defaults."""
resource = 'quota'

22
quantumclient/quantum/v2_0/router.py
View File

@ -20,12 +20,6 @@ import logging
from quantumclient.common import utils
from quantumclient.quantum import v2_0 as quantumv20
from quantumclient.quantum.v2_0 import CreateCommand
from quantumclient.quantum.v2_0 import DeleteCommand
from quantumclient.quantum.v2_0 import ListCommand
from quantumclient.quantum.v2_0 import QuantumCommand
from quantumclient.quantum.v2_0 import ShowCommand
from quantumclient.quantum.v2_0 import UpdateCommand
def _format_external_gateway_info(router):
@ -35,7 +29,7 @@ def _format_external_gateway_info(router):
return ''
class ListRouter(ListCommand):
class ListRouter(quantumv20.ListCommand):
"""List routers that belong to a given tenant."""
resource = 'router'
@ -46,14 +40,14 @@ class ListRouter(ListCommand):
sorting_support = True
class ShowRouter(ShowCommand):
class ShowRouter(quantumv20.ShowCommand):
"""Show information of a given router."""
resource = 'router'
log = logging.getLogger(__name__ + '.ShowRouter')
class CreateRouter(CreateCommand):
class CreateRouter(quantumv20.CreateCommand):
"""Create a router for a given tenant."""
resource = 'router'
@ -82,21 +76,21 @@ class CreateRouter(CreateCommand):
return body
class DeleteRouter(DeleteCommand):
class DeleteRouter(quantumv20.DeleteCommand):
"""Delete a given router."""
log = logging.getLogger(__name__ + '.DeleteRouter')
resource = 'router'
class UpdateRouter(UpdateCommand):
class UpdateRouter(quantumv20.UpdateCommand):
"""Update router's information."""
log = logging.getLogger(__name__ + '.UpdateRouter')
resource = 'router'
class RouterInterfaceCommand(QuantumCommand):
class RouterInterfaceCommand(quantumv20.QuantumCommand):
"""Based class to Add/Remove router interface."""
api = 'network'
@ -151,7 +145,7 @@ class RemoveInterfaceRouter(RouterInterfaceCommand):
_('Removed interface from router %s') % parsed_args.router_id)
class SetGatewayRouter(QuantumCommand):
class SetGatewayRouter(quantumv20.QuantumCommand):
"""Set the external network gateway for a router."""
log = logging.getLogger(__name__ + '.SetGatewayRouter')
@ -187,7 +181,7 @@ class SetGatewayRouter(QuantumCommand):
_('Set gateway for router %s') % parsed_args.router_id)
class RemoveGatewayRouter(QuantumCommand):
class RemoveGatewayRouter(quantumv20.QuantumCommand):
"""Remove an external network gateway from a router."""
log = logging.getLogger(__name__ + '.RemoveGatewayRouter')

15
quantumclient/quantum/v2_0/subnet.py
View File

@ -21,11 +21,6 @@ import logging
from quantumclient.common import exceptions
from quantumclient.common import utils
from quantumclient.quantum import v2_0 as quantumv20
from quantumclient.quantum.v2_0 import CreateCommand
from quantumclient.quantum.v2_0 import DeleteCommand
from quantumclient.quantum.v2_0 import ListCommand
from quantumclient.quantum.v2_0 import ShowCommand
from quantumclient.quantum.v2_0 import UpdateCommand
def _format_allocation_pools(subnet):
@ -52,7 +47,7 @@ def _format_host_routes(subnet):
return ''
class ListSubnet(ListCommand):
class ListSubnet(quantumv20.ListCommand):
"""List networks that belong to a given tenant."""
resource = 'subnet'
@ -65,14 +60,14 @@ class ListSubnet(ListCommand):
sorting_support = True
class ShowSubnet(ShowCommand):
class ShowSubnet(quantumv20.ShowCommand):
"""Show information of a given subnet."""
resource = 'subnet'
log = logging.getLogger(__name__ + '.ShowSubnet')
class CreateSubnet(CreateCommand):
class CreateSubnet(quantumv20.CreateCommand):
"""Create a subnet for a given tenant."""
resource = 'subnet'
@ -159,14 +154,14 @@ class CreateSubnet(CreateCommand):
return body
class DeleteSubnet(DeleteCommand):
class DeleteSubnet(quantumv20.DeleteCommand):
"""Delete a given subnet."""
resource = 'subnet'
log = logging.getLogger(__name__ + '.DeleteSubnet')
class UpdateSubnet(UpdateCommand):
class UpdateSubnet(quantumv20.UpdateCommand):
"""Update subnet's information."""
resource = 'subnet'

8
quantumclient/shell.py
View File

@ -25,8 +25,8 @@ import logging
import os
import sys
from cliff.app import App
from cliff.commandmanager import CommandManager
from cliff import app
from cliff import commandmanager
from quantumclient.common import clientmanager
from quantumclient.common import exceptions as exc
@ -287,7 +287,7 @@ class HelpAction(argparse.Action):
sys.exit(0)
class QuantumShell(App):
class QuantumShell(app.App):
CONSOLE_MESSAGE_FORMAT = '%(message)s'
DEBUG_MESSAGE_FORMAT = '%(levelname)s: %(name)s %(message)s'
@ -297,7 +297,7 @@ class QuantumShell(App):
super(QuantumShell, self).__init__(
description=__doc__.strip(),
version=VERSION,
command_manager=CommandManager('quantum.cli'), )
command_manager=commandmanager.CommandManager('quantum.cli'), )
self.commands = COMMANDS
for k, v in self.commands[apiversion].items():
self.command_manager.add_command(k, v)

4
quantumclient/v2_0/client.py
View File

@ -21,7 +21,7 @@ import time
import urllib
import urlparse
from quantumclient.client import HTTPClient
from quantumclient import client
from quantumclient.common import _
from quantumclient.common import constants
from quantumclient.common import exceptions
@ -887,7 +887,7 @@ class Client(object):
def __init__(self, **kwargs):
"""Initialize a new client for the Quantum v2.0 API."""
super(Client, self).__init__()
self.httpclient = HTTPClient(**kwargs)
self.httpclient = client.HTTPClient(**kwargs)
self.version = '2.0'
self.format = 'json'
self.action_prefix = "/v%s" % (self.version)

16
tests/unit/lb/test_cli20_healthmonitor.py
View File

@ -19,7 +19,7 @@
import sys
from mox import ContainsKeyValue
import mox
from quantumclient.quantum.v2_0.lb import healthmonitor
from tests.unit import test_cli20
@ -176,8 +176,8 @@ class CLITestV20LbHealthmonitorJSON(test_cli20.CLITestV20Base):
self.client.httpclient.request(
test_cli20.end_url(path), 'POST',
body=test_cli20.MyComparator(body, self.client),
headers=ContainsKeyValue('X-Auth-Token',
test_cli20.TOKEN)).AndReturn(return_tup)
headers=mox.ContainsKeyValue(
'X-Auth-Token', test_cli20.TOKEN)).AndReturn(return_tup)
self.mox.ReplayAll()
cmd_parser = cmd.get_parser('test_' + resource)
parsed_args = cmd_parser.parse_args(args)
@ -198,15 +198,15 @@ class CLITestV20LbHealthmonitorJSON(test_cli20.CLITestV20Base):
self.mox.StubOutWithMock(self.client.httpclient, "request")
cmd.get_client().MultipleTimes().AndReturn(self.client)
path = getattr(self.client,
"disassociate_pool_health_monitors_path") % \
{'pool': pool_id, 'health_monitor': health_monitor_id}
path = (getattr(self.client,
"disassociate_pool_health_monitors_path") %
{'pool': pool_id, 'health_monitor': health_monitor_id})
return_tup = (test_cli20.MyResp(204), None)
self.client.httpclient.request(
test_cli20.end_url(path), 'DELETE',
body=None,
headers=ContainsKeyValue('X-Auth-Token',
test_cli20.TOKEN)).AndReturn(return_tup)
headers=mox.ContainsKeyValue(
'X-Auth-Token', test_cli20.TOKEN)).AndReturn(return_tup)
self.mox.ReplayAll()
cmd_parser = cmd.get_parser('test_' + resource)
parsed_args = cmd_parser.parse_args(args)

6
tests/unit/lb/test_cli20_pool.py
View File

@ -19,7 +19,7 @@
import sys
from mox import ContainsKeyValue
import mox
from quantumclient.quantum.v2_0.lb import pool
from tests.unit import test_cli20
@ -152,8 +152,8 @@ class CLITestV20LbPoolJSON(test_cli20.CLITestV20Base):
self.client.httpclient.request(
test_cli20.end_url(path % my_id, query), 'GET',
body=None,
headers=ContainsKeyValue('X-Auth-Token',
test_cli20.TOKEN)).AndReturn(return_tup)
headers=mox.ContainsKeyValue(
'X-Auth-Token', test_cli20.TOKEN)).AndReturn(return_tup)
self.mox.ReplayAll()
cmd_parser = cmd.get_parser("test_" + resource)

138
tests/unit/test_auth.py
View File

@ -21,12 +21,10 @@ import json
import uuid
import mox
from mox import ContainsKeyValue, IsA, StrContains
import testtools
from quantumclient.client import exceptions
from quantumclient.client import HTTPClient
from quantumclient.client import ServiceCatalog
from quantumclient import client
from quantumclient.common import exceptions
USERNAME = 'testuser'
@ -73,9 +71,11 @@ class CLITestAuthKeystone(testtools.TestCase):
"""Prepare the test environment."""
super(CLITestAuthKeystone, self).setUp()
self.mox = mox.Mox()
self.client = HTTPClient(username=USERNAME, tenant_name=TENANT_NAME,
password=PASSWORD, auth_url=AUTH_URL,
region_name=REGION)
self.client = client.HTTPClient(username=USERNAME,
tenant_name=TENANT_NAME,
password=PASSWORD,
auth_url=AUTH_URL,
region_name=REGION)
self.addCleanup(self.mox.VerifyAll)
self.addCleanup(self.mox.UnsetStubs)
@ -85,12 +85,14 @@ class CLITestAuthKeystone(testtools.TestCase):
res200 = self.mox.CreateMock(httplib2.Response)
res200.status = 200
self.client.request(AUTH_URL + '/tokens', 'POST',
body=IsA(str), headers=IsA(dict)).\
AndReturn((res200, json.dumps(KS_TOKEN_RESULT)))
self.client.request(StrContains(ENDPOINT_URL + '/resource'), 'GET',
headers=ContainsKeyValue('X-Auth-Token', TOKEN)).\
AndReturn((res200, ''))
self.client.request(
AUTH_URL + '/tokens', 'POST',
body=mox.IsA(str), headers=mox.IsA(dict)
).AndReturn((res200, json.dumps(KS_TOKEN_RESULT)))
self.client.request(
mox.StrContains(ENDPOINT_URL + '/resource'), 'GET',
headers=mox.ContainsKeyValue('X-Auth-Token', TOKEN)
).AndReturn((res200, ''))
self.mox.ReplayAll()
self.client.do_request('/resource', 'GET')
@ -109,15 +111,18 @@ class CLITestAuthKeystone(testtools.TestCase):
res401.status = 401
# If a token is expired, quantum server retruns 401
self.client.request(StrContains(ENDPOINT_URL + '/resource'), 'GET',
headers=ContainsKeyValue('X-Auth-Token', TOKEN)).\
AndReturn((res401, ''))
self.client.request(AUTH_URL + '/tokens', 'POST',
body=IsA(str), headers=IsA(dict)).\
AndReturn((res200, json.dumps(KS_TOKEN_RESULT)))
self.client.request(StrContains(ENDPOINT_URL + '/resource'), 'GET',
headers=ContainsKeyValue('X-Auth-Token', TOKEN)).\
AndReturn((res200, ''))
self.client.request(
mox.StrContains(ENDPOINT_URL + '/resource'), 'GET',
headers=mox.ContainsKeyValue('X-Auth-Token', TOKEN)
).AndReturn((res401, ''))
self.client.request(
AUTH_URL + '/tokens', 'POST',
body=mox.IsA(str), headers=mox.IsA(dict)
).AndReturn((res200, json.dumps(KS_TOKEN_RESULT)))
self.client.request(
mox.StrContains(ENDPOINT_URL + '/resource'), 'GET',
headers=mox.ContainsKeyValue('X-Auth-Token', TOKEN)
).AndReturn((res200, ''))
self.mox.ReplayAll()
self.client.do_request('/resource', 'GET')
@ -129,20 +134,21 @@ class CLITestAuthKeystone(testtools.TestCase):
res200 = self.mox.CreateMock(httplib2.Response)
res200.status = 200
self.client.request(StrContains(AUTH_URL +
'/tokens/%s/endpoints' % TOKEN), 'GET',
headers=IsA(dict)). \
AndReturn((res200, json.dumps(ENDPOINTS_RESULT)))
self.client.request(StrContains(ENDPOINT_URL + '/resource'), 'GET',
headers=ContainsKeyValue('X-Auth-Token', TOKEN)). \
AndReturn((res200, ''))
self.client.request(
mox.StrContains(AUTH_URL + '/tokens/%s/endpoints' % TOKEN), 'GET',
headers=mox.IsA(dict)
).AndReturn((res200, json.dumps(ENDPOINTS_RESULT)))
self.client.request(
mox.StrContains(ENDPOINT_URL + '/resource'), 'GET',
headers=mox.ContainsKeyValue('X-Auth-Token', TOKEN)
).AndReturn((res200, ''))
self.mox.ReplayAll()
self.client.do_request('/resource', 'GET')
def test_get_endpoint_url_other(self):
self.client = HTTPClient(username=USERNAME, tenant_name=TENANT_NAME,
password=PASSWORD, auth_url=AUTH_URL,
region_name=REGION, endpoint_type='otherURL')
self.client = client.HTTPClient(
username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD,
auth_url=AUTH_URL, region_name=REGION, endpoint_type='otherURL')
self.mox.StubOutWithMock(self.client, "request")
self.client.auth_token = TOKEN
@ -150,10 +156,10 @@ class CLITestAuthKeystone(testtools.TestCase):
res200 = self.mox.CreateMock(httplib2.Response)
res200.status = 200
self.client.request(StrContains(AUTH_URL +
'/tokens/%s/endpoints' % TOKEN), 'GET',
headers=IsA(dict)). \
AndReturn((res200, json.dumps(ENDPOINTS_RESULT)))
self.client.request(
mox.StrContains(AUTH_URL + '/tokens/%s/endpoints' % TOKEN), 'GET',
headers=mox.IsA(dict)
).AndReturn((res200, json.dumps(ENDPOINTS_RESULT)))
self.mox.ReplayAll()
self.assertRaises(exceptions.EndpointTypeNotFound,
self.client.do_request,
@ -170,16 +176,18 @@ class CLITestAuthKeystone(testtools.TestCase):
res401 = self.mox.CreateMock(httplib2.Response)
res401.status = 401
self.client.request(StrContains(AUTH_URL +
'/tokens/%s/endpoints' % TOKEN), 'GET',
headers=IsA(dict)). \
AndReturn((res401, ''))
self.client.request(AUTH_URL + '/tokens', 'POST',
body=IsA(str), headers=IsA(dict)). \
AndReturn((res200, json.dumps(KS_TOKEN_RESULT)))
self.client.request(StrContains(ENDPOINT_URL + '/resource'), 'GET',
headers=ContainsKeyValue('X-Auth-Token', TOKEN)). \
AndReturn((res200, ''))
self.client.request(
mox.StrContains(AUTH_URL + '/tokens/%s/endpoints' % TOKEN), 'GET',
headers=mox.IsA(dict)
).AndReturn((res401, ''))
self.client.request(
AUTH_URL + '/tokens', 'POST',
body=mox.IsA(str), headers=mox.IsA(dict)
).AndReturn((res200, json.dumps(KS_TOKEN_RESULT)))
self.client.request(
mox.StrContains(ENDPOINT_URL + '/resource'), 'GET',
headers=mox.ContainsKeyValue('X-Auth-Token', TOKEN)
).AndReturn((res200, ''))
self.mox.ReplayAll()
self.client.do_request('/resource', 'GET')
@ -190,7 +198,7 @@ class CLITestAuthKeystone(testtools.TestCase):
endpoints['publicURL'] = 'public'
endpoints['internalURL'] = 'internal'
endpoints['adminURL'] = 'admin'
catalog = ServiceCatalog(resources)
catalog = client.ServiceCatalog(resources)
# endpoint_type not specified
url = catalog.url_for(attr='region',
@ -223,7 +231,7 @@ class CLITestAuthKeystone(testtools.TestCase):
# Test scenario with url_for when the service catalog only has publicURL.
def test_url_for_only_public_url(self):
resources = copy.deepcopy(KS_TOKEN_RESULT)
catalog = ServiceCatalog(resources)
catalog = client.ServiceCatalog(resources)
# Remove endpoints from the catalog.
endpoints = resources['access']['serviceCatalog'][0]['endpoints'][0]
@ -245,7 +253,7 @@ class CLITestAuthKeystone(testtools.TestCase):
# Test scenario with url_for when the service catalog only has adminURL.
def test_url_for_only_admin_url(self):
resources = copy.deepcopy(KS_TOKEN_RESULT)
catalog = ServiceCatalog(resources)
catalog = client.ServiceCatalog(resources)
endpoints = resources['access']['serviceCatalog'][0]['endpoints'][0]
del endpoints['internalURL']
del endpoints['publicURL']
@ -271,43 +279,41 @@ class CLITestAuthKeystone(testtools.TestCase):
endpoints['publicURL'] = 'public'
# Test default behavior is to choose public.
self.client = HTTPClient(username=USERNAME, tenant_name=TENANT_NAME,
password=PASSWORD, auth_url=AUTH_URL,
region_name=REGION)
self.client = client.HTTPClient(
username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD,
auth_url=AUTH_URL, region_name=REGION)
self.client._extract_service_catalog(resources)
self.assertEqual(self.client.endpoint_url, 'public')
# Test admin url
self.client = HTTPClient(username=USERNAME, tenant_name=TENANT_NAME,
password=PASSWORD, auth_url=AUTH_URL,
region_name=REGION, endpoint_type='adminURL')
self.client = client.HTTPClient(
username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD,
auth_url=AUTH_URL, region_name=REGION, endpoint_type='adminURL')
self.client._extract_service_catalog(resources)
self.assertEqual(self.client.endpoint_url, 'admin')
# Test public url
self.client = HTTPClient(username=USERNAME, tenant_name=TENANT_NAME,
password=PASSWORD, auth_url=AUTH_URL,
region_name=REGION, endpoint_type='publicURL')
self.client = client.HTTPClient(
username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD,
auth_url=AUTH_URL, region_name=REGION, endpoint_type='publicURL')
self.client._extract_service_catalog(resources)
self.assertEqual(self.client.endpoint_url, 'public')
# Test internal url
self.client = HTTPClient(username=USERNAME, tenant_name=TENANT_NAME,
password=PASSWORD, auth_url=AUTH_URL,
region_name=REGION,
endpoint_type='internalURL')
self.client = client.HTTPClient(
username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD,
auth_url=AUTH_URL, region_name=REGION, endpoint_type='internalURL')
self.client._extract_service_catalog(resources)
self.assertEqual(self.client.endpoint_url, 'internal')
# Test url that isn't found in the service catalog
self.client = HTTPClient(username=USERNAME, tenant_name=TENANT_NAME,
password=PASSWORD, auth_url=AUTH_URL,
region_name=REGION,
endpoint_type='privateURL')
self.client = client.HTTPClient(
username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD,
auth_url=AUTH_URL, region_name=REGION, endpoint_type='privateURL')
self.assertRaises(exceptions.EndpointTypeNotFound,
self.client._extract_service_catalog,

61
tests/unit/test_cli20.py
View File

@ -19,13 +19,11 @@ import urllib
import fixtures
import mox
from mox import Comparator
from mox import ContainsKeyValue
import testtools
from quantumclient.common import constants
from quantumclient import shell
from quantumclient.v2_0.client import Client
from quantumclient.v2_0 import client
API_VERSION = "2.0"
@ -64,7 +62,7 @@ def end_url(path, query=None, format=FORMAT):
return query and _url_str + "?" + query or _url_str
class MyUrlComparator(Comparator):
class MyUrlComparator(mox.Comparator):
def __init__(self, lhs, client):
self.lhs = lhs
self.client = client
@ -89,7 +87,7 @@ class MyUrlComparator(Comparator):
return str(self)
class MyComparator(Comparator):
class MyComparator(mox.Comparator):
def __init__(self, lhs, client):
self.lhs = lhs
self.client = client
@ -152,18 +150,18 @@ class CLITestV20Base(testtools.TestCase):
def _get_attr_metadata(self):
return self.metadata
Client.EXTED_PLURALS.update(constants.PLURALS)
Client.EXTED_PLURALS.update({'tags': 'tag'})
return {'plurals': Client.EXTED_PLURALS,
client.Client.EXTED_PLURALS.update(constants.PLURALS)
client.Client.EXTED_PLURALS.update({'tags': 'tag'})
return {'plurals': client.Client.EXTED_PLURALS,
'xmlns': constants.XML_NS_V20,
constants.EXT_NS: {'prefix': 'http://xxxx.yy.com'}}
def setUp(self, plurals={}):
"""Prepare the test environment."""
super(CLITestV20Base, self).setUp()
Client.EXTED_PLURALS.update(constants.PLURALS)
Client.EXTED_PLURALS.update(plurals)
self.metadata = {'plurals': Client.EXTED_PLURALS,
client.Client.EXTED_PLURALS.update(constants.PLURALS)
client.Client.EXTED_PLURALS.update(plurals)
self.metadata = {'plurals': client.Client.EXTED_PLURALS,
'xmlns': constants.XML_NS_V20,
constants.EXT_NS: {'prefix':
'http://xxxx.yy.com'}}
@ -177,7 +175,7 @@ class CLITestV20Base(testtools.TestCase):
self.useFixture(fixtures.MonkeyPatch(
'quantumclient.v2_0.client.Client.get_attr_metadata',
self._get_attr_metadata))
self.client = Client(token=TOKEN, endpoint_url=self.endurl)
self.client = client.Client(token=TOKEN, endpoint_url=self.endurl)
def _test_create_resource(self, resource, cmd,
name, myid, args,
@ -217,9 +215,8 @@ class CLITestV20Base(testtools.TestCase):
self.client.httpclient.request(
end_url(path, format=self.format), 'POST',
body=MyComparator(body, self.client),
headers=ContainsKeyValue('X-Auth-Token',
TOKEN)).AndReturn((MyResp(200),
resstr))
headers=mox.ContainsKeyValue(
'X-Auth-Token', TOKEN)).AndReturn((MyResp(200), resstr))
args.extend(['--request-format', self.format])
self.mox.ReplayAll()
cmd_parser = cmd.get_parser('create_' + resource)
@ -243,8 +240,8 @@ class CLITestV20Base(testtools.TestCase):
self.client.httpclient.request(
end_url(path, format=self.format), 'GET',
body=None,
headers=ContainsKeyValue('X-Auth-Token',
TOKEN)).AndReturn((MyResp(200), resstr))
headers=mox.ContainsKeyValue(
'X-Auth-Token', TOKEN)).AndReturn((MyResp(200), resstr))
args.extend(['--request-format', self.format])
self.mox.ReplayAll()
cmd_parser = cmd.get_parser("list_" + resources_collection)
@ -332,8 +329,8 @@ class CLITestV20Base(testtools.TestCase):
self.client),
'GET',
body=None,
headers=ContainsKeyValue('X-Auth-Token',
TOKEN)).AndReturn((MyResp(200), resstr))
headers=mox.ContainsKeyValue(
'X-Auth-Token', TOKEN)).AndReturn((MyResp(200), resstr))
self.mox.ReplayAll()
cmd_parser = cmd.get_parser("list_" + resources)
shell.run_command(cmd, cmd_parser, args)
@ -362,13 +359,13 @@ class CLITestV20Base(testtools.TestCase):
self.client.httpclient.request(
end_url(path, "", format=self.format), 'GET',
body=None,
headers=ContainsKeyValue('X-Auth-Token',
TOKEN)).AndReturn((MyResp(200), resstr1))
headers=mox.ContainsKeyValue(
'X-Auth-Token', TOKEN)).AndReturn((MyResp(200), resstr1))
self.client.httpclient.request(
end_url(path, fake_query, format=self.format), 'GET',
body=None,
headers=ContainsKeyValue('X-Auth-Token',
TOKEN)).AndReturn((MyResp(200), resstr2))
headers=mox.ContainsKeyValue(
'X-Auth-Token', TOKEN)).AndReturn((MyResp(200), resstr2))
self.mox.ReplayAll()
cmd_parser = cmd.get_parser("list_" + resources)
args = ['--request-format', self.format]
@ -389,8 +386,8 @@ class CLITestV20Base(testtools.TestCase):
self.client),
'PUT',
body=MyComparator(body, self.client),
headers=ContainsKeyValue('X-Auth-Token',
TOKEN)).AndReturn((MyResp(204), None))
headers=mox.ContainsKeyValue(
'X-Auth-Token', TOKEN)).AndReturn((MyResp(204), None))
args.extend(['--request-format', self.format])
self.mox.ReplayAll()
cmd_parser = cmd.get_parser("update_" + resource)
@ -414,8 +411,8 @@ class CLITestV20Base(testtools.TestCase):
self.client.httpclient.request(
end_url(path % myid, query, format=self.format), 'GET',
body=None,
headers=ContainsKeyValue('X-Auth-Token',
TOKEN)).AndReturn((MyResp(200), resstr))
headers=mox.ContainsKeyValue(
'X-Auth-Token', TOKEN)).AndReturn((MyResp(200), resstr))
args.extend(['--request-format', self.format])
self.mox.ReplayAll()
cmd_parser = cmd.get_parser("show_" + resource)
@ -434,8 +431,8 @@ class CLITestV20Base(testtools.TestCase):
self.client.httpclient.request(
end_url(path % myid, format=self.format), 'DELETE',
body=None,
headers=ContainsKeyValue('X-Auth-Token',
TOKEN)).AndReturn((MyResp(204), None))
headers=mox.ContainsKeyValue(
'X-Auth-Token', TOKEN)).AndReturn((MyResp(204), None))
args.extend(['--request-format', self.format])
self.mox.ReplayAll()
cmd_parser = cmd.get_parser("delete_" + resource)
@ -455,8 +452,8 @@ class CLITestV20Base(testtools.TestCase):
self.client.httpclient.request(
end_url(path % path_action, format=self.format), 'PUT',
body=MyComparator(body, self.client),
headers=ContainsKeyValue('X-Auth-Token',
TOKEN)).AndReturn((MyResp(204), None))
headers=mox.ContainsKeyValue(
'X-Auth-Token', TOKEN)).AndReturn((MyResp(204), None))
args.extend(['--request-format', self.format])
self.mox.ReplayAll()
cmd_parser = cmd.get_parser("delete_" + resource)
@ -489,7 +486,7 @@ class ClientV2UnicodeTestJson(CLITestV20Base):
self.client.httpclient.request(
end_url(expected_action, query=expect_query, format=self.format),
'PUT', body=expect_body,
headers=ContainsKeyValue(
headers=mox.ContainsKeyValue(
'X-Auth-Token',
expected_auth_token)).AndReturn((MyResp(200), expect_body))

34
tests/unit/test_cli20_floatingips.py
View File

@ -18,21 +18,15 @@
import sys
from quantumclient.quantum.v2_0.floatingip import AssociateFloatingIP
from quantumclient.quantum.v2_0.floatingip import CreateFloatingIP
from quantumclient.quantum.v2_0.floatingip import DeleteFloatingIP
from quantumclient.quantum.v2_0.floatingip import DisassociateFloatingIP
from quantumclient.quantum.v2_0.floatingip import ListFloatingIP
from quantumclient.quantum.v2_0.floatingip import ShowFloatingIP
from tests.unit.test_cli20 import CLITestV20Base
from tests.unit.test_cli20 import MyApp
from quantumclient.quantum.v2_0 import floatingip as fip
from tests.unit import test_cli20
class CLITestV20FloatingIpsJSON(CLITestV20Base):
class CLITestV20FloatingIpsJSON(test_cli20.CLITestV20Base):
def test_create_floatingip(self):
"""Create floatingip: fip1."""
resource = 'floatingip'
cmd = CreateFloatingIP(MyApp(sys.stdout), None)
cmd = fip.CreateFloatingIP(test_cli20.MyApp(sys.stdout), None)
name = 'fip1'
myid = 'myid'
args = [name]
@ -44,7 +38,7 @@ class CLITestV20FloatingIpsJSON(CLITestV20Base):
def test_create_floatingip_and_port(self):
"""Create floatingip: fip1."""
resource = 'floatingip'
cmd = CreateFloatingIP(MyApp(sys.stdout), None)
cmd = fip.CreateFloatingIP(test_cli20.MyApp(sys.stdout), None)
name = 'fip1'
myid = 'myid'
pid = 'mypid'
@ -63,7 +57,7 @@ class CLITestV20FloatingIpsJSON(CLITestV20Base):
def test_create_floatingip_and_port_and_address(self):
"""Create floatingip: fip1 with a given port and address."""
resource = 'floatingip'
cmd = CreateFloatingIP(MyApp(sys.stdout), None)
cmd = fip.CreateFloatingIP(test_cli20.MyApp(sys.stdout), None)
name = 'fip1'
myid = 'myid'
pid = 'mypid'
@ -82,12 +76,12 @@ class CLITestV20FloatingIpsJSON(CLITestV20Base):
def test_list_floatingips(self):
"""list floatingips: -D."""
resources = 'floatingips'
cmd = ListFloatingIP(MyApp(sys.stdout), None)
cmd = fip.ListFloatingIP(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resources, cmd, True)
def test_list_floatingips_pagination(self):
resources = 'floatingips'
cmd = ListFloatingIP(MyApp(sys.stdout), None)
cmd = fip.ListFloatingIP(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources_with_pagination(resources, cmd)
def test_list_floatingips_sort(self):
@ -95,7 +89,7 @@ class CLITestV20FloatingIpsJSON(CLITestV20Base):
--sort-key desc
"""
resources = 'floatingips'
cmd = ListFloatingIP(MyApp(sys.stdout), None)
cmd = fip.ListFloatingIP(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resources, cmd,
sort_key=["name", "id"],
sort_dir=["asc", "desc"])
@ -103,13 +97,13 @@ class CLITestV20FloatingIpsJSON(CLITestV20Base):
def test_list_floatingips_limit(self):
"""list floatingips: -P."""
resources = 'floatingips'
cmd = ListFloatingIP(MyApp(sys.stdout), None)
cmd = fip.ListFloatingIP(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resources, cmd, page_size=1000)
def test_delete_floatingip(self):
"""Delete floatingip: fip1."""
resource = 'floatingip'
cmd = DeleteFloatingIP(MyApp(sys.stdout), None)
cmd = fip.DeleteFloatingIP(test_cli20.MyApp(sys.stdout), None)
myid = 'myid'
args = [myid]
self._test_delete_resource(resource, cmd, myid, args)
@ -117,7 +111,7 @@ class CLITestV20FloatingIpsJSON(CLITestV20Base):
def test_show_floatingip(self):
"""Show floatingip: --fields id."""
resource = 'floatingip'
cmd = ShowFloatingIP(MyApp(sys.stdout), None)
cmd = fip.ShowFloatingIP(test_cli20.MyApp(sys.stdout), None)
args = ['--fields', 'id', self.test_id]
self._test_show_resource(resource, cmd, self.test_id,
args, ['id'])
@ -125,7 +119,7 @@ class CLITestV20FloatingIpsJSON(CLITestV20Base):
def test_disassociate_ip(self):
"""Disassociate floating IP: myid."""
resource = 'floatingip'
cmd = DisassociateFloatingIP(MyApp(sys.stdout), None)
cmd = fip.DisassociateFloatingIP(test_cli20.MyApp(sys.stdout), None)
args = ['myid']
self._test_update_resource(resource, cmd, 'myid',
args, {"port_id": None}
@ -134,7 +128,7 @@ class CLITestV20FloatingIpsJSON(CLITestV20Base):
def test_associate_ip(self):
"""Associate floating IP: myid portid."""
resource = 'floatingip'
cmd = AssociateFloatingIP(MyApp(sys.stdout), None)
cmd = fip.AssociateFloatingIP(test_cli20.MyApp(sys.stdout), None)
args = ['myid', 'portid']
self._test_update_resource(resource, cmd, 'myid',
args, {"port_id": "portid"}

127
tests/unit/test_cli20_network.py
View File

@ -16,30 +16,23 @@
import sys
from mox import (ContainsKeyValue, IgnoreArg, IsA)
import mox
from quantumclient.common import exceptions
from quantumclient.common import utils
from quantumclient.quantum.v2_0.network import CreateNetwork
from quantumclient.quantum.v2_0.network import DeleteNetwork
from quantumclient.quantum.v2_0.network import ListExternalNetwork
from quantumclient.quantum.v2_0.network import ListNetwork
from quantumclient.quantum.v2_0.network import ShowNetwork
from quantumclient.quantum.v2_0.network import UpdateNetwork
from quantumclient.quantum.v2_0 import network
from quantumclient import shell
from tests.unit import test_cli20
from tests.unit.test_cli20 import CLITestV20Base
from tests.unit.test_cli20 import MyApp
class CLITestV20NetworkJSON(CLITestV20Base):
class CLITestV20NetworkJSON(test_cli20.CLITestV20Base):
def setUp(self):
super(CLITestV20NetworkJSON, self).setUp(plurals={'tags': 'tag'})
def test_create_network(self):
"""Create net: myname."""
resource = 'network'
cmd = CreateNetwork(MyApp(sys.stdout), None)
cmd = network.CreateNetwork(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
args = [name, ]
@ -51,7 +44,7 @@ class CLITestV20NetworkJSON(CLITestV20Base):
def test_create_network_with_unicode(self):
"""Create net: u'\u7f51\u7edc'."""
resource = 'network'
cmd = CreateNetwork(MyApp(sys.stdout), None)
cmd = network.CreateNetwork(test_cli20.MyApp(sys.stdout), None)
name = u'\u7f51\u7edc'
myid = 'myid'
args = [name, ]
@ -63,7 +56,7 @@ class CLITestV20NetworkJSON(CLITestV20Base):
def test_create_network_tenant(self):
"""Create net: --tenant_id tenantid myname."""
resource = 'network'
cmd = CreateNetwork(MyApp(sys.stdout), None)
cmd = network.CreateNetwork(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
args = ['--tenant_id', 'tenantid', name]
@ -82,7 +75,7 @@ class CLITestV20NetworkJSON(CLITestV20Base):
def test_create_network_tags(self):
"""Create net: myname --tags a b."""
resource = 'network'
cmd = CreateNetwork(MyApp(sys.stdout), None)
cmd = network.CreateNetwork(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
args = [name, '--tags', 'a', 'b']
@ -95,7 +88,7 @@ class CLITestV20NetworkJSON(CLITestV20Base):
def test_create_network_state(self):
"""Create net: --admin_state_down myname."""
resource = 'network'
cmd = CreateNetwork(MyApp(sys.stdout), None)
cmd = network.CreateNetwork(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
args = ['--admin_state_down', name, ]
@ -113,11 +106,11 @@ class CLITestV20NetworkJSON(CLITestV20Base):
def test_list_nets_empty_with_column(self):
resources = "networks"
cmd = ListNetwork(MyApp(sys.stdout), None)
cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
self.mox.StubOutWithMock(cmd, "get_client")
self.mox.StubOutWithMock(self.client.httpclient, "request")
self.mox.StubOutWithMock(ListNetwork, "extend_list")
ListNetwork.extend_list(IsA(list), IgnoreArg())
self.mox.StubOutWithMock(network.ListNetwork, "extend_list")
network.ListNetwork.extend_list(mox.IsA(list), mox.IgnoreArg())
cmd.get_client().MultipleTimes().AndReturn(self.client)
reses = {resources: []}
resstr = self.client.serialize(reses)
@ -128,7 +121,7 @@ class CLITestV20NetworkJSON(CLITestV20Base):
self.client.httpclient.request(
test_cli20.end_url(path, query), 'GET',
body=None,
headers=test_cli20.ContainsKeyValue(
headers=mox.ContainsKeyValue(
'X-Auth-Token',
test_cli20.TOKEN)).AndReturn(
(test_cli20.MyResp(200), resstr))
@ -144,63 +137,63 @@ class CLITestV20NetworkJSON(CLITestV20Base):
fields_1=[], fields_2=[], page_size=None,
sort_key=[], sort_dir=[]):
resources = "networks"
self.mox.StubOutWithMock(ListNetwork, "extend_list")
ListNetwork.extend_list(IsA(list), IgnoreArg())
self.mox.StubOutWithMock(network.ListNetwork, "extend_list")
network.ListNetwork.extend_list(mox.IsA(list), mox.IgnoreArg())
self._test_list_resources(resources, cmd, detail, tags,
fields_1, fields_2, page_size=page_size,
sort_key=sort_key, sort_dir=sort_dir)
def test_list_nets_pagination(self):
cmd = ListNetwork(MyApp(sys.stdout), None)
self.mox.StubOutWithMock(ListNetwork, "extend_list")
ListNetwork.extend_list(IsA(list), IgnoreArg())
cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
self.mox.StubOutWithMock(network.ListNetwork, "extend_list")
network.ListNetwork.extend_list(mox.IsA(list), mox.IgnoreArg())
self._test_list_resources_with_pagination("networks", cmd)
def test_list_nets_sort(self):
"""list nets: --sort-key name --sort-key id --sort-dir asc
--sort-dir desc
"""
cmd = ListNetwork(MyApp(sys.stdout), None)
cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
self._test_list_networks(cmd, sort_key=['name', 'id'],
sort_dir=['asc', 'desc'])
def test_list_nets_sort_with_keys_more_than_dirs(self):
"""list nets: --sort-key name --sort-key id --sort-dir desc
"""
cmd = ListNetwork(MyApp(sys.stdout), None)
cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
self._test_list_networks(cmd, sort_key=['name', 'id'],
sort_dir=['desc'])
def test_list_nets_sort_with_dirs_more_than_keys(self):
"""list nets: --sort-key name --sort-dir desc --sort-dir asc
"""
cmd = ListNetwork(MyApp(sys.stdout), None)
cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
self._test_list_networks(cmd, sort_key=['name'],
sort_dir=['desc', 'asc'])
def test_list_nets_limit(self):
"""list nets: -P."""
cmd = ListNetwork(MyApp(sys.stdout), None)
cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
self._test_list_networks(cmd, page_size=1000)
def test_list_nets_detail(self):
"""list nets: -D."""
cmd = ListNetwork(MyApp(sys.stdout), None)
cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
self._test_list_networks(cmd, True)
def test_list_nets_tags(self):
"""List nets: -- --tags a b."""
cmd = ListNetwork(MyApp(sys.stdout), None)
cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
self._test_list_networks(cmd, tags=['a', 'b'])
def test_list_nets_tags_with_unicode(self):
"""List nets: -- --tags u'\u7f51\u7edc'."""
cmd = ListNetwork(MyApp(sys.stdout), None)
cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
self._test_list_networks(cmd, tags=[u'\u7f51\u7edc'])
def test_list_nets_detail_tags(self):
"""List nets: -D -- --tags a b."""
cmd = ListNetwork(MyApp(sys.stdout), None)
cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
self._test_list_networks(cmd, detail=True, tags=['a', 'b'])
def _test_list_nets_extend_subnets(self, data, expected):
@ -212,10 +205,10 @@ class CLITestV20NetworkJSON(CLITestV20Base):
self.client.httpclient.request(
test_cli20.end_url(path, query), 'GET',
body=None,
headers=ContainsKeyValue(
headers=mox.ContainsKeyValue(
'X-Auth-Token', test_cli20.TOKEN)).AndReturn(resp)
cmd = ListNetwork(test_cli20.MyApp(sys.stdout), None)
cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
self.mox.StubOutWithMock(cmd, 'get_client')
self.mox.StubOutWithMock(self.client.httpclient, 'request')
cmd.get_client().AndReturn(self.client)
@ -265,19 +258,19 @@ class CLITestV20NetworkJSON(CLITestV20Base):
def test_list_nets_fields(self):
"""List nets: --fields a --fields b -- --fields c d."""
cmd = ListNetwork(MyApp(sys.stdout), None)
cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
self._test_list_networks(cmd,
fields_1=['a', 'b'], fields_2=['c', 'd'])
def _test_list_nets_columns(self, cmd, returned_body,
args=['-f', 'json']):
resources = 'networks'
self.mox.StubOutWithMock(ListNetwork, "extend_list")
ListNetwork.extend_list(IsA(list), IgnoreArg())
self.mox.StubOutWithMock(network.ListNetwork, "extend_list")
network.ListNetwork.extend_list(mox.IsA(list), mox.IgnoreArg())
self._test_list_columns(cmd, resources, returned_body, args=args)
def test_list_nets_defined_column(self):
cmd = ListNetwork(MyApp(sys.stdout), None)
cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
returned_body = {"networks": [{"name": "buildname3",
"id": "id3",
"tenant_id": "tenant_3",
@ -287,12 +280,12 @@ class CLITestV20NetworkJSON(CLITestV20Base):
_str = self.fake_stdout.make_string()
returned_networks = utils.loads(_str)
self.assertEquals(1, len(returned_networks))
network = returned_networks[0]
self.assertEquals(1, len(network))
self.assertEquals("id", network.keys()[0])
net = returned_networks[0]
self.assertEquals(1, len(net))
self.assertEquals("id", net.keys()[0])
def test_list_nets_with_default_column(self):
cmd = ListNetwork(MyApp(sys.stdout), None)
cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
returned_body = {"networks": [{"name": "buildname3",
"id": "id3",
"tenant_id": "tenant_3",
@ -301,18 +294,17 @@ class CLITestV20NetworkJSON(CLITestV20Base):
_str = self.fake_stdout.make_string()
returned_networks = utils.loads(_str)
self.assertEquals(1, len(returned_networks))
network = returned_networks[0]
self.assertEquals(3, len(network))
self.assertEquals(0, len(set(network) ^
set(cmd.list_columns)))
net = returned_networks[0]
self.assertEquals(3, len(net))
self.assertEquals(0, len(set(net) ^ set(cmd.list_columns)))
def test_list_external_nets_empty_with_column(self):
resources = "networks"
cmd = ListExternalNetwork(MyApp(sys.stdout), None)
cmd = network.ListExternalNetwork(test_cli20.MyApp(sys.stdout), None)
self.mox.StubOutWithMock(cmd, "get_client")
self.mox.StubOutWithMock(self.client.httpclient, "request")
self.mox.StubOutWithMock(ListNetwork, "extend_list")
ListNetwork.extend_list(IsA(list), IgnoreArg())
self.mox.StubOutWithMock(network.ListNetwork, "extend_list")
network.ListNetwork.extend_list(mox.IsA(list), mox.IgnoreArg())
cmd.get_client().MultipleTimes().AndReturn(self.client)
reses = {resources: []}
resstr = self.client.serialize(reses)
@ -323,7 +315,7 @@ class CLITestV20NetworkJSON(CLITestV20Base):
self.client.httpclient.request(
test_cli20.end_url(path, query), 'GET',
body=None,
headers=test_cli20.ContainsKeyValue(
headers=mox.ContainsKeyValue(
'X-Auth-Token',
test_cli20.TOKEN)).AndReturn(
(test_cli20.MyResp(200), resstr))
@ -340,8 +332,8 @@ class CLITestV20NetworkJSON(CLITestV20Base):
fields_1=[], fields_2=[]):
self.mox.StubOutWithMock(cmd, "get_client")
self.mox.StubOutWithMock(self.client.httpclient, "request")
self.mox.StubOutWithMock(ListNetwork, "extend_list")
ListNetwork.extend_list(IsA(list), IgnoreArg())
self.mox.StubOutWithMock(network.ListNetwork, "extend_list")
network.ListNetwork.extend_list(mox.IsA(list), mox.IgnoreArg())
cmd.get_client().MultipleTimes().AndReturn(self.client)
reses = {resources: [{'id': 'myid1', },
{'id': 'myid2', }, ], }
@ -388,9 +380,8 @@ class CLITestV20NetworkJSON(CLITestV20Base):
self.client.httpclient.request(
test_cli20.end_url(path, query), 'GET',
body=None,
headers=ContainsKeyValue('X-Auth-Token',
test_cli20.TOKEN)).AndReturn(
(test_cli20.MyResp(200), resstr))
headers=mox.ContainsKeyValue('X-Auth-Token', test_cli20.TOKEN)
).AndReturn((test_cli20.MyResp(200), resstr))
self.mox.ReplayAll()
cmd_parser = cmd.get_parser("list_" + resources)
shell.run_command(cmd, cmd_parser, args)
@ -403,27 +394,27 @@ class CLITestV20NetworkJSON(CLITestV20Base):
def test_list_external_nets_detail(self):
"""list external nets: -D."""
resources = "networks"
cmd = ListExternalNetwork(MyApp(sys.stdout), None)
cmd = network.ListExternalNetwork(test_cli20.MyApp(sys.stdout), None)
self._test_list_external_nets(resources, cmd, True)
def test_list_external_nets_tags(self):
"""List external nets: -- --tags a b."""
resources = "networks"
cmd = ListExternalNetwork(MyApp(sys.stdout), None)
cmd = network.ListExternalNetwork(test_cli20.MyApp(sys.stdout), None)
self._test_list_external_nets(resources,
cmd, tags=['a', 'b'])
def test_list_external_nets_detail_tags(self):
"""List external nets: -D -- --tags a b."""
resources = "networks"
cmd = ListExternalNetwork(MyApp(sys.stdout), None)
cmd = network.ListExternalNetwork(test_cli20.MyApp(sys.stdout), None)
self._test_list_external_nets(resources, cmd,
detail=True, tags=['a', 'b'])
def test_list_externel_nets_fields(self):
"""List external nets: --fields a --fields b -- --fields c d."""
resources = "networks"
cmd = ListExternalNetwork(MyApp(sys.stdout), None)
cmd = network.ListExternalNetwork(test_cli20.MyApp(sys.stdout), None)
self._test_list_external_nets(resources, cmd,
fields_1=['a', 'b'],
fields_2=['c', 'd'])
@ -431,14 +422,14 @@ class CLITestV20NetworkJSON(CLITestV20Base):
def test_update_network_exception(self):
"""Update net: myid."""
resource = 'network'
cmd = UpdateNetwork(MyApp(sys.stdout), None)
cmd = network.UpdateNetwork(test_cli20.MyApp(sys.stdout), None)
self.assertRaises(exceptions.CommandError, self._test_update_resource,
resource, cmd, 'myid', ['myid'], {})
def test_update_network(self):
"""Update net: myid --name myname --tags a b."""
resource = 'network'
cmd = UpdateNetwork(MyApp(sys.stdout), None)
cmd = network.UpdateNetwork(test_cli20.MyApp(sys.stdout), None)
self._test_update_resource(resource, cmd, 'myid',
['myid', '--name', 'myname',
'--tags', 'a', 'b'],
@ -448,7 +439,7 @@ class CLITestV20NetworkJSON(CLITestV20Base):
def test_update_network_with_unicode(self):
"""Update net: myid --name u'\u7f51\u7edc' --tags a b."""
resource = 'network'
cmd = UpdateNetwork(MyApp(sys.stdout), None)
cmd = network.UpdateNetwork(test_cli20.MyApp(sys.stdout), None)
self._test_update_resource(resource, cmd, 'myid',
['myid', '--name', u'\u7f51\u7edc',
'--tags', 'a', 'b'],
@ -459,7 +450,7 @@ class CLITestV20NetworkJSON(CLITestV20Base):
def test_show_network(self):
"""Show net: --fields id --fields name myid."""
resource = 'network'
cmd = ShowNetwork(MyApp(sys.stdout), None)
cmd = network.ShowNetwork(test_cli20.MyApp(sys.stdout), None)
args = ['--fields', 'id', '--fields', 'name', self.test_id]
self._test_show_resource(resource, cmd, self.test_id, args,
['id', 'name'])
@ -467,7 +458,7 @@ class CLITestV20NetworkJSON(CLITestV20Base):
def test_delete_network(self):
"""Delete net: myid."""
resource = 'network'
cmd = DeleteNetwork(MyApp(sys.stdout), None)
cmd = network.DeleteNetwork(test_cli20.MyApp(sys.stdout), None)
myid = 'myid'
args = [myid]
self._test_delete_resource(resource, cmd, myid, args)
@ -478,7 +469,7 @@ class CLITestV20NetworkJSON(CLITestV20Base):
for i in range(0, 10)]
self.mox.StubOutWithMock(self.client.httpclient, "request")
path = getattr(self.client, 'subnets_path')
cmd = ListNetwork(MyApp(sys.stdout), None)
cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
self.mox.StubOutWithMock(cmd, "get_client")
cmd.get_client().MultipleTimes().AndReturn(self.client)
mox_calls(path, data)
@ -509,7 +500,7 @@ class CLITestV20NetworkJSON(CLITestV20Base):
test_cli20.end_url(path, 'fields=id&fields=cidr' + filters),
'GET',
body=None,
headers=ContainsKeyValue(
headers=mox.ContainsKeyValue(
'X-Auth-Token', test_cli20.TOKEN)).AndReturn(response)
self._test_extend_list(mox_calls)
@ -523,7 +514,7 @@ class CLITestV20NetworkJSON(CLITestV20Base):
test_cli20.end_url(path, 'fields=id&fields=cidr%s' % filters),
'GET',
body=None,
headers=ContainsKeyValue(
headers=mox.ContainsKeyValue(
'X-Auth-Token', test_cli20.TOKEN)).AndRaise(
exceptions.RequestURITooLong(excess=1))
for data in sub_data_lists:
@ -533,7 +524,7 @@ class CLITestV20NetworkJSON(CLITestV20Base):
'fields=id&fields=cidr%s' % filters),
'GET',
body=None,
headers=ContainsKeyValue(
headers=mox.ContainsKeyValue(
'X-Auth-Token', test_cli20.TOKEN)).AndReturn(response)
self._test_extend_list(mox_calls)

26
tests/unit/test_cli20_nvpnetworkgateway.py
View File

@ -17,12 +17,11 @@
import sys
from quantumclient.quantum.v2_0 import nvpnetworkgateway
from tests.unit.test_cli20 import CLITestV20Base
from tests.unit.test_cli20 import MyApp
from quantumclient.quantum.v2_0 import nvpnetworkgateway as nwgw
from tests.unit import test_cli20
class CLITestV20NetworkGatewayJSON(CLITestV20Base):
class CLITestV20NetworkGatewayJSON(test_cli20.CLITestV20Base):
resource = "network_gateway"
@ -32,7 +31,7 @@ class CLITestV20NetworkGatewayJSON(CLITestV20Base):
'network_gateways': 'network_gateway'})
def test_create_gateway(self):
cmd = nvpnetworkgateway.CreateNetworkGateway(MyApp(sys.stdout), None)
cmd = nwgw.CreateNetworkGateway(test_cli20.MyApp(sys.stdout), None)
name = 'gw-test'
myid = 'myid'
args = [name, ]
@ -42,7 +41,7 @@ class CLITestV20NetworkGatewayJSON(CLITestV20Base):
position_names, position_values)
def test_create_gateway_with_tenant(self):
cmd = nvpnetworkgateway.CreateNetworkGateway(MyApp(sys.stdout), None)
cmd = nwgw.CreateNetworkGateway(test_cli20.MyApp(sys.stdout), None)
name = 'gw-test'
myid = 'myid'
args = ['--tenant_id', 'tenantid', name]
@ -53,7 +52,7 @@ class CLITestV20NetworkGatewayJSON(CLITestV20Base):
tenant_id='tenantid')
def test_create_gateway_with_device(self):
cmd = nvpnetworkgateway.CreateNetworkGateway(MyApp(sys.stdout), None)
cmd = nwgw.CreateNetworkGateway(test_cli20.MyApp(sys.stdout), None)
name = 'gw-test'
myid = 'myid'
args = ['--device', 'device_id=test', name, ]
@ -65,29 +64,29 @@ class CLITestV20NetworkGatewayJSON(CLITestV20Base):
def test_list_gateways(self):
resources = '%ss' % self.resource
cmd = nvpnetworkgateway.ListNetworkGateway(MyApp(sys.stdout), None)
cmd = nwgw.ListNetworkGateway(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resources, cmd, True)
def test_update_gateway(self):
cmd = nvpnetworkgateway.UpdateNetworkGateway(MyApp(sys.stdout), None)
cmd = nwgw.UpdateNetworkGateway(test_cli20.MyApp(sys.stdout), None)
self._test_update_resource(self.resource, cmd, 'myid',
['myid', '--name', 'cavani'],
{'name': 'cavani'})
def test_delete_gateway(self):
cmd = nvpnetworkgateway.DeleteNetworkGateway(MyApp(sys.stdout), None)
cmd = nwgw.DeleteNetworkGateway(test_cli20.MyApp(sys.stdout), None)
myid = 'myid'
args = [myid]
self._test_delete_resource(self.resource, cmd, myid, args)
def test_show_gateway(self):
cmd = nvpnetworkgateway.ShowNetworkGateway(MyApp(sys.stdout), None)
cmd = nwgw.ShowNetworkGateway(test_cli20.MyApp(sys.stdout), None)
args = ['--fields', 'id', '--fields', 'name', self.test_id]
self._test_show_resource(self.resource, cmd, self.test_id, args,
['id', 'name'])
def test_connect_network_to_gateway(self):
cmd = nvpnetworkgateway.ConnectNetworkGateway(MyApp(sys.stdout), None)
cmd = nwgw.ConnectNetworkGateway(test_cli20.MyApp(sys.stdout), None)
args = ['gw_id', 'net_id',
'--segmentation-type', 'edi',
'--segmentation-id', '7']
@ -99,8 +98,7 @@ class CLITestV20NetworkGatewayJSON(CLITestV20Base):
'segmentation_id': '7'})
def test_disconnect_network_from_gateway(self):
cmd = nvpnetworkgateway.DisconnectNetworkGateway(MyApp(sys.stdout),
None)
cmd = nwgw.DisconnectNetworkGateway(test_cli20.MyApp(sys.stdout), None)
args = ['gw_id', 'net_id',
'--segmentation-type', 'edi',
'--segmentation-id', '7']

60
tests/unit/test_cli20_port.py
View File

@ -17,28 +17,21 @@
import sys
from mox import ContainsKeyValue
import mox
from quantumclient.quantum.v2_0.port import CreatePort
from quantumclient.quantum.v2_0.port import DeletePort
from quantumclient.quantum.v2_0.port import ListPort
from quantumclient.quantum.v2_0.port import ListRouterPort
from quantumclient.quantum.v2_0.port import ShowPort
from quantumclient.quantum.v2_0.port import UpdatePort
from quantumclient.quantum.v2_0 import port
from quantumclient import shell
from tests.unit import test_cli20
from tests.unit.test_cli20 import CLITestV20Base
from tests.unit.test_cli20 import MyApp
class CLITestV20PortJSON(CLITestV20Base):
class CLITestV20PortJSON(test_cli20.CLITestV20Base):
def setUp(self):
super(CLITestV20PortJSON, self).setUp(plurals={'tags': 'tag'})
def test_create_port(self):
"""Create port: netid."""
resource = 'port'
cmd = CreatePort(MyApp(sys.stdout), None)
cmd = port.CreatePort(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
netid = 'netid'
@ -52,7 +45,7 @@ class CLITestV20PortJSON(CLITestV20Base):
def test_create_port_full(self):
"""Create port: --mac_address mac --device_id deviceid netid."""
resource = 'port'
cmd = CreatePort(MyApp(sys.stdout), None)
cmd = port.CreatePort(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
netid = 'netid'
@ -71,7 +64,7 @@ class CLITestV20PortJSON(CLITestV20Base):
def test_create_port_tenant(self):
"""Create port: --tenant_id tenantid netid."""
resource = 'port'
cmd = CreatePort(MyApp(sys.stdout), None)
cmd = port.CreatePort(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
netid = 'netid'
@ -92,7 +85,7 @@ class CLITestV20PortJSON(CLITestV20Base):
def test_create_port_tags(self):
"""Create port: netid mac_address device_id --tags a b."""
resource = 'port'
cmd = CreatePort(MyApp(sys.stdout), None)
cmd = port.CreatePort(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
netid = 'netid'
@ -107,7 +100,7 @@ class CLITestV20PortJSON(CLITestV20Base):
def test_create_port_secgroup(self):
"""Create port: --security-group sg1_id netid."""
resource = 'port'
cmd = CreatePort(MyApp(sys.stdout), None)
cmd = port.CreatePort(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
netid = 'netid'
@ -124,7 +117,7 @@ class CLITestV20PortJSON(CLITestV20Base):
--security-group sg1_id --security-group sg2_id
"""
resource = 'port'
cmd = CreatePort(MyApp(sys.stdout), None)
cmd = port.CreatePort(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
netid = 'netid'
@ -139,12 +132,12 @@ class CLITestV20PortJSON(CLITestV20Base):
def test_list_ports(self):
"""List ports: -D."""
resources = "ports"
cmd = ListPort(MyApp(sys.stdout), None)
cmd = port.ListPort(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resources, cmd, True)
def test_list_ports_pagination(self):
resources = "ports"
cmd = ListPort(MyApp(sys.stdout), None)
cmd = port.ListPort(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources_with_pagination(resources, cmd)
def test_list_ports_sort(self):
@ -152,7 +145,7 @@ class CLITestV20PortJSON(CLITestV20Base):
--sort-key desc
"""
resources = "ports"
cmd = ListPort(MyApp(sys.stdout), None)
cmd = port.ListPort(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resources, cmd,
sort_key=["name", "id"],
sort_dir=["asc", "desc"])
@ -160,25 +153,25 @@ class CLITestV20PortJSON(CLITestV20Base):
def test_list_ports_limit(self):
"""list ports: -P."""
resources = "ports"
cmd = ListPort(MyApp(sys.stdout), None)
cmd = port.ListPort(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resources, cmd, page_size=1000)
def test_list_ports_tags(self):
"""List ports: -- --tags a b."""
resources = "ports"
cmd = ListPort(MyApp(sys.stdout), None)
cmd = port.ListPort(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resources, cmd, tags=['a', 'b'])
def test_list_ports_detail_tags(self):
"""List ports: -D -- --tags a b."""
resources = "ports"
cmd = ListPort(MyApp(sys.stdout), None)
cmd = port.ListPort(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resources, cmd, detail=True, tags=['a', 'b'])
def test_list_ports_fields(self):
"""List ports: --fields a --fields b -- --fields c d."""
resources = "ports"
cmd = ListPort(MyApp(sys.stdout), None)
cmd = port.ListPort(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resources, cmd,
fields_1=['a', 'b'], fields_2=['c', 'd'])
@ -232,9 +225,8 @@ class CLITestV20PortJSON(CLITestV20Base):
self.client.httpclient.request(
test_cli20.end_url(path, query % myid), 'GET',
body=None,
headers=ContainsKeyValue('X-Auth-Token',
test_cli20.TOKEN)).AndReturn(
(test_cli20.MyResp(200), resstr))
headers=mox.ContainsKeyValue('X-Auth-Token', test_cli20.TOKEN)
).AndReturn((test_cli20.MyResp(200), resstr))
self.mox.ReplayAll()
cmd_parser = cmd.get_parser("list_" + resources)
shell.run_command(cmd, cmd_parser, args)
@ -247,28 +239,28 @@ class CLITestV20PortJSON(CLITestV20Base):
def test_list_router_ports(self):
"""List router ports: -D."""
resources = "ports"
cmd = ListRouterPort(MyApp(sys.stdout), None)
cmd = port.ListRouterPort(test_cli20.MyApp(sys.stdout), None)
self._test_list_router_port(resources, cmd,
self.test_id, True)
def test_list_router_ports_tags(self):
"""List router ports: -- --tags a b."""
resources = "ports"
cmd = ListRouterPort(MyApp(sys.stdout), None)
cmd = port.ListRouterPort(test_cli20.MyApp(sys.stdout), None)
self._test_list_router_port(resources, cmd,
self.test_id, tags=['a', 'b'])
def test_list_router_ports_detail_tags(self):
"""List router ports: -D -- --tags a b."""
resources = "ports"
cmd = ListRouterPort(MyApp(sys.stdout), None)
cmd = port.ListRouterPort(test_cli20.MyApp(sys.stdout), None)
self._test_list_router_port(resources, cmd, self.test_id,
detail=True, tags=['a', 'b'])
def test_list_router_ports_fields(self):
"""List ports: --fields a --fields b -- --fields c d."""
resources = "ports"
cmd = ListRouterPort(MyApp(sys.stdout), None)
cmd = port.ListRouterPort(test_cli20.MyApp(sys.stdout), None)
self._test_list_router_port(resources, cmd, self.test_id,
fields_1=['a', 'b'],
fields_2=['c', 'd'])
@ -276,7 +268,7 @@ class CLITestV20PortJSON(CLITestV20Base):
def test_update_port(self):
"""Update port: myid --name myname --tags a b."""
resource = 'port'
cmd = UpdatePort(MyApp(sys.stdout), None)
cmd = port.UpdatePort(test_cli20.MyApp(sys.stdout), None)
self._test_update_resource(resource, cmd, 'myid',
['myid', '--name', 'myname',
'--tags', 'a', 'b'],
@ -286,7 +278,7 @@ class CLITestV20PortJSON(CLITestV20Base):
def test_update_port_security_group_off(self):
"""Update port: --no-security-groups myid."""
resource = 'port'
cmd = UpdatePort(MyApp(sys.stdout), None)
cmd = port.UpdatePort(test_cli20.MyApp(sys.stdout), None)
self._test_update_resource(resource, cmd, 'myid',
['--no-security-groups', 'myid'],
{'security_groups': None})
@ -294,7 +286,7 @@ class CLITestV20PortJSON(CLITestV20Base):
def test_show_port(self):
"""Show port: --fields id --fields name myid."""
resource = 'port'
cmd = ShowPort(MyApp(sys.stdout), None)
cmd = port.ShowPort(test_cli20.MyApp(sys.stdout), None)
args = ['--fields', 'id', '--fields', 'name', self.test_id]
self._test_show_resource(resource, cmd, self.test_id,
args, ['id', 'name'])
@ -302,7 +294,7 @@ class CLITestV20PortJSON(CLITestV20Base):
def test_delete_port(self):
"""Delete port: myid."""
resource = 'port'
cmd = DeletePort(MyApp(sys.stdout), None)
cmd = port.DeletePort(test_cli20.MyApp(sys.stdout), None)
myid = 'myid'
args = [myid]
self._test_delete_resource(resource, cmd, myid, args)

45
tests/unit/test_cli20_router.py
View File

@ -18,24 +18,15 @@
import sys
from quantumclient.common import exceptions
from quantumclient.quantum.v2_0.router import AddInterfaceRouter
from quantumclient.quantum.v2_0.router import CreateRouter
from quantumclient.quantum.v2_0.router import DeleteRouter
from quantumclient.quantum.v2_0.router import ListRouter
from quantumclient.quantum.v2_0.router import RemoveGatewayRouter
from quantumclient.quantum.v2_0.router import RemoveInterfaceRouter
from quantumclient.quantum.v2_0.router import SetGatewayRouter
from quantumclient.quantum.v2_0.router import ShowRouter
from quantumclient.quantum.v2_0.router import UpdateRouter
from tests.unit.test_cli20 import CLITestV20Base
from tests.unit.test_cli20 import MyApp
from quantumclient.quantum.v2_0 import router
from tests.unit import test_cli20
class CLITestV20RouterJSON(CLITestV20Base):
class CLITestV20RouterJSON(test_cli20.CLITestV20Base):
def test_create_router(self):
"""Create router: router1."""
resource = 'router'
cmd = CreateRouter(MyApp(sys.stdout), None)
cmd = router.CreateRouter(test_cli20.MyApp(sys.stdout), None)
name = 'router1'
myid = 'myid'
args = [name, ]
@ -47,7 +38,7 @@ class CLITestV20RouterJSON(CLITestV20Base):
def test_create_router_tenant(self):
"""Create router: --tenant_id tenantid myname."""
resource = 'router'
cmd = CreateRouter(MyApp(sys.stdout), None)
cmd = router.CreateRouter(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
args = ['--tenant_id', 'tenantid', name]
@ -60,7 +51,7 @@ class CLITestV20RouterJSON(CLITestV20Base):
def test_create_router_admin_state(self):
"""Create router: --admin_state_down myname."""
resource = 'router'
cmd = CreateRouter(MyApp(sys.stdout), None)
cmd = router.CreateRouter(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
args = ['--admin_state_down', name, ]
@ -73,12 +64,12 @@ class CLITestV20RouterJSON(CLITestV20Base):
def test_list_routers_detail(self):
"""list routers: -D."""
resources = "routers"
cmd = ListRouter(MyApp(sys.stdout), None)
cmd = router.ListRouter(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resources, cmd, True)
def test_list_routers_pagination(self):
resources = "routers"
cmd = ListRouter(MyApp(sys.stdout), None)
cmd = router.ListRouter(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources_with_pagination(resources, cmd)
def test_list_routers_sort(self):
@ -86,7 +77,7 @@ class CLITestV20RouterJSON(CLITestV20Base):
--sort-key desc
"""
resources = "routers"
cmd = ListRouter(MyApp(sys.stdout), None)
cmd = router.ListRouter(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resources, cmd,
sort_key=["name", "id"],
sort_dir=["asc", "desc"])
@ -94,20 +85,20 @@ class CLITestV20RouterJSON(CLITestV20Base):
def test_list_routers_limit(self):
"""list routers: -P."""
resources = "routers"
cmd = ListRouter(MyApp(sys.stdout), None)
cmd = router.ListRouter(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resources, cmd, page_size=1000)
def test_update_router_exception(self):
"""Update router: myid."""
resource = 'router'
cmd = UpdateRouter(MyApp(sys.stdout), None)
cmd = router.UpdateRouter(test_cli20.MyApp(sys.stdout), None)
self.assertRaises(exceptions.CommandError, self._test_update_resource,
resource, cmd, 'myid', ['myid'], {})
def test_update_router(self):
"""Update router: myid --name myname --tags a b."""
resource = 'router'
cmd = UpdateRouter(MyApp(sys.stdout), None)
cmd = router.UpdateRouter(test_cli20.MyApp(sys.stdout), None)
self._test_update_resource(resource, cmd, 'myid',
['myid', '--name', 'myname'],
{'name': 'myname'}
@ -116,7 +107,7 @@ class CLITestV20RouterJSON(CLITestV20Base):
def test_delete_router(self):
"""Delete router: myid."""
resource = 'router'
cmd = DeleteRouter(MyApp(sys.stdout), None)
cmd = router.DeleteRouter(test_cli20.MyApp(sys.stdout), None)
myid = 'myid'
args = [myid]
self._test_delete_resource(resource, cmd, myid, args)
@ -124,7 +115,7 @@ class CLITestV20RouterJSON(CLITestV20Base):
def test_show_router(self):
"""Show router: myid."""
resource = 'router'
cmd = ShowRouter(MyApp(sys.stdout), None)
cmd = router.ShowRouter(test_cli20.MyApp(sys.stdout), None)
args = ['--fields', 'id', '--fields', 'name', self.test_id]
self._test_show_resource(resource, cmd, self.test_id, args,
['id', 'name'])
@ -132,7 +123,7 @@ class CLITestV20RouterJSON(CLITestV20Base):
def test_add_interface(self):
"""Add interface to router: myid subnetid."""
resource = 'router'
cmd = AddInterfaceRouter(MyApp(sys.stdout), None)
cmd = router.AddInterfaceRouter(test_cli20.MyApp(sys.stdout), None)
args = ['myid', 'subnetid']
self._test_update_resource_action(resource, cmd, 'myid',
'add_router_interface',
@ -143,7 +134,7 @@ class CLITestV20RouterJSON(CLITestV20Base):
def test_del_interface(self):
"""Delete interface from router: myid subnetid."""
resource = 'router'
cmd = RemoveInterfaceRouter(MyApp(sys.stdout), None)
cmd = router.RemoveInterfaceRouter(test_cli20.MyApp(sys.stdout), None)
args = ['myid', 'subnetid']
self._test_update_resource_action(resource, cmd, 'myid',
'remove_router_interface',
@ -154,7 +145,7 @@ class CLITestV20RouterJSON(CLITestV20Base):
def test_set_gateway(self):
"""Set external gateway for router: myid externalid."""
resource = 'router'
cmd = SetGatewayRouter(MyApp(sys.stdout), None)
cmd = router.SetGatewayRouter(test_cli20.MyApp(sys.stdout), None)
args = ['myid', 'externalid']
self._test_update_resource(resource, cmd, 'myid',
args,
@ -166,7 +157,7 @@ class CLITestV20RouterJSON(CLITestV20Base):
def test_remove_gateway(self):
"""Remove external gateway from router: externalid."""
resource = 'router'
cmd = RemoveGatewayRouter(MyApp(sys.stdout), None)
cmd = router.RemoveGatewayRouter(test_cli20.MyApp(sys.stdout), None)
args = ['externalid']
self._test_update_resource(resource, cmd, 'externalid',
args, {"external_gateway_info": {}}

65
tests/unit/test_cli20_subnet.py
View File

@ -17,23 +17,18 @@
import sys
from quantumclient.quantum.v2_0.subnet import CreateSubnet
from quantumclient.quantum.v2_0.subnet import DeleteSubnet
from quantumclient.quantum.v2_0.subnet import ListSubnet
from quantumclient.quantum.v2_0.subnet import ShowSubnet
from quantumclient.quantum.v2_0.subnet import UpdateSubnet
from tests.unit.test_cli20 import CLITestV20Base
from tests.unit.test_cli20 import MyApp
from quantumclient.quantum.v2_0 import subnet
from tests.unit import test_cli20
class CLITestV20SubnetJSON(CLITestV20Base):
class CLITestV20SubnetJSON(test_cli20.CLITestV20Base):
def setUp(self):
super(CLITestV20SubnetJSON, self).setUp(plurals={'tags': 'tag'})
def test_create_subnet(self):
"""Create subnet: --gateway gateway netid cidr."""
resource = 'subnet'
cmd = CreateSubnet(MyApp(sys.stdout), None)
cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
netid = 'netid'
@ -48,7 +43,7 @@ class CLITestV20SubnetJSON(CLITestV20Base):
def test_create_subnet_with_no_gateway(self):
"""Create subnet: --no-gateway netid cidr."""
resource = 'subnet'
cmd = CreateSubnet(MyApp(sys.stdout), None)
cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
netid = 'netid'
@ -62,7 +57,7 @@ class CLITestV20SubnetJSON(CLITestV20Base):
def test_create_subnet_with_bad_gateway_option(self):
"""Create sbunet: --no-gateway netid cidr."""
resource = 'subnet'
cmd = CreateSubnet(MyApp(sys.stdout), None)
cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
netid = 'netid'
@ -81,7 +76,7 @@ class CLITestV20SubnetJSON(CLITestV20Base):
def test_create_subnet_tenant(self):
"""Create subnet: --tenant_id tenantid netid cidr."""
resource = 'subnet'
cmd = CreateSubnet(MyApp(sys.stdout), None)
cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
netid = 'netid'
@ -96,7 +91,7 @@ class CLITestV20SubnetJSON(CLITestV20Base):
def test_create_subnet_tags(self):
"""Create subnet: netid cidr --tags a b."""
resource = 'subnet'
cmd = CreateSubnet(MyApp(sys.stdout), None)
cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
netid = 'netid'
@ -113,7 +108,7 @@ class CLITestV20SubnetJSON(CLITestV20Base):
The <allocation_pool> is --allocation_pool start=1.1.1.10,end=1.1.1.20
"""
resource = 'subnet'
cmd = CreateSubnet(MyApp(sys.stdout), None)
cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
netid = 'netid'
@ -135,7 +130,7 @@ class CLITestV20SubnetJSON(CLITestV20Base):
--allocation_pool start=1.1.1.30,end=1.1.1.40
"""
resource = 'subnet'
cmd = CreateSubnet(MyApp(sys.stdout), None)
cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
netid = 'netid'
@ -159,7 +154,7 @@ class CLITestV20SubnetJSON(CLITestV20Base):
--host-route destination=172.16.1.0/24,nexthop=1.1.1.20
"""
resource = 'subnet'
cmd = CreateSubnet(MyApp(sys.stdout), None)
cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
netid = 'netid'
@ -182,7 +177,7 @@ class CLITestV20SubnetJSON(CLITestV20Base):
--host-route destination=172.17.7.0/24,nexthop=1.1.1.40
"""
resource = 'subnet'
cmd = CreateSubnet(MyApp(sys.stdout), None)
cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
netid = 'netid'
@ -206,7 +201,7 @@ class CLITestV20SubnetJSON(CLITestV20Base):
--dns-nameserver 1.1.1.20 and --dns-nameserver 1.1.1.40
"""
resource = 'subnet'
cmd = CreateSubnet(MyApp(sys.stdout), None)
cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
netid = 'netid'
@ -226,7 +221,7 @@ class CLITestV20SubnetJSON(CLITestV20Base):
def test_create_subnet_with_disable_dhcp(self):
"""Create subnet: --tenant-id tenantid --disable-dhcp netid cidr."""
resource = 'subnet'
cmd = CreateSubnet(MyApp(sys.stdout), None)
cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
netid = 'netid'
@ -243,7 +238,7 @@ class CLITestV20SubnetJSON(CLITestV20Base):
def test_create_subnet_merge_single_plurar(self):
resource = 'subnet'
cmd = CreateSubnet(MyApp(sys.stdout), None)
cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
netid = 'netid'
@ -264,7 +259,7 @@ class CLITestV20SubnetJSON(CLITestV20Base):
def test_create_subnet_merge_plurar(self):
resource = 'subnet'
cmd = CreateSubnet(MyApp(sys.stdout), None)
cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
netid = 'netid'
@ -283,7 +278,7 @@ class CLITestV20SubnetJSON(CLITestV20Base):
def test_create_subnet_merge_single_single(self):
resource = 'subnet'
cmd = CreateSubnet(MyApp(sys.stdout), None)
cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
myid = 'myid'
netid = 'netid'
@ -305,37 +300,37 @@ class CLITestV20SubnetJSON(CLITestV20Base):
def test_list_subnets_detail(self):
"""List subnets: -D."""
resources = "subnets"
cmd = ListSubnet(MyApp(sys.stdout), None)
cmd = subnet.ListSubnet(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resources, cmd, True)
def test_list_subnets_tags(self):
"""List subnets: -- --tags a b."""
resources = "subnets"
cmd = ListSubnet(MyApp(sys.stdout), None)
cmd = subnet.ListSubnet(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resources, cmd, tags=['a', 'b'])
def test_list_subnets_known_option_after_unknown(self):
"""List subnets: -- --tags a b --request-format xml."""
resources = "subnets"
cmd = ListSubnet(MyApp(sys.stdout), None)
cmd = subnet.ListSubnet(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resources, cmd, tags=['a', 'b'])
def test_list_subnets_detail_tags(self):
"""List subnets: -D -- --tags a b."""
resources = "subnets"
cmd = ListSubnet(MyApp(sys.stdout), None)
cmd = subnet.ListSubnet(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resources, cmd, detail=True, tags=['a', 'b'])
def test_list_subnets_fields(self):
"""List subnets: --fields a --fields b -- --fields c d."""
resources = "subnets"
cmd = ListSubnet(MyApp(sys.stdout), None)
cmd = subnet.ListSubnet(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resources, cmd,
fields_1=['a', 'b'], fields_2=['c', 'd'])
def test_list_subnets_pagination(self):
resources = "subnets"
cmd = ListSubnet(MyApp(sys.stdout), None)
cmd = subnet.ListSubnet(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources_with_pagination(resources, cmd)
def test_list_subnets_sort(self):
@ -343,7 +338,7 @@ class CLITestV20SubnetJSON(CLITestV20Base):
--sort-key desc
"""
resources = "subnets"
cmd = ListSubnet(MyApp(sys.stdout), None)
cmd = subnet.ListSubnet(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resources, cmd,
sort_key=["name", "id"],
sort_dir=["asc", "desc"])
@ -351,13 +346,13 @@ class CLITestV20SubnetJSON(CLITestV20Base):
def test_list_subnets_limit(self):
"""List subnets: -P."""
resources = "subnets"
cmd = ListSubnet(MyApp(sys.stdout), None)
cmd = subnet.ListSubnet(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resources, cmd, page_size=1000)
def test_update_subnet(self):
"""Update subnet: myid --name myname --tags a b."""
resource = 'subnet'
cmd = UpdateSubnet(MyApp(sys.stdout), None)
cmd = subnet.UpdateSubnet(test_cli20.MyApp(sys.stdout), None)
self._test_update_resource(resource, cmd, 'myid',
['myid', '--name', 'myname',
'--tags', 'a', 'b'],
@ -368,7 +363,7 @@ class CLITestV20SubnetJSON(CLITestV20Base):
"""Update subnet: --request-format json myid --name myname."""
# --request-format xml is known option
resource = 'subnet'
cmd = UpdateSubnet(MyApp(sys.stdout), None)
cmd = subnet.UpdateSubnet(test_cli20.MyApp(sys.stdout), None)
self._test_update_resource(resource, cmd, 'myid',
['--request-format', 'json',
'myid', '--name', 'myname'],
@ -379,7 +374,7 @@ class CLITestV20SubnetJSON(CLITestV20Base):
"""Update subnet: myid --name myname --request-format json."""
# --request-format xml is known option
resource = 'subnet'
cmd = UpdateSubnet(MyApp(sys.stdout), None)
cmd = subnet.UpdateSubnet(test_cli20.MyApp(sys.stdout), None)
self._test_update_resource(resource, cmd, 'myid',
['myid', '--name', 'myname',
'--request-format', 'json'],
@ -389,7 +384,7 @@ class CLITestV20SubnetJSON(CLITestV20Base):
def test_show_subnet(self):
"""Show subnet: --fields id --fields name myid."""
resource = 'subnet'
cmd = ShowSubnet(MyApp(sys.stdout), None)
cmd = subnet.ShowSubnet(test_cli20.MyApp(sys.stdout), None)
args = ['--fields', 'id', '--fields', 'name', self.test_id]
self._test_show_resource(resource, cmd, self.test_id,
args, ['id', 'name'])
@ -397,7 +392,7 @@ class CLITestV20SubnetJSON(CLITestV20Base):
def test_delete_subnet(self):
"""Delete subnet: subnetid."""
resource = 'subnet'
cmd = DeleteSubnet(MyApp(sys.stdout), None)
cmd = subnet.DeleteSubnet(test_cli20.MyApp(sys.stdout), None)
myid = 'myid'
args = [myid]
self._test_delete_resource(resource, cmd, myid, args)

36
tests/unit/test_name_or_id.py
View File

@ -18,12 +18,11 @@
import uuid
import mox
from mox import ContainsKeyValue
import testtools
from quantumclient.common import exceptions
from quantumclient.quantum import v2_0 as quantumv20
from quantumclient.v2_0.client import Client
from quantumclient.v2_0 import client
from tests.unit import test_cli20
@ -34,7 +33,8 @@ class CLITestNameorID(testtools.TestCase):
super(CLITestNameorID, self).setUp()
self.mox = mox.Mox()
self.endurl = test_cli20.ENDURL
self.client = Client(token=test_cli20.TOKEN, endpoint_url=self.endurl)
self.client = client.Client(token=test_cli20.TOKEN,
endpoint_url=self.endurl)
self.addCleanup(self.mox.VerifyAll)
self.addCleanup(self.mox.UnsetStubs)
@ -47,9 +47,8 @@ class CLITestNameorID(testtools.TestCase):
self.client.httpclient.request(
test_cli20.end_url(path, "fields=id&id=" + _id), 'GET',
body=None,
headers=ContainsKeyValue('X-Auth-Token',
test_cli20.TOKEN)).AndReturn(
(test_cli20.MyResp(200), resstr))
headers=mox.ContainsKeyValue('X-Auth-Token', test_cli20.TOKEN)
).AndReturn((test_cli20.MyResp(200), resstr))
self.mox.ReplayAll()
returned_id = quantumv20.find_resourceid_by_name_or_id(
self.client, 'network', _id)
@ -65,15 +64,13 @@ class CLITestNameorID(testtools.TestCase):
self.client.httpclient.request(
test_cli20.end_url(path, "fields=id&id=" + _id), 'GET',
body=None,
headers=ContainsKeyValue('X-Auth-Token',
test_cli20.TOKEN)).AndReturn(
(test_cli20.MyResp(200), resstr1))
headers=mox.ContainsKeyValue('X-Auth-Token', test_cli20.TOKEN)
).AndReturn((test_cli20.MyResp(200), resstr1))
self.client.httpclient.request(
test_cli20.end_url(path, "fields=id&name=" + _id), 'GET',
body=None,
headers=ContainsKeyValue('X-Auth-Token',
test_cli20.TOKEN)).AndReturn(
(test_cli20.MyResp(200), resstr))
headers=mox.ContainsKeyValue('X-Auth-Token', test_cli20.TOKEN)
).AndReturn((test_cli20.MyResp(200), resstr))
self.mox.ReplayAll()
returned_id = quantumv20.find_resourceid_by_name_or_id(
self.client, 'network', _id)
@ -89,9 +86,8 @@ class CLITestNameorID(testtools.TestCase):
self.client.httpclient.request(
test_cli20.end_url(path, "fields=id&name=" + name), 'GET',
body=None,
headers=ContainsKeyValue('X-Auth-Token',
test_cli20.TOKEN)).AndReturn(
(test_cli20.MyResp(200), resstr))
headers=mox.ContainsKeyValue('X-Auth-Token', test_cli20.TOKEN)
).AndReturn((test_cli20.MyResp(200), resstr))
self.mox.ReplayAll()
returned_id = quantumv20.find_resourceid_by_name_or_id(
self.client, 'network', name)
@ -107,9 +103,8 @@ class CLITestNameorID(testtools.TestCase):
self.client.httpclient.request(
test_cli20.end_url(path, "fields=id&name=" + name), 'GET',
body=None,
headers=ContainsKeyValue('X-Auth-Token',
test_cli20.TOKEN)).AndReturn(
(test_cli20.MyResp(200), resstr))
headers=mox.ContainsKeyValue('X-Auth-Token', test_cli20.TOKEN)
).AndReturn((test_cli20.MyResp(200), resstr))
self.mox.ReplayAll()
try:
quantumv20.find_resourceid_by_name_or_id(
@ -126,9 +121,8 @@ class CLITestNameorID(testtools.TestCase):
self.client.httpclient.request(
test_cli20.end_url(path, "fields=id&name=" + name), 'GET',
body=None,
headers=ContainsKeyValue('X-Auth-Token',
test_cli20.TOKEN)).AndReturn(
(test_cli20.MyResp(200), resstr))
headers=mox.ContainsKeyValue('X-Auth-Token', test_cli20.TOKEN)
).AndReturn((test_cli20.MyResp(200), resstr))
self.mox.ReplayAll()
try:
quantumv20.find_resourceid_by_name_or_id(