Try all Samsung TV websocket ports (#33001)
* Update bridge.py * add test * silence pylint * correct pylint * add some tests * Update test_media_player.pypull/33118/head
parent
197736f66b
commit
a129bc05ae
|
@ -206,6 +206,7 @@ class SamsungTVWSBridge(SamsungTVBridge):
|
|||
CONF_TIMEOUT: 31,
|
||||
}
|
||||
|
||||
result = None
|
||||
try:
|
||||
LOGGER.debug("Try config: %s", config)
|
||||
with SamsungTVWS(
|
||||
|
@ -223,9 +224,13 @@ class SamsungTVWSBridge(SamsungTVBridge):
|
|||
return RESULT_SUCCESS
|
||||
except WebSocketException:
|
||||
LOGGER.debug("Working but unsupported config: %s", config)
|
||||
return RESULT_NOT_SUPPORTED
|
||||
result = RESULT_NOT_SUPPORTED
|
||||
except (OSError, ConnectionFailure) as err:
|
||||
LOGGER.debug("Failing config: %s, error: %s", config, err)
|
||||
# pylint: disable=useless-else-on-loop
|
||||
else:
|
||||
if result:
|
||||
return result
|
||||
|
||||
return RESULT_NOT_SUCCESSFUL
|
||||
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
"""Tests for Samsung TV config flow."""
|
||||
from unittest.mock import call, patch
|
||||
from unittest.mock import Mock, PropertyMock, call, patch
|
||||
|
||||
from asynctest import mock
|
||||
import pytest
|
||||
|
@ -19,7 +19,7 @@ from homeassistant.components.ssdp import (
|
|||
ATTR_UPNP_MODEL_NAME,
|
||||
ATTR_UPNP_UDN,
|
||||
)
|
||||
from homeassistant.const import CONF_HOST, CONF_ID, CONF_METHOD, CONF_NAME
|
||||
from homeassistant.const import CONF_HOST, CONF_ID, CONF_METHOD, CONF_NAME, CONF_TOKEN
|
||||
|
||||
MOCK_USER_DATA = {CONF_HOST: "fake_host", CONF_NAME: "fake_name"}
|
||||
MOCK_SSDP_DATA = {
|
||||
|
@ -46,6 +46,20 @@ AUTODETECT_LEGACY = {
|
|||
"host": "fake_host",
|
||||
"timeout": 31,
|
||||
}
|
||||
AUTODETECT_WEBSOCKET_PLAIN = {
|
||||
"host": "fake_host",
|
||||
"name": "HomeAssistant",
|
||||
"port": 8001,
|
||||
"timeout": 31,
|
||||
"token": None,
|
||||
}
|
||||
AUTODETECT_WEBSOCKET_SSL = {
|
||||
"host": "fake_host",
|
||||
"name": "HomeAssistant",
|
||||
"port": 8002,
|
||||
"timeout": 31,
|
||||
"token": None,
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture(name="remote")
|
||||
|
@ -446,20 +460,48 @@ async def test_autodetect_websocket(hass, remote, remotews):
|
|||
with patch(
|
||||
"homeassistant.components.samsungtv.bridge.Remote", side_effect=OSError("Boom"),
|
||||
), patch("homeassistant.components.samsungtv.bridge.SamsungTVWS") as remotews:
|
||||
enter = Mock()
|
||||
type(enter).token = PropertyMock(return_value="123456789")
|
||||
remote = Mock()
|
||||
remote.__enter__ = Mock(return_value=enter)
|
||||
remote.__exit__ = Mock(return_value=False)
|
||||
remotews.return_value = remote
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
|
||||
)
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["data"][CONF_METHOD] == "websocket"
|
||||
assert result["data"][CONF_TOKEN] == "123456789"
|
||||
assert remotews.call_count == 1
|
||||
assert remotews.call_args_list == [call(**AUTODETECT_WEBSOCKET_PLAIN)]
|
||||
|
||||
|
||||
async def test_autodetect_websocket_ssl(hass, remote, remotews):
|
||||
"""Test for send key with autodetection of protocol."""
|
||||
with patch(
|
||||
"homeassistant.components.samsungtv.bridge.Remote", side_effect=OSError("Boom"),
|
||||
), patch(
|
||||
"homeassistant.components.samsungtv.bridge.SamsungTVWS",
|
||||
side_effect=[WebSocketProtocolException("Boom"), mock.DEFAULT],
|
||||
) as remotews:
|
||||
enter = Mock()
|
||||
type(enter).token = PropertyMock(return_value="123456789")
|
||||
remote = Mock()
|
||||
remote.__enter__ = Mock(return_value=enter)
|
||||
remote.__exit__ = Mock(return_value=False)
|
||||
remotews.return_value = remote
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
|
||||
)
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["data"][CONF_METHOD] == "websocket"
|
||||
assert result["data"][CONF_TOKEN] == "123456789"
|
||||
assert remotews.call_count == 2
|
||||
assert remotews.call_args_list == [
|
||||
call(
|
||||
host="fake_host",
|
||||
name="HomeAssistant",
|
||||
port=8001,
|
||||
timeout=31,
|
||||
token=None,
|
||||
)
|
||||
call(**AUTODETECT_WEBSOCKET_PLAIN),
|
||||
call(**AUTODETECT_WEBSOCKET_SSL),
|
||||
]
|
||||
|
||||
|
||||
|
@ -524,18 +566,6 @@ async def test_autodetect_none(hass, remote, remotews):
|
|||
]
|
||||
assert remotews.call_count == 2
|
||||
assert remotews.call_args_list == [
|
||||
call(
|
||||
host="fake_host",
|
||||
name="HomeAssistant",
|
||||
port=8001,
|
||||
timeout=31,
|
||||
token=None,
|
||||
),
|
||||
call(
|
||||
host="fake_host",
|
||||
name="HomeAssistant",
|
||||
port=8002,
|
||||
timeout=31,
|
||||
token=None,
|
||||
),
|
||||
call(**AUTODETECT_WEBSOCKET_PLAIN),
|
||||
call(**AUTODETECT_WEBSOCKET_SSL),
|
||||
]
|
||||
|
|
|
@ -34,8 +34,11 @@ from homeassistant.const import (
|
|||
ATTR_FRIENDLY_NAME,
|
||||
ATTR_SUPPORTED_FEATURES,
|
||||
CONF_HOST,
|
||||
CONF_IP_ADDRESS,
|
||||
CONF_METHOD,
|
||||
CONF_NAME,
|
||||
CONF_PORT,
|
||||
CONF_TOKEN,
|
||||
SERVICE_MEDIA_NEXT_TRACK,
|
||||
SERVICE_MEDIA_PAUSE,
|
||||
SERVICE_MEDIA_PLAY,
|
||||
|
@ -51,7 +54,7 @@ from homeassistant.const import (
|
|||
from homeassistant.setup import async_setup_component
|
||||
import homeassistant.util.dt as dt_util
|
||||
|
||||
from tests.common import async_fire_time_changed
|
||||
from tests.common import MockConfigEntry, async_fire_time_changed
|
||||
|
||||
ENTITY_ID = f"{DOMAIN}.fake"
|
||||
MOCK_CONFIG = {
|
||||
|
@ -64,17 +67,40 @@ MOCK_CONFIG = {
|
|||
}
|
||||
]
|
||||
}
|
||||
|
||||
MOCK_CONFIGWS = {
|
||||
SAMSUNGTV_DOMAIN: [
|
||||
{
|
||||
CONF_HOST: "fake",
|
||||
CONF_NAME: "fake",
|
||||
CONF_PORT: 8001,
|
||||
CONF_TOKEN: "123456789",
|
||||
CONF_ON_ACTION: [{"delay": "00:00:01"}],
|
||||
}
|
||||
]
|
||||
}
|
||||
MOCK_CALLS_WS = {
|
||||
"host": "fake",
|
||||
"port": 8001,
|
||||
"token": None,
|
||||
"timeout": 31,
|
||||
"name": "HomeAssistant",
|
||||
}
|
||||
|
||||
MOCK_ENTRY_WS = {
|
||||
CONF_IP_ADDRESS: "test",
|
||||
CONF_HOST: "fake",
|
||||
CONF_METHOD: "websocket",
|
||||
CONF_NAME: "fake",
|
||||
CONF_PORT: 8001,
|
||||
CONF_TOKEN: "abcde",
|
||||
}
|
||||
MOCK_CALLS_ENTRY_WS = {
|
||||
"host": "fake",
|
||||
"name": "HomeAssistant",
|
||||
"port": 8001,
|
||||
"timeout": 1,
|
||||
"token": "abcde",
|
||||
}
|
||||
|
||||
ENTITY_ID_NOTURNON = f"{DOMAIN}.fake_noturnon"
|
||||
MOCK_CONFIG_NOTURNON = {
|
||||
|
@ -155,6 +181,52 @@ async def test_setup_without_turnon(hass, remote):
|
|||
assert hass.states.get(ENTITY_ID_NOTURNON)
|
||||
|
||||
|
||||
async def test_setup_websocket(hass, remotews, mock_now):
|
||||
"""Test setup of platform."""
|
||||
with patch("homeassistant.components.samsungtv.bridge.SamsungTVWS") as remote_class:
|
||||
enter = mock.Mock()
|
||||
type(enter).token = mock.PropertyMock(return_value="987654321")
|
||||
remote = mock.Mock()
|
||||
remote.__enter__ = mock.Mock(return_value=enter)
|
||||
remote.__exit__ = mock.Mock()
|
||||
remote_class.return_value = remote
|
||||
|
||||
await setup_samsungtv(hass, MOCK_CONFIGWS)
|
||||
|
||||
assert remote_class.call_count == 1
|
||||
assert remote_class.call_args_list == [call(**MOCK_CALLS_WS)]
|
||||
assert hass.states.get(ENTITY_ID)
|
||||
|
||||
|
||||
async def test_setup_websocket_2(hass, mock_now):
|
||||
"""Test setup of platform from config entry."""
|
||||
entity_id = f"{DOMAIN}.fake"
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=SAMSUNGTV_DOMAIN, data=MOCK_ENTRY_WS, unique_id=entity_id,
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
config_entries = hass.config_entries.async_entries(SAMSUNGTV_DOMAIN)
|
||||
assert len(config_entries) == 1
|
||||
assert entry is config_entries[0]
|
||||
|
||||
assert await async_setup_component(hass, SAMSUNGTV_DOMAIN, {})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
next_update = mock_now + timedelta(minutes=5)
|
||||
with patch(
|
||||
"homeassistant.components.samsungtv.bridge.SamsungTVWS"
|
||||
) as remote, patch("homeassistant.util.dt.utcnow", return_value=next_update):
|
||||
async_fire_time_changed(hass, next_update)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(entity_id)
|
||||
assert state
|
||||
assert remote.call_count == 1
|
||||
assert remote.call_args_list == [call(**MOCK_CALLS_ENTRY_WS)]
|
||||
|
||||
|
||||
async def test_update_on(hass, remote, mock_now):
|
||||
"""Testing update tv on."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
|
|
Loading…
Reference in New Issue