core/tests/components/mailgun/test_init.py

"""Test the init file of Mailgun."""
import hashlib
import hmac
from unittest.mock import Mock

import pytest

from homeassistant import data_entry_flow
from homeassistant.components import mailgun, webhook
from homeassistant.const import CONF_API_KEY, CONF_DOMAIN
from homeassistant.core import callback
from homeassistant.setup import async_setup_component

API_KEY = "abc123"


@pytest.fixture
async def http_client(hass, aiohttp_client):
    """Initialize a Home Assistant Server for testing this module."""
    await async_setup_component(hass, webhook.DOMAIN, {})
    return await aiohttp_client(hass.http.app)


@pytest.fixture
async def webhook_id_with_api_key(hass):
    """Initialize the Mailgun component and get the webhook_id."""
    await async_setup_component(
        hass,
        mailgun.DOMAIN,
        {mailgun.DOMAIN: {CONF_API_KEY: API_KEY, CONF_DOMAIN: "example.com"}},
    )

    hass.config.api = Mock(base_url="http://example.com")
    result = await hass.config_entries.flow.async_init(
        "mailgun", context={"source": "user"}
    )
    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM, result

    result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
    assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY

    return result["result"].data["webhook_id"]


@pytest.fixture
async def webhook_id_without_api_key(hass):
    """Initialize the Mailgun component and get the webhook_id w/o API key."""
    await async_setup_component(hass, mailgun.DOMAIN, {})

    hass.config.api = Mock(base_url="http://example.com")
    result = await hass.config_entries.flow.async_init(
        "mailgun", context={"source": "user"}
    )
    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM, result

    result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
    assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY

    return result["result"].data["webhook_id"]


@pytest.fixture
async def mailgun_events(hass):
    """Return a list of mailgun_events triggered."""
    events = []

    @callback
    def handle_event(event):
        """Handle Mailgun event."""
        events.append(event)

    hass.bus.async_listen(mailgun.MESSAGE_RECEIVED, handle_event)

    return events


async def test_mailgun_webhook_with_missing_signature(
    http_client, webhook_id_with_api_key, mailgun_events
):
    """Test that webhook doesn't trigger an event without a signature."""
    event_count = len(mailgun_events)

    await http_client.post(
        "/api/webhook/{}".format(webhook_id_with_api_key),
        json={"hello": "mailgun", "signature": {}},
    )

    assert len(mailgun_events) == event_count

    await http_client.post(
        "/api/webhook/{}".format(webhook_id_with_api_key), json={"hello": "mailgun"}
    )

    assert len(mailgun_events) == event_count


async def test_mailgun_webhook_with_different_api_key(
    http_client, webhook_id_with_api_key, mailgun_events
):
    """Test that webhook doesn't trigger an event with a wrong signature."""
    timestamp = "1529006854"
    token = "a8ce0edb2dd8301dee6c2405235584e45aa91d1e9f979f3de0"

    event_count = len(mailgun_events)

    await http_client.post(
        "/api/webhook/{}".format(webhook_id_with_api_key),
        json={
            "hello": "mailgun",
            "signature": {
                "signature": hmac.new(
                    key=b"random_api_key",
                    msg=bytes("{}{}".format(timestamp, token), "utf-8"),
                    digestmod=hashlib.sha256,
                ).hexdigest(),
                "timestamp": timestamp,
                "token": token,
            },
        },
    )

    assert len(mailgun_events) == event_count


async def test_mailgun_webhook_event_with_correct_api_key(
    http_client, webhook_id_with_api_key, mailgun_events
):
    """Test that webhook triggers an event after validating a signature."""
    timestamp = "1529006854"
    token = "a8ce0edb2dd8301dee6c2405235584e45aa91d1e9f979f3de0"

    event_count = len(mailgun_events)

    await http_client.post(
        "/api/webhook/{}".format(webhook_id_with_api_key),
        json={
            "hello": "mailgun",
            "signature": {
                "signature": hmac.new(
                    key=bytes(API_KEY, "utf-8"),
                    msg=bytes("{}{}".format(timestamp, token), "utf-8"),
                    digestmod=hashlib.sha256,
                ).hexdigest(),
                "timestamp": timestamp,
                "token": token,
            },
        },
    )

    assert len(mailgun_events) == event_count + 1
    assert mailgun_events[-1].data["webhook_id"] == webhook_id_with_api_key
    assert mailgun_events[-1].data["hello"] == "mailgun"


async def test_mailgun_webhook_with_missing_signature_without_api_key(
    http_client, webhook_id_without_api_key, mailgun_events
):
    """Test that webhook triggers an event without a signature w/o API key."""
    event_count = len(mailgun_events)

    await http_client.post(
        "/api/webhook/{}".format(webhook_id_without_api_key),
        json={"hello": "mailgun", "signature": {}},
    )

    assert len(mailgun_events) == event_count + 1
    assert mailgun_events[-1].data["webhook_id"] == webhook_id_without_api_key
    assert mailgun_events[-1].data["hello"] == "mailgun"

    await http_client.post(
        "/api/webhook/{}".format(webhook_id_without_api_key), json={"hello": "mailgun"}
    )

    assert len(mailgun_events) == event_count + 1
    assert mailgun_events[-1].data["webhook_id"] == webhook_id_without_api_key
    assert mailgun_events[-1].data["hello"] == "mailgun"


async def test_mailgun_webhook_event_without_an_api_key(
    http_client, webhook_id_without_api_key, mailgun_events
):
    """Test that webhook triggers an event if there is no api key."""
    timestamp = "1529006854"
    token = "a8ce0edb2dd8301dee6c2405235584e45aa91d1e9f979f3de0"

    event_count = len(mailgun_events)

    await http_client.post(
        "/api/webhook/{}".format(webhook_id_without_api_key),
        json={
            "hello": "mailgun",
            "signature": {
                "signature": hmac.new(
                    key=bytes(API_KEY, "utf-8"),
                    msg=bytes("{}{}".format(timestamp, token), "utf-8"),
                    digestmod=hashlib.sha256,
                ).hexdigest(),
                "timestamp": timestamp,
                "token": token,
            },
        },
    )

    assert len(mailgun_events) == event_count + 1
    assert mailgun_events[-1].data["webhook_id"] == webhook_id_without_api_key
    assert mailgun_events[-1].data["hello"] == "mailgun"