Best Python code snippet using localstack_python
tests.py
Source:tests.py  
...167        A helper method to quickly have a superuser upload and download a file.168        """169        client = Client()170        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.super_user_token}'171        # Patch what get_federation_token() would return since we will not be calling any real AWS STS service.172        get_federation_token.return_value = self.fake_federation_token173        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/upload/?bucket={ALLOWED_BUCKET_NAME}')174        self.assertEqual(response.status_code, 200)175        folder_name = json.loads(response.content)["foldername"]176        file_name = json.loads(response.content)["filename"]177        # Pretend to upload the file to S3.178        file_content = io.StringIO()179        file_content.write("abcdefghijklmnopqrst")180        s3 = awsClient(service='s3')181        s3.upload_fileobj(file_content, ALLOWED_BUCKET_NAME, folder_name + '/' + file_name)182        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/download/')183    def test_groups_have_valid_bucket_permissions(self):184        # Confirm that the permissions added during test setUp applied.185        self.assertTrue('write_bucket' in get_perms(self.readers_group, self.allowed_bucket))186        self.assertTrue('write_bucket' in get_perms(self.writers_group, self.allowed_bucket))187        self.assertTrue('write_bucket' in get_perms(self.downloaders_group, self.allowed_bucket))188        self.assertTrue('write_bucket' in get_perms(self.uploaders_group, self.allowed_bucket))189        # Permissions were never given to the blocked bucket.190        self.assertFalse('write_bucket' in get_perms(self.readers_group, self.blocked_bucket))191        self.assertFalse('write_bucket' in get_perms(self.writers_group, self.blocked_bucket))192        self.assertFalse('write_bucket' in get_perms(self.downloaders_group, self.blocked_bucket))193        self.assertFalse('write_bucket' in get_perms(self.uploaders_group, self.blocked_bucket))194    def test_superuser_can_add_user(self):195        client = Client()196        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.super_user_token}'197        data = {"users": ["random_person@example.com"]}198        response = client.post('/filemaster/user/', data=json.dumps(data), content_type='application/json')199        self.assertEqual(response.status_code, 201)200    def test_miscellaneous_user_cannot_add_user(self):201        client = Client()202        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.readonly_user_token}'203        data = {"users": ["random_person@example.com"]}204        response = client.post('/filemaster/user/', data=json.dumps(data), content_type='application/json')205        self.assertEqual(response.status_code, 403)206    def test_superuser_can_add_user_to_group(self):207        client = Client()208        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.super_user_token}'209        user_email = "random_person@example.com"210        # First create the user.211        data = {"users": [user_email]}212        response = client.post('/filemaster/user/', data=json.dumps(data), content_type='application/json')213        self.assertEqual(response.status_code, 201)214        # Then add it to the group.215        data = {216            "users": [{"email": user_email}],217            "buckets": [{"name": ALLOWED_BUCKET_NAME}]218        }219        response = client.put(f'/filemaster/groups/{READERS_GROUP}/', data=json.dumps(data), content_type='application/json')220        self.assertEqual(response.status_code, 200)221        self.assertTrue(user_email in str(response.content))222    def test_superuser_can_remove_user_from_group(self):223        client = Client()224        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.super_user_token}'225        data = {"users": [{"email": READONLY_USER_EMAIL}]}226        response = client.delete(f'/filemaster/groups/{READERS_GROUP}/', data=json.dumps(data), content_type='application/json')227        self.assertEqual(response.status_code, 200)228        self.assertFalse(READONLY_USER_EMAIL in str(response.content))229    def test_miscellaneous_user_cannot_add_user_to_group(self):230        client = Client()231        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.readonly_user_token}'232        data = {233            "users": [{"email": READ_AND_WRITE_USER_EMAIL}],234            "buckets": [{"name": ALLOWED_BUCKET_NAME}]235        }236        response = client.put(f'/filemaster/groups/{UPLOADERS_GROUP}/', data=json.dumps(data), content_type='application/json')237        self.assertEqual(response.status_code, 403)238    def test_superuser_can_post_and_get_file_info(self):239        client = Client()240        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.super_user_token}'241        # Test that the superuser can POST a new file.242        response = client.post(f'/filemaster/api/file/', data=json.dumps(FILE_INSIDE_GROUP_INFO), content_type='application/json')243        self.assertEqual(response.status_code, 201)244        file_uuid = json.loads(response.content)['uuid']245        # Test that the superuser can GET information about that file.246        response = client.get(f'/filemaster/api/file/{file_uuid}/')247        self.assertEqual(response.status_code, 200)248        # Test that the superuser can POST a new file without permissions specified.249        response = client.post(f'/filemaster/api/file/', data=json.dumps(FILE_OUTSIDE_GROUP_INFO), content_type='application/json')250        self.assertEqual(response.status_code, 201)251        file_uuid = json.loads(response.content)['uuid']252        # Test that the superuser can GET information about that file even though there were no permissions with it.253        response = client.get(f'/filemaster/api/file/{file_uuid}/')254        self.assertEqual(response.status_code, 200)255    # TODO FS-72: users without write permissions to a group should not be able to post files to it256    # def test_noperms_user_cannot_post_file_to_group(self):257    #     client = Client()258    #     client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.readonly_user_token}'259    #     response = client.post(f'/filemaster/api/file/', data=json.dumps(FILE_INSIDE_GROUP_INFO), content_type='application/json')260    #     self.assertEqual(response.status_code, 403)261    def test_noperms_user_can_post_file_if_no_group_specified(self):262        client = Client()263        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.readonly_user_token}'264        response = client.post(f'/filemaster/api/file/', data=json.dumps(FILE_OUTSIDE_GROUP_INFO), content_type='application/json')265        self.assertEqual(response.status_code, 201)266    def test_noread_user_cannot_get_file(self):267        client = Client()268        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.writeonly_user_token}'269        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/')270        self.assertEqual(response.status_code, 403)271    def test_readonly_user_cannot_get_file_outside_group(self):272        client = Client()273        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.readonly_user_token}'274        response = client.get(f'/filemaster/api/file/{self.inaccessible_file_uuid}/')275        self.assertEqual(response.status_code, 403)276    def test_readonly_user_can_get_someone_elses_file(self):277        client = Client()278        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.readonly_user_token}'279        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/')280        self.assertEqual(response.status_code, 200)281    def test_user_cannot_access_file_after_being_removed_from_group(self):282        client = Client()283        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.readonly_user_token}'284        # First confirm the user can access the file.285        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/')286        self.assertEqual(response.status_code, 200)287        # Have the superuser remove the user from the group.288        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.super_user_token}'289        data = {290            "users": [{"email": READONLY_USER_EMAIL}],291        }292        response = client.delete(f'/filemaster/groups/{READERS_GROUP}/', data=json.dumps(data), content_type='application/json')293        self.assertEqual(response.status_code, 200)294        self.assertFalse(READONLY_USER_EMAIL in str(response.content))295        # Now confirm the user cannot access the file.296        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.readonly_user_token}'297        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/')298        self.assertEqual(response.status_code, 403)299    @patch('boto.sts.connection.STSConnection.get_federation_token')300    @override_settings(BUCKETS=BUCKETS_SETTING)301    def test_superuser_can_get_upload_url(self, get_federation_token):302        client = Client()303        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.super_user_token}'304        # Patch what get_federation_token() would return since we will not be calling any real AWS STS service.305        get_federation_token.return_value = self.fake_federation_token306        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/upload/?bucket={ALLOWED_BUCKET_NAME}')307        self.assertEqual(response.status_code, 200)308        self.assertTrue("url" in json.loads(response.content))309    @patch('boto.sts.connection.STSConnection.get_federation_token')310    @override_settings(BUCKETS=BUCKETS_SETTING)311    def test_upload_user_can_get_upload_url(self, get_federation_token):312        client = Client()313        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.everything_but_admin_user_token}'314        # Patch what get_federation_token() would return since we will not be calling any real AWS STS service.315        get_federation_token.return_value = self.fake_federation_token316        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/upload/?bucket={ALLOWED_BUCKET_NAME}')317        self.assertEqual(response.status_code, 200)318        self.assertTrue("url" in json.loads(response.content))319    @patch('boto.sts.connection.STSConnection.get_federation_token')320    @override_settings(BUCKETS=BUCKETS_SETTING)321    def test_upload_user_cannot_get_upload_url_for_file_outside_group(self, get_federation_token):322        client = Client()323        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.everything_but_admin_user_token}'324        # Patch what get_federation_token() would return since we will not be calling any real AWS STS service.325        get_federation_token.return_value = self.fake_federation_token326        response = client.get(f'/filemaster/api/file/{self.inaccessible_file_uuid}/upload/?bucket={ALLOWED_BUCKET_NAME}')327        self.assertEqual(response.status_code, 403)328    @patch('boto.sts.connection.STSConnection.get_federation_token')329    @override_settings(BUCKETS=BUCKETS_SETTING)330    def test_upload_user_cannot_get_upload_url_for_file_for_blocked_bucket(self, get_federation_token):331        client = Client()332        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.everything_but_admin_user_token}'333        # Patch what get_federation_token() would return since we will not be calling any real AWS STS service.334        get_federation_token.return_value = self.fake_federation_token335        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/upload/?bucket={BLOCKED_BUCKET_NAME}')336        self.assertEqual(response.status_code, 403)337    @patch('boto.sts.connection.STSConnection.get_federation_token')338    @override_settings(BUCKETS=BUCKETS_SETTING)339    def test_notupload_user_cannot_get_upload_url(self, get_federation_token):340        client = Client()341        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.readonly_user_token}'342        # Patch what get_federation_token() would return since we will not be calling any real AWS STS service.343        get_federation_token.return_value = self.fake_federation_token344        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/upload/?bucket={ALLOWED_BUCKET_NAME}')345        self.assertEqual(response.status_code, 403)346    # # TODO FS-59: 500 error if the file was not uploaded. Cannot run test in this case because of uncaught exception.347    # @patch('boto.sts.connection.STSConnection.get_federation_token')348    # @override_settings(BUCKETS=BUCKETS_SETTING)349    # def test_superuser_can_check_uploadcomplete_on_file_never_uploaded(self, get_federation_token):350    #     client = Client()351    #     client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.super_user_token}'352    #     # Patch what get_federation_token() would return since we will not be calling any real AWS STS service.353    #     get_federation_token.return_value = self.fake_federation_token354    #     response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/upload/?bucket={ALLOWED_BUCKET_NAME}')355    #     self.assertEqual(response.status_code, 200)356    #     file_location_id = json.loads(response.content)["locationid"]357    #     response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/uploadcomplete/?location={file_location_id}')358    #     self.assertEqual(response.status_code, 500)359    @patch('boto.sts.connection.STSConnection.get_federation_token')360    @override_settings(BUCKETS=BUCKETS_SETTING)361    def test_superuser_can_check_uploadcomplete_on_file_actually_uploaded(self, get_federation_token):362        client = Client()363        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.super_user_token}'364        # Patch what get_federation_token() would return since we will not be calling any real AWS STS service.365        get_federation_token.return_value = self.fake_federation_token366        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/upload/?bucket={ALLOWED_BUCKET_NAME}')367        self.assertEqual(response.status_code, 200)368        file_location_id = json.loads(response.content)["locationid"]369        folder_name = json.loads(response.content)["foldername"]370        file_name = json.loads(response.content)["filename"]371        # Pretend to upload the file to S3.372        file_content = io.StringIO()373        file_content.write("abcdefghijklmnopqrst")374        s3 = awsClient(service='s3')375        s3.upload_fileobj(file_content, ALLOWED_BUCKET_NAME, folder_name + '/' + file_name)376        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/uploadcomplete/?location={file_location_id}')377        self.assertEqual(response.status_code, 200)378    @patch('boto.sts.connection.STSConnection.get_federation_token')379    @override_settings(BUCKETS=BUCKETS_SETTING)380    def test_upload_user_can_check_uploadcomplete(self, get_federation_token):381        client = Client()382        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.everything_but_admin_user_token}'383        # Patch what get_federation_token() would return since we will not be calling any real AWS STS service.384        get_federation_token.return_value = self.fake_federation_token385        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/upload/?bucket={ALLOWED_BUCKET_NAME}')386        self.assertEqual(response.status_code, 200)387        file_location_id = json.loads(response.content)["locationid"]388        folder_name = json.loads(response.content)["foldername"]389        file_name = json.loads(response.content)["filename"]390        # Pretend to upload the file to S3.391        file_content = io.StringIO()392        file_content.write("abcdefghijklmnopqrst")393        s3 = awsClient(service='s3')394        s3.upload_fileobj(file_content, ALLOWED_BUCKET_NAME, folder_name + '/' + file_name)395        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/uploadcomplete/?location={file_location_id}')396        self.assertEqual(response.status_code, 200)397    @patch('boto.sts.connection.STSConnection.get_federation_token')398    @override_settings(BUCKETS=BUCKETS_SETTING)399    def test_notupload_user_cannot_check_uploadcomplete(self, get_federation_token):400        client = Client()401        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.readonly_user_token}'402        # Patch what get_federation_token() would return since we will not be calling any real AWS STS service.403        get_federation_token.return_value = self.fake_federation_token404        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/uploadcomplete/')405        self.assertEqual(response.status_code, 403)406    # # TODO FS-75 This should return an error code but currently breaks because of FS-74407    # @override_settings(BUCKETS=BUCKETS_SETTING)408    # def test_superuser_cannot_get_download_url_for_file_never_uploaded(self):409    #     client = Client()410    #     client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.super_user_token}'411    #     response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/download/')412    #     self.assertEqual(response.status_code, 404)413    @patch('boto.sts.connection.STSConnection.get_federation_token')414    @override_settings(BUCKETS=BUCKETS_SETTING)415    def test_superuser_can_get_download_url_for_uploaded_file(self, get_federation_token):416        client = Client()417        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.super_user_token}'418        # Patch what get_federation_token() would return since we will not be calling any real AWS STS service.419        get_federation_token.return_value = self.fake_federation_token420        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/upload/?bucket={ALLOWED_BUCKET_NAME}')421        self.assertEqual(response.status_code, 200)422        folder_name = json.loads(response.content)["foldername"]423        file_name = json.loads(response.content)["filename"]424        # Pretend to upload the file to S3.425        file_content = io.StringIO()426        file_content.write("abcdefghijklmnopqrst")427        s3 = awsClient(service='s3')428        s3.upload_fileobj(file_content, ALLOWED_BUCKET_NAME, folder_name + '/' + file_name)429        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/download/')430        self.assertEqual(response.status_code, 200)431        self.assertTrue("url" in json.loads(response.content))432    @patch('boto.sts.connection.STSConnection.get_federation_token')433    @override_settings(BUCKETS=BUCKETS_SETTING)434    def test_downloader_user_can_get_download_url_for_uploaded_file(self, get_federation_token):435        # First have a superuser upload the file436        client = Client()437        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.super_user_token}'438        # Patch what get_federation_token() would return since we will not be calling any real AWS STS service.439        get_federation_token.return_value = self.fake_federation_token440        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/upload/?bucket={ALLOWED_BUCKET_NAME}')441        self.assertEqual(response.status_code, 200)442        folder_name = json.loads(response.content)["foldername"]443        file_name = json.loads(response.content)["filename"]444        # Pretend to upload the file to S3.445        file_content = io.StringIO()446        file_content.write("abcdefghijklmnopqrst")447        s3 = awsClient(service='s3')448        s3.upload_fileobj(file_content, ALLOWED_BUCKET_NAME, folder_name + '/' + file_name)449        # Then have a downloader user try to download it.450        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.everything_but_admin_user_token}'451        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/download/')452        self.assertEqual(response.status_code, 200)453        self.assertTrue("url" in json.loads(response.content))454    @patch('boto.sts.connection.STSConnection.get_federation_token')455    @override_settings(BUCKETS=BUCKETS_SETTING)456    def test_notdownloader_user_cannot_get_download_url_for_uploaded_file(self, get_federation_token):457        # First have a superuser upload the file458        client = Client()459        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.super_user_token}'460        # Patch what get_federation_token() would return since we will not be calling any real AWS STS service.461        get_federation_token.return_value = self.fake_federation_token462        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/upload/?bucket={ALLOWED_BUCKET_NAME}')463        self.assertEqual(response.status_code, 200)464        folder_name = json.loads(response.content)["foldername"]465        file_name = json.loads(response.content)["filename"]466        # Pretend to upload the file to S3.467        file_content = io.StringIO()468        file_content.write("abcdefghijklmnopqrst")469        s3 = awsClient(service='s3')470        s3.upload_fileobj(file_content, ALLOWED_BUCKET_NAME, folder_name + '/' + file_name)471        # Then have a readonly user try to download it.472        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.readonly_user_token}'473        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/download/')474        self.assertEqual(response.status_code, 403)475    @patch('boto.sts.connection.STSConnection.get_federation_token')476    @override_settings(BUCKETS=BUCKETS_SETTING)477    def test_download_request_creates_download_log(self, get_federation_token):478        # First have a superuser upload the file479        client = Client()480        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.super_user_token}'481        # Patch what get_federation_token() would return since we will not be calling any real AWS STS service.482        get_federation_token.return_value = self.fake_federation_token483        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/upload/?bucket={ALLOWED_BUCKET_NAME}')484        self.assertEqual(response.status_code, 200)485        folder_name = json.loads(response.content)["foldername"]486        file_name = json.loads(response.content)["filename"]487        # Pretend to upload the file to S3.488        file_content = io.StringIO()489        file_content.write("abcdefghijklmnopqrst")490        s3 = awsClient(service='s3')491        s3.upload_fileobj(file_content, ALLOWED_BUCKET_NAME, folder_name + '/' + file_name)492        # Then have a downloader user try to download it.493        client.defaults['HTTP_AUTHORIZATION'] = f'Token {self.everything_but_admin_user_token}'494        response = client.get(f'/filemaster/api/file/{self.accessible_file_uuid}/download/')495        self.assertEqual(response.status_code, 200)...getConsole.py
Source:getConsole.py  
...79                        'Resource': '*'80                    }81                ]82            }83    get_federation_token = sts_client.get_federation_token(84        Name=username,85        Policy=json.dumps(policy),86        DurationSeconds=4320087    )88    tempCre = {89        "sessionId": get_federation_token['Credentials']['AccessKeyId'],90        "sessionKey": get_federation_token['Credentials']['SecretAccessKey'],91        "sessionToken": get_federation_token['Credentials']['SessionToken']92    }93    tempCreStr = json.dumps(tempCre)94    # Make request to AWS federation endpoint to get sign-in token95    req_para = "?Action=getSigninToken&SessionDuration=43200&Session=" + urllib.parse.quote_plus(tempCreStr)96    if region.startswith("cn-"):97        request_url = "https://signin.amazonaws.cn/federation" + req_para...sts_basic_application.py
Source:sts_basic_application.py  
...35    print 'Assumed Role ID: {!s}'.format(assumed_role.user.assume_role_id)36    credentials = assumed_role.credentials37    print 'Access Key: {!s}'.format(credentials.access_key)38    print 'Secret Key: {!s}'.format(credentials.secret_key)39def get_federation_token():40    '''41    Allows a given IAM user to request an STST token.42    '''43    logging.debug('get_federation_token called.')44    sts_interface = boto.connect_sts()45    token = sts_interface.get_federation_token('test')46    credentials = token.credentials47    print 'Federated User ID: {!s}'.format(token.federated_user_id)48    print 'Federated User ARN: {!s}'.format(token.federated_user_arn)49    print 'Access Key: {!s}'.format(credentials.access_key)50    print 'Secret Key: {!s}'.format(credentials.secret_key)51parser = argparse.ArgumentParser()52parser.add_argument('--username')53parser.add_argument('--password')54parser.add_argument('--mode', default='assumerole',55                    choices=['get_federation_token', 'assume_role'])56parser.add_argument('--role')57parser.add_argument('--log-level', dest='log_level', default='INFO',58                    choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'])59args = parser.parse_args()60# configure logging61log_format = '%(message)s'62log_level = str.upper(args.log_level)63logging.basicConfig(level=log_level, format=log_format)64if args.mode == 'get_federation_token':65    valid_user = validate_user(args.username, args.password)66    if valid_user is True:67        get_federation_token()68    else:69        logging.warning('User not valid. Token will not be issued.')70        exit(1)71elif args.mode == 'assume_role':72    assume_role(args.role)73else:...Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
