# All Rights Reserved # # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. # import itertools import sys from mox3 import mox from oslo_serialization import jsonutils from neutronclient.common import exceptions from neutronclient.neutron.v2_0 import network from neutronclient import shell from neutronclient.tests.unit import test_cli20 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 = network.CreateNetwork(test_cli20.MyApp(sys.stdout), None) name = 'myname' myid = 'myid' args = [name, ] position_names = ['name', ] position_values = [name, ] self._test_create_resource(resource, cmd, name, myid, args, position_names, position_values) def test_create_network_with_unicode(self): # Create net: u'\u7f51\u7edc'. resource = 'network' cmd = network.CreateNetwork(test_cli20.MyApp(sys.stdout), None) name = u'\u7f51\u7edc' myid = 'myid' args = [name, ] position_names = ['name', ] position_values = [name, ] self._test_create_resource(resource, cmd, name, myid, args, position_names, position_values) def test_create_network_description(self): # Create net: --tenant_id tenantid myname. resource = 'network' cmd = network.CreateNetwork(test_cli20.MyApp(sys.stdout), None) name = 'myname' myid = 'myid' args = ['--description', 'Nice network', name] position_names = ['name', ] position_values = [name, ] self._test_create_resource(resource, cmd, name, myid, args, position_names, position_values, description='Nice network') def test_create_network_tenant(self): # Create net: --tenant_id tenantid myname. resource = 'network' cmd = network.CreateNetwork(test_cli20.MyApp(sys.stdout), None) name = 'myname' myid = 'myid' args = ['--tenant_id', 'tenantid', name] position_names = ['name', ] position_values = [name, ] self._test_create_resource(resource, cmd, name, myid, args, position_names, position_values, tenant_id='tenantid') # Test dashed options args = ['--tenant-id', 'tenantid', name] self._test_create_resource(resource, cmd, name, myid, args, position_names, position_values, tenant_id='tenantid') def test_create_network_provider_args(self): # Create net: with --provider arguments. resource = 'network' cmd = network.CreateNetwork(test_cli20.MyApp(sys.stdout), None) name = 'myname' myid = 'myid' # Test --provider attributes before network name args = ['--provider:network_type', 'vlan', '--provider:physical_network', 'physnet1', '--provider:segmentation_id', '400', name] position_names = ['provider:network_type', 'provider:physical_network', 'provider:segmentation_id', 'name'] position_values = ['vlan', 'physnet1', '400', name] self._test_create_resource(resource, cmd, name, myid, args, position_names, position_values) def test_create_network_tags(self): # Create net: myname --tags a b. resource = 'network' cmd = network.CreateNetwork(test_cli20.MyApp(sys.stdout), None) name = 'myname' myid = 'myid' args = [name, '--tags', 'a', 'b'] position_names = ['name', ] position_values = [name, ] self._test_create_resource(resource, cmd, name, myid, args, position_names, position_values, tags=['a', 'b']) def test_create_network_state(self): # Create net: --admin_state_down myname. resource = 'network' cmd = network.CreateNetwork(test_cli20.MyApp(sys.stdout), None) name = 'myname' myid = 'myid' args = ['--admin_state_down', name, ] position_names = ['name', ] position_values = [name, ] self._test_create_resource(resource, cmd, name, myid, args, position_names, position_values, admin_state_up=False) # Test dashed options args = ['--admin-state-down', name, ] self._test_create_resource(resource, cmd, name, myid, args, position_names, position_values, admin_state_up=False) def test_create_network_vlan_transparent(self): # Create net: myname --vlan-transparent True. resource = 'network' cmd = network.CreateNetwork(test_cli20.MyApp(sys.stdout), None) name = 'myname' myid = 'myid' args = ['--vlan-transparent', 'True', name] vlantrans = {'vlan_transparent': 'True'} position_names = ['name', ] position_values = [name, ] self._test_create_resource(resource, cmd, name, myid, args, position_names, position_values, **vlantrans) def test_create_network_with_qos_policy(self): # Create net: --qos-policy mypolicy. resource = 'network' cmd = network.CreateNetwork(test_cli20.MyApp(sys.stdout), None) name = 'myname' myid = 'myid' qos_policy_name = 'mypolicy' args = [name, '--qos-policy', qos_policy_name] position_names = ['name', 'qos_policy_id'] position_values = [name, qos_policy_name] self._test_create_resource(resource, cmd, name, myid, args, position_names, position_values) def test_create_network_with_az_hint(self): # Create net: --availability-zone-hint zone1 # --availability-zone-hint zone2. resource = 'network' cmd = network.CreateNetwork(test_cli20.MyApp(sys.stdout), None) name = 'myname' myid = 'myid' args = ['--availability-zone-hint', 'zone1', '--availability-zone-hint', 'zone2', name] position_names = ['availability_zone_hints', 'name'] position_values = [['zone1', 'zone2'], name] self._test_create_resource(resource, cmd, name, myid, args, position_names, position_values) def test_create_network_with_dns_domain(self): # Create net: --dns-domain my-domain.org. resource = 'network' cmd = network.CreateNetwork(test_cli20.MyApp(sys.stdout), None) name = 'myname' myid = 'myid' dns_domain_name = 'my-domain.org.' args = [name, '--dns-domain', dns_domain_name] position_names = ['name', 'dns_domain'] position_values = [name, dns_domain_name] self._test_create_resource(resource, cmd, name, myid, args, position_names, position_values) def test_list_nets_empty_with_column(self): resources = "networks" 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(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) # url method body query = "id=myfakeid" args = ['-c', 'id', '--', '--id', 'myfakeid'] path = getattr(self.client, resources + "_path") self.client.httpclient.request( test_cli20.MyUrlComparator(test_cli20.end_url(path, query), self.client), 'GET', body=None, 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) self.mox.VerifyAll() self.mox.UnsetStubs() _str = self.fake_stdout.make_string() self.assertEqual('\n', _str) def _test_list_networks(self, cmd, detail=False, tags=(), fields_1=(), fields_2=(), page_size=None, sort_key=(), sort_dir=(), base_args=None, query=''): resources = "networks" 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, base_args=base_args, query=query) def test_list_nets_pagination(self): 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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): def setup_list_stub(resources, data, query): reses = {resources: data} resstr = self.client.serialize(reses) resp = (test_cli20.MyResp(200), resstr) path = getattr(self.client, resources + '_path') self.client.httpclient.request( test_cli20.MyUrlComparator( test_cli20.end_url(path, query), self.client), 'GET', body=None, headers=mox.ContainsKeyValue( 'X-Auth-Token', test_cli20.TOKEN)).AndReturn(resp) 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) setup_list_stub('networks', data, '') cmd.get_client().AndReturn(self.client) filters = '' for n in data: for s in n['subnets']: filters = filters + "&id=%s" % s setup_list_stub('subnets', [{'id': 'mysubid1', 'cidr': '192.168.1.0/24'}, {'id': 'mysubid2', 'cidr': '172.16.0.0/24'}, {'id': 'mysubid3', 'cidr': '10.1.1.0/24'}], query='fields=id&fields=cidr' + filters) self.mox.ReplayAll() args = [] cmd_parser = cmd.get_parser('list_networks') parsed_args = cmd_parser.parse_args(args) result = cmd.take_action(parsed_args) self.mox.VerifyAll() self.mox.UnsetStubs() _result = [x for x in result[1]] self.assertEqual(len(expected), len(_result)) for res, exp in zip(_result, expected): self.assertEqual(len(exp), len(res)) for obsrvd, expctd in zip(res, exp): self.assertEqual(expctd, obsrvd) def test_list_nets_extend_subnets(self): data = [{'id': 'netid1', 'name': 'net1', 'subnets': ['mysubid1']}, {'id': 'netid2', 'name': 'net2', 'subnets': ['mysubid2', 'mysubid3']}] # id, name, subnets expected = [('netid1', 'net1', 'mysubid1 192.168.1.0/24'), ('netid2', 'net2', 'mysubid2 172.16.0.0/24\nmysubid3 10.1.1.0/24')] self._test_list_nets_extend_subnets(data, expected) def test_list_nets_extend_subnets_no_subnet(self): data = [{'id': 'netid1', 'name': 'net1', 'subnets': ['mysubid1']}, {'id': 'netid2', 'name': 'net2', 'subnets': ['mysubid4']}] # id, name, subnets expected = [('netid1', 'net1', 'mysubid1 192.168.1.0/24'), ('netid2', 'net2', 'mysubid4 ')] self._test_list_nets_extend_subnets(data, expected) def test_list_nets_fields(self): # List nets: --fields a --fields b -- --fields c d. 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(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 = network.ListNetwork(test_cli20.MyApp(sys.stdout), None) returned_body = {"networks": [{"name": "buildname3", "id": "id3", "tenant_id": "tenant_3", "subnets": []}]} self._test_list_nets_columns(cmd, returned_body, args=['-f', 'json', '-c', 'id']) _str = self.fake_stdout.make_string() returned_networks = jsonutils.loads(_str) self.assertEqual(1, len(returned_networks)) net = returned_networks[0] self.assertEqual(1, len(net)) self.assertIn("id", net.keys()) def test_list_nets_with_default_column(self): cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None) returned_body = {"networks": [{"name": "buildname3", "id": "id3", "tenant_id": "tenant_3", "subnets": []}]} self._test_list_nets_columns(cmd, returned_body) _str = self.fake_stdout.make_string() returned_networks = jsonutils.loads(_str) self.assertEqual(1, len(returned_networks)) net = returned_networks[0] self.assertEqual(3, len(net)) self.assertEqual(0, len(set(net) ^ set(cmd.list_columns))) def test_list_external_nets_empty_with_column(self): resources = "networks" 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(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) # url method body query = "router%3Aexternal=True&id=myfakeid" args = ['-c', 'id', '--', '--id', 'myfakeid'] path = getattr(self.client, resources + "_path") self.client.httpclient.request( test_cli20.MyUrlComparator( test_cli20.end_url(path, query), self.client), 'GET', body=None, 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) self.mox.VerifyAll() self.mox.UnsetStubs() _str = self.fake_stdout.make_string() self.assertEqual('\n', _str) def _test_list_external_nets(self, resources, cmd, detail=False, tags=(), fields_1=(), fields_2=()): self.mox.StubOutWithMock(cmd, "get_client") self.mox.StubOutWithMock(self.client.httpclient, "request") 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', }, ], } resstr = self.client.serialize(reses) # url method body query = "" args = detail and ['-D', ] or [] if fields_1: for field in fields_1: args.append('--fields') args.append(field) if tags: args.append('--') args.append("--tag") for tag in tags: args.append(tag) if (not tags) and fields_2: args.append('--') if fields_2: args.append("--fields") for field in fields_2: args.append(field) for field in itertools.chain(fields_1, fields_2): if query: query += "&fields=" + field else: query = "fields=" + field if query: query += '&router%3Aexternal=True' else: query += 'router%3Aexternal=True' for tag in tags: if query: query += "&tag=" + tag else: query = "tag=" + tag if detail: query = query and query + '&verbose=True' or 'verbose=True' path = getattr(self.client, resources + "_path") self.client.httpclient.request( test_cli20.MyUrlComparator( test_cli20.end_url(path, query), self.client), 'GET', body=None, 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) self.mox.VerifyAll() self.mox.UnsetStubs() _str = self.fake_stdout.make_string() self.assertIn('myid1', _str) def test_list_external_nets_detail(self): # list external nets: -D. resources = "networks" 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 = 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 = 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 = network.ListExternalNetwork(test_cli20.MyApp(sys.stdout), None) self._test_list_external_nets(resources, cmd, fields_1=['a', 'b'], fields_2=['c', 'd']) def test_update_network_exception(self): # Update net: myid. resource = 'network' 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 = network.UpdateNetwork(test_cli20.MyApp(sys.stdout), None) self._test_update_resource(resource, cmd, 'myid', ['myid', '--name', 'myname', '--tags', 'a', 'b', '--description', 'This network takes the scenic route'], {'name': 'myname', 'tags': ['a', 'b'], 'description': 'This network takes the ' 'scenic route'}) def test_update_network_with_unicode(self): # Update net: myid --name u'\u7f51\u7edc' --tags a b. resource = 'network' cmd = network.UpdateNetwork(test_cli20.MyApp(sys.stdout), None) self._test_update_resource(resource, cmd, 'myid', ['myid', '--name', u'\u7f51\u7edc', '--tags', 'a', 'b'], {'name': u'\u7f51\u7edc', 'tags': ['a', 'b'], } ) def test_update_network_with_qos_policy(self): # Update net: myid --qos-policy mypolicy. resource = 'network' cmd = network.UpdateNetwork(test_cli20.MyApp(sys.stdout), None) self._test_update_resource(resource, cmd, 'myid', ['myid', '--qos-policy', 'mypolicy'], {'qos_policy_id': 'mypolicy', }) def test_update_network_with_no_qos_policy(self): # Update net: myid --no-qos-policy. resource = 'network' cmd = network.UpdateNetwork(test_cli20.MyApp(sys.stdout), None) self._test_update_resource(resource, cmd, 'myid', ['myid', '--no-qos-policy'], {'qos_policy_id': None, }) def test_update_network_with_dns_domain(self): # Update net: myid --dns-domain my-domain.org. resource = 'network' cmd = network.UpdateNetwork(test_cli20.MyApp(sys.stdout), None) self._test_update_resource(resource, cmd, 'myid', ['myid', '--dns-domain', 'my-domain.org.'], {'dns_domain': 'my-domain.org.', }) def test_update_network_with_no_dns_domain(self): # Update net: myid --no-dns-domain resource = 'network' cmd = network.UpdateNetwork(test_cli20.MyApp(sys.stdout), None) self._test_update_resource(resource, cmd, 'myid', ['myid', '--no-dns-domain'], {'dns_domain': "", }) def test_show_network(self): # Show net: --fields id --fields name myid. resource = 'network' 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']) def test_delete_network(self): # Delete net: myid. resource = 'network' cmd = network.DeleteNetwork(test_cli20.MyApp(sys.stdout), None) myid = 'myid' args = [myid] self._test_delete_resource(resource, cmd, myid, args) def test_bulk_delete_network(self): # Delete net: myid1 myid2. resource = 'network' cmd = network.DeleteNetwork(test_cli20.MyApp(sys.stdout), None) myid1 = 'myid1' myid2 = 'myid2' args = [myid1, myid2] self._test_delete_resource(resource, cmd, myid1, args, extra_id=myid2) def _test_extend_list(self, mox_calls): data = [{'id': 'netid%d' % i, 'name': 'net%d' % i, 'subnets': ['mysubid%d' % i]} for i in range(10)] self.mox.StubOutWithMock(self.client.httpclient, "request") path = getattr(self.client, 'subnets_path') 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) self.mox.ReplayAll() known_args, _vs = cmd.get_parser('create_subnets').parse_known_args() cmd.extend_list(data, known_args) self.mox.VerifyAll() def _build_test_data(self, data): subnet_ids = [] response = [] filters = "" for n in data: if 'subnets' in n: subnet_ids.extend(n['subnets']) for subnet_id in n['subnets']: filters = "%s&id=%s" % (filters, subnet_id) response.append({'id': subnet_id, 'cidr': '192.168.0.0/16'}) resp_str = self.client.serialize({'subnets': response}) resp = (test_cli20.MyResp(200), resp_str) return filters, resp def test_extend_list(self): def mox_calls(path, data): filters, response = self._build_test_data(data) self.client.httpclient.request( test_cli20.MyUrlComparator(test_cli20.end_url( path, 'fields=id&fields=cidr' + filters), self.client), 'GET', body=None, headers=mox.ContainsKeyValue( 'X-Auth-Token', test_cli20.TOKEN)).AndReturn(response) self._test_extend_list(mox_calls) def test_extend_list_exceed_max_uri_len(self): def mox_calls(path, data): sub_data_lists = [data[:len(data) - 1], data[len(data) - 1:]] filters, response = self._build_test_data(data) # 1 char of extra URI len will cause a split in 2 requests self.mox.StubOutWithMock(self.client.httpclient, "_check_uri_length") self.client.httpclient._check_uri_length(mox.IgnoreArg()).AndRaise( exceptions.RequestURITooLong(excess=1)) for data in sub_data_lists: filters, response = self._build_test_data(data) self.client.httpclient._check_uri_length( mox.IgnoreArg()).AndReturn(None) self.client.httpclient.request( test_cli20.MyUrlComparator( test_cli20.end_url( path, 'fields=id&fields=cidr%s' % filters), self.client), 'GET', body=None, headers=mox.ContainsKeyValue( 'X-Auth-Token', test_cli20.TOKEN)).AndReturn(response) self._test_extend_list(mox_calls) def test_list_shared_networks(self): # list nets : --shared False cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None) self._test_list_networks(cmd, base_args='--shared False'.split(), query='shared=False')