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

Copied to Clipboard!

How to use _s3_client_custom_config method in localstack

Best Python code snippet using localstack_python

test_s3.py

Source:test_s3.py

...2182        s3_endpoint_path_style = _endpoint_url()2183        s3_create_bucket(Bucket=bucket_name)2184        object_key = "temp.txt"2185        s3_config = {"addressing_style": "virtual"} if use_virtual_address else {}2186        client = _s3_client_custom_config(2187            Config(signature_version=signature_version, s3=s3_config),2188            endpoint_url=s3_endpoint_path_style,2189        )2190        upload_id = client.create_multipart_upload(2191            Bucket=bucket_name,2192            Key=object_key,2193        )["UploadId"]2194        data = to_bytes("hello this is a upload test")2195        upload_file_object = BytesIO(data)2196        signed_url = _generate_presigned_url(2197            client,2198            {2199                "Bucket": bucket_name,2200                "Key": object_key,2201                "UploadId": upload_id,2202                "PartNumber": 1,2203            },2204            expires=4,2205            client_method="upload_part",2206        )2207        response = requests.put(signed_url, data=upload_file_object)2208        assert response.status_code == 2002209        multipart_upload_parts = [{"ETag": response.headers["ETag"], "PartNumber": 1}]2210        response = client.complete_multipart_upload(2211            Bucket=bucket_name,2212            Key=object_key,2213            MultipartUpload={"Parts": multipart_upload_parts},2214            UploadId=upload_id,2215        )2216        assert 200 == response["ResponseMetadata"]["HTTPStatusCode"]2217        simple_params = {"Bucket": bucket_name, "Key": object_key}2218        response = requests.get(_generate_presigned_url(client, simple_params, 4))2219        assert 200 == response.status_code2220        assert response.content == data2221    @pytest.mark.parametrize(2222        "signature_version, use_virtual_address",2223        [2224            ("s3", False),2225            ("s3", True),2226            ("s3v4", False),2227            ("s3v4", True),2228        ],2229    )2230    @pytest.mark.aws_validated2231    def test_presigned_url_signature_authentication_expired(2232        self,2233        s3_client,2234        s3_create_bucket,2235        signature_version,2236        use_virtual_address,2237        monkeypatch,2238    ):2239        monkeypatch.setattr(config, "S3_SKIP_SIGNATURE_VALIDATION", False)2240        bucket_name = f"presign-{short_uid()}"2241        s3_endpoint_path_style = _endpoint_url()2242        s3_create_bucket(Bucket=bucket_name)2243        object_key = "temp.txt"2244        s3_client.put_object(Key=object_key, Bucket=bucket_name, Body="123")2245        s3_config = {"addressing_style": "virtual"} if use_virtual_address else {}2246        client = _s3_client_custom_config(2247            Config(signature_version=signature_version, s3=s3_config),2248            endpoint_url=s3_endpoint_path_style,2249        )2250        url = _generate_presigned_url(client, {"Bucket": bucket_name, "Key": object_key}, expires=1)2251        time.sleep(1)2252        assert 403 == requests.get(url).status_code2253    @pytest.mark.parametrize(2254        "signature_version, use_virtual_address",2255        [2256            ("s3", False),2257            ("s3", True),2258            ("s3v4", False),2259            ("s3v4", True),2260        ],2261    )2262    @pytest.mark.aws_validated2263    def test_presigned_url_signature_authentication(2264        self,2265        s3_client,2266        s3_create_bucket,2267        signature_version,2268        use_virtual_address,2269        monkeypatch,2270    ):2271        monkeypatch.setattr(config, "S3_SKIP_SIGNATURE_VALIDATION", False)2272        bucket_name = f"presign-{short_uid()}"2273        s3_endpoint_path_style = _endpoint_url()2274        s3_url = _bucket_url_vhost(bucket_name) if use_virtual_address else _bucket_url(bucket_name)2275        s3_create_bucket(Bucket=bucket_name)2276        object_key = "temp.txt"2277        s3_client.put_object(Key=object_key, Bucket=bucket_name, Body="123")2278        s3_config = {"addressing_style": "virtual"} if use_virtual_address else {}2279        client = _s3_client_custom_config(2280            Config(signature_version=signature_version, s3=s3_config),2281            endpoint_url=s3_endpoint_path_style,2282        )2283        expires = 42284        # GET requests2285        simple_params = {"Bucket": bucket_name, "Key": object_key}2286        response = requests.get(_generate_presigned_url(client, simple_params, expires))2287        assert 200 == response.status_code2288        assert response.content == b"123"2289        params = {2290            "Bucket": bucket_name,2291            "Key": object_key,2292            "ResponseContentType": "text/plain",2293            "ResponseContentDisposition": "attachment;  filename=test.txt",2294        }2295        presigned = _generate_presigned_url(client, params, expires)2296        response = requests.get(_generate_presigned_url(client, params, expires))2297        assert 200 == response.status_code2298        assert response.content == b"123"2299        object_data = "this should be found in when you download {}.".format(object_key)2300        # invalid requests2301        # TODO check how much sense it makes to make this url "invalid"...2302        assert (2303            4032304            == requests.get(2305                _make_url_invalid(s3_url, object_key, presigned),2306                data=object_data,2307                headers={"Content-Type": "my-fake-content/type"},2308            ).status_code2309        )2310        # put object valid2311        assert (2312            2002313            == requests.put(2314                _generate_presigned_url(client, simple_params, expires, client_method="put_object"),2315                data=object_data,2316            ).status_code2317        )2318        params = {2319            "Bucket": bucket_name,2320            "Key": object_key,2321            "ContentType": "text/plain",2322        }2323        presigned_put_url = _generate_presigned_url(2324            client, params, expires, client_method="put_object"2325        )2326        assert (2327            2002328            == requests.put(2329                presigned_put_url,2330                data=object_data,2331                headers={"Content-Type": "text/plain"},2332            ).status_code2333        )2334        # Invalid request2335        response = requests.put(2336            _make_url_invalid(s3_url, object_key, presigned_put_url),2337            data=object_data,2338            headers={"Content-Type": "my-fake-content/type"},2339        )2340        assert 403 == response.status_code2341        # DELETE requests2342        presigned_delete_url = _generate_presigned_url(2343            client, simple_params, expires, client_method="delete_object"2344        )2345        response = requests.delete(presigned_delete_url)2346        assert 204 == response.status_code2347class TestS3DeepArchive:2348    """2349    Test to cover DEEP_ARCHIVE Storage Class functionality.2350    """2351    @pytest.mark.aws_validated2352    def test_storage_class_deep_archive(self, s3_client, s3_resource, s3_bucket, tmpdir):2353        key = "my-key"2354        transfer_config = TransferConfig(multipart_threshold=5 * KB, multipart_chunksize=1 * KB)2355        def upload_file(size_in_kb: int):2356            file = tmpdir / f"test-file-{short_uid()}.bin"2357            data = b"1" * (size_in_kb * KB)2358            file.write(data=data, mode="w")2359            s3_client.upload_file(2360                Bucket=s3_bucket,2361                Key=key,2362                Filename=str(file.realpath()),2363                ExtraArgs={"StorageClass": "DEEP_ARCHIVE"},2364                Config=transfer_config,2365            )2366        upload_file(1)2367        upload_file(9)2368        upload_file(15)2369        objects = s3_resource.Bucket(s3_bucket).objects.all()2370        keys = []2371        for obj in objects:2372            keys.append(obj.key)2373            assert obj.storage_class == "DEEP_ARCHIVE"2374def _anon_client(service: str):2375    conf = Config(signature_version=UNSIGNED)2376    if os.environ.get("TEST_TARGET") == "AWS_CLOUD":2377        return boto3.client(service, config=conf, region_name=None)2378    return aws_stack.create_external_boto_client(service, config=conf)2379def _s3_client_custom_config(conf: Config, endpoint_url: str):2380    if os.environ.get("TEST_TARGET") == "AWS_CLOUD":2381        return boto3.client("s3", config=conf, endpoint_url=endpoint_url)2382    # TODO in future this should work with aws_stack.create_external_boto_client2383    #      currently it doesn't as authenticate_presign_url_signv2 requires the secret_key to be 'test'2384    # return aws_stack.create_external_boto_client(2385    #     "s3",2386    #     config=conf,2387    #     endpoint_url=endpoint_url,2388    #     aws_access_key_id=TEST_AWS_ACCESS_KEY_ID,2389    # )2390    return boto3.client(2391        "s3",2392        endpoint_url=endpoint_url,2393        config=conf,...

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.