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

Copied to Clipboard!

How to use s3_api method in localstack

Best Python code snippet using localstack_python

test_s3_api.py

Source:test_s3_api.py

1import json2from importlib.resources import open_text3from unittest import mock4from botocore.exceptions import ClientError5from django.test import TestCase6from djaws import s3_api7from djaws.s3.exceptions import S3Error, BucketAlreadyExists, BucketAlreadyOwnedByYouError8class CreatePublicAccessBucketTests(TestCase):9    @mock.patch.object(s3_api, 'create_bucket')10    @mock.patch.object(s3_api, 'put_public_access_block')11    @mock.patch.object(s3_api, 'put_bucket_cors')12    def test_ok(self, mock_create_bucket, mock_put_public_access_block, mock_put_bucket_cors):13        mock_create_bucket.return_value = None14        mock_put_public_access_block.return_value = True15        mock_put_bucket_cors.return_value = True16        r_bucket = s3_api.create_public_access_bucket(17            s3_api.s3_session_client(),18            bucket='my-test-bucket-2',19            region='us-west-2'20        )21        self.assertTrue(r_bucket)22class CreateBucketTests(TestCase):23    @classmethod24    def setUpTestData(cls):25        with open_text('djaws.tests.schemas.s3', 'create_bucket.json') as bucket_json:26            cls.r_bucket_data = json.loads(bucket_json.read())27    @mock.patch.object(s3_api, 's3_session_client')28    def test_ok(self, mock_client):29        mock_client.create_bucket.return_value = self.r_bucket_data30        r_bucket = s3_api.create_bucket(31            mock_client,32            bucket='bucket1',33            region='us-west-2',34        )35        self.assertEquals(36            r_bucket.location,37            'http://examplebucket.s3.amazonaws.com/'38        )39    @mock.patch.object(s3_api, 's3_session_client')40    def test_ok_no_region(self, mock_client):41        mock_client.create_bucket.return_value = self.r_bucket_data42        r_bucket = s3_api.create_bucket(43            mock_client,44            bucket='bucket1',45            region='us-west-2',46        )47        self.assertEquals(48            r_bucket.location,49            'http://examplebucket.s3.amazonaws.com/'50        )51    @mock.patch.object(s3_api, 's3_session_client')52    def test_BucketAlreadyExists(self, mock_client):53        mock_client.create_bucket.side_effect = ClientError(54            {55                'Error': {56                    'Code': 'BucketAlreadyExists',57                    'Message': 'Details/context around the exception or error'58                },59                'ResponseMetadata': {60                    'RequestId': '1234567890ABCDEF',61                    'HostId': 'host ID data will appear here as a hash',62                    'HTTPStatusCode': 400,63                    'HTTPHeaders': {'header metadata key/values will appear here'},64                    'RetryAttempts': 065                }66            },67            'generate_presigned_post'68        )69        with self.assertRaises(BucketAlreadyExists):70            s3_api.create_bucket(71                mock_client,72                bucket='bucket1',73                region='us-west-2',74            )75class PutPublicAccessBlockTests(TestCase):76    @classmethod77    def setUpTestData(cls):78        cls.access_block_data = {79            'bucket': 'bucket1',80            'config': s3_api.PublicAccessBlockConfiguration(81                block_public_acls=False,82                ignore_public_acls=False,83                block_public_policy=False,84                restrict_public_buckets=False,85            )86        }87    @mock.patch.object(s3_api, 's3_session_client')88    def test_put_public_access_block_ok(self, mock_client):89        mock_client.put_public_access_block.return_value = True90        r_access_block = s3_api.put_public_access_block(91            mock_client,92            **self.access_block_data,93        )94        self.assertTrue(r_access_block)95    @mock.patch.object(s3_api, 's3_session_client')96    def test_put_public_access_block_error(self, mock_client):97        mock_client.put_public_access_block.side_effect = ClientError(98            error_response=dict(),99            operation_name='put_public_access_block'100        )101        with self.assertRaises(S3Error):102            s3_api.put_public_access_block(103                mock_client,104                **self.access_block_data,105            )106class CreateBucketCORSTests(TestCase):107    @classmethod108    def setUpTestData(cls):109        cls.cors_data = {110            'bucket': 'bucket1',111            'cors_config': {112                'CORSRules': [113                    {114                        "AllowedHeaders": [115                            "*"116                        ],117                        "AllowedMethods": [118                            "GET"119                        ],120                        "AllowedOrigins": [121                            "*"122                        ],123                        "MaxAgeSeconds": 3000124                    },125                ],126            },127        }128    @mock.patch.object(s3_api, 's3_session_client')129    def test_put_bucket_cors_ok(self, mock_client):130        mock_client.put_bucket_cors.return_value = True131        r_cors = s3_api.put_bucket_cors(132            mock_client,133            **self.cors_data,134        )135        self.assertTrue(r_cors)136    @mock.patch.object(s3_api, 's3_session_client')137    def test_put_bucket_cors_error(self, mock_client):138        mock_client.put_bucket_cors.side_effect = ClientError(139            error_response=dict(),140            operation_name='put_bucket_cors'141        )142        with self.assertRaises(S3Error):143            s3_api.put_bucket_cors(144                mock_client,145                **self.cors_data,146            )147class UploadObjectTests(TestCase):148    @classmethod149    def setUpTestData(cls):150        cls.upload_data = {151            'data': b'test_image',152            'bucket': 'bucket1',153            'uri': 'https://example.com/test.jpg',154            'key': '998adad2-c217-402e-b259-ace44a26dff4.jpg'155        }156    @mock.patch.object(s3_api, 's3_session_client')157    def test_ok(self, mock_client):158        mock_client.put_object.return_value = None159        r_upload = s3_api.upload_object(160            mock_client,161            **self.upload_data,162        )163        self.assertEquals(164            r_upload.public_s3_url,165            'https://bucket1.s3.amazonaws.com/998adad2-c217-402e-b259-ace44a26dff4.jpg'166        )167    @mock.patch.object(s3_api, 's3_session_client')168    def test_ok_no_object_name(self, mock_client):169        mock_client.put_object.return_value = None170        upload_data = self.upload_data171        upload_data.pop('key')172        r_upload = s3_api.upload_object(173            mock_client,174            **upload_data,175        )176        self.assertTrue(r_upload.public_s3_url.startswith('https://bucket1.s3.amazonaws.com/'))177        self.assertTrue(r_upload.public_s3_url.endswith('.jpg'))178    @mock.patch.object(s3_api, 's3_session_client')179    def test_client_error(self, mock_client):180        mock_client.put_object.side_effect = ClientError(181            error_response=dict(),182            operation_name='put_object'183        )184        with self.assertRaises(S3Error):185            s3_api.upload_object(186                mock_client,187                **self.upload_data,188            )189class GeneratePresignedPostTests(TestCase):190    @classmethod191    def setUpTestData(cls):192        cls.generate_presigned_post = {193            'bucket': 'bucket1',194            'object_name': '998adad2-c217-402e-b259-ace44a26dff4.jpg'195        }196        with open_text('djaws.tests.schemas.s3', 'generate_presigned_post.json') as generate_presigned_post_json:197            cls.r_generate_presigned_post_data = json.loads(generate_presigned_post_json.read())198    @mock.patch.object(s3_api, 's3_session_client')199    def test_generate_presigned_post_ok(self, mock_client):200        mock_client.generate_presigned_post.return_value = self.r_generate_presigned_post_data201        r_generate_presigned_post = s3_api.generate_presigned_post(202            mock_client,203            **self.generate_presigned_post,204        )205        self.assertEquals(r_generate_presigned_post.url, 'https://mybucket.s3.amazonaws.com')206        self.assertEquals(r_generate_presigned_post.fields['acl'], 'public-read')207        self.assertEquals(r_generate_presigned_post.fields['key'], 'mykey')208        self.assertEquals(r_generate_presigned_post.fields['signature'], 'mysignature')209        self.assertEquals(r_generate_presigned_post.fields['policy'], 'mybase64 encoded policy')210    @mock.patch.object(s3_api, 's3_session_client')211    def test_upload_object_error(self, mock_client):212        mock_client.generate_presigned_post.side_effect = ClientError(213            {214                'Error': {215                    'Code': 'NotAuthorizedException',216                    'Message': 'Details/context around the exception or error'217                },218                'ResponseMetadata': {219                    'RequestId': '1234567890ABCDEF',220                    'HostId': 'host ID data will appear here as a hash',221                    'HTTPStatusCode': 400,222                    'HTTPHeaders': {'header metadata key/values will appear here'},223                    'RetryAttempts': 0224                }225            },226            'generate_presigned_post'227        )228        with self.assertRaises(S3Error):229            s3_api.generate_presigned_post(230                mock_client,231                **self.generate_presigned_post,...

s3.py

Source:s3.py

1from typing import List, Dict2from fastapi import UploadFile3from starlette.responses import StreamingResponse4from fastapi import APIRouter, status, HTTPException5from utils.s3 import S3Loader6from endpoints.base import prefix7router = APIRouter(prefix=prefix)8@router.get("/s3_api/file/{file_path:path}",9            status_code=status.HTTP_200_OK)10async def get_file(file_path: str):11    s3 = S3Loader(file_path=file_path)12    if s3.object is not None:13        return StreamingResponse(14            s3.streamer(), media_type=s3.object['ContentType'], headers={15                "Content-Disposition": f"attachment;filename={s3.file_name}"})16    raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,17                        detail="Resource Not Found")18@router.get("/s3_api/directory/{dir_path:path}",19            status_code=status.HTTP_200_OK)20async def get_list_files_in_directory(dir_path: str):21    s3 = S3Loader(dir_path=dir_path)22    if 'Contents' in s3.objects:23        return s3.get_file_names()24    raise HTTPException(status_code=status.HTTP_204_NO_CONTENT,25                        detail="Folder isn't exist")26@router.post("/s3_api/{dir_path:path}",27             response_model=Dict,28             status_code=status.HTTP_201_CREATED)29async def upload_file(files: List[UploadFile],30                      dir_path: str):31    s3 = S3Loader(files=files, file_path=dir_path)32    if s3.upload_files():33        return {34            "Files uploaded ": [35                dir_path +36                '/' +37                file.filename for file in files]}38    raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,39                        detail="Couldn't uploaded files")40@router.put("/s3_api/{file_path:path}",41            status_code=status.HTTP_200_OK)42async def update_file(files: UploadFile,43                      file_path: str):44    s3 = S3Loader(files=[files], file_path=file_path)45    s3.update_file()46    return {"Updated File": file_path}47@router.delete("/s3_api/{file_path:path}",48               status_code=status.HTTP_200_OK)49async def delete_file(file_path: str):50    s3 = S3Loader(file_path=file_path)51    s3.delete_file()...

