"Fossies" - the Fresh Open Source Software Archive

Member "ec2-api-12.0.0/ec2api/tests/unit/test_network_interface.py" (14 Apr 2021, 31455 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_network_interface.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 
   16 import copy
   17 from unittest import mock
   18 
   19 from neutronclient.common import exceptions as neutron_exception
   20 
   21 from ec2api.api import ec2utils
   22 from ec2api.tests.unit import base
   23 from ec2api.tests.unit import fakes
   24 from ec2api.tests.unit import matchers
   25 from ec2api.tests.unit import tools
   26 
   27 
   28 class NetworkInterfaceTestCase(base.ApiTestCase):
   29 
   30     def test_create_network_interface(self):
   31         self.set_mock_db_items(fakes.DB_SUBNET_1, fakes.DB_VPC_1,
   32                                fakes.DB_SECURITY_GROUP_1)
   33         self.db_api.add_item.return_value = fakes.DB_NETWORK_INTERFACE_1
   34         self.neutron.show_subnet.return_value = {'subnet': fakes.OS_SUBNET_1}
   35         self.neutron.create_port.return_value = {'port': fakes.OS_PORT_1}
   36         self.neutron.list_security_groups.return_value = (
   37             {'security_groups': [copy.deepcopy(fakes.OS_SECURITY_GROUP_1)]})
   38 
   39         def check_response(resp, auto_ips=False):
   40             self.assertThat(fakes.EC2_NETWORK_INTERFACE_1,
   41                             matchers.DictMatches(resp['networkInterface']))
   42             self.db_api.add_item.assert_called_once_with(
   43                 mock.ANY, 'eni',
   44                 tools.purge_dict(fakes.DB_NETWORK_INTERFACE_1, ('id',)))
   45             if auto_ips:
   46                 self.neutron.create_port.assert_called_once_with(
   47                     {'port':
   48                         {'network_id': fakes.ID_OS_NETWORK_1,
   49                          'fixed_ips':
   50                             [{'subnet_id': fakes.ID_OS_SUBNET_1}],
   51                          'security_groups': [fakes.ID_OS_SECURITY_GROUP_1]}})
   52             else:
   53                 self.neutron.create_port.assert_called_once_with(
   54                     {'port':
   55                         {'network_id': fakes.ID_OS_NETWORK_1,
   56                          'fixed_ips':
   57                             [{'ip_address': fakes.IP_NETWORK_INTERFACE_1}],
   58                          'security_groups': [fakes.ID_OS_SECURITY_GROUP_1]}})
   59             self.neutron.update_port.assert_called_once_with(
   60                 fakes.ID_OS_PORT_1,
   61                 {'port': {'name':
   62                           fakes.ID_EC2_NETWORK_INTERFACE_1}})
   63             self.neutron.reset_mock()
   64             self.db_api.reset_mock()
   65             self.neutron.list_security_groups.return_value = (
   66                 {'security_groups': [
   67                     copy.deepcopy(fakes.OS_SECURITY_GROUP_1)]})
   68 
   69         resp = self.execute(
   70             'CreateNetworkInterface',
   71             {'SubnetId': fakes.ID_EC2_SUBNET_1,
   72              'PrivateIpAddress': fakes.IP_NETWORK_INTERFACE_1,
   73              'Description': fakes.DESCRIPTION_NETWORK_INTERFACE_1})
   74         check_response(resp)
   75 
   76         resp = self.execute(
   77             'CreateNetworkInterface',
   78             {'SubnetId': fakes.ID_EC2_SUBNET_1,
   79              'PrivateIpAddresses.1.PrivateIpAddress': (
   80                     fakes.IP_NETWORK_INTERFACE_1),
   81              'PrivateIpAddresses.1.Primary': True,
   82              'Description': fakes.DESCRIPTION_NETWORK_INTERFACE_1})
   83         check_response(resp)
   84 
   85         resp = self.execute(
   86             'CreateNetworkInterface',
   87             {'SubnetId': fakes.ID_EC2_SUBNET_1,
   88              'Description': fakes.DESCRIPTION_NETWORK_INTERFACE_1})
   89         check_response(resp, True)
   90 
   91         resp = self.execute(
   92             'CreateNetworkInterface',
   93             {'SubnetId': fakes.ID_EC2_SUBNET_1,
   94              'SecondaryPrivateIpAddressCount': '1',
   95              'Description': fakes.DESCRIPTION_NETWORK_INTERFACE_1})
   96         check_response(resp, True)
   97 
   98         resp = self.execute(
   99             'CreateNetworkInterface',
  100             {'SubnetId': fakes.ID_EC2_SUBNET_1,
  101              'SecondaryPrivateIpAddressCount': '0',
  102              'Description': fakes.DESCRIPTION_NETWORK_INTERFACE_1})
  103         check_response(resp, True)
  104 
  105     def test_create_network_interface_multiple_ips(self):
  106         self.set_mock_db_items(fakes.DB_SUBNET_2, fakes.DB_VPC_1,
  107                                fakes.DB_SECURITY_GROUP_1)
  108         self.db_api.add_item.return_value = fakes.DB_NETWORK_INTERFACE_2
  109         self.neutron.show_subnet.return_value = {'subnet': fakes.OS_SUBNET_2}
  110         self.neutron.create_port.return_value = {'port': fakes.OS_PORT_2}
  111         self.neutron.list_security_groups.return_value = (
  112             {'security_groups': [copy.deepcopy(fakes.OS_SECURITY_GROUP_1)]})
  113         created_ec2_network_interface = tools.patch_dict(
  114             fakes.EC2_NETWORK_INTERFACE_2,
  115             {'privateIpAddressesSet': [
  116                 tools.purge_dict(s, ['association'])
  117                 for s in fakes.EC2_NETWORK_INTERFACE_2[
  118                     'privateIpAddressesSet']]},
  119             ['association'])
  120 
  121         def check_response(resp):
  122             self.assertThat(created_ec2_network_interface,
  123                             matchers.DictMatches(resp['networkInterface']))
  124             self.db_api.add_item.assert_called_once_with(
  125                 mock.ANY, 'eni',
  126                 tools.purge_dict(fakes.DB_NETWORK_INTERFACE_2,
  127                                  ('id',
  128                                   'device_index',
  129                                   'instance_id',
  130                                   'delete_on_termination',
  131                                   'attach_time')))
  132             self.neutron.update_port.assert_called_once_with(
  133                 fakes.ID_OS_PORT_2,
  134                 {'port': {'name':
  135                           fakes.ID_EC2_NETWORK_INTERFACE_2}})
  136             self.neutron.reset_mock()
  137             self.db_api.reset_mock()
  138             self.neutron.list_security_groups.return_value = (
  139                 {'security_groups': [
  140                     copy.deepcopy(fakes.OS_SECURITY_GROUP_1)]})
  141 
  142         resp = self.execute(
  143             'CreateNetworkInterface',
  144             {'SubnetId': fakes.ID_EC2_SUBNET_2,
  145              'SecondaryPrivateIpAddressCount': '3',
  146              'Description': fakes.DESCRIPTION_NETWORK_INTERFACE_2})
  147         self.neutron.create_port.assert_called_once_with(
  148             {'port': {'network_id': fakes.ID_OS_NETWORK_2,
  149                       'fixed_ips': [{'subnet_id': fakes.ID_OS_SUBNET_2},
  150                                     {'subnet_id': fakes.ID_OS_SUBNET_2},
  151                                     {'subnet_id': fakes.ID_OS_SUBNET_2}],
  152                       'security_groups': [fakes.ID_OS_SECURITY_GROUP_1]}})
  153         check_response(resp)
  154 
  155         resp = self.execute(
  156             'CreateNetworkInterface',
  157             {'SubnetId': fakes.ID_EC2_SUBNET_2,
  158              'PrivateIpAddress': fakes.IPS_NETWORK_INTERFACE_2[0],
  159              'PrivateIpAddresses.1.PrivateIpAddress':
  160                  fakes.IPS_NETWORK_INTERFACE_2[1],
  161              'PrivateIpAddresses.1.Primary': False,
  162              'PrivateIpAddresses.2.PrivateIpAddress':
  163                  fakes.IPS_NETWORK_INTERFACE_2[2],
  164              'PrivateIpAddresses.2.Primary': False,
  165              'Description': fakes.DESCRIPTION_NETWORK_INTERFACE_2})
  166         self.neutron.create_port.assert_called_once_with(
  167             {'port':
  168              {'network_id': fakes.ID_OS_NETWORK_2,
  169               'fixed_ips': [
  170                   {'ip_address': fakes.IPS_NETWORK_INTERFACE_2[0]},
  171                   {'ip_address': fakes.IPS_NETWORK_INTERFACE_2[1]},
  172                   {'ip_address': fakes.IPS_NETWORK_INTERFACE_2[2]}],
  173               'security_groups': [fakes.ID_OS_SECURITY_GROUP_1]}})
  174         check_response(resp)
  175 
  176         resp = self.execute(
  177             'CreateNetworkInterface',
  178             {'SubnetId': fakes.ID_EC2_SUBNET_2,
  179              'PrivateIpAddresses.1.PrivateIpAddress':
  180                  fakes.IPS_NETWORK_INTERFACE_2[0],
  181              'PrivateIpAddresses.1.Primary': True,
  182              'PrivateIpAddresses.2.PrivateIpAddress':
  183                  fakes.IPS_NETWORK_INTERFACE_2[1],
  184              'PrivateIpAddresses.2.Primary': False,
  185              'PrivateIpAddresses.3.PrivateIpAddress':
  186                  fakes.IPS_NETWORK_INTERFACE_2[2],
  187              'PrivateIpAddresses.3.Primary': False,
  188              'Description': fakes.DESCRIPTION_NETWORK_INTERFACE_2})
  189         self.neutron.create_port.assert_called_once_with(
  190             {'port':
  191              {'network_id': fakes.ID_OS_NETWORK_2,
  192               'fixed_ips': [
  193                   {'ip_address': fakes.IPS_NETWORK_INTERFACE_2[0]},
  194                   {'ip_address': fakes.IPS_NETWORK_INTERFACE_2[1]},
  195                   {'ip_address': fakes.IPS_NETWORK_INTERFACE_2[2]}],
  196               'security_groups': [fakes.ID_OS_SECURITY_GROUP_1]}})
  197         check_response(resp)
  198 
  199         resp = self.execute(
  200             'CreateNetworkInterface',
  201             {'SubnetId': fakes.ID_EC2_SUBNET_2,
  202              'PrivateIpAddress': fakes.IPS_NETWORK_INTERFACE_2[0],
  203              'PrivateIpAddresses.1.PrivateIpAddress':
  204                  fakes.IPS_NETWORK_INTERFACE_2[1],
  205              'PrivateIpAddresses.1.Primary': False,
  206              'SecondaryPrivateIpAddressCount': '1',
  207              'Description': fakes.DESCRIPTION_NETWORK_INTERFACE_2})
  208         self.neutron.create_port.assert_called_once_with(
  209             {'port':
  210              {'network_id': fakes.ID_OS_NETWORK_2,
  211               'fixed_ips': [
  212                   {'ip_address': fakes.IPS_NETWORK_INTERFACE_2[0]},
  213                   {'ip_address': fakes.IPS_NETWORK_INTERFACE_2[1]},
  214                   {'subnet_id': fakes.ID_OS_SUBNET_2}],
  215               'security_groups': [fakes.ID_OS_SECURITY_GROUP_1]}})
  216         check_response(resp)
  217 
  218     def test_create_network_interface_invalid_parameters(self):
  219         def do_check(args, error_code):
  220             self.neutron.reset_mock()
  221             self.db_api.reset_mock()
  222             self.assert_execution_error(
  223                 error_code, 'CreateNetworkInterface', args)
  224 
  225         self.set_mock_db_items()
  226         do_check({'SubnetId': fakes.ID_EC2_SUBNET_2},
  227                  'InvalidSubnetID.NotFound')
  228         self.db_api.get_item_by_id.assert_called_once_with(
  229             mock.ANY, fakes.ID_EC2_SUBNET_2)
  230 
  231         self.set_mock_db_items(fakes.DB_SUBNET_1, fakes.DB_VPC_1)
  232         self.neutron.show_subnet.return_value = {'subnet': fakes.OS_SUBNET_1}
  233         do_check({'SubnetId': fakes.ID_EC2_SUBNET_1,
  234                   'PrivateIpAddress': fakes.IP_NETWORK_INTERFACE_2},
  235                  'InvalidParameterValue')
  236 
  237         for cls in [neutron_exception.OverQuotaClient,
  238                     neutron_exception.IpAddressGenerationFailureClient]:
  239             self.neutron.create_port.side_effect = cls()
  240             do_check({'SubnetId': fakes.ID_EC2_SUBNET_1,
  241                       'PrivateIpAddress': fakes.IP_NETWORK_INTERFACE_1},
  242                      'InsufficientFreeAddressesInSubnet')
  243 
  244         for cls in [neutron_exception.IpAddressInUseClient,
  245                     neutron_exception.BadRequest]:
  246             self.neutron.create_port.side_effect = cls()
  247             do_check({'SubnetId': fakes.ID_EC2_SUBNET_1,
  248                       'PrivateIpAddress': fakes.IP_NETWORK_INTERFACE_1},
  249                      'InvalidParameterValue')
  250 
  251     @tools.screen_unexpected_exception_logs
  252     @mock.patch('ec2api.api.dhcp_options._add_dhcp_opts_to_port')
  253     def test_create_network_interface_rollback(self, _add_dhcp_opts_to_port):
  254         self.set_mock_db_items(
  255             tools.update_dict(
  256                 fakes.DB_VPC_1,
  257                 {'dhcp_options_id': fakes.ID_EC2_DHCP_OPTIONS_1}),
  258             fakes.DB_SUBNET_1, fakes.DB_DHCP_OPTIONS_1)
  259         self.db_api.add_item.return_value = fakes.DB_NETWORK_INTERFACE_1
  260         self.neutron.show_subnet.return_value = {'subnet': fakes.OS_SUBNET_1}
  261         self.neutron.create_port.return_value = {'port': fakes.OS_PORT_1}
  262         _add_dhcp_opts_to_port.side_effect = Exception()
  263 
  264         self.assert_execution_error(self.ANY_EXECUTE_ERROR,
  265                                     'CreateNetworkInterface',
  266                                     {'SubnetId': fakes.ID_EC2_SUBNET_1})
  267 
  268         self.neutron.delete_port.assert_called_once_with(fakes.ID_OS_PORT_1)
  269         self.db_api.delete_item.assert_called_once_with(
  270             mock.ANY, fakes.ID_EC2_NETWORK_INTERFACE_1)
  271 
  272     def test_delete_network_interface(self):
  273         self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_1)
  274         resp = self.execute(
  275             'DeleteNetworkInterface',
  276             {'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1})
  277         self.assertEqual(True, resp['return'])
  278         self.db_api.get_item_by_id.assert_any_call(
  279             mock.ANY,
  280             fakes.ID_EC2_NETWORK_INTERFACE_1)
  281         self.db_api.delete_item.assert_called_once_with(
  282             mock.ANY,
  283             fakes.ID_EC2_NETWORK_INTERFACE_1)
  284         self.neutron.delete_port.assert_called_once_with(
  285             fakes.ID_OS_PORT_1)
  286 
  287     def test_delete_network_interface_obsolete(self):
  288         self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_1)
  289         self.neutron.delete_port.side_effect = (
  290             neutron_exception.PortNotFoundClient())
  291         resp = self.execute(
  292             'DeleteNetworkInterface',
  293             {'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1})
  294         self.assertEqual(True, resp['return'])
  295 
  296     def test_delete_network_interface_no_network_interface(self):
  297         self.set_mock_db_items()
  298         self.assert_execution_error(
  299             'InvalidNetworkInterfaceID.NotFound', 'DeleteNetworkInterface',
  300             {'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1})
  301         self.assertEqual(0, self.neutron.delete_port.call_count)
  302 
  303     def test_delete_network_interface_is_in_use(self):
  304         self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_2)
  305         self.assert_execution_error(
  306             'InvalidParameterValue', 'DeleteNetworkInterface',
  307             {'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_2})
  308         self.assertEqual(0, self.neutron.delete_port.call_count)
  309 
  310     def test_delete_network_interface_with_public_ip(self):
  311         detached_network_interface_2 = fakes.gen_db_network_interface(
  312             fakes.ID_EC2_NETWORK_INTERFACE_2,
  313             fakes.ID_OS_PORT_2,
  314             fakes.ID_EC2_VPC_1,
  315             fakes.ID_EC2_SUBNET_2,
  316             fakes.IP_NETWORK_INTERFACE_2)
  317         self.set_mock_db_items(detached_network_interface_2,
  318                                fakes.DB_ADDRESS_1, fakes.DB_ADDRESS_2)
  319         resp = self.execute(
  320             'DeleteNetworkInterface',
  321             {'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_2})
  322         self.assertEqual(True, resp['return'])
  323         self.db_api.get_item_by_id.assert_any_call(
  324             mock.ANY,
  325             fakes.ID_EC2_NETWORK_INTERFACE_2)
  326         self.db_api.delete_item.assert_called_once_with(
  327             mock.ANY,
  328             fakes.ID_EC2_NETWORK_INTERFACE_2)
  329         self.neutron.delete_port.assert_called_once_with(
  330             fakes.ID_OS_PORT_2)
  331         self.db_api.update_item.assert_called_once_with(
  332             mock.ANY,
  333             tools.purge_dict(fakes.DB_ADDRESS_2,
  334                              ['network_interface_id',
  335                               'private_ip_address']))
  336 
  337     @tools.screen_unexpected_exception_logs
  338     def test_delete_network_interface_rollback(self):
  339         self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_1)
  340         self.neutron.delete_port.side_effect = Exception()
  341 
  342         self.assert_execution_error(
  343             self.ANY_EXECUTE_ERROR, 'DeleteNetworkInterface',
  344             {'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1})
  345 
  346         self.db_api.restore_item.assert_called_once_with(
  347             mock.ANY, 'eni', fakes.DB_NETWORK_INTERFACE_1)
  348 
  349     def test_describe_network_interfaces(self):
  350         self.set_mock_db_items(
  351             fakes.DB_NETWORK_INTERFACE_1, fakes.DB_NETWORK_INTERFACE_2,
  352             fakes.DB_ADDRESS_1, fakes.DB_ADDRESS_2,
  353             fakes.DB_INSTANCE_1, fakes.DB_INSTANCE_2,
  354             fakes.DB_SECURITY_GROUP_1)
  355         self.neutron.list_ports.return_value = (
  356             {'ports': [fakes.OS_PORT_1, fakes.OS_PORT_2]})
  357         self.neutron.list_floatingips.return_value = (
  358             {'floatingips': [fakes.OS_FLOATING_IP_1,
  359                              fakes.OS_FLOATING_IP_2]})
  360         self.neutron.list_security_groups.return_value = (
  361             {'security_groups': [copy.deepcopy(fakes.OS_SECURITY_GROUP_1)]})
  362 
  363         resp = self.execute('DescribeNetworkInterfaces', {})
  364         self.assertThat(resp['networkInterfaceSet'],
  365                         matchers.ListMatches(
  366                             [fakes.EC2_NETWORK_INTERFACE_1,
  367                              fakes.EC2_NETWORK_INTERFACE_2],
  368                             orderless_lists=True),
  369                         verbose=True)
  370 
  371         self.db_api.get_items_by_ids = tools.CopyingMock(
  372             return_value=[fakes.DB_NETWORK_INTERFACE_1])
  373         resp = self.execute(
  374             'DescribeNetworkInterfaces',
  375             {'NetworkInterfaceId.1': fakes.ID_EC2_NETWORK_INTERFACE_1})
  376         self.assertThat(resp['networkInterfaceSet'],
  377                         matchers.ListMatches(
  378                             [fakes.EC2_NETWORK_INTERFACE_1]))
  379         self.db_api.get_items_by_ids.assert_called_once_with(
  380             mock.ANY, set([fakes.ID_EC2_NETWORK_INTERFACE_1]))
  381 
  382         self.check_filtering(
  383             'DescribeNetworkInterfaces', 'networkInterfaceSet',
  384             [('addresses.private-ip-address',
  385               fakes.IP_NETWORK_INTERFACE_2_EXT_1,),
  386              ('addresses.primary', False),
  387              ('addresses.association.public-ip', fakes.IP_ADDRESS_2),
  388              ('addresses.association.owner-id', fakes.ID_OS_PROJECT),
  389              ('association.association-id', fakes.ID_EC2_ASSOCIATION_2),
  390              ('association.allocation-id', fakes.ID_EC2_ADDRESS_2),
  391              ('association.ip-owner-id', fakes.ID_OS_PROJECT),
  392              ('association.public-ip', fakes.IP_ADDRESS_2),
  393              ('attachment.attachment-id',
  394               fakes.ID_EC2_NETWORK_INTERFACE_2_ATTACH),
  395              ('attachment.instance-id', fakes.ID_EC2_INSTANCE_1),
  396              ('attachment.instance-owner-id', fakes.ID_OS_PROJECT),
  397              ('attachment.device-index', 0),
  398              ('attachment.status', 'attached'),
  399              ('attachment.attach.time', fakes.TIME_ATTACH_NETWORK_INTERFACE),
  400              ('attachment.delete-on-termination', False),
  401              ('description', fakes.DESCRIPTION_NETWORK_INTERFACE_1),
  402              ('group-id', fakes.ID_EC2_SECURITY_GROUP_1),
  403              ('group-name', fakes.NAME_DEFAULT_OS_SECURITY_GROUP),
  404              ('mac-address', fakes.MAC_ADDRESS),
  405              ('network-interface-id', fakes.ID_EC2_NETWORK_INTERFACE_1),
  406              ('owner-id', fakes.ID_OS_PROJECT),
  407              ('private-ip-address', fakes.IP_NETWORK_INTERFACE_1),
  408              ('requester-managed', False),
  409              ('source-dest-check', True),
  410              ('status', 'available'),
  411              ('vpc-id', fakes.ID_EC2_VPC_1),
  412              ('subnet-id', fakes.ID_EC2_SUBNET_2)])
  413         self.check_tag_support(
  414             'DescribeNetworkInterfaces', 'networkInterfaceSet',
  415             fakes.ID_EC2_NETWORK_INTERFACE_1, 'networkInterfaceId')
  416 
  417     def test_describe_network_interface_attribute(self):
  418         self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_1)
  419 
  420         resp = self.execute(
  421             'DescribeNetworkInterfaceAttribute',
  422             {'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1,
  423              'Attribute': 'description'})
  424         self.assertEqual(fakes.ID_EC2_NETWORK_INTERFACE_1,
  425                          resp['networkInterfaceId'])
  426         self.assertEqual(fakes.DESCRIPTION_NETWORK_INTERFACE_1,
  427                          resp['description'].get('value', None))
  428 
  429     def test_modify_network_interface_attribute(self):
  430         self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_1,
  431                                fakes.DB_NETWORK_INTERFACE_2)
  432 
  433         self.execute(
  434             'ModifyNetworkInterfaceAttribute',
  435             {'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1,
  436              'Description.Value': 'New description'})
  437         self.db_api.update_item.assert_called_once_with(
  438             mock.ANY,
  439             tools.update_dict(fakes.DB_NETWORK_INTERFACE_1,
  440                               {'description': 'New description'}))
  441 
  442         self.db_api.reset_mock()
  443         self.execute(
  444             'ModifyNetworkInterfaceAttribute',
  445             {'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_2,
  446              'Attachment.AttachmentId': (
  447                  fakes.ID_EC2_NETWORK_INTERFACE_2_ATTACH),
  448              'Attachment.DeleteOnTermination': 'True'})
  449         self.db_api.update_item.assert_called_once_with(
  450             mock.ANY,
  451             tools.update_dict(fakes.DB_NETWORK_INTERFACE_2,
  452                               {'delete_on_termination': True}))
  453 
  454     def test_modify_network_interface_attribute_invalid_parameters(self):
  455         self.assert_execution_error(
  456             'InvalidParameterCombination', 'ModifyNetworkInterfaceAttribute',
  457             {'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1,
  458              'Description.Value': 'New description',
  459              'SourceDestCheck.Value': 'True'})
  460 
  461         self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_1,
  462                                fakes.DB_NETWORK_INTERFACE_2)
  463 
  464         self.assert_execution_error(
  465             'MissingParameter', 'ModifyNetworkInterfaceAttribute',
  466             {'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_2,
  467              'Attachment.DeleteOnTermination': 'True'})
  468 
  469         self.assert_execution_error(
  470             'MissingParameter', 'ModifyNetworkInterfaceAttribute',
  471             {'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_2,
  472              'Attachment.AttachmentId': (
  473                  fakes.ID_EC2_NETWORK_INTERFACE_2_ATTACH)})
  474 
  475         self.assert_execution_error(
  476             'InvalidAttachmentID.NotFound', 'ModifyNetworkInterfaceAttribute',
  477             {'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1,
  478              'Attachment.AttachmentId': (
  479                  fakes.ID_EC2_NETWORK_INTERFACE_2_ATTACH),
  480              'Attachment.DeleteOnTermination': 'True'})
  481 
  482         self.assert_execution_error(
  483             'InvalidAttachmentID.NotFound', 'ModifyNetworkInterfaceAttribute',
  484             {'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_2,
  485              'Attachment.AttachmentId': ec2utils.change_ec2_id_kind(
  486                  fakes.ID_EC2_NETWORK_INTERFACE_1, 'eni-attach'),
  487              'Attachment.DeleteOnTermination': 'True'})
  488 
  489     def test_reset_network_interface_attribute(self):
  490         self.execute(
  491             'ResetNetworkInterfaceAttribute',
  492             {'NetworkInterfaceId':
  493              fakes.ID_EC2_NETWORK_INTERFACE_1,
  494              'Attribute': 'sourceDestCheck'})
  495 
  496     def test_attach_network_interface(self):
  497         self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_1,
  498                                fakes.DB_INSTANCE_1)
  499         self.neutron.show_port.return_value = (
  500             {'port': fakes.OS_PORT_1})
  501         self.isotime.return_value = fakes.TIME_ATTACH_NETWORK_INTERFACE
  502         self.execute(
  503             'AttachNetworkInterface',
  504             {'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1,
  505              'InstanceId': fakes.ID_EC2_INSTANCE_1,
  506              'DeviceIndex': '1'})
  507         self.nova.servers.interface_attach.assert_called_once_with(
  508             fakes.ID_OS_INSTANCE_1, fakes.ID_OS_PORT_1, None, None)
  509         self.db_api.update_item.assert_called_once_with(
  510             mock.ANY,
  511             tools.update_dict(
  512                 fakes.DB_NETWORK_INTERFACE_1,
  513                 {'device_index': 1,
  514                  'instance_id': fakes.ID_EC2_INSTANCE_1,
  515                  'delete_on_termination': False,
  516                  'attach_time': fakes.TIME_ATTACH_NETWORK_INTERFACE}))
  517 
  518     def test_attach_network_interface_invalid_parameters(self):
  519         # NOTE(ft): eni is already attached
  520         self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_2)
  521         self.assert_execution_error(
  522             'InvalidParameterValue', 'AttachNetworkInterface',
  523             {'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_2,
  524              'InstanceId': fakes.ID_EC2_INSTANCE_2,
  525              'DeviceIndex': '1'})
  526 
  527         # NOTE(ft): device index is in use
  528         self.set_mock_db_items(
  529             fakes.DB_NETWORK_INTERFACE_1, fakes.DB_NETWORK_INTERFACE_2,
  530             fakes.DB_INSTANCE_1)
  531         self.assert_execution_error(
  532             'InvalidParameterValue', 'AttachNetworkInterface',
  533             {'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1,
  534              'InstanceId': fakes.ID_EC2_INSTANCE_1,
  535              'DeviceIndex': '0'})
  536 
  537     @tools.screen_unexpected_exception_logs
  538     def test_attach_network_interface_rollback(self):
  539         self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_1,
  540                                fakes.DB_INSTANCE_1)
  541         self.neutron.show_port.return_value = (
  542             {'port': fakes.OS_PORT_2})
  543         self.isotime.return_value = fakes.TIME_ATTACH_NETWORK_INTERFACE
  544         self.nova.servers.interface_attach.side_effect = Exception()
  545 
  546         self.assert_execution_error(
  547             self.ANY_EXECUTE_ERROR, 'AttachNetworkInterface',
  548             {'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1,
  549              'InstanceId': fakes.ID_EC2_INSTANCE_1,
  550              'DeviceIndex': '1'})
  551 
  552         self.db_api.update_item.assert_any_call(
  553             mock.ANY, fakes.DB_NETWORK_INTERFACE_1)
  554 
  555     def test_detach_network_interface(self):
  556         network_interface = tools.update_dict(fakes.DB_NETWORK_INTERFACE_2,
  557                                               {'device_index': 1})
  558         self.set_mock_db_items(network_interface)
  559         self.neutron.show_port.return_value = (
  560             {'port': fakes.OS_PORT_2})
  561         self.execute(
  562             'DetachNetworkInterface',
  563             {'AttachmentId': ec2utils.change_ec2_id_kind(
  564                     fakes.ID_EC2_NETWORK_INTERFACE_2, 'eni-attach')})
  565         self.neutron.update_port.assert_called_once_with(
  566             fakes.ID_OS_PORT_2,
  567             {'port': {'device_id': '',
  568                       'device_owner': ''}}
  569         )
  570         self.db_api.update_item.assert_called_once_with(
  571             mock.ANY,
  572             tools.purge_dict(fakes.DB_NETWORK_INTERFACE_2,
  573                              {'device_index',
  574                               'instance_id',
  575                               'delete_on_termination',
  576                               'attach_time'}))
  577 
  578     def test_detach_network_interface_invalid_parameters(self):
  579         # NOTE(ft): eni is not found
  580         self.set_mock_db_items()
  581         self.assert_execution_error(
  582             'InvalidAttachmentID.NotFound', 'DetachNetworkInterface',
  583             {'AttachmentId': ec2utils.change_ec2_id_kind(
  584                     fakes.ID_EC2_NETWORK_INTERFACE_2, 'eni-attach')})
  585 
  586         # NOTE(ft): eni is attached with device index = 0
  587         self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_2)
  588         self.assert_execution_error(
  589             'OperationNotPermitted', 'DetachNetworkInterface',
  590             {'AttachmentId': ec2utils.change_ec2_id_kind(
  591                     fakes.ID_EC2_NETWORK_INTERFACE_2, 'eni-attach')})
  592 
  593     @tools.screen_unexpected_exception_logs
  594     def test_detach_network_interface_rollback(self):
  595         network_interface = tools.update_dict(fakes.DB_NETWORK_INTERFACE_2,
  596                                               {'device_index': 1})
  597         self.set_mock_db_items(network_interface)
  598         self.neutron.show_port.return_value = (
  599             {'port': fakes.OS_PORT_2})
  600         self.neutron.update_port.side_effect = Exception()
  601 
  602         self.assert_execution_error(
  603             self.ANY_EXECUTE_ERROR, 'DetachNetworkInterface',
  604             {'AttachmentId': fakes.ID_EC2_NETWORK_INTERFACE_2_ATTACH})
  605 
  606         self.db_api.update_item.assert_any_call(
  607             mock.ANY, network_interface)
  608 
  609     def test_assign_unassign_private_ip_addresses(self):
  610         self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_1, fakes.DB_SUBNET_1)
  611         self.neutron.show_subnet.return_value = (
  612             {'subnet': fakes.OS_SUBNET_1})
  613         self.neutron.show_port.return_value = (
  614             {'port': copy.deepcopy(fakes.OS_PORT_1)})
  615         self.execute(
  616             'AssignPrivateIpAddresses',
  617             {'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1,
  618              'PrivateIpAddress.1': '10.10.1.5',
  619              'PrivateIpAddress.2': '10.10.1.6',
  620              })
  621         self.neutron.update_port.assert_called_once_with(
  622             fakes.ID_OS_PORT_1,
  623             {'port':
  624              {'fixed_ips': [
  625                  {'subnet_id': fakes.ID_OS_SUBNET_1,
  626                   'ip_address': fakes.IP_NETWORK_INTERFACE_1},
  627                  {'ip_address': '10.10.1.5'},
  628                  {'ip_address': '10.10.1.6'}]}})
  629         self.execute(
  630             'UnassignPrivateIpAddresses',
  631             {'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1,
  632              'PrivateIpAddress.1': '10.10.1.5',
  633              'PrivateIpAddress.2': '10.10.1.6',
  634              })
  635         self.neutron.update_port.assert_any_call(
  636             fakes.ID_OS_PORT_1,
  637             {'port':
  638              {'fixed_ips': [
  639                  {'subnet_id': fakes.ID_OS_SUBNET_1,
  640                   'ip_address': fakes.IP_NETWORK_INTERFACE_1}]}})
  641 
  642     def test_assign_private_ip_addresses_invalid_parameters(self):
  643         self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_1, fakes.DB_SUBNET_1)
  644         self.neutron.show_subnet.return_value = (
  645             {'subnet': fakes.OS_SUBNET_1})
  646         self.neutron.show_port.return_value = (
  647             {'port': copy.deepcopy(fakes.OS_PORT_1)})
  648 
  649         def do_check(error_code):
  650             self.assert_execution_error(
  651                 error_code, 'AssignPrivateIpAddresses',
  652                 {'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1,
  653                  'PrivateIpAddress.1': '10.10.1.5',
  654                  'PrivateIpAddress.2': '10.10.1.6',
  655                  })
  656 
  657         self.neutron.update_port.side_effect = (
  658             neutron_exception.IpAddressGenerationFailureClient())
  659         do_check('InsufficientFreeAddressesInSubnet')
  660 
  661         self.neutron.update_port.side_effect = (
  662             neutron_exception.IpAddressInUseClient())
  663         do_check('InvalidParameterValue')
  664 
  665         self.neutron.update_port.side_effect = (
  666             neutron_exception.BadRequest())
  667         do_check('InvalidParameterValue')
  668 
  669     def test_unassign_private_ip_addresses_invalid_parameters(self):
  670         self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_2, fakes.DB_SUBNET_2)
  671         self.neutron.show_subnet.return_value = (
  672             {'subnet': fakes.OS_SUBNET_2})
  673         self.neutron.show_port.return_value = (
  674             {'port': copy.deepcopy(fakes.OS_PORT_2)})
  675 
  676         self.assert_execution_error(
  677             'InvalidParameterValue', 'UnassignPrivateIpAddresses',
  678             {'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_2,
  679              'PrivateIpAddress.1': '10.10.2.55'})