b1c0cabe6c
* Fix MQTT retained message not being re-dispatched * Fix tests * Use paho-mqtt for retained messages * Improve code style * Store list of subscribers * Fix lint error * Adhere to Home Assistant's logging standard "Try to avoid brackets and additional quotes around the output to make it easier for users to parse the log." - https://home-assistant.io/developers/development_guidelines/ * Add reconnect tests * Fix lint error * Introduce Subscription Tests still need to be updated * Use namedtuple for MQTT messages ... And fix issues Accessing the config manually at runtime isn't ideal * Fix MQTT __init__.py tests * Updated usage of Mocks * Moved tests that were testing subscriptions out of the MQTTComponent test, because of how mock.patch was used * Adjusted the remaining tests for the MQTT clients new behavior - e.g. self.progress was removed * Updated the async_fire_mqtt_message helper * ✅ Update MQTT tests * Re-introduce the MQTT subscriptions through the dispatcher for tests - quite ugly though... 🚧 * Update fixtures to use our new MQTT mock 🎨 * 📝 Update base code according to comments * 🔨 Adjust MQTT test base * 🔨 Update other MQTT tests * 🍎 Fix carriage return in source files Apparently test_mqtt_json.py and test_mqtt_template.py were written on Windows. In order to not mess up the diff, I'll just redo the carriage return. * 🎨 Remove unused import * 📝 Remove fire_mqtt_client_message * 🐛 Fix using python 3.6 method What's very interesting is that 3.4 didn't fail on travis... * 🐛 Fix using assert directly
108 lines
2.8 KiB
Python
108 lines
2.8 KiB
Python
"""Setup some common test helper things."""
|
|
import asyncio
|
|
import functools
|
|
import logging
|
|
import os
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
import pytest
|
|
import requests_mock as _requests_mock
|
|
|
|
from homeassistant import util
|
|
from homeassistant.util import location
|
|
|
|
from tests.common import async_test_home_assistant, INSTANCES, \
|
|
async_mock_mqtt_component
|
|
from tests.test_util.aiohttp import mock_aiohttp_client
|
|
from tests.mock.zwave import MockNetwork, MockOption
|
|
|
|
if os.environ.get('UVLOOP') == '1':
|
|
import uvloop
|
|
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
|
|
|
|
logging.basicConfig()
|
|
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)
|
|
|
|
|
|
def test_real(func):
|
|
"""Force a function to require a keyword _test_real to be passed in."""
|
|
@functools.wraps(func)
|
|
def guard_func(*args, **kwargs):
|
|
real = kwargs.pop('_test_real', None)
|
|
|
|
if not real:
|
|
raise Exception('Forgot to mock or pass "_test_real=True" to %s',
|
|
func.__name__)
|
|
|
|
return func(*args, **kwargs)
|
|
|
|
return guard_func
|
|
|
|
|
|
# Guard a few functions that would make network connections
|
|
location.detect_location_info = test_real(location.detect_location_info)
|
|
location.elevation = test_real(location.elevation)
|
|
util.get_local_ip = lambda: '127.0.0.1'
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def verify_cleanup():
|
|
"""Verify that the test has cleaned up resources correctly."""
|
|
yield
|
|
|
|
if len(INSTANCES) >= 2:
|
|
count = len(INSTANCES)
|
|
for inst in INSTANCES:
|
|
inst.stop()
|
|
pytest.exit("Detected non stopped instances "
|
|
"({}), aborting test run".format(count))
|
|
|
|
|
|
@pytest.fixture
|
|
def hass(loop):
|
|
"""Fixture to provide a test instance of HASS."""
|
|
hass = loop.run_until_complete(async_test_home_assistant(loop))
|
|
|
|
yield hass
|
|
|
|
loop.run_until_complete(hass.async_stop())
|
|
|
|
|
|
@pytest.fixture
|
|
def requests_mock():
|
|
"""Fixture to provide a requests mocker."""
|
|
with _requests_mock.mock() as m:
|
|
yield m
|
|
|
|
|
|
@pytest.fixture
|
|
def aioclient_mock():
|
|
"""Fixture to mock aioclient calls."""
|
|
with mock_aiohttp_client() as mock_session:
|
|
yield mock_session
|
|
|
|
|
|
@pytest.fixture
|
|
def mqtt_mock(loop, hass):
|
|
"""Fixture to mock MQTT."""
|
|
client = loop.run_until_complete(async_mock_mqtt_component(hass))
|
|
client.reset_mock()
|
|
return client
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_openzwave():
|
|
"""Mock out Open Z-Wave."""
|
|
base_mock = MagicMock()
|
|
libopenzwave = base_mock.libopenzwave
|
|
libopenzwave.__file__ = 'test'
|
|
base_mock.network.ZWaveNetwork = MockNetwork
|
|
base_mock.option.ZWaveOption = MockOption
|
|
|
|
with patch.dict('sys.modules', {
|
|
'libopenzwave': libopenzwave,
|
|
'openzwave.option': base_mock.option,
|
|
'openzwave.network': base_mock.network,
|
|
'openzwave.group': base_mock.group,
|
|
}):
|
|
yield base_mock
|