Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_s3_right_creds():
try:
bucket = hub.s3("snark-test", aws_creds_filepath=".secrets/s3").connect()
bucket.blob_set("temporary_blob.txt", bytes(1))
bucket.blob_get("temporary_blob.txt")
except:
pytest.fail("Unexpected error.. could not connect to s3")
def test_s3_wrong_creds():
with open("./data/cache/wrong_creds.txt", "w") as f:
f.write("empty")
with pytest.raises(Exception):
bucket = hub.s3(
"snark-test", aws_creds_filepath="./data/cache/wrong_creds.txt"
).connect()
bucket.blob_set("temporary_blob.txt", bytes(1))
bucket.blob_get("temporary_blob.txt")
def test_s3_blob():
bucket = hub.s3("snark-test", aws_creds_filepath=".secrets/s3").connect()
bucket.blob_set("temporary_blob.txt", bytes(1))
text = bucket.blob_get("temporary_blob.txt")
assert text == bytes(1)
def test_s3_delete_item():
bucket = hub.s3("snark-test", aws_creds_filepath=".secrets/s3").connect()
bucket.delete(name="test_array")
with pytest.raises(Exception):
bucket.open(name="test/example:5")
import hub
import numpy as np
import sys, os, time, random, uuid, itertools, json, traceback, io
# Create
conn = hub.s3(
'waymo-dataset-upload',
aws_creds_filepath='.creds/aws.json'
).connect()
x = conn.array_create(
shape = (50000, 250, 250, 3),
chunk=(4, 250, 250, 3),
name= os.path.join('test', f'{int(time.time())}'),
dtype='uint8',
)
# Upload
x[0] = np.ones((250, 250, 3), dtype='uint8')
# Download
def conn_setup(bucket: str = 'waymo-dataset-upload') -> hub.Bucket:
return hub.s3(bucket=bucket, aws_creds_filepath='.creds/aws.json').connect()