"Fossies" - the Fresh Open Source Software Archive

Member "ec2-api-12.0.0/ec2api/tests/unit/test_address.py" (14 Apr 2021, 30821 Bytes) of package /linux/misc/openstack/ec2-api-12.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. See also the last Fossies "Diffs" side-by-side code changes report for "test_address.py": 9.0.0_vs_10.0.0.

    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 from neutronclient.common import exceptions as neutron_exception
   16 from unittest import mock
   17 
   18 from ec2api.tests.unit import base
   19 from ec2api.tests.unit import fakes
   20 from ec2api.tests.unit import matchers
   21 from ec2api.tests.unit import tools
   22 
   23 
   24 class AddressTestCase(base.ApiTestCase):
   25 
   26     def setUp(self):
   27         super(AddressTestCase, self).setUp()
   28 
   29     def test_allocate_ec2_classic_address(self):
   30         self.configure(external_network=fakes.NAME_OS_PUBLIC_NETWORK)
   31         self.neutron.list_networks.return_value = (
   32             {'networks': [{'id': fakes.ID_OS_PUBLIC_NETWORK}]})
   33         self.neutron.create_floatingip.return_value = (
   34             {'floatingip': fakes.OS_FLOATING_IP_1})
   35 
   36         resp = self.execute('AllocateAddress', {})
   37         self.assertEqual(fakes.IP_ADDRESS_1, resp['publicIp'])
   38         self.assertEqual('standard', resp['domain'])
   39         self.assertNotIn('allocationId', resp)
   40         self.assertEqual(0, self.db_api.add_item.call_count)
   41         self.neutron.create_floatingip.assert_called_once_with(
   42             {'floatingip': {
   43                 'floating_network_id':
   44                 fakes.ID_OS_PUBLIC_NETWORK}})
   45 
   46     def test_allocate_vpc_address(self):
   47         self.configure(external_network=fakes.NAME_OS_PUBLIC_NETWORK)
   48         self.neutron.list_networks.return_value = (
   49             {'networks': [{'id': fakes.ID_OS_PUBLIC_NETWORK}]})
   50         self.neutron.create_floatingip.return_value = (
   51             {'floatingip': fakes.OS_FLOATING_IP_1})
   52         self.db_api.add_item.return_value = fakes.DB_ADDRESS_1
   53 
   54         resp = self.execute('AllocateAddress', {'Domain': 'vpc'})
   55 
   56         self.assertEqual(fakes.IP_ADDRESS_1, resp['publicIp'])
   57         self.assertEqual('vpc', resp['domain'])
   58         self.assertEqual(fakes.ID_EC2_ADDRESS_1,
   59                          resp['allocationId'])
   60         self.db_api.add_item.assert_called_once_with(
   61             mock.ANY, 'eipalloc',
   62             tools.purge_dict(fakes.DB_ADDRESS_1,
   63                              ('id', 'vpc_id')))
   64         self.neutron.create_floatingip.assert_called_once_with(
   65             {'floatingip': {
   66                 'floating_network_id':
   67                 fakes.ID_OS_PUBLIC_NETWORK}})
   68         self.neutron.list_networks.assert_called_once_with(
   69             **{'router:external': True,
   70                'name': fakes.NAME_OS_PUBLIC_NETWORK})
   71         self.db_api.reset_mock()
   72         self.neutron.create_floatingip.reset_mock()
   73         self.neutron.list_networks.reset_mock()
   74 
   75         self.configure(disable_ec2_classic=True)
   76         resp = self.execute('AllocateAddress', {})
   77 
   78         self.assertEqual(fakes.IP_ADDRESS_1, resp['publicIp'])
   79         self.assertEqual('vpc', resp['domain'])
   80         self.assertEqual(fakes.ID_EC2_ADDRESS_1,
   81                          resp['allocationId'])
   82         self.db_api.add_item.assert_called_once_with(
   83             mock.ANY, 'eipalloc',
   84             tools.purge_dict(fakes.DB_ADDRESS_1,
   85                              ('id', 'vpc_id')))
   86         self.neutron.create_floatingip.assert_called_once_with(
   87             {'floatingip': {
   88                 'floating_network_id':
   89                 fakes.ID_OS_PUBLIC_NETWORK}})
   90         self.neutron.list_networks.assert_called_once_with(
   91             **{'router:external': True,
   92                'name': fakes.NAME_OS_PUBLIC_NETWORK})
   93 
   94     def test_allocate_address_invalid_parameters(self):
   95         self.assert_execution_error('InvalidParameterValue', 'AllocateAddress',
   96                                     {'Domain': 'fake_domain'})
   97         self.assertEqual(0, self.db_api.add_item.call_count)
   98         self.assertEqual(0, self.neutron.create_floatingip.call_count)
   99 
  100     def test_allocate_address_overlimit(self):
  101         self.configure(external_network=fakes.NAME_OS_PUBLIC_NETWORK)
  102         self.neutron.list_networks.return_value = (
  103             {'networks': [{'id': fakes.ID_OS_PUBLIC_NETWORK}]})
  104         self.neutron.create_floatingip.side_effect = (
  105             neutron_exception.OverQuotaClient())
  106         self.assert_execution_error('AddressLimitExceeded', 'AllocateAddress',
  107                                     {'Domain': 'vpc'})
  108         self.assert_execution_error('AddressLimitExceeded', 'AllocateAddress',
  109                                     {})
  110 
  111     @tools.screen_unexpected_exception_logs
  112     def test_allocate_address_vpc_rollback(self):
  113         self.configure(external_network=fakes.NAME_OS_PUBLIC_NETWORK)
  114         self.neutron.list_networks.return_value = (
  115             {'networks': [{'id': fakes.ID_OS_PUBLIC_NETWORK}]})
  116         self.neutron.create_floatingip.return_value = (
  117             {'floatingip': fakes.OS_FLOATING_IP_1})
  118         self.db_api.add_item.side_effect = Exception()
  119 
  120         self.assert_execution_error(self.ANY_EXECUTE_ERROR, 'AllocateAddress',
  121                                     {'Domain': 'vpc'})
  122 
  123         self.neutron.delete_floatingip.assert_called_once_with(
  124             fakes.ID_OS_FLOATING_IP_1)
  125 
  126     # TODO(andrey-mp): api code has to be fixed
  127     # There is no add-floating-ip and remove-floating-ip command in
  128     # python-novaclient. Those command have been removed since 7.0.0
  129     # version (ocata) and ec2-api has version >9.1.0 since long.
  130     @base.skip_not_implemented
  131     def test_associate_address_ec2_classic(self):
  132         self.set_mock_db_items(fakes.DB_INSTANCE_1)
  133         self.neutron.list_floatingips.return_value = (
  134             {'floatingips': [fakes.OS_FLOATING_IP_1,
  135                              fakes.OS_FLOATING_IP_2]})
  136         self.nova.servers.add_floating_ip.return_value = True
  137 
  138         resp = self.execute('AssociateAddress',
  139                             {'PublicIp': fakes.IP_ADDRESS_1,
  140                              'InstanceId': fakes.ID_EC2_INSTANCE_1})
  141         self.assertEqual(True, resp['return'])
  142 
  143         self.nova.servers.add_floating_ip.assert_called_once_with(
  144             fakes.ID_OS_INSTANCE_1,
  145             fakes.IP_ADDRESS_1)
  146 
  147     def test_associate_address_vpc(self):
  148 
  149         def do_check(params, fixed_ip):
  150             resp = self.execute('AssociateAddress', params)
  151             self.assertEqual(True, resp['return'])
  152             self.assertEqual(fakes.ID_EC2_ASSOCIATION_1, resp['associationId'])
  153 
  154             self.neutron.update_floatingip.assert_called_once_with(
  155                 fakes.ID_OS_FLOATING_IP_1,
  156                 {'floatingip': {'port_id': fakes.ID_OS_PORT_2,
  157                                 'fixed_ip_address': fixed_ip}})
  158             self.db_api.update_item.assert_called_once_with(
  159                 mock.ANY,
  160                 tools.update_dict(
  161                     fakes.DB_ADDRESS_1,
  162                     {'network_interface_id':
  163                      fakes.ID_EC2_NETWORK_INTERFACE_2,
  164                      'private_ip_address': fixed_ip}))
  165 
  166             self.neutron.update_floatingip.reset_mock()
  167             self.db_api.update_item.reset_mock()
  168 
  169         self.set_mock_db_items(
  170             fakes.DB_ADDRESS_1, fakes.DB_IGW_1, fakes.DB_IGW_2,
  171             fakes.DB_NETWORK_INTERFACE_1, fakes.DB_NETWORK_INTERFACE_2)
  172         self.neutron.show_floatingip.return_value = (
  173             {'floatingip': fakes.OS_FLOATING_IP_1})
  174         do_check({'AllocationId': fakes.ID_EC2_ADDRESS_1,
  175                   'InstanceId': fakes.ID_EC2_INSTANCE_1},
  176                  fakes.IP_NETWORK_INTERFACE_2)
  177 
  178         do_check({'AllocationId': fakes.ID_EC2_ADDRESS_1,
  179                   'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_2},
  180                  fakes.IP_NETWORK_INTERFACE_2)
  181 
  182         do_check({'AllocationId': fakes.ID_EC2_ADDRESS_1,
  183                   'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_2,
  184                   'PrivateIpAddress': fakes.IP_NETWORK_INTERFACE_2_EXT_1},
  185                  fakes.IP_NETWORK_INTERFACE_2_EXT_1)
  186 
  187         assigned_db_address_1 = tools.update_dict(
  188             fakes.DB_ADDRESS_1,
  189             {'network_interface_id': fakes.ID_EC2_NETWORK_INTERFACE_1,
  190              'private_ip_address': fakes.IP_NETWORK_INTERFACE_1})
  191         self.add_mock_db_items(assigned_db_address_1)
  192         assigned_floating_ip_1 = tools.update_dict(
  193             fakes.OS_FLOATING_IP_1,
  194             {'fixed_port_id': fakes.ID_OS_PORT_1,
  195              'fixed_ip_address': fakes.IP_NETWORK_INTERFACE_1})
  196         self.neutron.show_floatingip.return_value = (
  197             {'floatingip': assigned_floating_ip_1})
  198         do_check({'AllocationId': fakes.ID_EC2_ADDRESS_1,
  199                   'InstanceId': fakes.ID_EC2_INSTANCE_1,
  200                   'AllowReassociation': 'True'},
  201                  fakes.IP_NETWORK_INTERFACE_2)
  202 
  203         self.configure(disable_ec2_classic=True)
  204         self.set_mock_db_items(
  205             fakes.DB_VPC_DEFAULT, fakes.DB_ADDRESS_1, fakes.DB_IGW_1,
  206             fakes.DB_NETWORK_INTERFACE_2)
  207         do_check({'PublicIp': fakes.IP_ADDRESS_1,
  208                   'InstanceId': fakes.ID_EC2_INSTANCE_1},
  209                  fakes.IP_NETWORK_INTERFACE_2)
  210 
  211     def test_associate_address_vpc_idempotent(self):
  212 
  213         def do_check(params):
  214             resp = self.execute('AssociateAddress', params)
  215             self.assertEqual(True, resp['return'])
  216             self.assertEqual(fakes.ID_EC2_ASSOCIATION_2, resp['associationId'])
  217 
  218         self.set_mock_db_items(fakes.DB_ADDRESS_2,
  219                                fakes.DB_NETWORK_INTERFACE_1,
  220                                fakes.DB_NETWORK_INTERFACE_2)
  221         self.neutron.show_floatingip.return_value = (
  222             {'floatingip': fakes.OS_FLOATING_IP_2})
  223 
  224         do_check({'AllocationId': fakes.ID_EC2_ADDRESS_2,
  225                   'InstanceId': fakes.ID_EC2_INSTANCE_1})
  226 
  227         do_check({'AllocationId': fakes.ID_EC2_ADDRESS_2,
  228                   'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_2})
  229 
  230         do_check({'AllocationId': fakes.ID_EC2_ADDRESS_2,
  231                   'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_2,
  232                   'PrivateIpAddress': fakes.IP_NETWORK_INTERFACE_2})
  233 
  234     def test_associate_address_invalid_main_parameters(self):
  235 
  236         def do_check(params, error):
  237             self.assert_execution_error(error, 'AssociateAddress', params)
  238 
  239         do_check({},
  240                  'MissingParameter')
  241 
  242         do_check({'PublicIp': '0.0.0.0',
  243                   'AllocationId': 'eipalloc-0'},
  244                  'InvalidParameterCombination')
  245 
  246         do_check({'PublicIp': '0.0.0.0'},
  247                  'MissingParameter')
  248 
  249         do_check({'AllocationId': 'eipalloc-0'},
  250                  'MissingParameter')
  251 
  252     def test_associate_address_invalid_ec2_classic_parameters(self):
  253         # NOTE(ft): ec2 classic instance vs allocation_id parameter
  254         self.set_mock_db_items(fakes.DB_INSTANCE_2)
  255         self.assert_execution_error('InvalidParameterCombination',
  256                                     'AssociateAddress',
  257                                     {'AllocationId': 'eipalloc-0',
  258                                      'InstanceId': fakes.ID_EC2_INSTANCE_2})
  259 
  260         # NOTE(ft): ec2 classic instance vs not existing public IP
  261         self.neutron.list_floatingips.return_value = {'floatingips': []}
  262         self.assert_execution_error('AuthFailure', 'AssociateAddress',
  263                                     {'PublicIp': fakes.IP_ADDRESS_1,
  264                                      'InstanceId': fakes.ID_EC2_INSTANCE_2})
  265 
  266         # NOTE(ft): ec2 classic instance vs vpc public ip
  267         self.add_mock_db_items(fakes.DB_ADDRESS_1, fakes.DB_ADDRESS_2)
  268         self.neutron.show_floatingip.return_value = (
  269             {'floatingip': fakes.OS_FLOATING_IP_1})
  270         self.assert_execution_error('AuthFailure', 'AssociateAddress',
  271                                     {'PublicIp': fakes.IP_ADDRESS_1,
  272                                      'InstanceId': fakes.ID_EC2_INSTANCE_2})
  273 
  274     def test_associate_address_invalid_vpc_parameters(self):
  275 
  276         def do_check(params, error):
  277             self.assert_execution_error(error, 'AssociateAddress', params)
  278 
  279         # NOTE(ft): not registered instance id vs vpc address
  280         self.set_mock_db_items()
  281         do_check({'AllocationId': fakes.ID_EC2_ADDRESS_1,
  282                   'InstanceId': fakes.ID_EC2_INSTANCE_1},
  283                  'InvalidInstanceID.NotFound')
  284 
  285         # NOTE(ft): vpc instance vs public ip parmeter
  286         self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_2)
  287         do_check({'PublicIp': '0.0.0.0',
  288                   'InstanceId': fakes.ID_EC2_INSTANCE_1},
  289                  'InvalidParameterCombination')
  290 
  291         # NOTE(ft): vpc instance vs not registered vpc address
  292         do_check({'AllocationId': fakes.ID_EC2_ADDRESS_1,
  293                   'InstanceId': fakes.ID_EC2_INSTANCE_1},
  294                  'InvalidAllocationID.NotFound')
  295 
  296         # NOTE(ft): not registered network interface id vs vpc address
  297         do_check({'AllocationId': fakes.ID_EC2_ADDRESS_1,
  298                   'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1},
  299                  'InvalidNetworkInterfaceID.NotFound')
  300 
  301         # NOTE(ft): vpc instance vs broken vpc address
  302         self.set_mock_db_items(fakes.DB_ADDRESS_1,
  303                                fakes.DB_NETWORK_INTERFACE_2)
  304         self.neutron.show_floatingip.side_effect = neutron_exception.NotFound
  305         do_check({'AllocationId': fakes.ID_EC2_ADDRESS_1,
  306                   'InstanceId': fakes.ID_EC2_INSTANCE_1},
  307                  'InvalidAllocationID.NotFound')
  308         self.neutron.show_floatingip.side_effect = None
  309 
  310         # NOTE(ft): already associated address vs network interface
  311         self.set_mock_db_items(fakes.DB_ADDRESS_1, fakes.DB_ADDRESS_2,
  312                                fakes.DB_NETWORK_INTERFACE_1)
  313         self.neutron.show_floatingip.return_value = (
  314             {'floatingip': fakes.OS_FLOATING_IP_2})
  315         self.assert_execution_error(
  316             'Resource.AlreadyAssociated', 'AssociateAddress',
  317             {'AllocationId': fakes.ID_EC2_ADDRESS_2,
  318              'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1})
  319 
  320         # NOTE(ft): already associated address vs vpc instance
  321         self.set_mock_db_items(
  322             fakes.DB_ADDRESS_2,
  323             fakes.gen_db_network_interface(
  324                 fakes.ID_EC2_NETWORK_INTERFACE_1,
  325                 fakes.ID_OS_PORT_1,
  326                 fakes.ID_EC2_VPC_1,
  327                 fakes.ID_EC2_SUBNET_1,
  328                 fakes.IP_NETWORK_INTERFACE_1,
  329                 instance_id=fakes.ID_EC2_INSTANCE_1))
  330         self.assert_execution_error('Resource.AlreadyAssociated',
  331                                     'AssociateAddress',
  332                                     {'AllocationId': fakes.ID_EC2_ADDRESS_2,
  333                                      'InstanceId': fakes.ID_EC2_INSTANCE_1})
  334 
  335         # NOTE(ft): multiple network interfaces in vpc instance
  336         # w/o network interface selection
  337         self.add_mock_db_items(fakes.DB_NETWORK_INTERFACE_2)
  338         self.assert_execution_error('InvalidInstanceID', 'AssociateAddress',
  339                                     {'AllocationId': fakes.ID_EC2_ADDRESS_1,
  340                                      'InstanceId': fakes.ID_EC2_INSTANCE_1})
  341 
  342         # NOTE(ft): internet gateway isn't attached to the vpc
  343         self.set_mock_db_items(fakes.DB_ADDRESS_1,
  344                                fakes.DB_NETWORK_INTERFACE_2)
  345         self.assert_execution_error('Gateway.NotAttached', 'AssociateAddress',
  346                                     {'AllocationId': fakes.ID_EC2_ADDRESS_1,
  347                                      'InstanceId': fakes.ID_EC2_INSTANCE_1})
  348 
  349         # NOTE(tikitavi): associate to wrong public ip
  350         self.configure(disable_ec2_classic=True)
  351         self.set_mock_db_items(
  352             fakes.DB_VPC_DEFAULT, fakes.DB_IGW_DEFAULT, fakes.DB_ADDRESS_1,
  353             fakes.DB_INSTANCE_DEFAULT, tools.update_dict(
  354                 fakes.DB_NETWORK_INTERFACE_DEFAULT,
  355                 {'instance_id': fakes.ID_EC2_INSTANCE_DEFAULT}))
  356         do_check({'PublicIp': '0.0.0.0',
  357                   'InstanceId': fakes.ID_EC2_INSTANCE_DEFAULT},
  358                  'AuthFailure')
  359 
  360     @tools.screen_unexpected_exception_logs
  361     def test_associate_address_vpc_rollback(self):
  362         self.set_mock_db_items(fakes.DB_ADDRESS_1, fakes.DB_IGW_1,
  363                                fakes.DB_NETWORK_INTERFACE_1,
  364                                fakes.DB_NETWORK_INTERFACE_2)
  365         self.neutron.show_floatingip.return_value = (
  366             {'floatingip': fakes.OS_FLOATING_IP_1})
  367         self.neutron.update_floatingip.side_effect = Exception()
  368 
  369         self.assert_execution_error(self.ANY_EXECUTE_ERROR, 'AssociateAddress',
  370                                     {'AllocationId': fakes.ID_EC2_ADDRESS_1,
  371                                      'InstanceId': fakes.ID_EC2_INSTANCE_1})
  372 
  373         self.db_api.update_item.assert_any_call(
  374             mock.ANY, fakes.DB_ADDRESS_1)
  375 
  376     # TODO(andrey-mp): api code has to be fixed
  377     # There is no add-floating-ip and remove-floating-ip command in
  378     # python-novaclient. Those command have been removed since 7.0.0
  379     # version (ocata) and ec2-api has version >9.1.0 since long.
  380     @base.skip_not_implemented
  381     def test_dissassociate_address_ec2_classic(self):
  382         self.set_mock_db_items(fakes.DB_INSTANCE_1)
  383         self.nova.servers.remove_floating_ip.return_value = True
  384         self.neutron.list_floatingips.return_value = (
  385             {'floatingips': [fakes.OS_FLOATING_IP_1,
  386                              fakes.OS_FLOATING_IP_2]})
  387         self.neutron.list_ports.return_value = (
  388             {'ports': [fakes.OS_PORT_1,
  389                        fakes.OS_PORT_2]})
  390 
  391         resp = self.execute('DisassociateAddress',
  392                             {'PublicIp': fakes.IP_ADDRESS_2})
  393         self.assertEqual(True, resp['return'])
  394         self.nova.servers.remove_floating_ip.assert_called_once_with(
  395             fakes.ID_OS_INSTANCE_1,
  396             fakes.IP_ADDRESS_2)
  397 
  398         # NOTE(Alex) Disassociate unassociated address in EC2 classic
  399         resp = self.execute('DisassociateAddress',
  400                             {'PublicIp': fakes.IP_ADDRESS_1})
  401         self.assertEqual(True, resp['return'])
  402         self.assertEqual(1, self.nova.servers.remove_floating_ip.call_count)
  403 
  404     def test_dissassociate_address_vpc(self):
  405         self.set_mock_db_items(fakes.DB_ADDRESS_2)
  406         self.neutron.show_floatingip.return_value = (
  407             {'floatingip': fakes.OS_FLOATING_IP_2})
  408 
  409         resp = self.execute('DisassociateAddress',
  410                             {'AssociationId': fakes.ID_EC2_ASSOCIATION_2})
  411         self.assertEqual(True, resp['return'])
  412 
  413         self.neutron.update_floatingip.assert_called_once_with(
  414             fakes.ID_OS_FLOATING_IP_2,
  415             {'floatingip': {'port_id': None}})
  416         self.db_api.update_item.assert_called_once_with(
  417             mock.ANY,
  418             tools.purge_dict(fakes.DB_ADDRESS_2, ['network_interface_id',
  419                                                   'private_ip_address']))
  420         self.neutron.update_floatingip.reset_mock()
  421         self.db_api.update_item.reset_mock()
  422 
  423         self.configure(disable_ec2_classic=True)
  424 
  425         resp = self.execute('DisassociateAddress',
  426                             {'PublicIp': fakes.IP_ADDRESS_2})
  427         self.assertEqual(True, resp['return'])
  428 
  429         self.neutron.update_floatingip.assert_called_once_with(
  430             fakes.ID_OS_FLOATING_IP_2,
  431             {'floatingip': {'port_id': None}})
  432         self.db_api.update_item.assert_called_once_with(
  433             mock.ANY,
  434             tools.purge_dict(fakes.DB_ADDRESS_2, ['network_interface_id',
  435                                                   'private_ip_address']))
  436 
  437     def test_dissassociate_address_vpc_idempotent(self):
  438         self.set_mock_db_items(fakes.DB_ADDRESS_1)
  439         self.neutron.show_floatingip.return_value = (
  440             {'floatingip': fakes.OS_FLOATING_IP_1})
  441 
  442         resp = self.execute('DisassociateAddress',
  443                             {'AssociationId': fakes.ID_EC2_ASSOCIATION_1})
  444         self.assertEqual(True, resp['return'])
  445 
  446         self.assertEqual(0, self.neutron.update_floatingip.call_count)
  447         self.assertEqual(0, self.db_api.update_item.call_count)
  448 
  449     def test_disassociate_address_invalid_parameters(self):
  450 
  451         def do_check(params, error):
  452             self.assert_execution_error(error, 'DisassociateAddress', params)
  453 
  454         do_check({},
  455                  'MissingParameter')
  456 
  457         do_check({'PublicIp': '0.0.0.0',
  458                   'AssociationId': 'eipassoc-0'},
  459                  'InvalidParameterCombination')
  460 
  461         # NOTE(ft): EC2 Classic public IP does not exists
  462         self.set_mock_db_items()
  463         self.neutron.list_floatingips.return_value = {'floatingips': []}
  464 
  465         self.assert_execution_error('AuthFailure', 'DisassociateAddress',
  466                                     {'PublicIp': fakes.IP_ADDRESS_2})
  467 
  468         # NOTE(ft): vpc address vs public ip parameter
  469         self.set_mock_db_items(fakes.DB_ADDRESS_1)
  470         self.neutron.show_floatingip.return_value = (
  471             {'floatingip': fakes.OS_FLOATING_IP_1})
  472         do_check({'PublicIp': fakes.IP_ADDRESS_1},
  473                  'InvalidParameterValue')
  474 
  475         # NOTE(ft): not registered address
  476         self.set_mock_db_items()
  477         do_check({'AssociationId': fakes.ID_EC2_ASSOCIATION_1},
  478                  'InvalidAssociationID.NotFound')
  479 
  480         # NOTE(ft): registered broken vpc address
  481         self.set_mock_db_items(fakes.DB_ADDRESS_2)
  482         self.neutron.show_floatingip.side_effect = neutron_exception.NotFound
  483         do_check({'AssociationId': fakes.ID_EC2_ASSOCIATION_2},
  484                  'InvalidAssociationID.NotFound')
  485 
  486         # NOTE(tikitavi): disassociate to wrong public ip
  487         self.configure(disable_ec2_classic=True)
  488         self.set_mock_db_items()
  489         self.assert_execution_error('AuthFailure', 'DisassociateAddress',
  490                                     {'PublicIp': fakes.IP_ADDRESS_2})
  491 
  492         # NOTE(tikitavi): disassociate to unassociated ip
  493         self.set_mock_db_items(fakes.DB_ADDRESS_1)
  494         self.assert_execution_error('InvalidParameterValue',
  495                                     'DisassociateAddress',
  496                                     {'PublicIp': fakes.IP_ADDRESS_1})
  497 
  498     @tools.screen_unexpected_exception_logs
  499     def test_dissassociate_address_vpc_rollback(self):
  500         self.set_mock_db_items(fakes.DB_ADDRESS_2)
  501         self.neutron.show_floatingip.return_value = (
  502             {'floatingip': fakes.OS_FLOATING_IP_2})
  503         self.neutron.update_floatingip.side_effect = Exception()
  504 
  505         self.assert_execution_error(
  506             self.ANY_EXECUTE_ERROR, 'DisassociateAddress',
  507             {'AssociationId': fakes.ID_EC2_ASSOCIATION_2})
  508 
  509         self.db_api.update_item.assert_any_call(
  510             mock.ANY, fakes.DB_ADDRESS_2)
  511 
  512     def test_release_address_ec2_classic(self):
  513         self.set_mock_db_items()
  514         self.neutron.delete_floatingip.return_value = True
  515         self.neutron.list_floatingips.return_value = (
  516             {'floatingips': [fakes.OS_FLOATING_IP_1,
  517                              fakes.OS_FLOATING_IP_2]})
  518 
  519         resp = self.execute('ReleaseAddress',
  520                             {'PublicIp': fakes.IP_ADDRESS_1})
  521         self.assertEqual(True, resp['return'])
  522 
  523         self.neutron.delete_floatingip.assert_called_once_with(
  524             fakes.OS_FLOATING_IP_1['id'])
  525 
  526     def test_release_address_vpc(self):
  527         self.set_mock_db_items(fakes.DB_ADDRESS_1)
  528         self.neutron.show_floatingip.return_value = (
  529             {'floatingip': fakes.OS_FLOATING_IP_1})
  530 
  531         resp = self.execute('ReleaseAddress',
  532                             {'AllocationId': fakes.ID_EC2_ADDRESS_1})
  533         self.assertEqual(True, resp['return'])
  534 
  535         self.neutron.delete_floatingip.assert_called_once_with(
  536             fakes.ID_OS_FLOATING_IP_1)
  537         self.db_api.delete_item.assert_called_once_with(
  538             mock.ANY, fakes.ID_EC2_ADDRESS_1)
  539 
  540     @mock.patch('ec2api.api.address.AddressEngineNeutron.disassociate_address')
  541     def test_release_address_default_vpc(self, disassociate_address):
  542         self.configure(disable_ec2_classic=True)
  543         self.set_mock_db_items(fakes.DB_VPC_DEFAULT,
  544                                fakes.DB_ADDRESS_DEFAULT,
  545                                fakes.DB_NETWORK_INTERFACE_DEFAULT)
  546         self.neutron.show_floatingip.return_value = (
  547             {'floatingip': fakes.OS_FLOATING_IP_2})
  548 
  549         resp = self.execute('ReleaseAddress',
  550                             {'AllocationId': fakes.ID_EC2_ADDRESS_DEFAULT})
  551         self.assertEqual(True, resp['return'])
  552 
  553         disassociate_address.assert_called_once_with(
  554             mock.ANY, association_id=fakes.ID_EC2_ASSOCIATION_DEFAULT)
  555         self.neutron.delete_floatingip.assert_called_once_with(
  556             fakes.ID_OS_FLOATING_IP_2)
  557         self.db_api.delete_item.assert_called_once_with(
  558             mock.ANY, fakes.ID_EC2_ADDRESS_DEFAULT)
  559 
  560     def test_release_address_invalid_parameters(self):
  561 
  562         def do_check(params, error):
  563             self.assert_execution_error(error, 'ReleaseAddress', params)
  564 
  565         do_check({},
  566                  'MissingParameter')
  567 
  568         do_check({'PublicIp': '0.0.0.0',
  569                   'AllocationId': 'eipalloc-0'},
  570                  'InvalidParameterCombination')
  571 
  572         # NOTE(ft): EC2 Classic public IP is not found
  573         self.neutron.list_floatingips.return_value = {'floatingips': []}
  574         do_check({'PublicIp': fakes.IP_ADDRESS_1},
  575                  'AuthFailure')
  576 
  577         # NOTE(ft): vpc address vs public ip parameter
  578         self.set_mock_db_items(fakes.DB_ADDRESS_1)
  579         self.neutron.show_floatingip.return_value = (
  580             {'floatingip': fakes.OS_FLOATING_IP_1})
  581         do_check({'PublicIp': fakes.IP_ADDRESS_1},
  582                  'InvalidParameterValue')
  583 
  584         # NOTE(ft): not registered address
  585         self.set_mock_db_items()
  586         do_check({'AllocationId': fakes.ID_EC2_ADDRESS_1},
  587                  'InvalidAllocationID.NotFound')
  588 
  589         # NOTE(ft): registered broken vpc address
  590         self.set_mock_db_items(fakes.DB_ADDRESS_1)
  591         self.neutron.show_floatingip.side_effect = neutron_exception.NotFound
  592         do_check({'AllocationId': fakes.ID_EC2_ADDRESS_1},
  593                  'InvalidAllocationID.NotFound')
  594         self.neutron.show_floatingip.side_effect = None
  595 
  596         # NOTE(ft): address is in use
  597         self.set_mock_db_items(fakes.DB_ADDRESS_2)
  598         self.neutron.show_floatingip.return_value = (
  599             {'floatingip': fakes.OS_FLOATING_IP_2})
  600         do_check({'AllocationId': fakes.ID_EC2_ADDRESS_2},
  601                  'InvalidIPAddress.InUse')
  602 
  603         # NOTE(tikitavi): address is in use in not default vpc
  604         self.configure(disable_ec2_classic=True)
  605         self.set_mock_db_items(fakes.DB_VPC_DEFAULT,
  606                                fakes.DB_VPC_1,
  607                                fakes.DB_ADDRESS_2,
  608                                fakes.DB_NETWORK_INTERFACE_2)
  609         self.neutron.show_floatingip.return_value = (
  610             {'floatingip': fakes.OS_FLOATING_IP_2})
  611 
  612         do_check({'AllocationId': fakes.ID_EC2_ADDRESS_2},
  613                  'InvalidIPAddress.InUse')
  614 
  615     @tools.screen_unexpected_exception_logs
  616     def test_release_address_vpc_rollback(self):
  617         self.set_mock_db_items(fakes.DB_ADDRESS_1)
  618         self.neutron.show_floatingip.return_value = (
  619             {'floatingip': fakes.OS_FLOATING_IP_1})
  620         self.neutron.delete_floatingip.side_effect = Exception()
  621 
  622         self.assert_execution_error(self.ANY_EXECUTE_ERROR, 'ReleaseAddress',
  623                                     {'AllocationId': fakes.ID_EC2_ADDRESS_1})
  624 
  625         self.db_api.restore_item.assert_called_once_with(
  626             mock.ANY, 'eipalloc', fakes.DB_ADDRESS_1)
  627 
  628     def test_describe_addresses_vpc(self):
  629         self.neutron.list_floatingips.return_value = (
  630             {'floatingips': [fakes.OS_FLOATING_IP_1,
  631                              fakes.OS_FLOATING_IP_2]})
  632         self.neutron.list_ports.return_value = (
  633             {'ports': [fakes.OS_PORT_1,
  634                        fakes.OS_PORT_2]})
  635         self.set_mock_db_items(
  636             fakes.DB_ADDRESS_1, fakes.DB_ADDRESS_2, fakes.DB_INSTANCE_1,
  637             fakes.DB_NETWORK_INTERFACE_1, fakes.DB_NETWORK_INTERFACE_2)
  638 
  639         resp = self.execute('DescribeAddresses', {})
  640         self.assertThat(resp['addressesSet'],
  641                         matchers.ListMatches([fakes.EC2_ADDRESS_1,
  642                                               fakes.EC2_ADDRESS_2]))
  643 
  644         self.db_api.get_items_by_ids = tools.CopyingMock(
  645             return_value=[fakes.DB_ADDRESS_1])
  646         resp = self.execute('DescribeAddresses',
  647                             {'AllocationId.1': fakes.ID_EC2_ADDRESS_1})
  648         self.assertThat(resp['addressesSet'],
  649                         matchers.ListMatches([fakes.EC2_ADDRESS_1]))
  650         self.db_api.get_items_by_ids.assert_called_once_with(
  651             mock.ANY, set([fakes.ID_EC2_ADDRESS_1]))
  652 
  653         self.check_filtering(
  654              'DescribeAddresses', 'addressesSet',
  655              [('allocation-id', fakes.ID_EC2_ADDRESS_1),
  656               ('association-id', fakes.ID_EC2_ASSOCIATION_2),
  657               ('domain', 'vpc'),
  658               ('instance-id', fakes.ID_EC2_INSTANCE_1),
  659               ('network-interface-id', fakes.ID_EC2_NETWORK_INTERFACE_2),
  660               ('network-interface-owner-id', fakes.ID_OS_PROJECT),
  661               ('private-ip-address', fakes.IP_NETWORK_INTERFACE_2),
  662               ('public-ip', fakes.IP_ADDRESS_2)])
  663 
  664     def test_describe_addresses_ec2_classic(self):
  665         self.set_mock_db_items(fakes.DB_INSTANCE_1)
  666         self.neutron.list_floatingips.return_value = (
  667             {'floatingips': [fakes.OS_FLOATING_IP_1,
  668                              fakes.OS_FLOATING_IP_2]})
  669         self.neutron.list_ports.return_value = (
  670             {'ports': [fakes.OS_PORT_1,
  671                        fakes.OS_PORT_2]})
  672         resp = self.execute('DescribeAddresses', {})
  673         self.assertThat(resp['addressesSet'],
  674                         matchers.ListMatches([fakes.EC2_ADDRESS_CLASSIC_1,
  675                                               fakes.EC2_ADDRESS_CLASSIC_2]))
  676         resp = self.execute('DescribeAddresses', {'PublicIp.1':
  677                                                   fakes.IP_ADDRESS_2})
  678         self.assertThat(resp['addressesSet'],
  679                         matchers.ListMatches([fakes.EC2_ADDRESS_CLASSIC_2]))