core/tests/components/aws/test_init.py

"""Tests for the aws component config and setup."""
from asynctest import patch as async_patch, MagicMock, CoroutineMock

from homeassistant.components import aws
from homeassistant.setup import async_setup_component


class MockAioSession:
    """Mock AioSession."""

    def __init__(self, *args, **kwargs):
        """Init a mock session."""

    def create_client(self, *args, **kwargs):  # pylint: disable=no-self-use
        """Create a mocked client."""
        return MagicMock(
            __aenter__=CoroutineMock(return_value=CoroutineMock(
                get_user=CoroutineMock(),      # iam
                invoke=CoroutineMock(),        # lambda
                publish=CoroutineMock(),       # sns
                send_message=CoroutineMock(),  # sqs
            )),
            __aexit__=CoroutineMock()
        )


async def test_empty_config(hass):
    """Test a default config will be create for empty config."""
    with async_patch('aiobotocore.AioSession', new=MockAioSession):
        await async_setup_component(hass, 'aws', {
            'aws': {}
        })
        await hass.async_block_till_done()

    sessions = hass.data[aws.DATA_SESSIONS]
    assert sessions is not None
    assert len(sessions) == 1
    assert isinstance(sessions.get('default'), MockAioSession)


async def test_empty_credential(hass):
    """Test a default config will be create for empty credential section."""
    with async_patch('aiobotocore.AioSession', new=MockAioSession):
        await async_setup_component(hass, 'aws', {
            'aws': {
                'notify': [{
                    'service': 'lambda',
                    'name': 'New Lambda Test',
                    'region_name': 'us-east-1',
                }]
            }
        })
        await hass.async_block_till_done()

    sessions = hass.data[aws.DATA_SESSIONS]
    assert sessions is not None
    assert len(sessions) == 1
    assert isinstance(sessions.get('default'), MockAioSession)

    assert hass.services.has_service('notify', 'new_lambda_test') is True
    await hass.services.async_call(
        'notify',
        'new_lambda_test',
        {'message': 'test', 'target': 'ARN'},
        blocking=True
    )


async def test_profile_credential(hass):
    """Test credentials with profile name."""
    with async_patch('aiobotocore.AioSession', new=MockAioSession):
        await async_setup_component(hass, 'aws', {
            'aws': {
                'credentials': {
                    'name': 'test',
                    'profile_name': 'test-profile',
                },
                'notify': [{
                    'service': 'sns',
                    'credential_name': 'test',
                    'name': 'SNS Test',
                    'region_name': 'us-east-1',
                }]
            }
        })
        await hass.async_block_till_done()

    sessions = hass.data[aws.DATA_SESSIONS]
    assert sessions is not None
    assert len(sessions) == 1
    assert isinstance(sessions.get('test'), MockAioSession)

    assert hass.services.has_service('notify', 'sns_test') is True
    await hass.services.async_call(
        'notify',
        'sns_test',
        {'title': 'test', 'message': 'test', 'target': 'ARN'},
        blocking=True
    )


async def test_access_key_credential(hass):
    """Test credentials with access key."""
    with async_patch('aiobotocore.AioSession', new=MockAioSession):
        await async_setup_component(hass, 'aws', {
            'aws': {
                'credentials': [
                    {
                        'name': 'test',
                        'profile_name': 'test-profile',
                    },
                    {
                        'name': 'key',
                        'aws_access_key_id': 'test-key',
                        'aws_secret_access_key': 'test-secret',
                    },
                ],
                'notify': [{
                    'service': 'sns',
                    'credential_name': 'key',
                    'name': 'SNS Test',
                    'region_name': 'us-east-1',
                }]
            }
        })
        await hass.async_block_till_done()

    sessions = hass.data[aws.DATA_SESSIONS]
    assert sessions is not None
    assert len(sessions) == 2
    assert isinstance(sessions.get('key'), MockAioSession)

    assert hass.services.has_service('notify', 'sns_test') is True
    await hass.services.async_call(
        'notify',
        'sns_test',
        {'title': 'test', 'message': 'test', 'target': 'ARN'},
        blocking=True
    )


async def test_notify_credential(hass):
    """Test notify service can use access key directly."""
    with async_patch('aiobotocore.AioSession', new=MockAioSession):
        await async_setup_component(hass, 'aws', {
            'aws': {
                'notify': [{
                    'service': 'sqs',
                    'credential_name': 'test',
                    'name': 'SQS Test',
                    'region_name': 'us-east-1',
                    'aws_access_key_id': 'some-key',
                    'aws_secret_access_key': 'some-secret',
                }]
            }
        })
        await hass.async_block_till_done()

    sessions = hass.data[aws.DATA_SESSIONS]
    assert sessions is not None
    assert len(sessions) == 1
    assert isinstance(sessions.get('default'), MockAioSession)

    assert hass.services.has_service('notify', 'sqs_test') is True
    await hass.services.async_call(
        'notify',
        'sqs_test',
        {'message': 'test', 'target': 'ARN'},
        blocking=True
    )


async def test_notify_credential_profile(hass):
    """Test notify service can use profile directly."""
    with async_patch('aiobotocore.AioSession', new=MockAioSession):
        await async_setup_component(hass, 'aws', {
            'aws': {
                'notify': [{
                    'service': 'sqs',
                    'name': 'SQS Test',
                    'region_name': 'us-east-1',
                    'profile_name': 'test',
                }]
            }
        })
        await hass.async_block_till_done()

    sessions = hass.data[aws.DATA_SESSIONS]
    assert sessions is not None
    assert len(sessions) == 1
    assert isinstance(sessions.get('default'), MockAioSession)

    assert hass.services.has_service('notify', 'sqs_test') is True
    await hass.services.async_call(
        'notify',
        'sqs_test',
        {'message': 'test', 'target': 'ARN'},
        blocking=True
    )