Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use put_object_retention method in localstack

Best Python code snippet using localstack_python

test_s3_object_lock_x.py

Source:test_s3_object_lock_x.py

...225        key = 'file1'226        response = client.put_object(Bucket=bucket_name, Body='abc', Key=key)227        version_id = response['VersionId']228        retention = {'Mode': 'GOVERNANCE', 'RetainUntilDate': datetime.datetime(2030, 1, 1, tzinfo=pytz.UTC)}229        response = client.put_object_retention(Bucket=bucket_name, Key=key, Retention=retention)230        self.eq(response['ResponseMetadata']['HTTPStatusCode'], 200)231        client.delete_object(Bucket=bucket_name, Key=key, VersionId=version_id, BypassGovernanceRetention=True)232    def test_object_lock_put_obj_retention_invalid_bucket(self, s3cfg_global_unique):233        """234        (operation='Test put object retention with bucket object lock not enabled')235        (assertion='fails')236        """237        client = get_client(s3cfg_global_unique)238        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)239        client.create_bucket(Bucket=bucket_name)240        key = 'file1'241        client.put_object(Bucket=bucket_name, Body='abc', Key=key)242        retention = {'Mode': 'GOVERNANCE', 'RetainUntilDate': datetime.datetime(2030, 1, 1, tzinfo=pytz.UTC)}243        e = assert_raises(ClientError, client.put_object_retention, Bucket=bucket_name, Key=key, Retention=retention)244        status, error_code = self.get_status_and_error_code(e.response)245        self.eq(status, 400)246        self.eq(error_code, 'InvalidRequest')247    def test_object_lock_put_obj_retention_invalid_mode(self, s3cfg_global_unique):248        """249        (operation='Test put object retention with invalid mode')250        (assertion='fails')251        """252        client = get_client(s3cfg_global_unique)253        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)254        client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True)255        key = 'file1'256        client.put_object(Bucket=bucket_name, Body='abc', Key=key)257        retention = {'Mode': 'governance', 'RetainUntilDate': datetime.datetime(2030, 1, 1, tzinfo=pytz.UTC)}258        e = assert_raises(ClientError, client.put_object_retention, Bucket=bucket_name, Key=key, Retention=retention)259        status, error_code = self.get_status_and_error_code(e.response)260        self.eq(status, 400)261        self.eq(error_code, 'MalformedXML')262        retention = {'Mode': 'abc', 'RetainUntilDate': datetime.datetime(2030, 1, 1, tzinfo=pytz.UTC)}263        e = assert_raises(ClientError, client.put_object_retention, Bucket=bucket_name, Key=key, Retention=retention)264        status, error_code = self.get_status_and_error_code(e.response)265        self.eq(status, 400)266        self.eq(error_code, 'MalformedXML')267    def test_object_lock_get_obj_retention(self, s3cfg_global_unique):268        """269        (operation='Test get object retention')270        (assertion='success')271        """272        client = get_client(s3cfg_global_unique)273        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)274        client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True)275        key = 'file1'276        response = client.put_object(Bucket=bucket_name, Body='abc', Key=key)277        version_id = response['VersionId']278        retention = {'Mode': 'GOVERNANCE', 'RetainUntilDate': datetime.datetime(2030, 1, 1, tzinfo=pytz.UTC)}279        client.put_object_retention(Bucket=bucket_name, Key=key, Retention=retention)280        response = client.get_object_retention(Bucket=bucket_name, Key=key)281        self.eq(response['Retention'], retention)282        client.delete_object(Bucket=bucket_name, Key=key, VersionId=version_id, BypassGovernanceRetention=True)283    def test_object_lock_get_obj_retention_iso8601(self, s3cfg_global_unique):284        """285        (operation='Test object retention date formatting')286        (assertion='success')287        """288        client = get_client(s3cfg_global_unique)289        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)290        client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True)291        key = 'file1'292        response = client.put_object(Bucket=bucket_name, Body='abc', Key=key)293        version_id = response['VersionId']294        date = datetime.datetime.today() + datetime.timedelta(days=365)295        retention = {'Mode': 'GOVERNANCE', 'RetainUntilDate': date}296        http_response = None297        def get_http_response(**kwargs):298            nonlocal http_response299            http_response = kwargs['http_response'].__dict__300        client.put_object_retention(Bucket=bucket_name, Key=key, Retention=retention)301        client.meta.events.register('after-call.s3.HeadObject', get_http_response)302        client.head_object(Bucket=bucket_name, VersionId=version_id, Key=key)303        retain_date = http_response['headers']['x-amz-object-lock-retain-until-date']304        isodate.parse_datetime(retain_date)305        client.delete_object(Bucket=bucket_name, Key=key, VersionId=version_id, BypassGovernanceRetention=True)306    def test_object_lock_get_obj_retention_invalid_bucket(self, s3cfg_global_unique):307        """308        (operation='Test get object retention with invalid bucket')309        (assertion='fails')310        """311        client = get_client(s3cfg_global_unique)312        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)313        client.create_bucket(Bucket=bucket_name)314        key = 'file1'315        client.put_object(Bucket=bucket_name, Body='abc', Key=key)316        e = assert_raises(ClientError, client.get_object_retention, Bucket=bucket_name, Key=key)317        status, error_code = self.get_status_and_error_code(e.response)318        self.eq(status, 400)319        self.eq(error_code, 'InvalidRequest')320    def test_object_lock_put_obj_retention_version_id(self, s3cfg_global_unique):321        """322        (operation='Test put object retention with version id')323        (assertion='success')324        """325        client = get_client(s3cfg_global_unique)326        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)327        client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True)328        key = 'file1'329        client.put_object(Bucket=bucket_name, Body='abc', Key=key)330        response = client.put_object(Bucket=bucket_name, Body='abc', Key=key)331        version_id = response['VersionId']332        retention = {'Mode': 'GOVERNANCE', 'RetainUntilDate': datetime.datetime(2030, 1, 1, tzinfo=pytz.UTC)}333        client.put_object_retention(Bucket=bucket_name, Key=key, VersionId=version_id, Retention=retention)334        response = client.get_object_retention(Bucket=bucket_name, Key=key, VersionId=version_id)335        self.eq(response['Retention'], retention)336        client.delete_object(Bucket=bucket_name, Key=key, VersionId=version_id, BypassGovernanceRetention=True)337    def test_object_lock_put_obj_retention_override_default_retention(self, s3cfg_global_unique):338        """339        (operation='Test put object retention to override default retention')340        (assertion='success')341        """342        client = get_client(s3cfg_global_unique)343        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)344        client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True)345        conf = {'ObjectLockEnabled': 'Enabled',346                'Rule': {347                    'DefaultRetention': {348                        'Mode': 'GOVERNANCE',349                        'Days': 1350                    }351                }}352        client.put_object_lock_configuration(353            Bucket=bucket_name,354            ObjectLockConfiguration=conf)355        key = 'file1'356        response = client.put_object(Bucket=bucket_name, Body='abc', Key=key)357        version_id = response['VersionId']358        retention = {'Mode': 'GOVERNANCE', 'RetainUntilDate': datetime.datetime(2030, 1, 1, tzinfo=pytz.UTC)}359        client.put_object_retention(Bucket=bucket_name, Key=key, Retention=retention)360        response = client.get_object_retention(Bucket=bucket_name, Key=key)361        self.eq(response['Retention'], retention)362        client.delete_object(Bucket=bucket_name, Key=key, VersionId=version_id, BypassGovernanceRetention=True)363    def test_object_lock_put_obj_retention_increase_period(self, s3cfg_global_unique):364        """365        (operation='Test put object retention to increase retention period')366        (assertion='success')367        """368        client = get_client(s3cfg_global_unique)369        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)370        client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True)371        key = 'file1'372        response = client.put_object(Bucket=bucket_name, Body='abc', Key=key)373        version_id = response['VersionId']374        retention1 = {'Mode': 'GOVERNANCE', 'RetainUntilDate': datetime.datetime(2030, 1, 1, tzinfo=pytz.UTC)}375        client.put_object_retention(Bucket=bucket_name, Key=key, Retention=retention1)376        retention2 = {'Mode': 'GOVERNANCE', 'RetainUntilDate': datetime.datetime(2030, 1, 3, tzinfo=pytz.UTC)}377        client.put_object_retention(Bucket=bucket_name, Key=key, Retention=retention2)378        response = client.get_object_retention(Bucket=bucket_name, Key=key)379        self.eq(response['Retention'], retention2)380        client.delete_object(Bucket=bucket_name, Key=key, VersionId=version_id, BypassGovernanceRetention=True)381    def test_object_lock_put_obj_retention_shorten_period(self, s3cfg_global_unique):382        """383        (operation='Test put object retention to shorten period')384        (assertion='fails')385        """386        client = get_client(s3cfg_global_unique)387        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)388        client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True)389        key = 'file1'390        response = client.put_object(Bucket=bucket_name, Body='abc', Key=key)391        version_id = response['VersionId']392        retention = {'Mode': 'GOVERNANCE', 'RetainUntilDate': datetime.datetime(2030, 1, 3, tzinfo=pytz.UTC)}393        client.put_object_retention(Bucket=bucket_name, Key=key, Retention=retention)394        retention = {'Mode': 'GOVERNANCE', 'RetainUntilDate': datetime.datetime(2030, 1, 1, tzinfo=pytz.UTC)}395        e = assert_raises(ClientError, client.put_object_retention, Bucket=bucket_name, Key=key, Retention=retention)396        status, error_code = self.get_status_and_error_code(e.response)397        self.eq(status, 403)398        self.eq(error_code, 'AccessDenied')399        client.delete_object(Bucket=bucket_name, Key=key, VersionId=version_id, BypassGovernanceRetention=True)400    def test_object_lock_put_obj_retention_shorten_period_bypass(self, s3cfg_global_unique):401        """402        (operation='Test put object retention to shorten period with bypass header')403        (assertion='success')404        """405        client = get_client(s3cfg_global_unique)406        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)407        client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True)408        key = 'file1'409        response = client.put_object(Bucket=bucket_name, Body='abc', Key=key)410        version_id = response['VersionId']411        retention = {'Mode': 'GOVERNANCE', 'RetainUntilDate': datetime.datetime(2030, 1, 3, tzinfo=pytz.UTC)}412        client.put_object_retention(Bucket=bucket_name, Key=key, Retention=retention)413        retention = {'Mode': 'GOVERNANCE', 'RetainUntilDate': datetime.datetime(2030, 1, 1, tzinfo=pytz.UTC)}414        client.put_object_retention(Bucket=bucket_name, Key=key, Retention=retention, BypassGovernanceRetention=True)415        response = client.get_object_retention(Bucket=bucket_name, Key=key)416        self.eq(response['Retention'], retention)417        client.delete_object(Bucket=bucket_name, Key=key, VersionId=version_id, BypassGovernanceRetention=True)418    def test_object_lock_delete_object_with_retention(self, s3cfg_global_unique):419        """420        (operation='Test delete object with retention')421        (assertion='retention period make effects')422        """423        client = get_client(s3cfg_global_unique)424        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)425        client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True)426        key = 'file1'427        response = client.put_object(Bucket=bucket_name, Body='abc', Key=key)428        retention = {'Mode': 'GOVERNANCE', 'RetainUntilDate': datetime.datetime(2030, 1, 1, tzinfo=pytz.UTC)}429        client.put_object_retention(Bucket=bucket_name, Key=key, Retention=retention)430        e = assert_raises(ClientError, client.delete_object, Bucket=bucket_name, Key=key,431                          VersionId=response['VersionId'])432        status, error_code = self.get_status_and_error_code(e.response)433        self.eq(status, 403)434        self.eq(error_code, 'AccessDenied')435        response = client.delete_object(Bucket=bucket_name, Key=key, VersionId=response['VersionId'],436                                        BypassGovernanceRetention=True)437        self.eq(response['ResponseMetadata']['HTTPStatusCode'], 204)438    def test_object_lock_multi_delete_object_with_retention(self, s3cfg_global_unique):439        """440        (operation='Test multi-delete object with retention')441        (assertion='retention period make effects')442        """443        client = get_client(s3cfg_global_unique)444        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)445        client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True)446        key1 = 'file1'447        key2 = 'file2'448        response1 = client.put_object(Bucket=bucket_name, Body='abc', Key=key1)449        response2 = client.put_object(Bucket=bucket_name, Body='abc', Key=key2)450        version_id1 = response1['VersionId']451        version_id2 = response2['VersionId']452        # key1 is under retention, but key2 isn't.453        retention = {'Mode': 'GOVERNANCE', 'RetainUntilDate': datetime.datetime(2030, 1, 1, tzinfo=pytz.UTC)}454        client.put_object_retention(Bucket=bucket_name, Key=key1, Retention=retention)455        delete_response = client.delete_objects(456            Bucket=bucket_name,457            Delete={458                'Objects': [459                    {460                        'Key': key1,461                        'VersionId': version_id1462                    },463                    {464                        'Key': key2,465                        'VersionId': version_id2466                    }467                ]468            }469        )470        self.eq(len(delete_response['Deleted']), 1)471        self.eq(len(delete_response['Errors']), 1)472        failed_object = delete_response['Errors'][0]473        self.eq(failed_object['Code'], 'AccessDenied')474        self.eq(failed_object['Key'], key1)475        self.eq(failed_object['VersionId'], version_id1)476        deleted_object = delete_response['Deleted'][0]477        self.eq(deleted_object['Key'], key2)478        self.eq(deleted_object['VersionId'], version_id2)479        delete_response = client.delete_objects(480            Bucket=bucket_name,481            Delete={482                'Objects': [483                    {484                        'Key': key1,485                        'VersionId': version_id1486                    }487                ]488            },489            BypassGovernanceRetention=True490        )491        assert (('Errors' not in delete_response) or (len(delete_response['Errors']) == 0))492        self.eq(len(delete_response['Deleted']), 1)493        deleted_object = delete_response['Deleted'][0]494        self.eq(deleted_object['Key'], key1)495        self.eq(deleted_object['VersionId'], version_id1)496    def test_object_lock_put_legal_hold(self, s3cfg_global_unique):497        """498        (operation='Test put legal hold')499        (assertion='success')500        """501        client = get_client(s3cfg_global_unique)502        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)503        client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True)504        key = 'file1'505        client.put_object(Bucket=bucket_name, Body='abc', Key=key)506        legal_hold = {'Status': 'ON'}507        response = client.put_object_legal_hold(Bucket=bucket_name, Key=key, LegalHold=legal_hold)508        self.eq(response['ResponseMetadata']['HTTPStatusCode'], 200)509        response = client.put_object_legal_hold(Bucket=bucket_name, Key=key, LegalHold={'Status': 'OFF'})510        self.eq(response['ResponseMetadata']['HTTPStatusCode'], 200)511    def test_object_lock_put_legal_hold_invalid_bucket(self, s3cfg_global_unique):512        """513        (operation='Test put legal hold with invalid bucket')514        (assertion='fails')515        """516        client = get_client(s3cfg_global_unique)517        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)518        client.create_bucket(Bucket=bucket_name)519        key = 'file1'520        client.put_object(Bucket=bucket_name, Body='abc', Key=key)521        legal_hold = {'Status': 'ON'}522        e = assert_raises(ClientError, client.put_object_legal_hold, Bucket=bucket_name, Key=key, LegalHold=legal_hold)523        status, error_code = self.get_status_and_error_code(e.response)524        self.eq(status, 400)525        self.eq(error_code, 'InvalidRequest')526    def test_object_lock_put_legal_hold_invalid_status(self, s3cfg_global_unique):527        """528        (operation='Test put legal hold with invalid status')529        (assertion='fails')530        """531        client = get_client(s3cfg_global_unique)532        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)533        client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True)534        key = 'file1'535        client.put_object(Bucket=bucket_name, Body='abc', Key=key)536        legal_hold = {'Status': 'abc'}537        e = assert_raises(ClientError, client.put_object_legal_hold, Bucket=bucket_name, Key=key, LegalHold=legal_hold)538        status, error_code = self.get_status_and_error_code(e.response)539        self.eq(status, 400)540        self.eq(error_code, 'MalformedXML')541    def test_object_lock_get_legal_hold(self, s3cfg_global_unique):542        """543        (operation='Test get legal hold')544        (assertion='success')545        """546        client = get_client(s3cfg_global_unique)547        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)548        client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True)549        key = 'file1'550        client.put_object(Bucket=bucket_name, Body='abc', Key=key)551        legal_hold = {'Status': 'ON'}552        client.put_object_legal_hold(Bucket=bucket_name, Key=key, LegalHold=legal_hold)553        response = client.get_object_legal_hold(Bucket=bucket_name, Key=key)554        self.eq(response['LegalHold'], legal_hold)555        legal_hold_off = {'Status': 'OFF'}556        client.put_object_legal_hold(Bucket=bucket_name, Key=key, LegalHold=legal_hold_off)557        response = client.get_object_legal_hold(Bucket=bucket_name, Key=key)558        self.eq(response['LegalHold'], legal_hold_off)559    def test_object_lock_get_legal_hold_invalid_bucket(self, s3cfg_global_unique):560        """561        (operation='Test get legal hold with invalid bucket')562        (assertion='fails')563        """564        client = get_client(s3cfg_global_unique)565        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)566        client.create_bucket(Bucket=bucket_name)567        key = 'file1'568        client.put_object(Bucket=bucket_name, Body='abc', Key=key)569        e = assert_raises(ClientError, client.get_object_legal_hold, Bucket=bucket_name, Key=key)570        status, error_code = self.get_status_and_error_code(e.response)571        self.eq(status, 400)572        self.eq(error_code, 'InvalidRequest')573    def test_object_lock_delete_object_with_legal_hold_on(self, s3cfg_global_unique):574        """575        (operation='Test delete object with legal hold on')576        (assertion='fails')577        """578        client = get_client(s3cfg_global_unique)579        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)580        client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True)581        key = 'file1'582        response = client.put_object(Bucket=bucket_name, Body='abc', Key=key)583        client.put_object_legal_hold(Bucket=bucket_name, Key=key, LegalHold={'Status': 'ON'})584        e = assert_raises(ClientError, client.delete_object, Bucket=bucket_name, Key=key,585                          VersionId=response['VersionId'])586        status, error_code = self.get_status_and_error_code(e.response)587        self.eq(status, 403)588        self.eq(error_code, 'AccessDenied')589        client.put_object_legal_hold(Bucket=bucket_name, Key=key, LegalHold={'Status': 'OFF'})590    def test_object_lock_delete_object_with_legal_hold_off(self, s3cfg_global_unique):591        """592        (operation='Test delete object with legal hold off')593        (assertion='fails')594        """595        client = get_client(s3cfg_global_unique)596        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)597        client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True)598        key = 'file1'599        response = client.put_object(Bucket=bucket_name, Body='abc', Key=key)600        client.put_object_legal_hold(Bucket=bucket_name, Key=key, LegalHold={'Status': 'OFF'})601        response = client.delete_object(Bucket=bucket_name, Key=key, VersionId=response['VersionId'])602        self.eq(response['ResponseMetadata']['HTTPStatusCode'], 204)603    def test_object_lock_get_obj_metadata(self, s3cfg_global_unique):604        """605        (operation='Test get object metadata')606        (assertion='success')607        """608        client = get_client(s3cfg_global_unique)609        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)610        client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True)611        key = 'file1'612        client.put_object(Bucket=bucket_name, Body='abc', Key=key)613        legal_hold = {'Status': 'ON'}614        client.put_object_legal_hold(Bucket=bucket_name, Key=key, LegalHold=legal_hold)615        retention = {'Mode': 'GOVERNANCE', 'RetainUntilDate': datetime.datetime(2030, 1, 1, tzinfo=pytz.UTC)}616        client.put_object_retention(Bucket=bucket_name, Key=key, Retention=retention)617        response = client.head_object(Bucket=bucket_name, Key=key)618        self.eq(response['ObjectLockMode'], retention['Mode'])619        self.eq(response['ObjectLockRetainUntilDate'], retention['RetainUntilDate'])620        self.eq(response['ObjectLockLegalHoldStatus'], legal_hold['Status'])621        client.put_object_legal_hold(Bucket=bucket_name, Key=key, LegalHold={'Status': 'OFF'})622        client.delete_object(Bucket=bucket_name, Key=key, VersionId=response['VersionId'],623                             BypassGovernanceRetention=True)624    def test_object_lock_uploading_obj(self, s3cfg_global_unique):625        """626        (operation='Test put legal hold and retention when uploading object')627        (assertion='success')628        """629        client = get_client(s3cfg_global_unique)630        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)631        client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True)632        key = 'file1'633        client.put_object(Bucket=bucket_name, Body='abc', Key=key, ObjectLockMode='GOVERNANCE',634                          ObjectLockRetainUntilDate=datetime.datetime(2030, 1, 1, tzinfo=pytz.UTC),635                          ObjectLockLegalHoldStatus='ON')636        response = client.head_object(Bucket=bucket_name, Key=key)637        self.eq(response['ObjectLockMode'], 'GOVERNANCE')638        self.eq(response['ObjectLockRetainUntilDate'], datetime.datetime(2030, 1, 1, tzinfo=pytz.UTC))639        self.eq(response['ObjectLockLegalHoldStatus'], 'ON')640        client.put_object_legal_hold(Bucket=bucket_name, Key=key, LegalHold={'Status': 'OFF'})641        client.delete_object(Bucket=bucket_name, Key=key, VersionId=response['VersionId'],642                             BypassGovernanceRetention=True)643    def test_object_lock_changing_mode_from_governance_with_bypass(self, s3cfg_global_unique):644        """645        (operation='Test changing object retention mode from GOVERNANCE to COMPLIANCE with bypass')646        (assertion='succeeds')647        """648        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)649        client = get_client(s3cfg_global_unique)650        key = 'file1'651        client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True)652        # upload object with mode=GOVERNANCE653        retain_until = datetime.datetime.now(pytz.utc) + datetime.timedelta(seconds=10)654        client.put_object(Bucket=bucket_name, Body='abc', Key=key, ObjectLockMode='GOVERNANCE',655                          ObjectLockRetainUntilDate=retain_until)656        # change mode to COMPLIANCE657        retention = {'Mode': 'COMPLIANCE', 'RetainUntilDate': retain_until}658        client.put_object_retention(Bucket=bucket_name, Key=key, Retention=retention, BypassGovernanceRetention=True)659    def test_object_lock_changing_mode_from_governance_without_bypass(self, s3cfg_global_unique):660        """661        (operation='Test changing object retention mode from GOVERNANCE to COMPLIANCE without bypass')662        (assertion='fails')663        """664        bucket_name = self.get_new_bucket_name(s3cfg_global_unique)665        client = get_client(s3cfg_global_unique)666        key = 'file1'667        client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True)668        # upload object with mode=GOVERNANCE669        retain_until = datetime.datetime.now(pytz.utc) + datetime.timedelta(seconds=10)670        client.put_object(Bucket=bucket_name, Body='abc', Key=key, ObjectLockMode='GOVERNANCE',671                          ObjectLockRetainUntilDate=retain_until)672        # try to change mode to COMPLIANCE...

