"Fossies" - the Fresh Open Source Software Archive

Member "ec2-api-9.0.0/ec2api/tests/unit/test_subnet.py" (16 Oct 2019, 15025 Bytes) of package /linux/misc/openstack/ec2-api-9.0.0.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file.

    1 # Copyright 2014
    2 # The Cloudscaling Group, Inc.
    3 #
    4 # Licensed under the Apache License, Version 2.0 (the "License");
    5 # you may not use this file except in compliance with the License.
    6 # You may obtain a copy of the License at
    7 # http://www.apache.org/licenses/LICENSE-2.0
    8 #
    9 # Unless required by applicable law or agreed to in writing, software
   10 # distributed under the License is distributed on an "AS IS" BASIS,
   11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   12 # See the License for the specific language governing permissions and
   13 # limitations under the License.
   14 
   15 import mock
   16 from neutronclient.common import exceptions as neutron_exception
   17 
   18 from ec2api.api import common
   19 from ec2api.tests.unit import base
   20 from ec2api.tests.unit import fakes
   21 from ec2api.tests.unit import matchers
   22 from ec2api.tests.unit import tools
   23 
   24 
   25 class SubnetTestCase(base.ApiTestCase):
   26 
   27     def setUp(self):
   28         super(SubnetTestCase, self).setUp()
   29         self.vpn_gateway_api = self.mock('ec2api.api.subnet.vpn_gateway_api')
   30 
   31     def test_create_subnet(self):
   32         self.set_mock_db_items(fakes.DB_VPC_1, fakes.DB_ROUTE_TABLE_1)
   33         self.db_api.add_item.side_effect = (
   34                 tools.get_db_api_add_item(fakes.ID_EC2_SUBNET_1))
   35         self.neutron.create_network.side_effect = (
   36                 tools.get_neutron_create('network', fakes.ID_OS_NETWORK_1,
   37                                          {'status': 'available'}))
   38         self.neutron.create_subnet.side_effect = (
   39                 tools.get_neutron_create('subnet', fakes.ID_OS_SUBNET_1))
   40         subnet_1 = tools.purge_dict(fakes.DB_SUBNET_1, ('os_vpnservice_id',))
   41 
   42         def check_response(resp):
   43             self.assertThat(fakes.EC2_SUBNET_1, matchers.DictMatches(
   44                     resp['subnet']))
   45             self.db_api.add_item.assert_called_once_with(
   46                     mock.ANY, 'subnet',
   47                     tools.purge_dict(subnet_1, ('id',)))
   48             self.neutron.create_network.assert_called_once_with(
   49                     {'network': {'name': 'subnet-0'}})
   50             self.neutron.update_network.assert_called_once_with(
   51                     fakes.ID_OS_NETWORK_1,
   52                     {'network': {'name': fakes.ID_EC2_SUBNET_1}})
   53             self.neutron.create_subnet.assert_called_once_with(
   54                     {'subnet': tools.purge_dict(fakes.OS_SUBNET_1,
   55                                                 ('id', 'name', 'gateway_ip'))})
   56             self.neutron.update_subnet.assert_called_once_with(
   57                     fakes.ID_OS_SUBNET_1,
   58                     {'subnet': {'name': fakes.ID_EC2_SUBNET_1,
   59                                 'gateway_ip': None}})
   60             self.neutron.add_interface_router.assert_called_once_with(
   61                     fakes.ID_OS_ROUTER_1,
   62                     {'subnet_id': fakes.ID_OS_SUBNET_1})
   63             self.vpn_gateway_api._start_vpn_in_subnet.assert_called_once_with(
   64                 mock.ANY, self.neutron, mock.ANY, subnet_1,
   65                 fakes.DB_VPC_1, fakes.DB_ROUTE_TABLE_1)
   66             self.assertIsInstance(
   67                 self.vpn_gateway_api._start_vpn_in_subnet.call_args[0][2],
   68                 common.OnCrashCleaner)
   69 
   70         resp = self.execute('CreateSubnet', {'VpcId': fakes.ID_EC2_VPC_1,
   71                                              'CidrBlock': fakes.CIDR_SUBNET_1})
   72         check_response(resp)
   73 
   74         self.neutron.reset_mock()
   75         self.db_api.reset_mock()
   76         self.vpn_gateway_api.reset_mock()
   77 
   78         resp = self.execute('CreateSubnet', {'VpcId': fakes.ID_EC2_VPC_1,
   79                                              'CidrBlock': fakes.CIDR_SUBNET_1,
   80                                              'AvailabilityZone': 'nova'})
   81         check_response(resp)
   82 
   83     def test_create_subnet_invalid_parameters(self):
   84         def do_check(args, error_code):
   85             self.neutron.reset_mock()
   86             self.db_api.reset_mock()
   87             self.assert_execution_error(error_code, 'CreateSubnet', args)
   88             self.assertEqual(0, self.neutron.create_network.call_count)
   89             self.assertEqual(0, self.neutron.create_subnet.call_count)
   90             self.assertEqual(0, self.neutron.add_interface_router.call_count)
   91 
   92         self.set_mock_db_items()
   93         do_check({'VpcId': fakes.ID_EC2_VPC_1,
   94                   'CidrBlock': fakes.CIDR_SUBNET_1},
   95                  'InvalidVpcID.NotFound')
   96         self.db_api.get_item_by_id.assert_called_once_with(mock.ANY,
   97                                                            fakes.ID_EC2_VPC_1)
   98 
   99         self.set_mock_db_items(fakes.DB_VPC_1)
  100         do_check({'VpcId': fakes.ID_EC2_VPC_1,
  101                   'CidrBlock': 'invalid_cidr'},
  102                  'InvalidParameterValue')
  103         self.assertEqual(0, self.db_api.get_item_by_id.call_count)
  104 
  105         do_check({'VpcId': fakes.ID_EC2_VPC_1,
  106                   'CidrBlock': '10.10.0.0/30'},
  107                  'InvalidSubnet.Range')
  108         self.assertEqual(0, self.db_api.get_item_by_id.call_count)
  109 
  110         do_check({'VpcId': fakes.ID_EC2_VPC_1,
  111                   'CidrBlock': '10.20.0.0/24'},
  112                  'InvalidSubnet.Range')
  113         self.db_api.get_item_by_id.assert_called_once_with(mock.ANY,
  114                                                            fakes.ID_EC2_VPC_1)
  115 
  116     def test_create_subnet_overlapped(self):
  117         self.set_mock_db_items(fakes.DB_VPC_1, fakes.DB_ROUTE_TABLE_1)
  118         self.neutron.create_network.side_effect = (
  119                 tools.get_neutron_create('network', fakes.ID_OS_NETWORK_1,
  120                                          {'status': 'available'}))
  121         self.neutron.create_subnet.side_effect = (
  122                 tools.get_neutron_create('subnet', fakes.ID_OS_SUBNET_1))
  123         self.neutron.add_interface_router.side_effect = (
  124                 neutron_exception.BadRequest())
  125 
  126         self.assert_execution_error('InvalidSubnet.Conflict', 'CreateSubnet',
  127                                     {'VpcId': fakes.ID_EC2_VPC_1,
  128                                      'CidrBlock': fakes.CIDR_SUBNET_1})
  129 
  130     def test_create_subnet_overlimit(self):
  131         self.set_mock_db_items(fakes.DB_VPC_1, fakes.DB_ROUTE_TABLE_1)
  132         self.neutron.create_network.side_effect = (
  133                 tools.get_neutron_create('network', fakes.ID_OS_NETWORK_1,
  134                                          {'status': 'available'}))
  135         self.neutron.create_subnet.side_effect = (
  136                 tools.get_neutron_create('subnet', fakes.ID_OS_SUBNET_1))
  137 
  138         def test_overlimit(func):
  139             self.neutron.reset_mock()
  140             saved_side_effect = func.side_effect
  141             func.side_effect = neutron_exception.OverQuotaClient
  142 
  143             self.assert_execution_error('SubnetLimitExceeded', 'CreateSubnet',
  144                                         {'VpcId': fakes.ID_EC2_VPC_1,
  145                                          'CidrBlock': fakes.CIDR_SUBNET_1})
  146             func.side_effect = saved_side_effect
  147 
  148         test_overlimit(self.neutron.create_network)
  149         test_overlimit(self.neutron.create_subnet)
  150 
  151     @tools.screen_unexpected_exception_logs
  152     def test_create_subnet_rollback(self):
  153         self.set_mock_db_items(fakes.DB_VPC_1, fakes.DB_ROUTE_TABLE_1)
  154         self.db_api.add_item.side_effect = (
  155                 tools.get_db_api_add_item(fakes.ID_EC2_SUBNET_1))
  156         self.neutron.create_network.side_effect = (
  157                 tools.get_neutron_create('network', fakes.ID_OS_NETWORK_1,
  158                                          {'status': 'available'}))
  159         self.neutron.create_subnet.side_effect = (
  160                 tools.get_neutron_create('subnet', fakes.ID_OS_SUBNET_1))
  161         self.neutron.update_network.side_effect = Exception()
  162 
  163         self.assert_execution_error(self.ANY_EXECUTE_ERROR, 'CreateSubnet',
  164                                     {'VpcId': fakes.ID_EC2_VPC_1,
  165                                      'CidrBlock': fakes.CIDR_SUBNET_1})
  166 
  167         self.neutron.assert_has_calls([
  168             mock.call.remove_interface_router(
  169                 fakes.ID_OS_ROUTER_1, {'subnet_id': fakes.ID_OS_SUBNET_1}),
  170             mock.call.delete_subnet(fakes.ID_OS_SUBNET_1),
  171             mock.call.delete_network(fakes.ID_OS_NETWORK_1)])
  172         self.db_api.delete_item.assert_called_once_with(
  173                 mock.ANY, fakes.ID_EC2_SUBNET_1)
  174 
  175     def test_delete_subnet(self):
  176         self.set_mock_db_items(fakes.DB_VPC_1, fakes.DB_SUBNET_1)
  177         self.neutron.show_subnet.return_value = (
  178                 {'subnet': fakes.OS_SUBNET_1})
  179 
  180         resp = self.execute('DeleteSubnet',
  181                             {'SubnetId': fakes.ID_EC2_SUBNET_1})
  182 
  183         self.assertEqual(True, resp['return'])
  184         self.db_api.delete_item.assert_called_once_with(
  185                 mock.ANY,
  186                 fakes.ID_EC2_SUBNET_1)
  187         self.neutron.remove_interface_router.assert_called_once_with(
  188                 fakes.ID_OS_ROUTER_1,
  189                 {'subnet_id': fakes.ID_OS_SUBNET_1})
  190         self.neutron.delete_network.assert_called_once_with(
  191                 fakes.ID_OS_NETWORK_1)
  192         self.assertTrue(
  193             self.neutron.mock_calls.index(
  194                 mock.call.delete_network(fakes.ID_OS_NETWORK_1)) >
  195             self.neutron.mock_calls.index(
  196                 mock.call.remove_interface_router(
  197                     fakes.ID_OS_ROUTER_1,
  198                     {'subnet_id': fakes.ID_OS_SUBNET_1})))
  199         self.vpn_gateway_api._stop_vpn_in_subnet.assert_called_once_with(
  200             mock.ANY, self.neutron, mock.ANY, fakes.DB_SUBNET_1)
  201         self.assertIsInstance(
  202             self.vpn_gateway_api._stop_vpn_in_subnet.call_args[0][2],
  203             common.OnCrashCleaner)
  204 
  205     def test_delete_subnet_inconsistent_os(self):
  206         self.set_mock_db_items(fakes.DB_VPC_1, fakes.DB_SUBNET_1)
  207         self.neutron.remove_interface_router.side_effect = (
  208                 neutron_exception.NotFound())
  209         self.neutron.show_subnet.return_value = (
  210                 {'subnet': fakes.OS_SUBNET_1})
  211         self.neutron.delete_network.side_effect = (
  212                 neutron_exception.NetworkInUseClient())
  213 
  214         resp = self.execute('DeleteSubnet',
  215                             {'SubnetId': fakes.ID_EC2_SUBNET_1})
  216         self.assertEqual(True, resp['return'])
  217 
  218         self.neutron.show_subnet.side_effect = neutron_exception.NotFound()
  219 
  220         resp = self.execute('DeleteSubnet',
  221                             {'SubnetId': fakes.ID_EC2_SUBNET_1})
  222         self.assertEqual(True, resp['return'])
  223 
  224     def test_delete_subnet_invalid_parameters(self):
  225         self.set_mock_db_items()
  226         self.neutron.show_subnet.return_value = fakes.OS_SUBNET_1
  227         self.neutron.show_network.return_value = fakes.OS_NETWORK_1
  228 
  229         self.assert_execution_error('InvalidSubnetID.NotFound', 'DeleteSubnet',
  230                                     {'SubnetId': fakes.ID_EC2_SUBNET_1})
  231         self.assertEqual(0, self.neutron.delete_network.call_count)
  232         self.assertEqual(0, self.neutron.delete_subnet.call_count)
  233         self.assertEqual(0, self.neutron.remove_interface_router.call_count)
  234 
  235     @mock.patch('ec2api.api.network_interface.describe_network_interfaces')
  236     def test_delete_subnet_not_empty(self, describe_network_interfaces):
  237         self.set_mock_db_items(fakes.DB_VPC_1, fakes.DB_SUBNET_1)
  238         describe_network_interfaces.return_value = (
  239                 {'networkInterfaceSet': [fakes.EC2_NETWORK_INTERFACE_1]})
  240         self.assert_execution_error('DependencyViolation', 'DeleteSubnet',
  241                                     {'SubnetId': fakes.ID_EC2_SUBNET_1})
  242 
  243     @tools.screen_unexpected_exception_logs
  244     def test_delete_subnet_rollback(self):
  245         self.set_mock_db_items(fakes.DB_VPC_1, fakes.DB_SUBNET_1)
  246         self.neutron.show_subnet.side_effect = Exception()
  247 
  248         self.assert_execution_error(self.ANY_EXECUTE_ERROR, 'DeleteSubnet',
  249                                     {'SubnetId': fakes.ID_EC2_SUBNET_1})
  250 
  251         self.db_api.restore_item.assert_called_once_with(
  252                 mock.ANY, 'subnet', fakes.DB_SUBNET_1)
  253         self.neutron.add_interface_router.assert_called_once_with(
  254                 fakes.ID_OS_ROUTER_1, {'subnet_id': fakes.ID_OS_SUBNET_1})
  255 
  256     def test_describe_subnets(self):
  257         self.set_mock_db_items(fakes.DB_SUBNET_1, fakes.DB_SUBNET_2)
  258         self.neutron.list_subnets.return_value = (
  259                 {'subnets': [fakes.OS_SUBNET_1, fakes.OS_SUBNET_2]})
  260         self.neutron.list_networks.return_value = (
  261                 {'networks': [fakes.OS_NETWORK_1, fakes.OS_NETWORK_2]})
  262 
  263         resp = self.execute('DescribeSubnets', {})
  264         self.assertThat(resp['subnetSet'],
  265                         matchers.ListMatches([fakes.EC2_SUBNET_1,
  266                                               fakes.EC2_SUBNET_2]))
  267 
  268         self.db_api.get_items_by_ids = tools.CopyingMock(
  269             return_value=[fakes.DB_SUBNET_2])
  270         resp = self.execute('DescribeSubnets',
  271                             {'SubnetId.1': fakes.ID_EC2_SUBNET_2})
  272         self.assertThat(resp['subnetSet'],
  273                         matchers.ListMatches([fakes.EC2_SUBNET_2]))
  274         self.db_api.get_items_by_ids.assert_called_once_with(
  275             mock.ANY, set([fakes.ID_EC2_SUBNET_2]))
  276 
  277         self.check_filtering(
  278             'DescribeSubnets', 'subnetSet',
  279             [
  280              # TODO(ft): declare a constant for the count in fakes
  281              ('available-ip-address-count', 253),
  282              ('cidr', fakes.CIDR_SUBNET_2),
  283              ('cidrBlock', fakes.CIDR_SUBNET_2),
  284              ('cidr-block', fakes.CIDR_SUBNET_2),
  285              ('subnet-id', fakes.ID_EC2_SUBNET_2),
  286              ('state', 'available'),
  287              ('vpc-id', fakes.ID_EC2_VPC_1)])
  288         self.check_tag_support(
  289             'DescribeSubnets', 'subnetSet',
  290             fakes.ID_EC2_SUBNET_2, 'subnetId')
  291 
  292     def test_describe_subnets_not_consistent_os_subnet(self):
  293         self.set_mock_db_items(fakes.DB_SUBNET_1, fakes.DB_SUBNET_2)
  294         self.neutron.list_subnets.return_value = (
  295                 {'subnets': [fakes.OS_SUBNET_2]})
  296         self.neutron.list_networks.return_value = (
  297                 {'networks': [fakes.OS_NETWORK_1]})
  298 
  299         resp = self.execute('DescribeSubnets', {})
  300         self.assertEqual([], resp['subnetSet'])
  301 
  302     @mock.patch('ec2api.api.ec2utils.check_and_create_default_vpc')
  303     def test_describe_subnets_no_default_vpc(self, check_and_create):
  304         self.configure(disable_ec2_classic=True)
  305 
  306         def mock_check_and_create(context):
  307             self.set_mock_db_items(fakes.DB_VPC_DEFAULT,
  308                                    fakes.DB_SUBNET_DEFAULT)
  309             self.neutron.list_subnets.return_value = (
  310                 {'subnets': [fakes.OS_SUBNET_DEFAULT]})
  311             self.neutron.list_networks.return_value = (
  312                 {'networks': [fakes.OS_NETWORK_DEFAULT]})
  313         check_and_create.side_effect = mock_check_and_create
  314 
  315         resp = self.execute('DescribeSubnets', {})
  316         self.assertEqual(resp['subnetSet'], [fakes.EC2_SUBNET_DEFAULT])
  317 
  318         check_and_create.assert_called_once_with(mock.ANY)