Remove test that only runs locally
This commit is contained in:
parent
5e4736e233
commit
410d9ee901
@ -4393,78 +4393,3 @@ def test_s3_config_dict():
|
||||
assert not logging_bucket["supplementaryConfiguration"].get(
|
||||
"BucketTaggingConfiguration"
|
||||
)
|
||||
|
||||
|
||||
@mock_s3
|
||||
def test_delete_downloaded_file():
|
||||
# SET UP
|
||||
filename = 'some_large_file.pdf'
|
||||
file = open(filename, 'rb')
|
||||
uploader = PdfFileUploader(file)
|
||||
boto3.client('s3').create_bucket(Bucket=uploader.bucket_name())
|
||||
uploader.upload()
|
||||
|
||||
downloader = PdfFileDownloader(uploader.full_bucket_file_name())
|
||||
downloader.download()
|
||||
|
||||
downloader.delete_downloaded_file()
|
||||
|
||||
|
||||
from pathlib import Path
|
||||
import re
|
||||
import os
|
||||
class PdfFileDownloader:
|
||||
def __init__(self, full_bucket_file_name):
|
||||
self.bucket_name, self.file_name = self.extract(full_bucket_file_name)
|
||||
self.s3 = boto3.client('s3')
|
||||
|
||||
def download(self):
|
||||
try:
|
||||
self.s3.download_file(self.bucket_name, self.file_name, self.local_path())
|
||||
|
||||
return self.local_path()
|
||||
except ClientError as exc:
|
||||
raise exc
|
||||
|
||||
def local_path(self):
|
||||
return '/tmp/' + self.file_name.replace('/', '')
|
||||
|
||||
def delete_downloaded_file(self):
|
||||
if Path(self.local_path()).is_file():
|
||||
print("Removing " + str(self.local_path()))
|
||||
os.remove(self.local_path())
|
||||
|
||||
def file(self):
|
||||
return open(self.local_path(), 'rb')
|
||||
|
||||
def extract(self, full_bucket_file_name):
|
||||
match = re.search(r'([\.a-zA-Z_-]+)\/(.*)', full_bucket_file_name)
|
||||
|
||||
if match and len(match.groups()) == 2:
|
||||
return (match.groups()[0], match.groups()[1])
|
||||
else:
|
||||
raise RuntimeError(f"Cannot determine bucket and file name for {full_bucket_file_name}")
|
||||
|
||||
|
||||
import binascii
|
||||
class PdfFileUploader:
|
||||
def __init__(self, file):
|
||||
self.file = file
|
||||
date = datetime.datetime.now().strftime('%Y%m%d%H%M%S')
|
||||
random_hex = binascii.b2a_hex(os.urandom(16)).decode('ascii')
|
||||
self.bucket_file_name = f"{date}_{random_hex}.pdf"
|
||||
|
||||
def upload(self):
|
||||
self.file.seek(0)
|
||||
boto3.client('s3').upload_fileobj(self.file, self.bucket_name(), self.bucket_file_name)
|
||||
|
||||
return (self.original_file_name(), self.full_bucket_file_name())
|
||||
|
||||
def original_file_name(self):
|
||||
return os.path.basename(self.file.name)
|
||||
|
||||
def bucket_name(self):
|
||||
return 'test_bucket' #os.environ['AWS_BUCKET_NAME']
|
||||
|
||||
def full_bucket_file_name(self):
|
||||
return f"{self.bucket_name()}/{self.bucket_file_name}"
|
||||
|
Loading…
Reference in New Issue
Block a user