main.py

Source:main.py

1from fastapi import FastAPI, File, Header, Request, Response2from fastapi.middleware.cors import CORSMiddleware3from fastapi.responses import StreamingResponse4from typing import Optional5import socket6import os7import model8import object_store9import s3_api10api = FastAPI()11api.add_middleware(12    CORSMiddleware,13    allow_origins=["*"],14    allow_credentials=True,15    allow_methods=["*"],16    allow_headers=["*"],17)18s3 = FastAPI()19s3.add_middleware(20    CORSMiddleware,21    allow_origins=["*"],22    allow_credentials=True,23    allow_methods=["*"],24    allow_headers=["*"],25)26# S3 API27s3.include_router(s3_api.api)28@s3.exception_handler(s3_api.S3ApiException)29async def s3_api_exception_handler(req: Request, ex: s3_api.S3ApiException):30    return Response(ex.body, status_code=ex.status_code, headers=s3_api.XML_HEADERS)31# API for UI32@api.get("/api")33def index():34    return {35        "objects": model.list_objects(),36    }37@api.get("/api/{object_id}")38def object_index(object_id, query=None):39    return {40        "ticketCount": model.get_ticket_count(object_id),41        "nodes": model.get_object_nodes(object_id),42        "tickets": model.get_object_tickets(object_id),43        "size": model.get_object_size(object_id),44        "contentType": model.get_content_type(object_id) or "text/plain",45    }46@api.get("/api/{object_id}/stream")47async def object_stream(object_id):48    return StreamingResponse(49        object_store.stream(object_id),50        headers={51            "Content-Disposition": f"attachment; filename={object_id}",  # TODO: Support object names52            "Content-Type": "application/octet-stream",  # TODO: Support mimetypes53        },...

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.