<s> from __future__ import unicode_literals <EOL> from decimal import Decimal <EOL> from django . core . urlresolvers import reverse <EOL> from mock import patch , Mock <EOL> from rest_framework import status <EOL> from rest_framework import test <EOL> from nodeconductor . core . models import SynchronizationStates <EOL> from nodeconductor . backup import models as backup_models <EOL> from nodeconductor . backup . tests import factories as backup_factories <EOL> from nodeconductor . core . fields import comma_separated_string_list_re as ips_regex <EOL> from nodeconductor . iaas . models import Instance , CloudProjectMembership , FloatingIP <EOL> from nodeconductor . iaas . tests import factories <EOL> from nodeconductor . structure . models import ProjectRole , ProjectGroupRole <EOL> from nodeconductor . structure . tests import factories as structure_factories <EOL> class UrlResolverMixin ( object ) : <EOL> def _get_flavor_url ( self , flavor ) : <EOL> return '<STR_LIT>' + reverse ( '<STR_LIT>' , kwargs = { '<STR_LIT>' : flavor . uuid } ) <EOL> def _get_project_url ( self , project ) : <EOL> return '<STR_LIT>' + reverse ( '<STR_LIT>' , kwargs = { '<STR_LIT>' : project . uuid } ) <EOL> def _get_template_url ( self , template ) : <EOL> return '<STR_LIT>' + reverse ( '<STR_LIT>' , kwargs = { '<STR_LIT>' : template . uuid } ) <EOL> def _get_ssh_public_key_url ( self , key ) : <EOL> return '<STR_LIT>' + reverse ( '<STR_LIT>' , kwargs = { '<STR_LIT>' : key . uuid } ) <EOL> class InstanceApiPermissionTest ( UrlResolverMixin , test . APITransactionTestCase ) : <EOL> def setUp ( self ) : <EOL> self . user = structure_factories . UserFactory ( ) <EOL> self . staff = structure_factories . UserFactory ( is_staff = True ) <EOL> self . admined_instance = factories . InstanceFactory ( state = Instance . States . OFFLINE ) <EOL> self . managed_instance = factories . InstanceFactory ( state = Instance . States . OFFLINE ) <EOL> admined_project = self . admined_instance . cloud_project_membership . project <EOL> admined_project . add_user ( self . user , ProjectRole . ADMINISTRATOR ) <EOL> project = self . managed_instance . cloud_project_membership . project <EOL> managed_project_group = structure_factories . ProjectGroupFactory ( ) <EOL> managed_project_group . projects . add ( project ) <EOL> managed_project_group . add_user ( self . user , ProjectGroupRole . MANAGER ) <EOL> def test_anonymous_user_cannot_list_instances ( self ) : <EOL> response = self . client . get ( reverse ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_401_UNAUTHORIZED ) <EOL> def test_user_can_list_instances_of_projects_he_is_administrator_of ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> response = self . client . get ( reverse ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> instance_url = factories . InstanceFactory . get_url ( self . admined_instance ) <EOL> self . assertIn ( instance_url , [ instance [ '<STR_LIT:url>' ] for instance in response . data ] ) <EOL> def test_user_can_list_instances_of_projects_he_is_manager_of ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> response = self . client . get ( reverse ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> instance_url = factories . InstanceFactory . get_url ( self . managed_instance ) <EOL> self . assertIn ( instance_url , [ instance [ '<STR_LIT:url>' ] for instance in response . data ] ) <EOL> def test_user_cannot_list_instances_of_projects_he_has_no_role_in ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> inaccessible_instance = factories . InstanceFactory ( ) <EOL> response = self . client . get ( reverse ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> instance_url = factories . InstanceFactory . get_url ( inaccessible_instance ) <EOL> self . assertNotIn ( instance_url , [ instance [ '<STR_LIT:url>' ] for instance in response . data ] ) <EOL> def test_anonymous_user_cannot_access_instance ( self ) : <EOL> instance = factories . InstanceFactory ( ) <EOL> response = self . client . get ( factories . InstanceFactory . get_url ( instance ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_401_UNAUTHORIZED ) <EOL> def test_user_can_access_instances_of_projects_he_is_administrator_of ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> response = self . client . get ( factories . InstanceFactory . get_url ( self . admined_instance ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> def test_user_can_access_instances_of_projects_he_is_manager_of ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> response = self . client . get ( factories . InstanceFactory . get_url ( self . managed_instance ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> def test_user_cannot_access_instances_of_projects_he_has_no_role_in ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> inaccessible_instance = factories . InstanceFactory ( ) <EOL> response = self . client . get ( self . _get_project_url ( inaccessible_instance ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_404_NOT_FOUND ) <EOL> def test_anonymous_user_cannot_delete_instances ( self ) : <EOL> response = self . client . delete ( self . _get_project_url ( factories . InstanceFactory ( ) ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_401_UNAUTHORIZED ) <EOL> def test_user_can_delete_offline_instances_of_projects_he_is_administrator_of ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> instance = self . admined_instance <EOL> instance . state = Instance . States . OFFLINE <EOL> instance . save ( ) <EOL> response = self . client . delete ( factories . InstanceFactory . get_url ( instance ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_202_ACCEPTED ) <EOL> reread_instance = Instance . objects . get ( pk = instance . pk ) <EOL> self . assertEqual ( reread_instance . state , Instance . States . DELETION_SCHEDULED , <EOL> '<STR_LIT>' ) <EOL> def test_user_cannot_delete_non_offline_instances_of_projects_he_is_administrator_of ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> forbidden_states = [ <EOL> state <EOL> for ( state , _ ) in Instance . States . CHOICES <EOL> if state != Instance . States . OFFLINE <EOL> ] <EOL> for state in forbidden_states : <EOL> instance = factories . InstanceFactory ( state = state ) <EOL> instance . cloud_project_membership . project . add_user ( self . user , ProjectRole . ADMINISTRATOR ) <EOL> response = self . client . delete ( factories . InstanceFactory . get_url ( instance ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_409_CONFLICT ) <EOL> try : <EOL> reread_instance = Instance . objects . get ( pk = instance . pk ) <EOL> except Instance . DoesNotExist : <EOL> self . fail ( '<STR_LIT>' ) <EOL> else : <EOL> self . assertEqual ( <EOL> reread_instance . state , instance . state , <EOL> '<STR_LIT>' , <EOL> ) <EOL> def test_user_cannot_delete_instances_of_projects_he_is_manager_of ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> response = self . client . delete ( factories . InstanceFactory . get_url ( self . managed_instance ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_403_FORBIDDEN ) <EOL> def test_user_can_change_description_of_instance_he_is_administrator_of ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> data = self . _get_valid_payload ( self . admined_instance ) <EOL> data [ '<STR_LIT:description>' ] = '<STR_LIT>' <EOL> response = self . client . put ( factories . InstanceFactory . get_url ( self . admined_instance ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> changed_instance = Instance . objects . get ( pk = self . admined_instance . pk ) <EOL> self . assertEqual ( changed_instance . description , '<STR_LIT>' ) <EOL> def test_user_can_change_security_groups_of_instance_he_is_administrator_of ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> data = self . _get_valid_payload ( self . admined_instance ) <EOL> cloud_project_membership = self . admined_instance . cloud_project_membership <EOL> security_group = factories . SecurityGroupFactory ( cloud_project_membership = cloud_project_membership ) <EOL> data [ '<STR_LIT>' ] = [ <EOL> { '<STR_LIT:url>' : factories . SecurityGroupFactory . get_url ( security_group ) } <EOL> ] <EOL> response = self . client . put ( factories . InstanceFactory . get_url ( self . admined_instance ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> def test_user_cannot_change_description_of_instance_he_is_manager_of ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> data = self . _get_valid_payload ( self . managed_instance ) <EOL> data [ '<STR_LIT:description>' ] = '<STR_LIT>' <EOL> response = self . client . put ( factories . InstanceFactory . get_url ( self . managed_instance ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_403_FORBIDDEN ) <EOL> def test_user_cannot_change_description_of_instance_he_has_no_role_in ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> inaccessible_instance = factories . InstanceFactory ( ) <EOL> data = self . _get_valid_payload ( inaccessible_instance ) <EOL> data [ '<STR_LIT:description>' ] = '<STR_LIT>' <EOL> response = self . client . put ( factories . InstanceFactory . get_url ( inaccessible_instance ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_404_NOT_FOUND ) <EOL> def test_user_can_change_description_single_field_of_instance_he_is_administrator_of ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> data = { <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> } <EOL> response = self . client . patch ( factories . InstanceFactory . get_url ( self . admined_instance ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> changed_instance = Instance . objects . get ( pk = self . admined_instance . pk ) <EOL> self . assertEqual ( changed_instance . description , '<STR_LIT>' ) <EOL> def test_user_cannot_change_description_single_field_of_instance_he_is_manager_of ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> data = { <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> } <EOL> response = self . client . patch ( factories . InstanceFactory . get_url ( self . managed_instance ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_403_FORBIDDEN ) <EOL> def test_user_cannot_change_description_single_field_of_instance_he_has_no_role_in ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> inaccessible_instance = factories . InstanceFactory ( ) <EOL> data = { <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> } <EOL> response = self . client . patch ( factories . InstanceFactory . get_url ( inaccessible_instance ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_404_NOT_FOUND ) <EOL> def test_user_can_change_flavor_of_stopped_instance_he_is_administrator_of ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> new_flavor = factories . FlavorFactory ( <EOL> cloud = self . admined_instance . cloud_project_membership . cloud , <EOL> disk = self . admined_instance . system_volume_size + <NUM_LIT:1> , <EOL> ) <EOL> data = { '<STR_LIT>' : self . _get_flavor_url ( new_flavor ) } <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( self . admined_instance , action = '<STR_LIT>' ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_202_ACCEPTED , response . data ) <EOL> reread_instance = Instance . objects . get ( pk = self . admined_instance . pk ) <EOL> self . assertEqual ( reread_instance . system_volume_size , self . admined_instance . system_volume_size , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( reread_instance . state , Instance . States . RESIZING_SCHEDULED , <EOL> '<STR_LIT>' ) <EOL> def test_user_can_change_flavor_to_flavor_with_less_cpu_if_result_cpu_quota_usage_is_less_then_cpu_limit ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> instance = self . admined_instance <EOL> instance . cores = <NUM_LIT:5> <EOL> instance . save ( ) <EOL> membership = instance . cloud_project_membership <EOL> membership . set_quota_usage ( '<STR_LIT>' , instance . cores ) <EOL> membership . set_quota_limit ( '<STR_LIT>' , instance . cores ) <EOL> membership . set_quota_limit ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> membership . set_quota_limit ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> new_flavor = factories . FlavorFactory ( <EOL> cloud = self . admined_instance . cloud_project_membership . cloud , <EOL> disk = self . admined_instance . system_volume_size + <NUM_LIT:1> , <EOL> cores = instance . cores - <NUM_LIT:1> , <EOL> ) <EOL> data = { '<STR_LIT>' : self . _get_flavor_url ( new_flavor ) } <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( self . admined_instance , action = '<STR_LIT>' ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_202_ACCEPTED , response . data ) <EOL> reread_instance = Instance . objects . get ( pk = self . admined_instance . pk ) <EOL> self . assertEqual ( reread_instance . state , Instance . States . RESIZING_SCHEDULED , <EOL> '<STR_LIT>' ) <EOL> def test_user_cannot_resize_instance_without_flavor_and_disk_size_in_request ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( self . admined_instance , action = '<STR_LIT>' ) , { } ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> def test_user_can_change_flavor_to_flavor_with_less_ram_if_result_ram_quota_usage_is_less_then_ram_limit ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> instance = self . admined_instance <EOL> instance . cores = <NUM_LIT:5> <EOL> instance . save ( ) <EOL> membership = instance . cloud_project_membership <EOL> membership . set_quota_usage ( '<STR_LIT>' , instance . cores ) <EOL> membership . set_quota_limit ( '<STR_LIT>' , instance . ram ) <EOL> membership . set_quota_limit ( '<STR_LIT>' , instance . cores ) <EOL> membership . set_quota_limit ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> membership . set_quota_limit ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> new_flavor = factories . FlavorFactory ( <EOL> cloud = self . admined_instance . cloud_project_membership . cloud , <EOL> disk = self . admined_instance . system_volume_size + <NUM_LIT:1> , <EOL> ram = instance . ram - <NUM_LIT:1> , <EOL> ) <EOL> data = { '<STR_LIT>' : self . _get_flavor_url ( new_flavor ) } <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( self . admined_instance , action = '<STR_LIT>' ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_202_ACCEPTED , response . data ) <EOL> reread_instance = Instance . objects . get ( pk = self . admined_instance . pk ) <EOL> self . assertEqual ( reread_instance . state , Instance . States . RESIZING_SCHEDULED , <EOL> '<STR_LIT>' ) <EOL> def test_user_cannot_change_flavor_of_stopped_instance_he_is_administrator_of_if_quota_would_be_exceeded ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> membership = self . admined_instance . cloud_project_membership <EOL> membership . set_quota_limit ( '<STR_LIT>' , <NUM_LIT> ) <EOL> big_ram_flavor = factories . FlavorFactory ( <EOL> cloud = membership . cloud , <EOL> ram = membership . quotas . get ( name = '<STR_LIT>' ) . limit + self . admined_instance . ram + <NUM_LIT:1> , <EOL> ) <EOL> data = { '<STR_LIT>' : self . _get_flavor_url ( big_ram_flavor ) } <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( self . admined_instance , action = '<STR_LIT>' ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST , response . data ) <EOL> many_core_flavor = factories . FlavorFactory ( <EOL> cloud = membership . cloud , <EOL> cores = membership . quotas . get ( name = '<STR_LIT>' ) . limit + self . admined_instance . cores + <NUM_LIT:1> , <EOL> ) <EOL> data = { '<STR_LIT>' : self . _get_flavor_url ( many_core_flavor ) } <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( self . admined_instance , action = '<STR_LIT>' ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST , response . data ) <EOL> def test_user_cannot_modify_instance_connected_to_failing_cloud_project_membership ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> data = { <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> } <EOL> self . admined_instance . cloud_project_membership . set_erred ( ) <EOL> self . admined_instance . cloud_project_membership . save ( ) <EOL> response = self . client . patch ( factories . InstanceFactory . get_url ( self . admined_instance ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_409_CONFLICT ) <EOL> def test_user_cannot_change_flavor_to_flavor_from_different_cloud ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> instance = self . admined_instance <EOL> new_flavor = factories . FlavorFactory ( disk = self . admined_instance . system_volume_size + <NUM_LIT:1> ) <EOL> CloudProjectMembership . objects . create ( <EOL> project = instance . cloud_project_membership . project , <EOL> cloud = new_flavor . cloud , <EOL> ) <EOL> data = { '<STR_LIT>' : self . _get_flavor_url ( new_flavor ) } <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( instance , action = '<STR_LIT>' ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> self . assertDictContainsSubset ( { '<STR_LIT>' : '<STR_LIT>' } , <EOL> response . data ) <EOL> reread_instance = Instance . objects . get ( pk = instance . pk ) <EOL> self . assertEqual ( reread_instance . system_volume_size , instance . system_volume_size , <EOL> '<STR_LIT>' ) <EOL> def test_user_cannot_set_disk_size_greater_than_resource_quota ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> instance = self . admined_instance <EOL> membership = instance . cloud_project_membership <EOL> data = { <EOL> '<STR_LIT>' : membership . quotas . get ( name = '<STR_LIT>' ) . limit + <NUM_LIT:1> + instance . data_volume_size <EOL> } <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( instance , action = '<STR_LIT>' ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> reread_instance = Instance . objects . get ( pk = instance . pk ) <EOL> self . assertEqual ( reread_instance . data_volume_size , instance . data_volume_size , <EOL> '<STR_LIT>' ) <EOL> def test_user_cannot_change_flavor_of_stopped_instance_he_is_manager_of ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> instance = self . managed_instance <EOL> new_flavor = factories . FlavorFactory ( <EOL> cloud = instance . cloud_project_membership . cloud , <EOL> disk = self . admined_instance . system_volume_size + <NUM_LIT:1> , <EOL> ) <EOL> data = { '<STR_LIT>' : self . _get_flavor_url ( new_flavor ) } <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( instance , action = '<STR_LIT>' ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_403_FORBIDDEN ) <EOL> reread_instance = Instance . objects . get ( pk = instance . pk ) <EOL> self . assertEqual ( reread_instance . system_volume_size , instance . system_volume_size , <EOL> '<STR_LIT>' ) <EOL> def test_user_cannot_change_flavor_of_instance_he_has_no_role_in ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> inaccessible_instance = factories . InstanceFactory ( ) <EOL> new_flavor = factories . FlavorFactory ( <EOL> cloud = inaccessible_instance . cloud_project_membership . cloud , <EOL> disk = self . admined_instance . system_volume_size + <NUM_LIT:1> , <EOL> ) <EOL> data = { '<STR_LIT>' : self . _get_flavor_url ( new_flavor ) } <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( inaccessible_instance , action = '<STR_LIT>' ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_404_NOT_FOUND ) <EOL> reread_instance = Instance . objects . get ( pk = inaccessible_instance . pk ) <EOL> self . assertEqual ( reread_instance . system_volume_size , inaccessible_instance . system_volume_size , <EOL> '<STR_LIT>' ) <EOL> def test_user_cannot_modify_instance_in_provisioning_scheduled_state ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> instance = factories . InstanceFactory ( state = Instance . States . PROVISIONING_SCHEDULED ) <EOL> project = instance . cloud_project_membership . project <EOL> project . add_user ( self . user , ProjectRole . ADMINISTRATOR ) <EOL> url = factories . InstanceFactory . get_url ( instance ) <EOL> for action in '<STR_LIT>' , '<STR_LIT:start>' , '<STR_LIT>' : <EOL> response = self . client . post ( '<STR_LIT>' % ( url , action ) , { } ) <EOL> self . assertEqual ( response . status_code , status . HTTP_409_CONFLICT ) <EOL> response = self . client . put ( url , { } ) <EOL> self . assertEqual ( response . status_code , status . HTTP_409_CONFLICT ) <EOL> response = self . client . patch ( url , { } ) <EOL> self . assertEqual ( response . status_code , status . HTTP_409_CONFLICT ) <EOL> response = self . client . delete ( url ) <EOL> self . assertEqual ( response . status_code , status . HTTP_409_CONFLICT ) <EOL> def test_user_cannot_modify_in_unstable_state ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> for state in Instance . States . UNSTABLE_STATES : <EOL> instance = factories . InstanceFactory ( state = state ) <EOL> project = instance . cloud_project_membership . project <EOL> project . add_user ( self . user , ProjectRole . ADMINISTRATOR ) <EOL> url = factories . InstanceFactory . get_url ( instance ) <EOL> for method in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> func = getattr ( self . client , method . lower ( ) ) <EOL> response = func ( url ) <EOL> self . assertEqual ( response . status_code , status . HTTP_409_CONFLICT ) <EOL> def test_user_cannot_change_flavor_of_non_offline_instance ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> forbidden_states = [ <EOL> state <EOL> for ( state , _ ) in Instance . States . CHOICES <EOL> if state not in ( Instance . States . DELETING , Instance . States . OFFLINE ) <EOL> ] <EOL> for state in forbidden_states : <EOL> instance = factories . InstanceFactory ( state = state ) <EOL> membership = instance . cloud_project_membership <EOL> membership . project . add_user ( self . user , ProjectRole . ADMINISTRATOR ) <EOL> changed_flavor = factories . FlavorFactory ( cloud = membership . cloud ) <EOL> data = { '<STR_LIT>' : self . _get_flavor_url ( changed_flavor ) } <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( instance ) + '<STR_LIT>' , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_409_CONFLICT ) <EOL> reread_instance = Instance . objects . get ( pk = instance . pk ) <EOL> self . assertEqual ( reread_instance . system_volume_size , instance . system_volume_size , <EOL> '<STR_LIT>' ) <EOL> def test_user_cannot_change_flavor_of_running_instance_he_is_manager_of ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> forbidden_states = [ <EOL> state <EOL> for ( state , _ ) in Instance . States . CHOICES <EOL> ] <EOL> for state in forbidden_states : <EOL> managed_instance = factories . InstanceFactory ( state = state ) <EOL> membership = managed_instance . cloud_project_membership <EOL> membership . project . add_user ( self . user , ProjectRole . MANAGER ) <EOL> new_flavor = factories . FlavorFactory ( cloud = membership . cloud ) <EOL> data = { '<STR_LIT>' : self . _get_flavor_url ( new_flavor ) } <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( managed_instance , action = '<STR_LIT>' ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_403_FORBIDDEN ) <EOL> def test_user_cannot_change_flavor_and_disk_size_simultaneously ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> instance = factories . InstanceFactory ( state = Instance . States . OFFLINE ) <EOL> cloud = instance . cloud_project_membership . cloud <EOL> project = instance . cloud_project_membership . project <EOL> project . add_user ( self . user , ProjectRole . MANAGER ) <EOL> project . add_user ( self . user , ProjectRole . ADMINISTRATOR ) <EOL> new_flavor = factories . FlavorFactory ( cloud = cloud ) <EOL> data = { <EOL> '<STR_LIT>' : self . _get_flavor_url ( new_flavor ) , <EOL> '<STR_LIT>' : <NUM_LIT:100> , <EOL> } <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( instance , action = '<STR_LIT>' ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> self . assertDictContainsSubset ( <EOL> { '<STR_LIT>' : [ '<STR_LIT>' ] } , response . data ) <EOL> def test_user_cannot_resize_with_empty_parameters ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> instance = factories . InstanceFactory ( state = Instance . States . OFFLINE ) <EOL> project = instance . cloud_project_membership . project <EOL> project . add_user ( self . user , ProjectRole . MANAGER ) <EOL> project . add_user ( self . user , ProjectRole . ADMINISTRATOR ) <EOL> data = { } <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( instance , action = '<STR_LIT>' ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> self . assertDictContainsSubset ( <EOL> { '<STR_LIT>' : [ '<STR_LIT>' ] } , response . data ) <EOL> def test_user_can_resize_disk_of_flavor_of_instance_he_is_administrator_of ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> instance = self . admined_instance <EOL> instance . cloud_project_membership . project . add_user ( self . user , ProjectRole . ADMINISTRATOR ) <EOL> new_size = instance . data_volume_size + <NUM_LIT> <EOL> data = { '<STR_LIT>' : new_size } <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( instance , action = '<STR_LIT>' ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_202_ACCEPTED ) <EOL> reread_instance = Instance . objects . get ( pk = instance . pk ) <EOL> self . assertEqual ( reread_instance . data_volume_size , new_size ) <EOL> def test_user_cannot_resize_disk_of_flavor_of_instance_he_is_manager_of ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> managed_instance = factories . InstanceFactory ( ) <EOL> managed_instance . cloud_project_membership . project . add_user ( self . user , ProjectRole . MANAGER ) <EOL> self . _ensure_cannot_resize_disk_of_flavor ( managed_instance , status . HTTP_403_FORBIDDEN ) <EOL> def test_user_cannot_resize_disk_of_flavor_of_instance_he_has_no_role_in ( self ) : <EOL> self . client . force_authenticate ( user = self . user ) <EOL> inaccessible_instance = factories . InstanceFactory ( ) <EOL> self . _ensure_cannot_resize_disk_of_flavor ( inaccessible_instance , status . HTTP_404_NOT_FOUND ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_user_cannot_assign_floating_ip_to_instance_in_unstable_state ( self , mocked_task ) : <EOL> self . client . force_authenticate ( user = self . staff ) <EOL> cpm = factories . CloudProjectMembershipFactory ( external_network_id = '<STR_LIT>' , state = SynchronizationStates . IN_SYNC ) <EOL> floating_ip = factories . FloatingIPFactory ( cloud_project_membership = cpm , <EOL> backend_network_id = cpm . external_network_id , <EOL> status = '<STR_LIT>' ) <EOL> instance = factories . InstanceFactory ( state = Instance . States . ERRED , cloud_project_membership = cpm ) <EOL> data = { '<STR_LIT>' : floating_ip . uuid . hex } <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( instance , action = '<STR_LIT>' ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_409_CONFLICT ) <EOL> self . assertEqual ( response . data [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertFalse ( mocked_task . delay . called ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_user_cannot_assign_floating_ip_to_instance_with_cpm_without_external_network_id ( self , mocked_task ) : <EOL> self . client . force_authenticate ( user = self . staff ) <EOL> cpm = factories . CloudProjectMembershipFactory ( state = SynchronizationStates . IN_SYNC ) <EOL> floating_ip = factories . FloatingIPFactory ( cloud_project_membership = cpm , status = '<STR_LIT>' ) <EOL> instance = factories . InstanceFactory ( state = Instance . States . OFFLINE , cloud_project_membership = cpm ) <EOL> data = { '<STR_LIT>' : floating_ip . uuid . hex } <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( instance , action = '<STR_LIT>' ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_409_CONFLICT ) <EOL> self . assertEqual ( response . data [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertFalse ( mocked_task . delay . called ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_user_cannot_assign_floating_ip_to_instance_with_cpm_in_unstable_state ( self , mocked_task ) : <EOL> self . client . force_authenticate ( user = self . staff ) <EOL> cpm = factories . CloudProjectMembershipFactory ( external_network_id = '<STR_LIT>' , state = SynchronizationStates . ERRED ) <EOL> floating_ip = factories . FloatingIPFactory ( cloud_project_membership = cpm , <EOL> backend_network_id = cpm . external_network_id , <EOL> status = '<STR_LIT>' ) <EOL> instance = factories . InstanceFactory ( state = Instance . States . OFFLINE , cloud_project_membership = cpm ) <EOL> data = { '<STR_LIT>' : floating_ip . uuid . hex } <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( instance , action = '<STR_LIT>' ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_409_CONFLICT ) <EOL> self . assertEqual ( response . data [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertFalse ( mocked_task . delay . called ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_user_cannot_assign_not_existing_ip_to_the_instance ( self , mocked_task ) : <EOL> self . client . force_authenticate ( user = self . staff ) <EOL> data = { '<STR_LIT>' : '<STR_LIT>' } <EOL> instance = factories . InstanceFactory ( state = Instance . States . OFFLINE ) <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( instance , action = '<STR_LIT>' ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> self . assertEqual ( response . data [ '<STR_LIT>' ] , [ '<STR_LIT>' ] ) <EOL> self . assertFalse ( mocked_task . delay . called ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_user_cannot_assign_used_ip_to_the_instance ( self , mocked_task ) : <EOL> self . client . force_authenticate ( user = self . staff ) <EOL> cpm = factories . CloudProjectMembershipFactory ( external_network_id = '<STR_LIT>' , state = SynchronizationStates . IN_SYNC ) <EOL> floating_ip = factories . FloatingIPFactory ( cloud_project_membership = cpm , <EOL> backend_network_id = cpm . external_network_id , <EOL> status = '<STR_LIT>' ) <EOL> instance = factories . InstanceFactory ( state = Instance . States . OFFLINE , cloud_project_membership = cpm ) <EOL> data = { '<STR_LIT>' : floating_ip . uuid . hex } <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( instance , action = '<STR_LIT>' ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> self . assertEqual ( response . data [ '<STR_LIT>' ] , [ '<STR_LIT>' ] ) <EOL> self . assertFalse ( mocked_task . delay . called ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_user_cannot_assign_ip_from_different_cpm_to_the_instance ( self , mocked_task ) : <EOL> self . client . force_authenticate ( user = self . staff ) <EOL> cpm = factories . CloudProjectMembershipFactory ( external_network_id = '<STR_LIT>' , state = SynchronizationStates . IN_SYNC ) <EOL> floating_ip = factories . FloatingIPFactory ( status = '<STR_LIT>' ) <EOL> instance = factories . InstanceFactory ( state = Instance . States . OFFLINE , cloud_project_membership = cpm ) <EOL> data = { '<STR_LIT>' : floating_ip . uuid . hex } <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( instance , action = '<STR_LIT>' ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> self . assertEqual ( response . data [ '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' ] ) <EOL> self . assertFalse ( mocked_task . delay . called ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_user_can_assign_floating_ip_to_instance_with_satisfied_requirements ( self , mocked_task ) : <EOL> self . client . force_authenticate ( user = self . staff ) <EOL> cpm = factories . CloudProjectMembershipFactory ( external_network_id = '<STR_LIT>' , state = SynchronizationStates . IN_SYNC ) <EOL> floating_ip = factories . FloatingIPFactory ( cloud_project_membership = cpm , <EOL> backend_network_id = cpm . external_network_id , <EOL> status = '<STR_LIT>' ) <EOL> instance = factories . InstanceFactory ( state = Instance . States . OFFLINE , cloud_project_membership = cpm ) <EOL> data = { '<STR_LIT>' : floating_ip . uuid . hex } <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( instance , action = '<STR_LIT>' ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_202_ACCEPTED ) <EOL> self . assertEqual ( response . data [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertTrue ( mocked_task . delay . called ) <EOL> def _get_valid_payload ( self , resource = None ) : <EOL> resource = resource or factories . InstanceFactory ( ) <EOL> ssh_public_key = structure_factories . SshPublicKeyFactory ( user = self . user ) <EOL> membership = resource . cloud_project_membership <EOL> flavor = factories . FlavorFactory ( cloud = membership . cloud ) <EOL> return { <EOL> '<STR_LIT:name>' : resource . name , <EOL> '<STR_LIT:description>' : resource . description , <EOL> '<STR_LIT>' : self . _get_project_url ( membership . project ) , <EOL> '<STR_LIT>' : self . _get_template_url ( resource . template ) , <EOL> '<STR_LIT>' : self . _get_flavor_url ( flavor ) , <EOL> '<STR_LIT>' : self . _get_ssh_public_key_url ( ssh_public_key ) <EOL> } <EOL> def _ensure_cannot_resize_disk_of_flavor ( self , instance , expected_status ) : <EOL> data = { '<STR_LIT>' : <NUM_LIT> } <EOL> response = self . client . post ( factories . InstanceFactory . get_url ( instance , action = '<STR_LIT>' ) , data ) <EOL> self . assertEqual ( response . status_code , expected_status ) <EOL> reread_instance = Instance . objects . get ( uuid = instance . uuid ) <EOL> self . assertNotEqual ( reread_instance . system_volume_size , data [ '<STR_LIT>' ] ) <EOL> class InstanceProvisioningTest ( UrlResolverMixin , test . APITransactionTestCase ) : <EOL> def setUp ( self ) : <EOL> self . user = structure_factories . UserFactory . create ( ) <EOL> self . client . force_authenticate ( user = self . user ) <EOL> self . instance_list_url = reverse ( '<STR_LIT>' ) <EOL> self . cloud = factories . CloudFactory ( ) <EOL> self . template = factories . TemplateFactory ( ) <EOL> self . flavor = factories . FlavorFactory ( cloud = self . cloud ) <EOL> self . project = structure_factories . ProjectFactory ( ) <EOL> self . membership = factories . CloudProjectMembershipFactory ( cloud = self . cloud , project = self . project ) <EOL> self . ssh_public_key = structure_factories . SshPublicKeyFactory ( user = self . user ) <EOL> self . project . add_user ( self . user , ProjectRole . ADMINISTRATOR ) <EOL> factories . ImageFactory ( template = self . template , cloud = self . cloud ) <EOL> def assert_field_required ( self , field_name ) : <EOL> data = self . get_valid_data ( ) <EOL> del data [ field_name ] <EOL> response = self . client . post ( self . instance_list_url , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> self . assertDictContainsSubset ( { field_name : [ '<STR_LIT>' ] } , response . data ) <EOL> def assert_field_non_empty ( self , field_name , empty_value = '<STR_LIT>' ) : <EOL> data = self . get_valid_data ( ) <EOL> data [ field_name ] = empty_value <EOL> response = self . client . post ( self . instance_list_url , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> self . assertIn ( field_name , response . data ) <EOL> def test_can_create_instance_without_description ( self ) : <EOL> data = self . get_valid_data ( ) <EOL> del data [ '<STR_LIT:description>' ] <EOL> response = self . client . post ( self . instance_list_url , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_201_CREATED ) <EOL> def test_created_instance_has_agreed_sla_from_template ( self ) : <EOL> data = self . get_valid_data ( ) <EOL> response = self . client . post ( factories . InstanceFactory . get_list_url ( ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_201_CREATED , '<STR_LIT>' % response . data ) <EOL> sla_level = self . template . sla_level <EOL> created_instance = self . client . get ( factories . InstanceFactory . get_list_url ( ) + response . data [ '<STR_LIT>' ] + '<STR_LIT:/>' ) <EOL> self . assertEqual ( sla_level , Decimal ( created_instance . data [ '<STR_LIT>' ] ) ) <EOL> def test_can_create_instance_with_empty_description ( self ) : <EOL> data = self . get_valid_data ( ) <EOL> data [ '<STR_LIT:description>' ] = '<STR_LIT>' <EOL> response = self . client . post ( self . instance_list_url , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_201_CREATED ) <EOL> def test_can_create_instance_without_ssh_public_key ( self ) : <EOL> data = self . get_valid_data ( ) <EOL> del data [ '<STR_LIT>' ] <EOL> response = self . client . post ( self . instance_list_url , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_201_CREATED ) <EOL> def test_cannot_create_instance_with_flavor_not_from_supplied_project ( self ) : <EOL> data = self . get_valid_data ( ) <EOL> another_flavor = factories . FlavorFactory ( ) <EOL> another_project = structure_factories . ProjectFactory ( ) <EOL> factories . CloudProjectMembershipFactory ( project = another_project , cloud = another_flavor . cloud ) <EOL> another_project . add_user ( self . user , ProjectRole . ADMINISTRATOR ) <EOL> data [ '<STR_LIT>' ] = self . _get_flavor_url ( another_flavor ) <EOL> response = self . client . post ( self . instance_list_url , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> self . assertDictContainsSubset ( { '<STR_LIT>' : "<STR_LIT>" } , response . data ) <EOL> def test_cannot_create_instance_with_flavor_not_from_clouds_allowed_for_users_projects ( self ) : <EOL> data = self . get_valid_data ( ) <EOL> others_flavor = factories . FlavorFactory ( ) <EOL> data [ '<STR_LIT>' ] = self . _get_flavor_url ( others_flavor ) <EOL> response = self . client . post ( self . instance_list_url , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> self . assertDictContainsSubset ( { '<STR_LIT>' : [ '<STR_LIT>' ] } , response . data ) <EOL> def test_cannot_create_instance_with_project_not_from_users_projects ( self ) : <EOL> data = self . get_valid_data ( ) <EOL> others_project = structure_factories . ProjectFactory ( ) <EOL> data [ '<STR_LIT>' ] = self . _get_project_url ( others_project ) <EOL> response = self . client . post ( self . instance_list_url , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> self . assertDictContainsSubset ( { '<STR_LIT>' : [ '<STR_LIT>' ] } , response . data ) <EOL> def test_cannot_create_instance_with_empty_name ( self ) : <EOL> self . assert_field_non_empty ( '<STR_LIT:name>' ) <EOL> def test_cannot_create_instance_without_name ( self ) : <EOL> self . assert_field_required ( '<STR_LIT:name>' ) <EOL> def test_cannot_create_instance_with_empty_template_name ( self ) : <EOL> self . assert_field_non_empty ( '<STR_LIT>' ) <EOL> def test_cannot_create_instance_without_template_name ( self ) : <EOL> self . assert_field_required ( '<STR_LIT>' ) <EOL> def test_cannot_create_instance_without_flavor ( self ) : <EOL> self . assert_field_required ( '<STR_LIT>' ) <EOL> def test_cannot_create_instance_with_empty_flavor ( self ) : <EOL> self . assert_field_non_empty ( '<STR_LIT>' ) <EOL> def test_cannot_create_instance_with_data_volume_size_lower_then_one_gb ( self ) : <EOL> data = self . get_valid_data ( ) <EOL> data [ '<STR_LIT>' ] = <NUM_LIT> <EOL> response = self . client . post ( self . instance_list_url , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> self . assertIn ( '<STR_LIT>' , response . data ) <EOL> def test_cannot_create_instance_with_system_volume_size_lower_image_disk_image ( self ) : <EOL> image = self . template . images . first ( ) <EOL> image . min_disk = <NUM_LIT:10> * <NUM_LIT> <EOL> image . save ( ) <EOL> data = self . get_valid_data ( ) <EOL> data [ '<STR_LIT>' ] = <NUM_LIT:5> * <NUM_LIT> <EOL> response = self . client . post ( self . instance_list_url , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> def test_cannot_create_instance_with_security_groups_from_other_project ( self ) : <EOL> security_groups = [ factories . SecurityGroupFactory ( ) for _ in range ( <NUM_LIT:3> ) ] <EOL> data = self . get_valid_data ( ) <EOL> data [ '<STR_LIT>' ] = [ { '<STR_LIT:url>' : factories . SecurityGroupFactory . get_url ( sg ) } for sg in security_groups ] <EOL> response = self . client . post ( self . instance_list_url , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> def test_instance_factory_generates_valid_internal_ips_field ( self ) : <EOL> instance = factories . InstanceFactory ( ) <EOL> internal_ips = instance . internal_ips <EOL> self . assertTrue ( ips_regex . match ( internal_ips ) ) <EOL> def test_instance_factory_generates_valid_external_ips_field ( self ) : <EOL> instance = factories . InstanceFactory ( ) <EOL> external_ips = instance . internal_ips <EOL> self . assertTrue ( ips_regex . match ( external_ips ) ) <EOL> def test_can_create_instance_with_external_ips_from_floating_pool_set ( self ) : <EOL> data = self . get_valid_data ( ) <EOL> address = '<STR_LIT:127.0.0.1>' <EOL> data [ '<STR_LIT>' ] = [ address ] <EOL> FloatingIP . objects . create ( status = '<STR_LIT>' , cloud_project_membership = self . membership , address = address ) <EOL> response = self . client . post ( self . instance_list_url , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_201_CREATED , '<STR_LIT>' % response . data ) <EOL> def test_assigning_floating_ip_on_provisioning_marks_it_as_booked ( self ) : <EOL> data = self . get_valid_data ( ) <EOL> address = '<STR_LIT:127.0.0.1>' <EOL> data [ '<STR_LIT>' ] = [ address ] <EOL> FloatingIP . objects . create ( status = '<STR_LIT>' , cloud_project_membership = self . membership , address = address ) <EOL> response = self . client . post ( self . instance_list_url , data ) <EOL> floating_ip_booked = FloatingIP . objects . filter ( status = '<STR_LIT>' , <EOL> cloud_project_membership = self . membership , <EOL> address = address ) . exists ( ) <EOL> self . assertEqual ( response . status_code , status . HTTP_201_CREATED , '<STR_LIT>' % response . data ) <EOL> self . assertTrue ( floating_ip_booked ) <EOL> def test_can_create_instance_with_defined_volume_size ( self ) : <EOL> data = self . get_valid_data ( ) <EOL> data [ '<STR_LIT>' ] = <NUM_LIT:2> * <NUM_LIT> <EOL> data [ '<STR_LIT>' ] = <NUM_LIT:10> * <NUM_LIT> <EOL> response = self . client . post ( self . instance_list_url , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_201_CREATED ) <EOL> self . assertTrue ( Instance . objects . filter ( <EOL> data_volume_size = data [ '<STR_LIT>' ] , system_volume_size = data [ '<STR_LIT>' ] ) . exists ( ) ) <EOL> def test_can_create_instance_with_external_ips_set_to_empty_string ( self ) : <EOL> data = self . get_valid_data ( ) <EOL> data [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> response = self . client . post ( self . instance_list_url , data ) <EOL> self . assertEqual ( <EOL> response . status_code , status . HTTP_201_CREATED , <EOL> '<STR_LIT>' % ( response . status_code , response . data ) <EOL> ) <EOL> def test_can_create_instance_with_external_ips_set_to_null ( self ) : <EOL> data = self . get_valid_data ( ) <EOL> data [ '<STR_LIT>' ] = None <EOL> response = self . client . post ( self . instance_list_url , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_201_CREATED ) <EOL> def test_can_create_instance_with_external_ips_missing ( self ) : <EOL> data = self . get_valid_data ( ) <EOL> try : <EOL> del data [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> pass <EOL> response = self . client . post ( self . instance_list_url , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_201_CREATED ) <EOL> def test_instance_api_contains_valid_external_ips_field ( self ) : <EOL> instance = factories . InstanceFactory ( ) <EOL> instance . cloud_project_membership . project . add_user ( self . user , ProjectRole . ADMINISTRATOR ) <EOL> response = self . client . get ( factories . InstanceFactory . get_url ( instance ) ) <EOL> external_ips = response . data [ '<STR_LIT>' ] <EOL> for ip in external_ips : <EOL> self . assertTrue ( ips_regex . match ( ip ) ) <EOL> def test_paas_inctance_cannot_be_created_if_memebership_external_network_id_is_empty ( self ) : <EOL> data = self . get_valid_data ( ) <EOL> data [ '<STR_LIT:type>' ] = Instance . Services . PAAS <EOL> membership = CloudProjectMembership . objects . get ( project = self . project , cloud = self . flavor . cloud ) <EOL> membership . external_network_id = '<STR_LIT>' <EOL> membership . save ( ) <EOL> response = self . client . post ( self . instance_list_url , data = data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> def test_instance_api_contains_valid_internal_ips_field ( self ) : <EOL> instance = factories . InstanceFactory ( ) <EOL> instance . cloud_project_membership . project . add_user ( self . user , ProjectRole . ADMINISTRATOR ) <EOL> response = self . client . get ( factories . InstanceFactory . get_url ( instance ) ) <EOL> internal_ips = response . data [ '<STR_LIT>' ] <EOL> for ip in internal_ips : <EOL> self . assertTrue ( ips_regex . match ( ip ) ) <EOL> def test_instance_licenses_added_on_instance_creation ( self ) : <EOL> template_license = factories . TemplateLicenseFactory ( ) <EOL> self . template . template_licenses . add ( template_license ) <EOL> response = self . client . post ( self . instance_list_url , self . get_valid_data ( ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_201_CREATED ) <EOL> self . assertEqual ( template_license . instance_licenses . count ( ) , <NUM_LIT:1> ) <EOL> def test_instance_licenses_exist_in_instance_retrieve_request ( self ) : <EOL> instance = factories . InstanceFactory ( ) <EOL> instance . cloud_project_membership . project . add_user ( self . user , ProjectRole . ADMINISTRATOR ) <EOL> instance_license = factories . InstanceLicenseFactory ( instance = instance ) <EOL> response = self . client . get ( factories . InstanceFactory . get_url ( instance ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> self . assertIn ( '<STR_LIT>' , response . data ) <EOL> self . assertEqual ( response . data [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:name>' ] , instance_license . template_license . name ) <EOL> def test_flavor_fields_is_copied_to_instance_on_instance_creation ( self ) : <EOL> response = self . client . post ( self . instance_list_url , self . get_valid_data ( ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_201_CREATED ) <EOL> instance = Instance . objects . get ( uuid = response . data [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( instance . system_volume_size , self . flavor . disk ) <EOL> self . assertEqual ( instance . ram , self . flavor . ram ) <EOL> self . assertEqual ( instance . cores , self . flavor . cores ) <EOL> def test_ssh_public_key_is_copied_to_instance_on_instance_creation ( self ) : <EOL> response = self . client . post ( self . instance_list_url , self . get_valid_data ( ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_201_CREATED ) <EOL> instance = Instance . objects . get ( uuid = response . data [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( instance . key_name , self . ssh_public_key . name ) <EOL> self . assertEqual ( instance . key_fingerprint , self . ssh_public_key . fingerprint ) <EOL> def test_instance_size_can_not_be_bigger_than_quota ( self ) : <EOL> data = self . get_valid_data ( ) <EOL> data [ '<STR_LIT>' ] = self . membership . quotas . get ( name = '<STR_LIT>' ) . limit + <NUM_LIT:1> <EOL> response = self . client . post ( self . instance_list_url , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> def test_user_cannot_create_instance_with_template_not_connected_to_projects_cloud ( self ) : <EOL> templates = { <EOL> '<STR_LIT>' : factories . ImageFactory ( ) . template , <EOL> '<STR_LIT>' : factories . TemplateFactory ( ) , <EOL> } <EOL> data = self . get_valid_data ( ) <EOL> for t in templates : <EOL> data [ '<STR_LIT>' ] = self . _get_template_url ( templates [ t ] ) <EOL> response = self . client . post ( self . instance_list_url , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> self . assertDictContainsSubset ( { '<STR_LIT>' : [ '<STR_LIT>' ] } , response . data ) <EOL> def test_external_ips_have_to_from_membership_floating_ips ( self ) : <EOL> random_address = '<STR_LIT:127.0.0.1>' <EOL> data = self . get_valid_data ( ) <EOL> data [ '<STR_LIT>' ] = [ random_address ] <EOL> response = self . client . post ( self . instance_list_url , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> self . assertDictContainsSubset ( <EOL> { '<STR_LIT>' : [ '<STR_LIT>' ] } , response . data ) <EOL> def test_instance_can_not_be_created_if_customer_resource_quota_exceeded ( self ) : <EOL> self . project . customer . set_quota_limit ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> data = self . get_valid_data ( ) <EOL> response = self . client . post ( factories . InstanceFactory . get_list_url ( ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_409_CONFLICT ) <EOL> def test_zabbix_host_visible_name_is_updated_when_instance_is_renamed ( self ) : <EOL> instance = factories . InstanceFactory ( state = Instance . States . OFFLINE ) <EOL> instance . cloud_project_membership . project . add_user ( self . user , ProjectRole . ADMINISTRATOR ) <EOL> with patch ( '<STR_LIT>' ) as mocked_task : <EOL> data = { '<STR_LIT:name>' : '<STR_LIT>' } <EOL> response = self . client . put ( factories . InstanceFactory . get_url ( instance ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> mocked_task . assert_called_with ( instance . uuid . hex ) <EOL> def test_zabbix_host_visible_name_is_not_updated_when_instance_is_not_renamed ( self ) : <EOL> instance = factories . InstanceFactory ( state = Instance . States . OFFLINE ) <EOL> instance . cloud_project_membership . project . add_user ( self . user , ProjectRole . ADMINISTRATOR ) <EOL> with patch ( '<STR_LIT>' ) as mocked_task : <EOL> data = { '<STR_LIT:name>' : instance . name } <EOL> response = self . client . put ( factories . InstanceFactory . get_url ( instance ) , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> self . assertFalse ( mocked_task . called ) <EOL> def get_valid_data ( self ) : <EOL> return { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . _get_project_url ( self . project ) , <EOL> '<STR_LIT>' : self . _get_template_url ( self . template ) , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : self . _get_flavor_url ( self . flavor ) , <EOL> '<STR_LIT>' : self . _get_ssh_public_key_url ( self . ssh_public_key ) <EOL> } <EOL> def test_quotas_increase_on_instance_creation ( self ) : <EOL> data = self . get_valid_data ( ) <EOL> response = self . client . post ( self . instance_list_url , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_201_CREATED ) <EOL> instance = Instance . objects . get ( uuid = response . data [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( self . membership . quotas . get ( name = '<STR_LIT>' ) . usage , <NUM_LIT:1> ) <EOL> self . assertEqual ( self . membership . quotas . get ( name = '<STR_LIT>' ) . usage , instance . cores ) <EOL> self . assertEqual ( self . membership . quotas . get ( name = '<STR_LIT>' ) . usage , instance . ram ) <EOL> self . assertEqual ( <EOL> self . membership . quotas . get ( name = '<STR_LIT>' ) . usage , instance . data_volume_size + instance . system_volume_size ) <EOL> class InstanceListRetrieveTest ( test . APITransactionTestCase ) : <EOL> def setUp ( self ) : <EOL> self . staff = structure_factories . UserFactory ( is_staff = True ) <EOL> self . instance = factories . InstanceFactory ( ) <EOL> def test_user_does_not_receive_deleted_instances_backups ( self ) : <EOL> self . client . force_authenticate ( self . staff ) <EOL> backup_factories . BackupFactory ( state = backup_models . Backup . States . DELETED , backup_source = self . instance ) <EOL> backup = backup_factories . BackupFactory ( backup_source = self . instance ) <EOL> url = factories . InstanceFactory . get_url ( self . instance ) <EOL> response = self . client . get ( url ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> self . assertEqual ( len ( response . data [ '<STR_LIT>' ] ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( response . data [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:url>' ] , backup_factories . BackupFactory . get_url ( backup ) ) <EOL> def test_ascending_sort_by_start_time_puts_instances_with_null_value_first ( self ) : <EOL> self . client . force_authenticate ( self . staff ) <EOL> factories . InstanceFactory . create_batch ( <NUM_LIT:2> , start_time = None ) <EOL> factories . InstanceFactory ( ) <EOL> response = self . client . get ( factories . InstanceFactory . get_list_url ( ) , <EOL> data = { '<STR_LIT:o>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> for i in ( <NUM_LIT:0> , <NUM_LIT:1> ) : <EOL> self . assertEqual ( response . data [ i ] [ '<STR_LIT>' ] , None ) <EOL> self . assertTrue ( response . data [ <NUM_LIT:2> ] [ '<STR_LIT>' ] < response . data [ <NUM_LIT:3> ] [ '<STR_LIT>' ] ) <EOL> def test_descending_sort_by_start_time_puts_instances_with_null_value_last ( self ) : <EOL> self . client . force_authenticate ( self . staff ) <EOL> factories . InstanceFactory . create_batch ( <NUM_LIT:2> , start_time = None ) <EOL> factories . InstanceFactory ( ) <EOL> response = self . client . get ( factories . InstanceFactory . get_list_url ( ) , <EOL> data = { '<STR_LIT:o>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> self . assertTrue ( response . data [ <NUM_LIT:0> ] [ '<STR_LIT>' ] > response . data [ <NUM_LIT:1> ] [ '<STR_LIT>' ] ) <EOL> for i in ( <NUM_LIT:2> , <NUM_LIT:3> ) : <EOL> self . assertEqual ( response . data [ i ] [ '<STR_LIT>' ] , None ) <EOL> class InstanceUsageTest ( test . APITransactionTestCase ) : <EOL> def setUp ( self ) : <EOL> self . staff = structure_factories . UserFactory ( is_staff = True ) <EOL> self . instance = factories . InstanceFactory ( state = Instance . States . OFFLINE ) <EOL> self . url = factories . InstanceFactory . get_url ( self . instance , action = '<STR_LIT>' ) <EOL> def test_instance_does_not_have_backend_id ( self ) : <EOL> self . client . force_authenticate ( self . staff ) <EOL> instance = factories . InstanceFactory ( backend_id = '<STR_LIT>' ) <EOL> url = factories . InstanceFactory . get_url ( instance , action = '<STR_LIT>' ) <EOL> response = self . client . get ( url ) <EOL> self . assertEqual ( response . status_code , status . HTTP_404_NOT_FOUND ) <EOL> def test_instance_installation_state_is_faile_instance_is_not_online ( self ) : <EOL> instance = factories . InstanceFactory ( installation_state = '<STR_LIT:OK>' , state = Instance . States . OFFLINE ) <EOL> self . client . force_authenticate ( self . staff ) <EOL> response = self . client . get ( factories . InstanceFactory . get_url ( instance ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> self . assertEqual ( response . data [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_instance_in_provisioning_scheduled_state ( self ) : <EOL> self . client . force_authenticate ( self . staff ) <EOL> instance = factories . InstanceFactory ( state = Instance . States . PROVISIONING_SCHEDULED ) <EOL> url = factories . InstanceFactory . get_url ( instance , action = '<STR_LIT>' ) <EOL> response = self . client . get ( url ) <EOL> self . assertEqual ( response . status_code , status . HTTP_404_NOT_FOUND ) <EOL> def test_instance_in_provisioning_state ( self ) : <EOL> self . client . force_authenticate ( self . staff ) <EOL> instance = factories . InstanceFactory ( state = Instance . States . PROVISIONING ) <EOL> url = factories . InstanceFactory . get_url ( instance , action = '<STR_LIT>' ) <EOL> response = self . client . get ( url ) <EOL> self . assertEqual ( response . status_code , status . HTTP_404_NOT_FOUND ) <EOL> def test_item_parameter_have_to_be_defined ( self ) : <EOL> self . client . force_authenticate ( self . staff ) <EOL> response = self . client . get ( self . url ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> def test_item_parameter_have_to_be_one_of_zabbix_db_client_items ( self ) : <EOL> self . client . force_authenticate ( self . staff ) <EOL> response = self . client . get ( self . url , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> def test_cpu_usage ( self ) : <EOL> self . client . force_authenticate ( self . staff ) <EOL> patched_cliend = Mock ( ) <EOL> patched_cliend . items = { '<STR_LIT>' : { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT:value>' : '<STR_LIT>' } } <EOL> expected_data = [ <EOL> { '<STR_LIT>' : <NUM_LIT:1> L , '<STR_LIT:to>' : <NUM_LIT> L , '<STR_LIT:value>' : <NUM_LIT:0> } , <EOL> { '<STR_LIT>' : <NUM_LIT> L , '<STR_LIT:to>' : <NUM_LIT> L , '<STR_LIT:value>' : <NUM_LIT:0> } , <EOL> { '<STR_LIT>' : <NUM_LIT> L , '<STR_LIT:to>' : <NUM_LIT> L , '<STR_LIT:value>' : <NUM_LIT> } <EOL> ] <EOL> patched_cliend . get_item_stats = Mock ( return_value = expected_data ) <EOL> with patch ( '<STR_LIT>' , return_value = patched_cliend ) as patched : <EOL> patched . items = { '<STR_LIT>' : { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } } <EOL> data = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:1> L , '<STR_LIT:to>' : <NUM_LIT> L , '<STR_LIT>' : <NUM_LIT:3> } <EOL> response = self . client . get ( self . url , data ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> self . assertEqual ( response . data , expected_data ) <EOL> patched_cliend . get_item_stats . assert_called_once_with ( <EOL> [ self . instance ] , data [ '<STR_LIT>' ] , data [ '<STR_LIT>' ] , data [ '<STR_LIT:to>' ] , data [ '<STR_LIT>' ] ) </s>
