How to use fake s3 bucket

Question

I tried to use aiobotocore for async work with Ceph, but i have no access to real bucket, then i need to mock s3 bucket to create a fake one

I'm trying to use moto but it anyway want to connect a real bucket

    @moto.mock_aws
    async def _load_asl_by_type(self, file: File, typ: ContentType) -> AsyncIterable[Tuple[str, Content]]:
        session = get_session()
        async with session.create_client('s3', region_name='us-east-1', aws_secret_access_key='moto',
                                         aws_access_key_id='moto') as s3:
            bucket_name = 'y-bucket'
            await s3.create_bucket(Bucket=bucket_name)
            await s3.put_object(Bucket=bucket_name, Key=file.path)
            try:
                async with AsyncFileIO(io=await s3.get_object(Bucket=bucket_name, Key=file.path)) as f:
                    file.in_progress()
                    collection = json.load(f)
                    for name, data in collection.items():
                        def _spec_load() -> Any:
                            _logger.info(f"Loading {typ.__qualname__} from {file.path}: {name}")
                            return spec.load(typ, data, key=name, options=self._options)

                        obj = await asyncio.get_running_loop().run_in_executor(None, _spec_load)
                        file.loaded()
                        yield name, obj
            except Exception as error:
                _logger.error(f"Failed to load file {file.path}: {error}")
                file.failed()
                raise error

Traceback:

 raise ClientConnectorCertificateError(req.connection_key, exc) from exc
aiohttp.client_exceptions.ClientConnectorCertificateError: Cannot connect to host y-bucket.s3.amazonaws.com:443 ssl:True [SSLCertVerificationError: (1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1133)')]

raise SSLError(endpoint_url=request.url, error=e)
botocore.exceptions.SSLError: SSL validation failed for https://y-bucket.s3.amazonaws.com/ Cannot connect to host y-bucket.s3.amazonaws.com:443 ssl:True [SSLCertVerificationError: (1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1133)')]

Answer 1

I think it'd be a good idea to encapsulate the s3 interaction logic in an S3 client with a specific interface. I'll use get_item_by_id instead of _load_asl_by_type method just to simplify the example:

import uuid
from abc import ABC, abstractmethod

from pydantic import UUID4


class S3ClientInterface(ABC):
    """S3 client interface."""

    @abstractmethod
    async def get_item_by_id(self, item_id: UUID4) -> Item:
        """Get item by item id provided."""


class S3Client(S3ClientInterface):
    """S3 client."""

    def __init__(self) -> None:
        """."""

        self.session = ...

    async def get_item_by_id(self, item_id: UUID4) -> Item:
        """Get item from real S3 by id provided."""

        async with self.session.create_client(...):
            ...
            return item

Now you can make your "upper level" e.g. controller dependent from the S3ClientInterface, but not the S3Client:

class ItemController:
    """Item controller."""
    
    def __init__(self, s3_client: S3ClientInterface) -> None:
        """."""
        
        self.s3_client = s3_client
        
    async def process_item(self, item_id: UUID4) -> Item:
        """Get item from s3 and postprocess somehow."""
        
        item = await self.s3_client.get_item_by_id(item_id=item_id)
        
        # Do something with item
        # Do more
        
        return item

And during testing your controller you can mock the whole S3 client with a testing one implementing the same interface:

class MockS3Client(S3ClientInterface):
    """Mock S3 client."""

    async def get_item_by_id(self, item_id: UUID4) -> Item:
        """Get mocked item from mocked S3 by id provided."""
        
        # Return mocked item
        return Item( 
            id=item_id,
            name="Item",
            size=5
        )

    
# Tests
controller = ItemController(s3_client=MockS3Client())
controller.process_item(item_id=...)  # Will use mocked get_item_by_id