KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
