Async tests for remaining device trackers (#18682)

This commit is contained in:
Adam Mills 2018-11-24 15:10:57 -05:00 committed by GitHub
parent 6ebdc7dabc
commit 50a30d4dc9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 213 additions and 237 deletions

View File

@ -1,7 +1,7 @@
"""The tests for the tplink device tracker platform.""" """The tests for the tplink device tracker platform."""
import os import os
import unittest import pytest
from homeassistant.components import device_tracker from homeassistant.components import device_tracker
from homeassistant.components.device_tracker.tplink import Tplink4DeviceScanner from homeassistant.components.device_tracker.tplink import Tplink4DeviceScanner
@ -9,27 +9,19 @@ from homeassistant.const import (CONF_PLATFORM, CONF_PASSWORD, CONF_USERNAME,
CONF_HOST) CONF_HOST)
import requests_mock import requests_mock
from tests.common import get_test_home_assistant
@pytest.fixture(autouse=True)
def setup_comp(hass):
"""Initialize components."""
yaml_devices = hass.config.path(device_tracker.YAML_DEVICES)
yield
if os.path.isfile(yaml_devices):
os.remove(yaml_devices)
class TestTplink4DeviceScanner(unittest.TestCase): async def test_get_mac_addresses_from_both_bands(hass):
"""Tests for the Tplink4DeviceScanner class.""" """Test grabbing the mac addresses from 2.4 and 5 GHz clients pages."""
with requests_mock.Mocker() as m:
def setUp(self): # pylint: disable=invalid-name
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
def tearDown(self): # pylint: disable=invalid-name
"""Stop everything that was started."""
self.hass.stop()
try:
os.remove(self.hass.config.path(device_tracker.YAML_DEVICES))
except FileNotFoundError:
pass
@requests_mock.mock()
def test_get_mac_addresses_from_both_bands(self, m):
"""Test grabbing the mac addresses from 2.4 and 5 GHz clients pages."""
conf_dict = { conf_dict = {
CONF_PLATFORM: 'tplink', CONF_PLATFORM: 'tplink',
CONF_HOST: 'fake-host', CONF_HOST: 'fake-host',

View File

@ -1,14 +1,12 @@
"""The tests for the Unifi direct device tracker platform.""" """The tests for the Unifi direct device tracker platform."""
import os import os
from datetime import timedelta from datetime import timedelta
import unittest from asynctest import mock, patch
from unittest import mock
from unittest.mock import patch
import pytest import pytest
import voluptuous as vol import voluptuous as vol
from homeassistant.setup import setup_component from homeassistant.setup import async_setup_component
from homeassistant.components import device_tracker from homeassistant.components import device_tracker
from homeassistant.components.device_tracker import ( from homeassistant.components.device_tracker import (
CONF_CONSIDER_HOME, CONF_TRACK_NEW, CONF_AWAY_HIDE, CONF_CONSIDER_HOME, CONF_TRACK_NEW, CONF_AWAY_HIDE,
@ -19,133 +17,129 @@ from homeassistant.const import (CONF_PLATFORM, CONF_PASSWORD, CONF_USERNAME,
CONF_HOST) CONF_HOST)
from tests.common import ( from tests.common import (
get_test_home_assistant, assert_setup_component, assert_setup_component, mock_component, load_fixture)
mock_component, load_fixture)
scanner_path = 'homeassistant.components.device_tracker.' + \
'unifi_direct.UnifiDeviceScanner'
class TestComponentsDeviceTrackerUnifiDirect(unittest.TestCase): @pytest.fixture(autouse=True)
"""Tests for the Unifi direct device tracker platform.""" def setup_comp(hass):
"""Initialize components."""
mock_component(hass, 'zone')
yaml_devices = hass.config.path(device_tracker.YAML_DEVICES)
yield
if os.path.isfile(yaml_devices):
os.remove(yaml_devices)
hass = None
scanner_path = 'homeassistant.components.device_tracker.' + \
'unifi_direct.UnifiDeviceScanner'
def setup_method(self, _): @patch(scanner_path, return_value=mock.MagicMock())
"""Set up things to be run when tests are started.""" async def test_get_scanner(unifi_mock, hass):
self.hass = get_test_home_assistant() """Test creating an Unifi direct scanner with a password."""
mock_component(self.hass, 'zone') conf_dict = {
DOMAIN: {
def teardown_method(self, _): CONF_PLATFORM: 'unifi_direct',
"""Stop everything that was started.""" CONF_HOST: 'fake_host',
self.hass.stop() CONF_USERNAME: 'fake_user',
try: CONF_PASSWORD: 'fake_pass',
os.remove(self.hass.config.path(device_tracker.YAML_DEVICES)) CONF_TRACK_NEW: True,
except FileNotFoundError: CONF_CONSIDER_HOME: timedelta(seconds=180),
pass CONF_NEW_DEVICE_DEFAULTS: {
@mock.patch(scanner_path,
return_value=mock.MagicMock())
def test_get_scanner(self, unifi_mock):
"""Test creating an Unifi direct scanner with a password."""
conf_dict = {
DOMAIN: {
CONF_PLATFORM: 'unifi_direct',
CONF_HOST: 'fake_host',
CONF_USERNAME: 'fake_user',
CONF_PASSWORD: 'fake_pass',
CONF_TRACK_NEW: True, CONF_TRACK_NEW: True,
CONF_CONSIDER_HOME: timedelta(seconds=180), CONF_AWAY_HIDE: False
CONF_NEW_DEVICE_DEFAULTS: {
CONF_TRACK_NEW: True,
CONF_AWAY_HIDE: False
}
} }
} }
}
with assert_setup_component(1, DOMAIN): with assert_setup_component(1, DOMAIN):
assert setup_component(self.hass, DOMAIN, conf_dict) assert await async_setup_component(hass, DOMAIN, conf_dict)
conf_dict[DOMAIN][CONF_PORT] = 22 conf_dict[DOMAIN][CONF_PORT] = 22
assert unifi_mock.call_args == mock.call(conf_dict[DOMAIN]) assert unifi_mock.call_args == mock.call(conf_dict[DOMAIN])
@patch('pexpect.pxssh.pxssh')
def test_get_device_name(self, mock_ssh): @patch('pexpect.pxssh.pxssh')
"""Testing MAC matching.""" async def test_get_device_name(mock_ssh, hass):
conf_dict = { """Testing MAC matching."""
DOMAIN: { conf_dict = {
CONF_PLATFORM: 'unifi_direct', DOMAIN: {
CONF_HOST: 'fake_host', CONF_PLATFORM: 'unifi_direct',
CONF_USERNAME: 'fake_user', CONF_HOST: 'fake_host',
CONF_PASSWORD: 'fake_pass', CONF_USERNAME: 'fake_user',
CONF_PORT: 22, CONF_PASSWORD: 'fake_pass',
CONF_TRACK_NEW: True, CONF_PORT: 22,
CONF_CONSIDER_HOME: timedelta(seconds=180) CONF_TRACK_NEW: True,
} CONF_CONSIDER_HOME: timedelta(seconds=180)
} }
mock_ssh.return_value.before = load_fixture('unifi_direct.txt') }
scanner = get_scanner(self.hass, conf_dict) mock_ssh.return_value.before = load_fixture('unifi_direct.txt')
devices = scanner.scan_devices() scanner = get_scanner(hass, conf_dict)
assert 23 == len(devices) devices = scanner.scan_devices()
assert "iPhone" == \ assert 23 == len(devices)
scanner.get_device_name("98:00:c6:56:34:12") assert "iPhone" == \
assert "iPhone" == \ scanner.get_device_name("98:00:c6:56:34:12")
scanner.get_device_name("98:00:C6:56:34:12") assert "iPhone" == \
scanner.get_device_name("98:00:C6:56:34:12")
@patch('pexpect.pxssh.pxssh.logout')
@patch('pexpect.pxssh.pxssh.login')
def test_failed_to_log_in(self, mock_login, mock_logout):
"""Testing exception at login results in False."""
from pexpect import exceptions
conf_dict = { @patch('pexpect.pxssh.pxssh.logout')
DOMAIN: { @patch('pexpect.pxssh.pxssh.login')
CONF_PLATFORM: 'unifi_direct', async def test_failed_to_log_in(mock_login, mock_logout, hass):
CONF_HOST: 'fake_host', """Testing exception at login results in False."""
CONF_USERNAME: 'fake_user', from pexpect import exceptions
CONF_PASSWORD: 'fake_pass',
CONF_PORT: 22, conf_dict = {
CONF_TRACK_NEW: True, DOMAIN: {
CONF_CONSIDER_HOME: timedelta(seconds=180) CONF_PLATFORM: 'unifi_direct',
} CONF_HOST: 'fake_host',
CONF_USERNAME: 'fake_user',
CONF_PASSWORD: 'fake_pass',
CONF_PORT: 22,
CONF_TRACK_NEW: True,
CONF_CONSIDER_HOME: timedelta(seconds=180)
} }
}
mock_login.side_effect = exceptions.EOF("Test") mock_login.side_effect = exceptions.EOF("Test")
scanner = get_scanner(self.hass, conf_dict) scanner = get_scanner(hass, conf_dict)
assert not scanner assert not scanner
@patch('pexpect.pxssh.pxssh.logout')
@patch('pexpect.pxssh.pxssh.login', autospec=True) @patch('pexpect.pxssh.pxssh.logout')
@patch('pexpect.pxssh.pxssh.prompt') @patch('pexpect.pxssh.pxssh.login', autospec=True)
@patch('pexpect.pxssh.pxssh.sendline') @patch('pexpect.pxssh.pxssh.prompt')
def test_to_get_update(self, mock_sendline, mock_prompt, mock_login, @patch('pexpect.pxssh.pxssh.sendline')
mock_logout): async def test_to_get_update(mock_sendline, mock_prompt, mock_login,
"""Testing exception in get_update matching.""" mock_logout, hass):
conf_dict = { """Testing exception in get_update matching."""
DOMAIN: { conf_dict = {
CONF_PLATFORM: 'unifi_direct', DOMAIN: {
CONF_HOST: 'fake_host', CONF_PLATFORM: 'unifi_direct',
CONF_USERNAME: 'fake_user', CONF_HOST: 'fake_host',
CONF_PASSWORD: 'fake_pass', CONF_USERNAME: 'fake_user',
CONF_PORT: 22, CONF_PASSWORD: 'fake_pass',
CONF_TRACK_NEW: True, CONF_PORT: 22,
CONF_CONSIDER_HOME: timedelta(seconds=180) CONF_TRACK_NEW: True,
} CONF_CONSIDER_HOME: timedelta(seconds=180)
} }
}
scanner = get_scanner(self.hass, conf_dict) scanner = get_scanner(hass, conf_dict)
# mock_sendline.side_effect = AssertionError("Test") # mock_sendline.side_effect = AssertionError("Test")
mock_prompt.side_effect = AssertionError("Test") mock_prompt.side_effect = AssertionError("Test")
devices = scanner._get_update() # pylint: disable=protected-access devices = scanner._get_update() # pylint: disable=protected-access
assert devices is None assert devices is None
def test_good_response_parses(self):
"""Test that the response form the AP parses to JSON correctly."""
response = _response_to_json(load_fixture('unifi_direct.txt'))
assert response != {}
def test_bad_response_returns_none(self): def test_good_response_parses(hass):
"""Test that a bad response form the AP parses to JSON correctly.""" """Test that the response form the AP parses to JSON correctly."""
assert _response_to_json("{(}") == {} response = _response_to_json(load_fixture('unifi_direct.txt'))
assert response != {}
def test_bad_response_returns_none(hass):
"""Test that a bad response form the AP parses to JSON correctly."""
assert _response_to_json("{(}") == {}
def test_config_error(): def test_config_error():

View File

@ -1,8 +1,6 @@
"""The tests for the Xiaomi router device tracker platform.""" """The tests for the Xiaomi router device tracker platform."""
import logging import logging
import unittest from asynctest import mock, patch
from unittest import mock
from unittest.mock import patch
import requests import requests
@ -10,7 +8,6 @@ from homeassistant.components.device_tracker import DOMAIN, xiaomi as xiaomi
from homeassistant.components.device_tracker.xiaomi import get_scanner from homeassistant.components.device_tracker.xiaomi import get_scanner
from homeassistant.const import (CONF_HOST, CONF_USERNAME, CONF_PASSWORD, from homeassistant.const import (CONF_HOST, CONF_USERNAME, CONF_PASSWORD,
CONF_PLATFORM) CONF_PLATFORM)
from tests.common import get_test_home_assistant
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -152,113 +149,106 @@ def mocked_requests(*args, **kwargs):
_LOGGER.debug('UNKNOWN ROUTE') _LOGGER.debug('UNKNOWN ROUTE')
class TestXiaomiDeviceScanner(unittest.TestCase): @patch(
"""Xiaomi device scanner test class.""" 'homeassistant.components.device_tracker.xiaomi.XiaomiDeviceScanner',
return_value=mock.MagicMock())
async def test_config(xiaomi_mock, hass):
"""Testing minimal configuration."""
config = {
DOMAIN: xiaomi.PLATFORM_SCHEMA({
CONF_PLATFORM: xiaomi.DOMAIN,
CONF_HOST: '192.168.0.1',
CONF_PASSWORD: 'passwordTest'
})
}
xiaomi.get_scanner(hass, config)
assert xiaomi_mock.call_count == 1
assert xiaomi_mock.call_args == mock.call(config[DOMAIN])
call_arg = xiaomi_mock.call_args[0][0]
assert call_arg['username'] == 'admin'
assert call_arg['password'] == 'passwordTest'
assert call_arg['host'] == '192.168.0.1'
assert call_arg['platform'] == 'device_tracker'
def setUp(self):
"""Initialize values for this testcase class."""
self.hass = get_test_home_assistant()
def tearDown(self): @patch(
"""Stop everything that was started.""" 'homeassistant.components.device_tracker.xiaomi.XiaomiDeviceScanner',
self.hass.stop() return_value=mock.MagicMock())
async def test_config_full(xiaomi_mock, hass):
"""Testing full configuration."""
config = {
DOMAIN: xiaomi.PLATFORM_SCHEMA({
CONF_PLATFORM: xiaomi.DOMAIN,
CONF_HOST: '192.168.0.1',
CONF_USERNAME: 'alternativeAdminName',
CONF_PASSWORD: 'passwordTest'
})
}
xiaomi.get_scanner(hass, config)
assert xiaomi_mock.call_count == 1
assert xiaomi_mock.call_args == mock.call(config[DOMAIN])
call_arg = xiaomi_mock.call_args[0][0]
assert call_arg['username'] == 'alternativeAdminName'
assert call_arg['password'] == 'passwordTest'
assert call_arg['host'] == '192.168.0.1'
assert call_arg['platform'] == 'device_tracker'
@mock.patch(
'homeassistant.components.device_tracker.xiaomi.XiaomiDeviceScanner',
return_value=mock.MagicMock())
def test_config(self, xiaomi_mock):
"""Testing minimal configuration."""
config = {
DOMAIN: xiaomi.PLATFORM_SCHEMA({
CONF_PLATFORM: xiaomi.DOMAIN,
CONF_HOST: '192.168.0.1',
CONF_PASSWORD: 'passwordTest'
})
}
xiaomi.get_scanner(self.hass, config)
assert xiaomi_mock.call_count == 1
assert xiaomi_mock.call_args == mock.call(config[DOMAIN])
call_arg = xiaomi_mock.call_args[0][0]
assert call_arg['username'] == 'admin'
assert call_arg['password'] == 'passwordTest'
assert call_arg['host'] == '192.168.0.1'
assert call_arg['platform'] == 'device_tracker'
@mock.patch( @patch('requests.get', side_effect=mocked_requests)
'homeassistant.components.device_tracker.xiaomi.XiaomiDeviceScanner', @patch('requests.post', side_effect=mocked_requests)
return_value=mock.MagicMock()) async def test_invalid_credential(mock_get, mock_post, hass):
def test_config_full(self, xiaomi_mock): """Testing invalid credential handling."""
"""Testing full configuration.""" config = {
config = { DOMAIN: xiaomi.PLATFORM_SCHEMA({
DOMAIN: xiaomi.PLATFORM_SCHEMA({ CONF_PLATFORM: xiaomi.DOMAIN,
CONF_PLATFORM: xiaomi.DOMAIN, CONF_HOST: '192.168.0.1',
CONF_HOST: '192.168.0.1', CONF_USERNAME: INVALID_USERNAME,
CONF_USERNAME: 'alternativeAdminName', CONF_PASSWORD: 'passwordTest'
CONF_PASSWORD: 'passwordTest' })
}) }
} assert get_scanner(hass, config) is None
xiaomi.get_scanner(self.hass, config)
assert xiaomi_mock.call_count == 1
assert xiaomi_mock.call_args == mock.call(config[DOMAIN])
call_arg = xiaomi_mock.call_args[0][0]
assert call_arg['username'] == 'alternativeAdminName'
assert call_arg['password'] == 'passwordTest'
assert call_arg['host'] == '192.168.0.1'
assert call_arg['platform'] == 'device_tracker'
@patch('requests.get', side_effect=mocked_requests)
@patch('requests.post', side_effect=mocked_requests)
def test_invalid_credential(self, mock_get, mock_post):
"""Testing invalid credential handling."""
config = {
DOMAIN: xiaomi.PLATFORM_SCHEMA({
CONF_PLATFORM: xiaomi.DOMAIN,
CONF_HOST: '192.168.0.1',
CONF_USERNAME: INVALID_USERNAME,
CONF_PASSWORD: 'passwordTest'
})
}
assert get_scanner(self.hass, config) is None
@patch('requests.get', side_effect=mocked_requests) @patch('requests.get', side_effect=mocked_requests)
@patch('requests.post', side_effect=mocked_requests) @patch('requests.post', side_effect=mocked_requests)
def test_valid_credential(self, mock_get, mock_post): async def test_valid_credential(mock_get, mock_post, hass):
"""Testing valid refresh.""" """Testing valid refresh."""
config = { config = {
DOMAIN: xiaomi.PLATFORM_SCHEMA({ DOMAIN: xiaomi.PLATFORM_SCHEMA({
CONF_PLATFORM: xiaomi.DOMAIN, CONF_PLATFORM: xiaomi.DOMAIN,
CONF_HOST: '192.168.0.1', CONF_HOST: '192.168.0.1',
CONF_USERNAME: 'admin', CONF_USERNAME: 'admin',
CONF_PASSWORD: 'passwordTest' CONF_PASSWORD: 'passwordTest'
}) })
} }
scanner = get_scanner(self.hass, config) scanner = get_scanner(hass, config)
assert scanner is not None assert scanner is not None
assert 2 == len(scanner.scan_devices()) assert 2 == len(scanner.scan_devices())
assert "Device1" == \ assert "Device1" == \
scanner.get_device_name("23:83:BF:F6:38:A0") scanner.get_device_name("23:83:BF:F6:38:A0")
assert "Device2" == \ assert "Device2" == \
scanner.get_device_name("1D:98:EC:5E:D5:A6") scanner.get_device_name("1D:98:EC:5E:D5:A6")
@patch('requests.get', side_effect=mocked_requests)
@patch('requests.post', side_effect=mocked_requests)
def test_token_timed_out(self, mock_get, mock_post):
"""Testing refresh with a timed out token.
New token is requested and list is downloaded a second time. @patch('requests.get', side_effect=mocked_requests)
""" @patch('requests.post', side_effect=mocked_requests)
config = { async def test_token_timed_out(mock_get, mock_post, hass):
DOMAIN: xiaomi.PLATFORM_SCHEMA({ """Testing refresh with a timed out token.
CONF_PLATFORM: xiaomi.DOMAIN,
CONF_HOST: '192.168.0.1', New token is requested and list is downloaded a second time.
CONF_USERNAME: TOKEN_TIMEOUT_USERNAME, """
CONF_PASSWORD: 'passwordTest' config = {
}) DOMAIN: xiaomi.PLATFORM_SCHEMA({
} CONF_PLATFORM: xiaomi.DOMAIN,
scanner = get_scanner(self.hass, config) CONF_HOST: '192.168.0.1',
assert scanner is not None CONF_USERNAME: TOKEN_TIMEOUT_USERNAME,
assert 2 == len(scanner.scan_devices()) CONF_PASSWORD: 'passwordTest'
assert "Device1" == \ })
scanner.get_device_name("23:83:BF:F6:38:A0") }
assert "Device2" == \ scanner = get_scanner(hass, config)
scanner.get_device_name("1D:98:EC:5E:D5:A6") assert scanner is not None
assert 2 == len(scanner.scan_devices())
assert "Device1" == \
scanner.get_device_name("23:83:BF:F6:38:A0")
assert "Device2" == \
scanner.get_device_name("1D:98:EC:5E:D5:A6")