feat(backup): wrapper de stockage S3 (boto3) + faux stockage de test (#89)
This commit is contained in:
@@ -0,0 +1,15 @@
|
||||
class FakeStorage:
|
||||
def __init__(self):
|
||||
self.objects: dict[str, bytes] = {}
|
||||
|
||||
def upload_bytes(self, key, data):
|
||||
self.objects[key] = data
|
||||
|
||||
def download_bytes(self, key):
|
||||
return self.objects[key]
|
||||
|
||||
def list_keys(self, prefix):
|
||||
return [k for k in self.objects if k.startswith(prefix)]
|
||||
|
||||
def delete(self, key):
|
||||
self.objects.pop(key, None)
|
||||
@@ -0,0 +1,31 @@
|
||||
import boto3
|
||||
import pytest
|
||||
from moto import mock_aws
|
||||
|
||||
from src.backup.storage import S3Storage
|
||||
|
||||
BUCKET = "decp-backups"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def s3_storage():
|
||||
with mock_aws():
|
||||
boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET)
|
||||
yield S3Storage(
|
||||
bucket=BUCKET,
|
||||
endpoint_url=None,
|
||||
access_key="testing",
|
||||
secret_key="testing",
|
||||
)
|
||||
|
||||
|
||||
def test_upload_list_download_delete(s3_storage):
|
||||
s3_storage.upload_bytes("backups/a.enc", b"alpha")
|
||||
s3_storage.upload_bytes("backups/b.enc", b"beta")
|
||||
s3_storage.upload_bytes("autre/c.enc", b"gamma")
|
||||
|
||||
assert set(s3_storage.list_keys("backups/")) == {"backups/a.enc", "backups/b.enc"}
|
||||
assert s3_storage.download_bytes("backups/a.enc") == b"alpha"
|
||||
|
||||
s3_storage.delete("backups/a.enc")
|
||||
assert s3_storage.list_keys("backups/") == ["backups/b.enc"]
|
||||
Reference in New Issue
Block a user