fonctions.py

Source:fonctions.py

...57    fileName = os.path.basename(path)58    try:59        s3_client.upload_file(Filename= path, Bucket=container, Key= fileName)60        if retention !=0 :61    	    s3_client.put_object_retention(Bucket=container, Key= fileName, 62    	    Retention={63        	    'Mode': 'GOVERNANCE',64       		    'RetainUntilDate': datetime.today()+timedelta(retention)65    	            }66            )67    except FileNotFoundError:68        print("No such file : "+path)69def test_connection( client):70    try:71        # use the JSON library's dump() method for indentation72        info = client.info()73        # pass client object to info() method74        print ("Elasticsearch client info():", info)75    except ConnectionError as err:76        # print ConnectionError for Elasticsearch77        print ("\nElasticsearch info() ERROR:", err)78        print ("\nThe client host:", config['elasticsearchDomain'], "is invalid or cluster is not running")79        # change the client's value to 'None' if ConnectionError80        client = None81#function to delete a file in the container82def deleteFileInContainer(container, fileName):83    """Function called to delete a certain file (thanks to its name) in a defined container"""84    try:85         s3_client.delete_object(Bucket=container, Key= fileName)86    except s3_client.exceptions.NoSuchBucket:87          print("No container named "+container+".")88#function to copy an entire folder from sever to the OpenIO container89def uploadFolder( container, folder_path, retention=0):90    """Function called to upload an entire folder (thanks to its path) from your computer to a defined container"""91    try:92        for file_name_ext in os.listdir(folder_path):93            file_path_ext=str(folder_path)+'/'+file_name_ext94            s3_client.upload_file(Filename= file_path_ext, Bucket= container, Key=file_name_ext)95            if retention !=0 :96                s3_client.put_object_retention(Bucket=container, Key=file_name_ext, 97    		Retention={98        		'Mode': 'GOVERNANCE',99       			'RetainUntilDate': datetime.today()+timedelta(retention)100    		}101	    )102    except FileNotFoundError:103        print("No such directory : "+folder_path)104#function to list all containers105def listBuckets():106    print(s3_client.list_buckets())107#function to list all data inside a container108def listDataForAGivenPeriod( container, period):109    """Function called to list all data that a container has for a given period of time in days"""110    objects = []111    t = timedelta(days=period)112    utc=pytz.UTC113    today = utc.localize(datetime.today())114    try:115        for element in s3_client.list_objects(Bucket=container)['Contents']:116           creationDate = element['LastModified'].replace(tzinfo=utc)117           duration = today - t118           if(creationDate >= duration):119              objects.append(element)120        print(objects)121    except s3_client.exceptions.NoSuchBucket:122          print("No container named "+container+".")123#function to retrieve all data from a container124def retrieveAllDataFromContainer(container):125    """Function called to retrieve and copy in the folder of the python executable all the files contained in a126    certain container """127    try:128         for element in s3_client.list_objects(Bucket=container)['Contents']:129              s3_client.download_file(Bucket=container, Key=element['Key'], Filename= element['Key'])130    except s3_client.exceptions.NoSuchBucket:131          print("No container named "+container+".")132#function to copy an entire folder from ElasticSearch to the OpenIO container133def elasticUploadFolder(container, index, retention=0):134    """Function called to copy an entire folder  """135    if os.name == 'posix':136        slash = "/" # for Linux and macOS137    else:138        slash = chr(92) # '\' for Windows139    host = str(config['elasticsearchDomain']) + ":" + str(config['elasticsearchPort'])140    client = Elasticsearch(host)141    test_connection(client)142    response = client.search(index=index, body={}, size=100)143    elastic_docs = response["hits"]["hits"]144    for num, doc in enumerate(elastic_docs):145        # get _source data dict from document146        source_data = doc["_source"]147        s3_client.upload_file(Filename= source_data[config['elasticsearchFilenameAttribut']],Bucket=container, Key= source_data[config['elasticsearchFilenameAttribut']])148        if retention !=0 :149                s3_client.put_object_retention(Bucket=container, Key= source_data[config['elasticsearchFilenameAttribut']], 150    		Retention={151        		'Mode': 'GOVERNANCE',152       			'RetainUntilDate': datetime.today()+timedelta(retention)153    		}154	)155#function to create a container156def addContainer(container, newACL='private'):157    if newACL=='private' or newACL=='public-read' or newACL=='public-read-write' or newACL=='authenticated-read':158          try:159               s3_client.create_bucket(ACL=newACL, Bucket=container)160          except s3_client.exceptions.BucketAlreadyExists:161               print("Bucket "+container+" already exists!")162          except s3_client.exceptions.ClientError:163                print("Bucket name is not valid.")164                print("Bucket names must be between 3 and 63 characters long.")165                print("Bucket names can consist only of lowercase letters, numbers, dots (.), and hyphens (-).")166                print("Bucket names must begin and end with a letter or number.")167                print("Bucket names must not be formatted as an IP address (for example, 192.168.5.4).")168                print("Bucket names can't begin with xn-- (for buckets created after February 2020).")169    else:170          print("ACL argument is not valid. It must be 'private'|'public-read'|'public-read-write'|'authenticated-read' for a container.")171#function to get the Acess Control List policy of a container172def getBucketACL(container):173    try:174          bucket_acl = s3_client.get_bucket_acl(Bucket=container)175          print(bucket_acl)176    except s3_client.exceptions.NoSuchBucket:177          print("No container named "+container+".")178#function to modify the Acess Control List policy of a container179def putBucketACL(container,newACL):180    if newACL=='private' or newACL=='public-read' or newACL=='public-read-write' or newACL=='authenticated-read':181          try:182                bucket_acl = s3_client.put_bucket_acl(ACL=newACL, Bucket=container)183          except s3_client.exceptions.NoSuchBucket:184                print("No container named "+container+".")185    else:186          print("ACL argument is not valid. It must be 'private'|'public-read'|'public-read-write'|'authenticated-read' for a container.")187#function to get the Acess Control List policy of a file188def getObjectACL(container,filename):189    try:190          object_acl = s3_client.get_object_acl(Bucket=container, Key= filename)191          print(object_acl)192    except s3_client.exceptions.NoSuchBucket:193          print("No container named "+container+".")194    except s3_client.exceptions.NoSuchKey:195          print("No file named "+filename+" in the container "+container+".")196#function to modify the Acess Control List policy of a file197def putObjectACL(container,filename, newACL):198    if newACL=='private' or newACL=='public-read' or newACL=='public-read-write' or newACL=='authenticated-read' or newACL=='bucket-owner-read' or newACL=='bucket-owner-full-control':199          try:200                object_acl = s3_client.put_object_acl(ACL=newACL, Bucket=container, Key= filename)201          except s3_client.exceptions.NoSuchKey:202                print("No file named "+filename+" in the container "+container+".")203          except s3_client.exceptions.NoSuchBucket:204                print("No container named "+container+".")205    else:206          print("ACL argument is not valid. It must be 'private'|'public-read'|'public-read-write'|'authenticated-read'|'aws-exec-read'|'bucket-owner-read'|'bucket-owner-full-control' for an object.")207#function to get the retention policy of an object208def getRetention(container, filename):209    try:210          object_retention = s3_client.get_object_retention(Bucket=container, Key= filename)211          print(object_retention)212    except botocore.parsers.ResponseParserError:213          print("No retention policy for this object")214    except s3_client.exceptions.NoSuchBucket:215          print("No container named "+container+".")216    except s3_client.exceptions.NoSuchKey:217          print("No file named "+filename+" in the container "+container+".")218   219#function to modify the retention policy of an object220def putRetention(container, filename, retention):221    try:222          object_retention = s3_client.put_object_retention(Bucket=container, Key= filename,223          Retention={224             'Mode': 'GOVERNANCE',225             'RetainUntilDate': datetime.today()+timedelta(retention)226           }227           )228    except s3_client.exceptions.NoSuchBucket:229          print("No container named "+container+".")230    except s3_client.exceptions.NoSuchKey:...

Automation Testing Tutorials

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.