def _raise(self, *args, **kwargs):
network_type = kwargs.get('network_type', None)
msg = 'Invalid network type %s.\nValid types: %s' % (
- network_type, ' , '.join(this.net_types))
+ network_type, ' '.join(this.net_types))
assert network_type in this.net_types, msg
return foo(self, *args, **kwargs)
return _raise
from kamaki.cli import command
from kamaki.cli.command_tree import CommandTree
-from kamaki.cli.errors import (
- CLISyntaxError, CLIBaseUrlError, CLIInvalidArgument)
+from kamaki.cli.errors import CLISyntaxError, CLIBaseUrlError
from kamaki.clients.cyclades import CycladesNetworkClient
from kamaki.cli.argument import FlagArgument, ValueArgument
from kamaki.cli.commands import _command_init, errors, addLogSettings
@errors.generic.all
@errors.cyclades.connection
def _run(self):
- nets = self.client.list_networks()
+ nets = self.client.list_networks(detail=self['detail'])
nets = self._filter_by_name(nets)
nets = self._filter_by_id(nets)
- if not self['detail']:
- nets = [dict(id=net['id'], name=net['name']) for net in nets]
kwargs = dict()
if self['more']:
kwargs['out'] = StringIO()
@command(network_cmds)
class network_create(_init_network, _optional_json):
"""Create a new network
- Valid network types: CUSTOM, MAC_FILTERED, IP_LESS_ROUTED, PHYSICAL_VLAN
+ Valid network types: CUSTOM MAC_FILTERED IP_LESS_ROUTED PHYSICAL_VLAN
"""
arguments = dict(
from kamaki.clients.cyclades.rest_api import CycladesRestClient
from kamaki.clients.network import NetworkClient
+from kamaki.clients.utils import path4url
from kamaki.clients import ClientError
network_types = (
'CUSTOM', 'MAC_FILTERED', 'IP_LESS_ROUTED', 'PHYSICAL_VLAN')
+ def list_networks(self, detail=None):
+ path = path4url('networks', 'detail' if detail else '')
+ r = self.get(path, success=200)
+ return r.json['networks']
+
def create_network(self, type, name=None, shared=None):
req = dict(network=dict(type=type, admin_state_up=True))
if name:
FR.json = vm_recv
del self.client
+ @patch('kamaki.clients.Client.get', return_value=FR)
+ def test_list_networks(self, get):
+ FR.json = dict(networks='ret val')
+ for detail in (True, None):
+ self.assertEqual(self.client.list_networks(detail), 'ret val')
+ path = '/networks/detail' if detail else '/networks'
+ self.assertEqual(get.mock_calls[-1], call(path, success=200))
+
@patch(
- 'kamaki.clients.network.NetworkClient.networks_post',
+ 'kamaki.clients.network.rest_api.NetworkRestClient.networks_post',
return_value=FR())
def test_create_network(self, networks_post):
for name, shared in product((None, 'net name'), (None, True)):