<s> from __future__ import unicode_literals <EOL> from rest_framework import status <EOL> from rest_framework import test <EOL> from nodeconductor . iaas . models import Cloud , OpenStackSettings <EOL> from nodeconductor . iaas . tests import factories <EOL> from nodeconductor . core . models import SynchronizationStates <EOL> from nodeconductor . structure . models import ProjectRole , CustomerRole , ProjectGroupRole <EOL> from nodeconductor . structure . tests import factories as structure_factories <EOL> class CloudPermissionTest ( test . APITransactionTestCase ) : <EOL> def setUp ( self ) : <EOL> self . customers = { <EOL> '<STR_LIT>' : structure_factories . CustomerFactory ( ) , <EOL> '<STR_LIT>' : structure_factories . CustomerFactory ( ) , <EOL> '<STR_LIT>' : structure_factories . CustomerFactory ( ) , <EOL> '<STR_LIT>' : structure_factories . CustomerFactory ( ) , <EOL> } <EOL> self . users = { <EOL> '<STR_LIT>' : structure_factories . UserFactory ( ) , <EOL> '<STR_LIT>' : structure_factories . UserFactory ( ) , <EOL> '<STR_LIT>' : structure_factories . UserFactory ( ) , <EOL> '<STR_LIT>' : structure_factories . UserFactory ( ) , <EOL> '<STR_LIT>' : structure_factories . UserFactory ( ) , <EOL> } <EOL> self . projects = { <EOL> '<STR_LIT>' : structure_factories . ProjectFactory ( customer = self . customers [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : structure_factories . ProjectFactory ( customer = self . customers [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : structure_factories . ProjectFactory ( customer = self . customers [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : structure_factories . ProjectFactory ( <EOL> customer = self . customers [ '<STR_LIT>' ] ) , <EOL> } <EOL> self . clouds = { <EOL> '<STR_LIT>' : factories . CloudFactory ( <EOL> state = SynchronizationStates . IN_SYNC , customer = self . customers [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : factories . CloudFactory ( <EOL> state = SynchronizationStates . IN_SYNC , customer = self . customers [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : factories . CloudFactory ( <EOL> state = SynchronizationStates . IN_SYNC , customer = self . customers [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : factories . CloudFactory ( <EOL> state = SynchronizationStates . IN_SYNC , customer = self . customers [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : factories . CloudFactory ( ) , <EOL> } <EOL> self . customers [ '<STR_LIT>' ] . add_user ( self . users [ '<STR_LIT>' ] , CustomerRole . OWNER ) <EOL> self . projects [ '<STR_LIT>' ] . add_user ( self . users [ '<STR_LIT>' ] , ProjectRole . ADMINISTRATOR ) <EOL> self . projects [ '<STR_LIT>' ] . add_user ( self . users [ '<STR_LIT>' ] , ProjectRole . MANAGER ) <EOL> project_group = structure_factories . ProjectGroupFactory ( ) <EOL> project_group . projects . add ( self . projects [ '<STR_LIT>' ] ) <EOL> project_group . add_user ( self . users [ '<STR_LIT>' ] , ProjectGroupRole . MANAGER ) <EOL> factories . CloudProjectMembershipFactory ( cloud = self . clouds [ '<STR_LIT>' ] , project = self . projects [ '<STR_LIT>' ] ) <EOL> factories . CloudProjectMembershipFactory ( cloud = self . clouds [ '<STR_LIT>' ] , project = self . projects [ '<STR_LIT>' ] ) <EOL> factories . CloudProjectMembershipFactory ( <EOL> cloud = self . clouds [ '<STR_LIT>' ] , project = self . projects [ '<STR_LIT>' ] ) <EOL> OpenStackSettings . objects . update_or_create ( <EOL> auth_url = '<STR_LIT>' , <EOL> defaults = { <EOL> '<STR_LIT:username>' : '<STR_LIT>' , <EOL> '<STR_LIT:password>' : '<STR_LIT:password>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> def test_anonymous_user_cannot_list_clouds ( self ) : <EOL> response = self . client . get ( factories . CloudFactory . get_list_url ( ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_401_UNAUTHORIZED ) <EOL> def test_user_can_list_clouds_of_projects_he_is_administrator_of ( self ) : <EOL> self . client . force_authenticate ( user = self . users [ '<STR_LIT>' ] ) <EOL> response = self . client . get ( factories . CloudFactory . get_list_url ( ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> cloud_url = factories . CloudFactory . get_url ( self . clouds [ '<STR_LIT>' ] ) <EOL> self . assertIn ( cloud_url , [ instance [ '<STR_LIT:url>' ] for instance in response . data ] ) <EOL> def test_user_can_list_clouds_of_projects_he_is_manager_of ( self ) : <EOL> self . client . force_authenticate ( user = self . users [ '<STR_LIT>' ] ) <EOL> response = self . client . get ( factories . CloudFactory . get_list_url ( ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> cloud_url = factories . CloudFactory . get_url ( self . clouds [ '<STR_LIT>' ] ) <EOL> self . assertIn ( cloud_url , [ instance [ '<STR_LIT:url>' ] for instance in response . data ] ) <EOL> def test_user_can_list_clouds_of_projects_he_is_group_manager_of ( self ) : <EOL> self . client . force_authenticate ( user = self . users [ '<STR_LIT>' ] ) <EOL> response = self . client . get ( factories . CloudFactory . get_list_url ( ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> cloud_url = factories . CloudFactory . get_url ( self . clouds [ '<STR_LIT>' ] ) <EOL> self . assertIn ( cloud_url , [ instance [ '<STR_LIT:url>' ] for instance in response . data ] ) <EOL> def test_user_can_list_clouds_of_projects_he_is_customer_owner_of ( self ) : <EOL> self . client . force_authenticate ( user = self . users [ '<STR_LIT>' ] ) <EOL> response = self . client . get ( factories . CloudFactory . get_list_url ( ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> cloud_url = factories . CloudFactory . get_url ( self . clouds [ '<STR_LIT>' ] ) <EOL> self . assertIn ( cloud_url , [ instance [ '<STR_LIT:url>' ] for instance in response . data ] ) <EOL> def test_user_cannot_list_clouds_of_projects_he_has_no_role_in ( self ) : <EOL> self . client . force_authenticate ( user = self . users [ '<STR_LIT>' ] ) <EOL> response = self . client . get ( factories . CloudFactory . get_list_url ( ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> for cloud_type in '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' : <EOL> cloud_url = factories . CloudFactory . get_url ( self . clouds [ cloud_type ] ) <EOL> self . assertNotIn ( <EOL> cloud_url , <EOL> [ instance [ '<STR_LIT:url>' ] for instance in response . data ] , <EOL> '<STR_LIT>' + cloud_type + '<STR_LIT:)>' , <EOL> ) <EOL> def test_anonymous_user_cannot_access_cloud ( self ) : <EOL> for cloud_type in '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' : <EOL> response = self . client . get ( factories . CloudFactory . get_url ( self . clouds [ cloud_type ] ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_401_UNAUTHORIZED ) <EOL> def test_user_can_access_cloud_allowed_for_project_he_is_administrator_of ( self ) : <EOL> self . client . force_authenticate ( user = self . users [ '<STR_LIT>' ] ) <EOL> response = self . client . get ( factories . CloudFactory . get_url ( self . clouds [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> def test_user_can_access_cloud_allowed_for_project_he_is_manager_of ( self ) : <EOL> self . client . force_authenticate ( user = self . users [ '<STR_LIT>' ] ) <EOL> response = self . client . get ( factories . CloudFactory . get_url ( self . clouds [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> def test_user_can_access_cloud_allowed_for_project_he_is_group_manager_of ( self ) : <EOL> self . client . force_authenticate ( user = self . users [ '<STR_LIT>' ] ) <EOL> response = self . client . get ( factories . CloudFactory . get_url ( self . clouds [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> def test_user_can_see_clouds_customer_name ( self ) : <EOL> self . client . force_authenticate ( user = self . users [ '<STR_LIT>' ] ) <EOL> response = self . client . get ( factories . CloudFactory . get_url ( self . clouds [ '<STR_LIT>' ] ) ) <EOL> customer = self . clouds [ '<STR_LIT>' ] . customer <EOL> self . assertIn ( '<STR_LIT>' , response . data ) <EOL> self . assertEqual ( structure_factories . CustomerFactory . get_url ( customer ) , response . data [ '<STR_LIT>' ] ) <EOL> self . assertIn ( '<STR_LIT>' , response . data ) <EOL> self . assertEqual ( customer . name , response . data [ '<STR_LIT>' ] ) <EOL> def test_user_cannot_access_cloud_allowed_for_project_he_has_no_role_in ( self ) : <EOL> self . client . force_authenticate ( user = self . users [ '<STR_LIT>' ] ) <EOL> for cloud_type in '<STR_LIT>' , '<STR_LIT>' : <EOL> response = self . client . get ( factories . CloudFactory . get_url ( self . clouds [ cloud_type ] ) ) <EOL> self . assertEqual ( <EOL> response . status_code , <EOL> status . HTTP_404_NOT_FOUND , <EOL> '<STR_LIT>' + cloud_type + '<STR_LIT:)>' , <EOL> ) <EOL> def test_user_cannot_access_cloud_not_allowed_for_any_project ( self ) : <EOL> for user_role in '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' : <EOL> self . client . force_authenticate ( user = self . users [ user_role ] ) <EOL> response = self . client . get ( factories . CloudFactory . get_url ( self . clouds [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( <EOL> response . status_code , <EOL> status . HTTP_404_NOT_FOUND , <EOL> '<STR_LIT>' + user_role + '<STR_LIT>' , <EOL> ) <EOL> def test_user_can_see_flavors_within_cloud ( self ) : <EOL> for user_role , cloud_type in { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } . iteritems ( ) : <EOL> self . client . force_authenticate ( user = self . users [ user_role ] ) <EOL> seen_flavor = factories . FlavorFactory ( cloud = self . clouds [ cloud_type ] ) <EOL> response = self . client . get ( factories . CloudFactory . get_url ( self . clouds [ cloud_type ] ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> self . assertIn ( <EOL> '<STR_LIT>' , <EOL> response . data , <EOL> '<STR_LIT>' + cloud_type + '<STR_LIT>' , <EOL> ) <EOL> flavor_urls = set ( [ flavor [ '<STR_LIT:url>' ] for flavor in response . data [ '<STR_LIT>' ] ] ) <EOL> self . assertIn ( <EOL> factories . FlavorFactory . get_url ( seen_flavor ) , <EOL> flavor_urls , <EOL> '<STR_LIT>' + user_role + '<STR_LIT>' , <EOL> ) <EOL> def test_user_can_add_cloud_to_the_customer_he_owns ( self ) : <EOL> self . client . force_authenticate ( user = self . users [ '<STR_LIT>' ] ) <EOL> new_cloud = factories . CloudFactory . build ( customer = self . customers [ '<STR_LIT>' ] ) <EOL> response = self . client . post ( factories . CloudFactory . get_list_url ( ) , self . _get_valid_payload ( new_cloud ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_201_CREATED ) <EOL> def test_user_cannot_add_cloud_to_the_customer_he_sees_but_doesnt_own ( self ) : <EOL> for user_role , customer_type in { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } . items ( ) : <EOL> self . client . force_authenticate ( user = self . users [ user_role ] ) <EOL> new_cloud = factories . CloudFactory . build ( customer = self . customers [ customer_type ] ) <EOL> response = self . client . post ( factories . CloudFactory . get_list_url ( ) , self . _get_valid_payload ( new_cloud ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_403_FORBIDDEN ) <EOL> def test_user_cannot_add_cloud_to_the_customer_he_has_no_role_in ( self ) : <EOL> self . client . force_authenticate ( user = self . users [ '<STR_LIT>' ] ) <EOL> new_cloud = factories . CloudFactory . build ( customer = self . customers [ '<STR_LIT>' ] ) <EOL> response = self . client . post ( factories . CloudFactory . get_list_url ( ) , self . _get_valid_payload ( new_cloud ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_400_BAD_REQUEST ) <EOL> def test_user_cannot_change_auth_url_of_cloud_he_owns ( self ) : <EOL> self . client . force_authenticate ( user = self . users [ '<STR_LIT>' ] ) <EOL> cloud = self . clouds [ '<STR_LIT>' ] <EOL> payload = { '<STR_LIT>' : '<STR_LIT>' } <EOL> response = self . client . patch ( factories . CloudFactory . get_url ( cloud ) , <EOL> data = payload ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> reread_cloud = Cloud . objects . get ( pk = cloud . pk ) <EOL> self . assertEqual ( reread_cloud . auth_url , cloud . auth_url ) <EOL> def test_user_cannot_change_customer_of_cloud_he_owns ( self ) : <EOL> user = self . users [ '<STR_LIT>' ] <EOL> self . client . force_authenticate ( user = user ) <EOL> cloud = self . clouds [ '<STR_LIT>' ] <EOL> new_customer = structure_factories . CustomerFactory ( ) <EOL> new_customer . add_user ( user , CustomerRole . OWNER ) <EOL> payload = { '<STR_LIT>' : structure_factories . CustomerFactory . get_url ( new_customer ) } <EOL> response = self . client . patch ( factories . CloudFactory . get_url ( cloud ) , data = payload ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> reread_cloud = Cloud . objects . get ( pk = cloud . pk ) <EOL> self . assertEqual ( reread_cloud . customer , cloud . customer ) <EOL> def test_user_can_change_cloud_name_of_cloud_he_owns ( self ) : <EOL> self . client . force_authenticate ( user = self . users [ '<STR_LIT>' ] ) <EOL> cloud = self . clouds [ '<STR_LIT>' ] <EOL> payload = { '<STR_LIT:name>' : '<STR_LIT>' } <EOL> response = self . client . patch ( factories . CloudFactory . get_url ( cloud ) , data = payload ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) <EOL> reread_cloud = Cloud . objects . get ( pk = cloud . pk ) <EOL> self . assertEqual ( reread_cloud . name , '<STR_LIT>' ) <EOL> def test_user_cannot_create_cloud_with_auth_url_not_listed_in_settings ( self ) : <EOL> self . client . force_authenticate ( user = self . users [ '<STR_LIT>' ] ) <EOL> new_cloud = factories . CloudFactory . build ( <EOL> customer = self . customers [ '<STR_LIT>' ] , <EOL> auth_url = '<STR_LIT>' , <EOL> ) <EOL> payload = self . _get_valid_payload ( new_cloud ) <EOL> response = self . client . post ( factories . CloudFactory . get_list_url ( ) , payload ) <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_create_cloud_with_auth_url_listed_in_settings ( self ) : <EOL> self . client . force_authenticate ( user = self . users [ '<STR_LIT>' ] ) <EOL> new_cloud = factories . CloudFactory . build ( customer = self . customers [ '<STR_LIT>' ] ) <EOL> response = self . client . post ( factories . CloudFactory . get_list_url ( ) , self . _get_valid_payload ( new_cloud ) ) <EOL> self . assertEqual ( response . status_code , status . HTTP_201_CREATED ) <EOL> def test_user_cannot_modify_in_unstable_state ( self ) : <EOL> self . client . force_authenticate ( user = self . users [ '<STR_LIT>' ] ) <EOL> for state in SynchronizationStates . UNSTABLE_STATES : <EOL> cloud = factories . CloudFactory ( state = state , customer = self . customers [ '<STR_LIT>' ] ) <EOL> url = factories . CloudFactory . get_url ( cloud ) <EOL> for method in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> if method == '<STR_LIT>' and state == SynchronizationStates . NEW : <EOL> continue <EOL> func = getattr ( self . client , method . lower ( ) ) <EOL> response = func ( url ) <EOL> self . assertEqual ( response . status_code , status . HTTP_409_CONFLICT ) <EOL> def _get_valid_payload ( self , resource ) : <EOL> return { <EOL> '<STR_LIT:name>' : resource . name , <EOL> '<STR_LIT>' : structure_factories . CustomerFactory . get_url ( resource . customer ) , <EOL> '<STR_LIT>' : resource . auth_url , <EOL> } <EOL> class CloudFilteringTest ( test . APITransactionTestCase ) : <EOL> def test_cloud_filtering_does_not_raise_internal_error ( self ) : <EOL> fields = [ <EOL> '<STR_LIT:name>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] <EOL> user = structure_factories . UserFactory ( is_staff = True ) <EOL> self . client . force_authenticate ( user = user ) <EOL> for field in fields : <EOL> response = self . client . get ( factories . CloudFactory . get_list_url ( ) , { field : '<STR_LIT>' } ) <EOL> self . assertEqual ( response . status_code , status . HTTP_200_OK ) </s>
