core/tests/components/http/test_init.py

"""The tests for the Home Assistant HTTP component."""
import asyncio
import requests
from unittest.mock import MagicMock

from homeassistant import bootstrap, const
import homeassistant.components.http as http

from tests.common import get_test_instance_port, get_test_home_assistant

API_PASSWORD = 'test1234'
SERVER_PORT = get_test_instance_port()
HTTP_BASE = '127.0.0.1:{}'.format(SERVER_PORT)
HTTP_BASE_URL = 'http://{}'.format(HTTP_BASE)
HA_HEADERS = {
    const.HTTP_HEADER_HA_AUTH: API_PASSWORD,
    const.HTTP_HEADER_CONTENT_TYPE: const.CONTENT_TYPE_JSON,
}
CORS_ORIGINS = [HTTP_BASE_URL, HTTP_BASE]

hass = None


def _url(path=''):
    """Helper method to generate URLs."""
    return HTTP_BASE_URL + path


# pylint: disable=invalid-name
def setUpModule():
    """Initialize a Home Assistant server."""
    global hass

    hass = get_test_home_assistant()

    bootstrap.setup_component(
        hass, http.DOMAIN, {
            http.DOMAIN: {
                http.CONF_API_PASSWORD: API_PASSWORD,
                http.CONF_SERVER_PORT: SERVER_PORT,
                http.CONF_CORS_ORIGINS: CORS_ORIGINS,
            }
        }
    )

    bootstrap.setup_component(hass, 'api')

    # Registering static path as it caused CORS to blow up
    hass.http.register_static_path(
        '/custom_components', hass.config.path('custom_components'))

    hass.start()


# pylint: disable=invalid-name
def tearDownModule():
    """Stop the Home Assistant server."""
    hass.stop()


class TestCors:
    """Test HTTP component."""

    def test_cors_allowed_with_password_in_url(self):
        """Test cross origin resource sharing with password in url."""
        req = requests.get(_url(const.URL_API),
                           params={'api_password': API_PASSWORD},
                           headers={const.HTTP_HEADER_ORIGIN: HTTP_BASE_URL})

        allow_origin = const.HTTP_HEADER_ACCESS_CONTROL_ALLOW_ORIGIN

        assert req.status_code == 200
        assert req.headers.get(allow_origin) == HTTP_BASE_URL

    def test_cors_allowed_with_password_in_header(self):
        """Test cross origin resource sharing with password in header."""
        headers = {
            const.HTTP_HEADER_HA_AUTH: API_PASSWORD,
            const.HTTP_HEADER_ORIGIN: HTTP_BASE_URL
        }
        req = requests.get(_url(const.URL_API), headers=headers)

        allow_origin = const.HTTP_HEADER_ACCESS_CONTROL_ALLOW_ORIGIN

        assert req.status_code == 200
        assert req.headers.get(allow_origin) == HTTP_BASE_URL

    def test_cors_denied_without_origin_header(self):
        """Test cross origin resource sharing with password in header."""
        headers = {
            const.HTTP_HEADER_HA_AUTH: API_PASSWORD
        }
        req = requests.get(_url(const.URL_API), headers=headers)

        allow_origin = const.HTTP_HEADER_ACCESS_CONTROL_ALLOW_ORIGIN
        allow_headers = const.HTTP_HEADER_ACCESS_CONTROL_ALLOW_HEADERS

        assert req.status_code == 200
        assert allow_origin not in req.headers
        assert allow_headers not in req.headers

    def test_cors_preflight_allowed(self):
        """Test cross origin resource sharing preflight (OPTIONS) request."""
        headers = {
            const.HTTP_HEADER_ORIGIN: HTTP_BASE_URL,
            'Access-Control-Request-Method': 'GET',
            'Access-Control-Request-Headers': 'x-ha-access'
        }
        req = requests.options(_url(const.URL_API), headers=headers)

        allow_origin = const.HTTP_HEADER_ACCESS_CONTROL_ALLOW_ORIGIN
        allow_headers = const.HTTP_HEADER_ACCESS_CONTROL_ALLOW_HEADERS

        assert req.status_code == 200
        assert req.headers.get(allow_origin) == HTTP_BASE_URL
        assert req.headers.get(allow_headers) == \
            const.HTTP_HEADER_HA_AUTH.upper()


class TestView(http.HomeAssistantView):
    """Test the HTTP views."""

    name = 'test'
    url = '/hello'

    @asyncio.coroutine
    def get(self, request):
        """Return a get request."""
        return 'hello'


@asyncio.coroutine
def test_registering_view_while_running(hass, test_client):
    """Test that we can register a view while the server is running."""
    yield from bootstrap.async_setup_component(
        hass, http.DOMAIN, {
            http.DOMAIN: {
                http.CONF_SERVER_PORT: get_test_instance_port(),
            }
        }
    )

    yield from bootstrap.async_setup_component(hass, 'api')

    yield from hass.async_start()

    yield from hass.async_block_till_done()

    hass.http.register_view(TestView)

    client = yield from test_client(hass.http.app)

    resp = yield from client.get('/hello')
    assert resp.status == 200

    text = yield from resp.text()
    assert text == 'hello'


def test_api_base_url(loop):
    """Test setting api url."""
    hass = MagicMock()
    hass.loop = loop

    assert loop.run_until_complete(
        bootstrap.async_setup_component(hass, 'http', {
            'http': {
                'base_url': 'example.com'
            }
        })
    )

    assert hass.config.api.base_url == 'http://example.com'

    assert loop.run_until_complete(
        bootstrap.async_setup_component(hass, 'http', {
            'http': {
                'server_host': '1.1.1.1'
            }
        })
    )

    assert hass.config.api.base_url == 'http://1.1.1.1:8123'

    assert loop.run_until_complete(
        bootstrap.async_setup_component(hass, 'http', {
            'http': {
                'server_host': '1.1.1.1'
            }
        })
    )

    assert hass.config.api.base_url == 'http://1.1.1.1:8123'

    assert loop.run_until_complete(
        bootstrap.async_setup_component(hass, 'http', {
            'http': {
            }
        })
    )

    assert hass.config.api.base_url == 'http://127.0.0.1:8123'