Class FakeSession

tests/unit/test_clidriver.py:221–296

class FakeSession:
    def __init__(self, emitter=None):
        self.operation = None
        if emitter is None:
            emitter = HierarchicalEmitter()
        self.emitter = emitter
        self.profile = None
        self.stream_logger_args = None
        self.credentials = 'fakecredentials'
        self.session_vars = {}
        self.config_store = self._register_config_store()
        self.user_agent_name = 'aws-cli'
        self.user_agent_version = '100.100.100'

    def _register_config_store(self):
        chain_builder = ConfigChainFactory(session=self)
        config_store = ConfigValueStore(
            mapping=create_botocore_default_config_mapping(chain_builder)
        )
        return config_store

    def register(self, event_name, handler):
        self.emitter.register(event_name, handler)

    def emit(self, event_name, **kwargs):
        return self.emitter.emit(event_name, **kwargs)

    def emit_first_non_none_response(self, event_name, **kwargs):
        responses = self.emitter.emit(event_name, **kwargs)
        for _, response in responses:
            if response is not None:
                return response

    def get_component(self, name):
        if name == 'event_emitter':
            return self.emitter
        if name == 'config_store':
            return self.config_store

    def create_client(self, *args, **kwargs):
        client = mock.Mock()
        client.list_objects.return_value = {}
        client.can_paginate.return_value = False
        return client

    def get_available_services(self):
        return ['s3']

    def get_data(self, name):
        return GET_DATA[name]

    def get_config_variable(self, name):
        if name in GET_VARIABLE:
            return GET_VARIABLE[name]
        return self.session_vars[name]

    def get_service_model(self, name, api_version=None):
        return botocore.model.ServiceModel(MINI_SERVICE, service_name='s3')