mirror of
https://github.com/home-assistant/core.git
synced 2025-04-23 08:47:57 +00:00
Address Kodi code review follow up (#39104)
* Code review follow up * Update config_flow.py * Update config_flow.py * Update strings.json * Update config_flow.py * Update config_flow.py * Update config_flow.py * Update config_flow.py * Update config_flow.py * Update config_flow.py * Update util.py * Update test_config_flow.py * Update config_flow.py * Update util.py * Update test_config_flow.py * Update test_config_flow.py * Update test_config_flow.py * Update test_config_flow.py * Update test_config_flow.py * Update test_config_flow.py * Update util.py * Update test_config_flow.py * Update config_flow.py * Update config_flow.py * Update config_flow.py * Update config_flow.py * Update test_config_flow.py * Update test_config_flow.py * Update config_flow.py * Update test_config_flow.py * Update test_config_flow.py * Update test_config_flow.py * Update test_config_flow.py * Update config_flow.py * Update test_config_flow.py * Update test_config_flow.py * Update test_config_flow.py * Update test_config_flow.py * Update test_config_flow.py Co-authored-by: Chris Talkington <chris@talkingtontech.com>
This commit is contained in:
On Freund
GitHub
parent
3c1c6069da
commit
f075742a86
@ -40,7 +40,7 @@ async def validate_http(hass: core.HomeAssistant, data):
|
||||
ssl = data.get(CONF_SSL)
|
||||
session = async_get_clientsession(hass)
|
||||
|
||||
_LOGGER.debug("Connecting to %s:%s over HTTP.", host, port)
|
||||
_LOGGER.debug("Connecting to %s:%s over HTTP", host, port)
|
||||
khc = get_kodi_connection(
|
||||
host, port, None, username, password, ssl, session=session
|
||||
)
|
||||
@ -67,19 +67,19 @@ async def validate_ws(hass: core.HomeAssistant, data):
|
||||
|
||||
session = async_get_clientsession(hass)
|
||||
|
||||
_LOGGER.debug("Connecting to %s:%s over WebSocket.", host, ws_port)
|
||||
_LOGGER.debug("Connecting to %s:%s over WebSocket", host, ws_port)
|
||||
kwc = get_kodi_connection(
|
||||
host, port, ws_port, username, password, ssl, session=session
|
||||
)
|
||||
try:
|
||||
await kwc.connect()
|
||||
if not kwc.connected:
|
||||
_LOGGER.warning("Cannot connect to %s:%s over WebSocket.", host, ws_port)
|
||||
raise CannotConnect()
|
||||
_LOGGER.warning("Cannot connect to %s:%s over WebSocket", host, ws_port)
|
||||
raise WSCannotConnect()
|
||||
kodi = Kodi(kwc)
|
||||
await kodi.ping()
|
||||
except CannotConnectError as error:
|
||||
raise CannotConnect from error
|
||||
raise WSCannotConnect from error
|
||||
|
||||
|
||||
class KodiConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
@ -91,8 +91,8 @@ class KodiConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
def __init__(self):
|
||||
"""Initialize flow."""
|
||||
self._host: Optional[str] = None
|
||||
self._port: Optional[int] = None
|
||||
self._ws_port: Optional[int] = None
|
||||
self._port: Optional[int] = DEFAULT_PORT
|
||||
self._ws_port: Optional[int] = DEFAULT_WS_PORT
|
||||
self._name: Optional[str] = None
|
||||
self._username: Optional[str] = None
|
||||
self._password: Optional[str] = None
|
||||
@ -116,83 +116,170 @@ class KodiConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
}
|
||||
)
|
||||
|
||||
try:
|
||||
await validate_http(self.hass, self._get_data())
|
||||
await validate_ws(self.hass, self._get_data())
|
||||
except InvalidAuth:
|
||||
return await self.async_step_credentials()
|
||||
except WSCannotConnect:
|
||||
return await self.async_step_ws_port()
|
||||
except CannotConnect:
|
||||
return self.async_abort(reason="cannot_connect")
|
||||
except Exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Unexpected exception")
|
||||
return self.async_abort(reason="unknown")
|
||||
|
||||
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
|
||||
self.context.update({"title_placeholders": {CONF_NAME: self._name}})
|
||||
return await self.async_step_discovery_confirm()
|
||||
|
||||
async def async_step_discovery_confirm(self, user_input=None):
|
||||
"""Handle user-confirmation of discovered node."""
|
||||
if user_input is not None:
|
||||
return await self.async_step_credentials()
|
||||
if user_input is None:
|
||||
return self.async_show_form(
|
||||
step_id="discovery_confirm",
|
||||
description_placeholders={"name": self._name},
|
||||
)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="discovery_confirm", description_placeholders={"name": self._name}
|
||||
)
|
||||
return self._create_entry()
|
||||
|
||||
async def async_step_user(self, user_input=None):
|
||||
"""Handle the initial step."""
|
||||
return await self.async_step_host(user_input)
|
||||
|
||||
async def async_step_host(self, user_input=None, errors=None):
|
||||
"""Handle host name and port input."""
|
||||
if not errors:
|
||||
errors = {}
|
||||
errors = {}
|
||||
|
||||
if user_input is not None:
|
||||
self._host = user_input[CONF_HOST]
|
||||
self._port = user_input[CONF_PORT]
|
||||
self._ssl = user_input[CONF_SSL]
|
||||
return await self.async_step_credentials()
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="host", data_schema=self._host_schema(), errors=errors
|
||||
)
|
||||
|
||||
async def async_step_credentials(self, user_input=None):
|
||||
"""Handle username and password input."""
|
||||
errors = {}
|
||||
if user_input is not None:
|
||||
self._username = user_input.get(CONF_USERNAME)
|
||||
self._password = user_input.get(CONF_PASSWORD)
|
||||
try:
|
||||
await validate_http(self.hass, self._get_data())
|
||||
return await self.async_step_ws_port()
|
||||
except InvalidAuth:
|
||||
errors["base"] = "invalid_auth"
|
||||
except CannotConnect:
|
||||
if self._discovery_name:
|
||||
return self.async_abort(reason="cannot_connect")
|
||||
return await self.async_step_host(errors={"base": "cannot_connect"})
|
||||
except Exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Unexpected exception")
|
||||
errors["base"] = "unknown"
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="credentials", data_schema=self._credentials_schema(), errors=errors
|
||||
)
|
||||
|
||||
async def async_step_ws_port(self, user_input=None):
|
||||
"""Handle websocket port of discovered node."""
|
||||
errors = {}
|
||||
if user_input is not None:
|
||||
self._ws_port = user_input.get(CONF_WS_PORT)
|
||||
try:
|
||||
await validate_ws(self.hass, self._get_data())
|
||||
return self._create_entry()
|
||||
except InvalidAuth:
|
||||
return await self.async_step_credentials()
|
||||
except WSCannotConnect:
|
||||
return await self.async_step_ws_port()
|
||||
except CannotConnect:
|
||||
errors["base"] = "cannot_connect"
|
||||
except Exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Unexpected exception")
|
||||
errors["base"] = "unknown"
|
||||
else:
|
||||
return self._create_entry()
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="ws_port", data_schema=self._ws_port_schema(), errors=errors
|
||||
)
|
||||
return self._show_user_form(errors)
|
||||
|
||||
async def async_step_credentials(self, user_input=None):
|
||||
"""Handle username and password input."""
|
||||
errors = {}
|
||||
|
||||
if user_input is not None:
|
||||
self._username = user_input.get(CONF_USERNAME)
|
||||
self._password = user_input.get(CONF_PASSWORD)
|
||||
|
||||
try:
|
||||
await validate_http(self.hass, self._get_data())
|
||||
await validate_ws(self.hass, self._get_data())
|
||||
except InvalidAuth:
|
||||
errors["base"] = "invalid_auth"
|
||||
except WSCannotConnect:
|
||||
return await self.async_step_ws_port()
|
||||
except CannotConnect:
|
||||
errors["base"] = "cannot_connect"
|
||||
except Exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Unexpected exception")
|
||||
errors["base"] = "unknown"
|
||||
else:
|
||||
return self._create_entry()
|
||||
|
||||
return self._show_credentials_form(errors)
|
||||
|
||||
async def async_step_ws_port(self, user_input=None):
|
||||
"""Handle websocket port of discovered node."""
|
||||
errors = {}
|
||||
|
||||
if user_input is not None:
|
||||
self._ws_port = user_input.get(CONF_WS_PORT)
|
||||
|
||||
try:
|
||||
await validate_ws(self.hass, self._get_data())
|
||||
except WSCannotConnect:
|
||||
errors["base"] = "cannot_connect"
|
||||
except Exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Unexpected exception")
|
||||
errors["base"] = "unknown"
|
||||
else:
|
||||
return self._create_entry()
|
||||
|
||||
return self._show_ws_port_form(errors)
|
||||
|
||||
async def async_step_import(self, data):
|
||||
"""Handle import from YAML."""
|
||||
# We assume that the imported values work and just create the entry
|
||||
return self.async_create_entry(title=data[CONF_NAME], data=data)
|
||||
reason = None
|
||||
try:
|
||||
await validate_http(self.hass, data)
|
||||
await validate_ws(self.hass, data)
|
||||
except InvalidAuth:
|
||||
_LOGGER.exception("Invalid Kodi credentials")
|
||||
reason = "invalid_auth"
|
||||
except CannotConnect:
|
||||
_LOGGER.exception("Cannot connect to Kodi")
|
||||
reason = "cannot_connect"
|
||||
except Exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Unexpected exception")
|
||||
reason = "unknown"
|
||||
else:
|
||||
return self.async_create_entry(title=data[CONF_NAME], data=data)
|
||||
|
||||
return self.async_abort(reason=reason)
|
||||
|
||||
@callback
|
||||
def _show_credentials_form(self, errors=None):
|
||||
schema = vol.Schema(
|
||||
{
|
||||
vol.Optional(
|
||||
CONF_USERNAME, description={"suggested_value": self._username}
|
||||
): str,
|
||||
vol.Optional(
|
||||
CONF_PASSWORD, description={"suggested_value": self._password}
|
||||
): str,
|
||||
}
|
||||
)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="credentials", data_schema=schema, errors=errors or {}
|
||||
)
|
||||
|
||||
@callback
|
||||
def _show_user_form(self, errors=None):
|
||||
default_port = self._port or DEFAULT_PORT
|
||||
default_ssl = self._ssl or DEFAULT_SSL
|
||||
schema = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_HOST, default=self._host): str,
|
||||
vol.Required(CONF_PORT, default=default_port): int,
|
||||
vol.Required(CONF_SSL, default=default_ssl): bool,
|
||||
}
|
||||
)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user", data_schema=schema, errors=errors or {}
|
||||
)
|
||||
|
||||
@callback
|
||||
def _show_ws_port_form(self, errors=None):
|
||||
suggestion = self._ws_port or DEFAULT_WS_PORT
|
||||
schema = vol.Schema(
|
||||
{
|
||||
vol.Optional(
|
||||
CONF_WS_PORT, description={"suggested_value": suggestion}
|
||||
): int
|
||||
}
|
||||
)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="ws_port", data_schema=schema, errors=errors or {}
|
||||
)
|
||||
|
||||
@callback
|
||||
def _create_entry(self):
|
||||
@ -215,42 +302,6 @@ class KodiConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
|
||||
return data
|
||||
|
||||
@callback
|
||||
def _ws_port_schema(self):
|
||||
suggestion = self._ws_port or DEFAULT_WS_PORT
|
||||
return vol.Schema(
|
||||
{
|
||||
vol.Optional(
|
||||
CONF_WS_PORT, description={"suggested_value": suggestion}
|
||||
): int
|
||||
}
|
||||
)
|
||||
|
||||
@callback
|
||||
def _host_schema(self):
|
||||
default_port = self._port or DEFAULT_PORT
|
||||
default_ssl = self._ssl or DEFAULT_SSL
|
||||
return vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_HOST, default=self._host): str,
|
||||
vol.Required(CONF_PORT, default=default_port): int,
|
||||
vol.Required(CONF_SSL, default=default_ssl): bool,
|
||||
}
|
||||
)
|
||||
|
||||
@callback
|
||||
def _credentials_schema(self):
|
||||
return vol.Schema(
|
||||
{
|
||||
vol.Optional(
|
||||
CONF_USERNAME, description={"suggested_value": self._username}
|
||||
): str,
|
||||
vol.Optional(
|
||||
CONF_PASSWORD, description={"suggested_value": self._password}
|
||||
): str,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class CannotConnect(exceptions.HomeAssistantError):
|
||||
"""Error to indicate we cannot connect."""
|
||||
@ -258,3 +309,7 @@ class CannotConnect(exceptions.HomeAssistantError):
|
||||
|
||||
class InvalidAuth(exceptions.HomeAssistantError):
|
||||
"""Error to indicate there is invalid auth."""
|
||||
|
||||
|
||||
class WSCannotConnect(exceptions.HomeAssistantError):
|
||||
"""Error to indicate we cannot connect to websocket."""
|
||||
|
||||
@ -2,7 +2,7 @@
|
||||
"config": {
|
||||
"flow_title": "Kodi: {name}",
|
||||
"step": {
|
||||
"host": {
|
||||
"user": {
|
||||
"description": "Kodi connection information. Please make sure to enable \"Allow control of Kodi via HTTP\" in System/Settings/Network/Services.",
|
||||
"data": {
|
||||
"host": "[%key:common::config_flow::data::host%]",
|
||||
@ -35,7 +35,9 @@
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
|
||||
"cannot_connect": "Cannot connect to discovered Kodi"
|
||||
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
|
||||
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
|
||||
"unknown": "[%key:common::config_flow::error::unknown%]"
|
||||
}
|
||||
},
|
||||
"device_automation": {
|
||||
@ -44,4 +46,4 @@
|
||||
"turn_off": "{entity_name} was requested to turn off"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -27,6 +27,8 @@ async def init_integration(hass) -> MockConfigEntry:
|
||||
CONF_SSL: False,
|
||||
}
|
||||
entry = MockConfigEntry(domain=DOMAIN, data=entry_data, title="name")
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
with patch("homeassistant.components.kodi.Kodi.ping", return_value=True), patch(
|
||||
"homeassistant.components.kodi.Kodi.get_application_properties",
|
||||
return_value={"version": {"major": 1, "minor": 1}},
|
||||
@ -34,7 +36,6 @@ async def init_integration(hass) -> MockConfigEntry:
|
||||
"homeassistant.components.kodi.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
):
|
||||
entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
@ -16,9 +16,11 @@ from .util import (
|
||||
TEST_WS_PORT,
|
||||
UUID,
|
||||
MockConnection,
|
||||
MockWSConnection,
|
||||
get_kodi_connection,
|
||||
)
|
||||
|
||||
from tests.async_mock import AsyncMock, patch
|
||||
from tests.async_mock import AsyncMock, PropertyMock, patch
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
@ -30,32 +32,55 @@ async def user_flow(hass):
|
||||
)
|
||||
assert result["type"] == "form"
|
||||
assert result["errors"] == {}
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], TEST_HOST
|
||||
)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["errors"] == {}
|
||||
|
||||
return result["flow_id"]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def discovery_flow(hass):
|
||||
"""Return a discovery flow after confirmation."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "zeroconf"}, data=TEST_DISCOVERY
|
||||
)
|
||||
assert result["type"] == "form"
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["errors"] == {}
|
||||
return result["flow_id"]
|
||||
|
||||
|
||||
async def test_user_flow(hass, user_flow):
|
||||
"""Test a successful user initiated flow."""
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping", return_value=True,
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
), patch(
|
||||
"homeassistant.components.kodi.async_setup", return_value=True
|
||||
) as mock_setup, patch(
|
||||
"homeassistant.components.kodi.async_setup_entry", return_value=True,
|
||||
) as mock_setup_entry:
|
||||
result = await hass.config_entries.flow.async_configure(user_flow, TEST_HOST)
|
||||
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["title"] == TEST_HOST["host"]
|
||||
assert result["data"] == {
|
||||
**TEST_HOST,
|
||||
**TEST_WS_PORT,
|
||||
"password": None,
|
||||
"username": None,
|
||||
"name": None,
|
||||
"timeout": DEFAULT_TIMEOUT,
|
||||
}
|
||||
|
||||
await hass.async_block_till_done()
|
||||
assert len(mock_setup.mock_calls) == 1
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_form_valid_auth(hass, user_flow):
|
||||
"""Test we handle valid auth."""
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping",
|
||||
side_effect=InvalidAuthError,
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(user_flow, TEST_HOST)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "credentials"
|
||||
assert result["errors"] == {}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping", return_value=True,
|
||||
), patch(
|
||||
@ -67,22 +92,61 @@ async def test_user_flow(hass, user_flow):
|
||||
"homeassistant.components.kodi.async_setup_entry", return_value=True,
|
||||
) as mock_setup_entry:
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
user_flow, TEST_CREDENTIALS
|
||||
result["flow_id"], TEST_CREDENTIALS
|
||||
)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["errors"] == {}
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["title"] == TEST_HOST["host"]
|
||||
assert result["data"] == {
|
||||
**TEST_HOST,
|
||||
**TEST_WS_PORT,
|
||||
**TEST_CREDENTIALS,
|
||||
"name": None,
|
||||
"timeout": DEFAULT_TIMEOUT,
|
||||
}
|
||||
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
await hass.async_block_till_done()
|
||||
assert len(mock_setup.mock_calls) == 1
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_form_valid_ws_port(hass, user_flow):
|
||||
"""Test we handle valid websocket port."""
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping", return_value=True,
|
||||
), patch.object(
|
||||
MockWSConnection, "connect", AsyncMock(side_effect=CannotConnectError),
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
new=get_kodi_connection,
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(user_flow, TEST_HOST)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "ws_port"
|
||||
assert result["errors"] == {}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping", return_value=True,
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
), patch(
|
||||
"homeassistant.components.kodi.async_setup", return_value=True
|
||||
) as mock_setup, patch(
|
||||
"homeassistant.components.kodi.async_setup_entry", return_value=True,
|
||||
) as mock_setup_entry:
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], TEST_WS_PORT
|
||||
)
|
||||
|
||||
assert result2["type"] == "create_entry"
|
||||
assert result2["title"] == TEST_HOST["host"]
|
||||
assert result2["data"] == {
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["title"] == TEST_HOST["host"]
|
||||
assert result["data"] == {
|
||||
**TEST_HOST,
|
||||
**TEST_CREDENTIALS,
|
||||
**TEST_WS_PORT,
|
||||
"password": None,
|
||||
"username": None,
|
||||
"name": None,
|
||||
"timeout": DEFAULT_TIMEOUT,
|
||||
}
|
||||
@ -94,6 +158,19 @@ async def test_user_flow(hass, user_flow):
|
||||
|
||||
async def test_form_invalid_auth(hass, user_flow):
|
||||
"""Test we handle invalid auth."""
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping",
|
||||
side_effect=InvalidAuthError,
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(user_flow, TEST_HOST)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "credentials"
|
||||
assert result["errors"] == {}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping",
|
||||
side_effect=InvalidAuthError,
|
||||
@ -102,12 +179,58 @@ async def test_form_invalid_auth(hass, user_flow):
|
||||
return_value=MockConnection(),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
user_flow, TEST_CREDENTIALS
|
||||
result["flow_id"], TEST_CREDENTIALS
|
||||
)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "credentials"
|
||||
assert result["errors"] == {"base": "invalid_auth"}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping",
|
||||
side_effect=CannotConnectError,
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], TEST_CREDENTIALS
|
||||
)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "credentials"
|
||||
assert result["errors"] == {"base": "cannot_connect"}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping", side_effect=Exception,
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], TEST_CREDENTIALS
|
||||
)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "credentials"
|
||||
assert result["errors"] == {"base": "unknown"}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping", return_value=True,
|
||||
), patch.object(
|
||||
MockWSConnection, "connect", AsyncMock(side_effect=CannotConnectError),
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
new=get_kodi_connection,
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], TEST_CREDENTIALS
|
||||
)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "ws_port"
|
||||
assert result["errors"] == {}
|
||||
|
||||
|
||||
async def test_form_cannot_connect_http(hass, user_flow):
|
||||
"""Test we handle cannot connect over HTTP error."""
|
||||
@ -118,11 +241,10 @@ async def test_form_cannot_connect_http(hass, user_flow):
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
user_flow, TEST_CREDENTIALS
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(user_flow, TEST_HOST)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {"base": "cannot_connect"}
|
||||
|
||||
|
||||
@ -134,11 +256,10 @@ async def test_form_exception_http(hass, user_flow):
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
user_flow, TEST_CREDENTIALS
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(user_flow, TEST_HOST)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {"base": "unknown"}
|
||||
|
||||
|
||||
@ -146,108 +267,114 @@ async def test_form_cannot_connect_ws(hass, user_flow):
|
||||
"""Test we handle cannot connect over WebSocket error."""
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping", return_value=True,
|
||||
), patch.object(
|
||||
MockWSConnection, "connect", AsyncMock(side_effect=CannotConnectError),
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
new=get_kodi_connection,
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(user_flow, TEST_HOST)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "ws_port"
|
||||
assert result["errors"] == {}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping", return_value=True,
|
||||
), patch.object(
|
||||
MockWSConnection, "connected", new_callable=PropertyMock(return_value=False)
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
new=get_kodi_connection,
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
user_flow, TEST_CREDENTIALS
|
||||
)
|
||||
|
||||
with patch.object(
|
||||
MockConnection, "connect", AsyncMock(side_effect=CannotConnectError)
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], TEST_WS_PORT
|
||||
)
|
||||
|
||||
assert result2["type"] == "form"
|
||||
assert result2["errors"] == {"base": "cannot_connect"}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(connected=False),
|
||||
):
|
||||
result3 = await hass.config_entries.flow.async_configure(
|
||||
result2["flow_id"], TEST_WS_PORT
|
||||
)
|
||||
|
||||
assert result3["type"] == "form"
|
||||
assert result3["errors"] == {"base": "cannot_connect"}
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "ws_port"
|
||||
assert result["errors"] == {"base": "cannot_connect"}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping",
|
||||
side_effect=CannotConnectError,
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
new=get_kodi_connection,
|
||||
):
|
||||
result4 = await hass.config_entries.flow.async_configure(
|
||||
result3["flow_id"], TEST_WS_PORT
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], TEST_WS_PORT
|
||||
)
|
||||
|
||||
assert result4["type"] == "form"
|
||||
assert result4["errors"] == {"base": "cannot_connect"}
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "ws_port"
|
||||
assert result["errors"] == {"base": "cannot_connect"}
|
||||
|
||||
|
||||
async def test_form_exception_ws(hass, user_flow):
|
||||
"""Test we handle generic exception over WebSocket."""
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping", return_value=True,
|
||||
), patch.object(
|
||||
MockWSConnection, "connect", AsyncMock(side_effect=CannotConnectError),
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
new=get_kodi_connection,
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
user_flow, TEST_CREDENTIALS
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(user_flow, TEST_HOST)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "ws_port"
|
||||
assert result["errors"] == {}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping", side_effect=Exception,
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping", return_value=True,
|
||||
), patch.object(
|
||||
MockWSConnection, "connect", AsyncMock(side_effect=Exception)
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
new=get_kodi_connection,
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], TEST_WS_PORT
|
||||
)
|
||||
|
||||
assert result2["type"] == "form"
|
||||
assert result2["errors"] == {"base": "unknown"}
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "ws_port"
|
||||
assert result["errors"] == {"base": "unknown"}
|
||||
|
||||
|
||||
async def test_discovery(hass, discovery_flow):
|
||||
async def test_discovery(hass):
|
||||
"""Test discovery flow works."""
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping", return_value=True,
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
), patch(
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "zeroconf"}, data=TEST_DISCOVERY
|
||||
)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "discovery_confirm"
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.kodi.async_setup", return_value=True
|
||||
) as mock_setup, patch(
|
||||
"homeassistant.components.kodi.async_setup_entry", return_value=True,
|
||||
) as mock_setup_entry:
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
discovery_flow, TEST_CREDENTIALS
|
||||
flow_id=result["flow_id"], user_input={}
|
||||
)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["errors"] == {}
|
||||
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], TEST_WS_PORT
|
||||
)
|
||||
|
||||
assert result2["type"] == "create_entry"
|
||||
assert result2["title"] == "hostname"
|
||||
assert result2["data"] == {
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["title"] == "hostname"
|
||||
assert result["data"] == {
|
||||
**TEST_HOST,
|
||||
**TEST_CREDENTIALS,
|
||||
**TEST_WS_PORT,
|
||||
"password": None,
|
||||
"username": None,
|
||||
"name": "hostname",
|
||||
"timeout": DEFAULT_TIMEOUT,
|
||||
}
|
||||
@ -257,7 +384,7 @@ async def test_discovery(hass, discovery_flow):
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_discovery_cannot_connect_http(hass, discovery_flow):
|
||||
async def test_discovery_cannot_connect_http(hass):
|
||||
"""Test discovery aborts if cannot connect."""
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping",
|
||||
@ -266,19 +393,86 @@ async def test_discovery_cannot_connect_http(hass, discovery_flow):
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
discovery_flow, TEST_CREDENTIALS
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "zeroconf"}, data=TEST_DISCOVERY
|
||||
)
|
||||
|
||||
assert result["type"] == "abort"
|
||||
assert result["reason"] == "cannot_connect"
|
||||
|
||||
|
||||
async def test_discovery_duplicate_data(hass, discovery_flow):
|
||||
async def test_discovery_cannot_connect_ws(hass):
|
||||
"""Test discovery aborts if cannot connect to websocket."""
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping", return_value=True,
|
||||
), patch.object(
|
||||
MockWSConnection, "connect", AsyncMock(side_effect=CannotConnectError),
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
new=get_kodi_connection,
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "zeroconf"}, data=TEST_DISCOVERY
|
||||
)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "ws_port"
|
||||
assert result["errors"] == {}
|
||||
|
||||
|
||||
async def test_discovery_exception_http(hass, user_flow):
|
||||
"""Test we handle generic exception during discovery validation."""
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping", side_effect=Exception,
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "zeroconf"}, data=TEST_DISCOVERY
|
||||
)
|
||||
|
||||
assert result["type"] == "abort"
|
||||
assert result["reason"] == "unknown"
|
||||
|
||||
|
||||
async def test_discovery_invalid_auth(hass):
|
||||
"""Test we handle invalid auth during discovery."""
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping",
|
||||
side_effect=InvalidAuthError,
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "zeroconf"}, data=TEST_DISCOVERY
|
||||
)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "credentials"
|
||||
assert result["errors"] == {}
|
||||
|
||||
|
||||
async def test_discovery_duplicate_data(hass):
|
||||
"""Test discovery aborts if same mDNS packet arrives."""
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping", return_value=True,
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "zeroconf"}, data=TEST_DISCOVERY
|
||||
)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "discovery_confirm"
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "zeroconf"}, data=TEST_DISCOVERY
|
||||
)
|
||||
|
||||
assert result["type"] == "abort"
|
||||
assert result["reason"] == "already_in_progress"
|
||||
|
||||
@ -308,6 +502,11 @@ async def test_discovery_updates_unique_id(hass):
|
||||
async def test_form_import(hass):
|
||||
"""Test we get the form with import source."""
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping", return_value=True,
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
), patch(
|
||||
"homeassistant.components.kodi.async_setup", return_value=True
|
||||
) as mock_setup, patch(
|
||||
"homeassistant.components.kodi.async_setup_entry", return_value=True,
|
||||
@ -323,3 +522,53 @@ async def test_form_import(hass):
|
||||
await hass.async_block_till_done()
|
||||
assert len(mock_setup.mock_calls) == 1
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_form_import_invalid_auth(hass):
|
||||
"""Test we handle invalid auth on import."""
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping",
|
||||
side_effect=InvalidAuthError,
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=TEST_IMPORT,
|
||||
)
|
||||
|
||||
assert result["type"] == "abort"
|
||||
assert result["reason"] == "invalid_auth"
|
||||
|
||||
|
||||
async def test_form_import_cannot_connect(hass):
|
||||
"""Test we handle cannot connect on import."""
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping",
|
||||
side_effect=CannotConnectError,
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=TEST_IMPORT,
|
||||
)
|
||||
|
||||
assert result["type"] == "abort"
|
||||
assert result["reason"] == "cannot_connect"
|
||||
|
||||
|
||||
async def test_form_import_exception(hass):
|
||||
"""Test we handle unknown exception on import."""
|
||||
with patch(
|
||||
"homeassistant.components.kodi.config_flow.Kodi.ping", side_effect=Exception,
|
||||
), patch(
|
||||
"homeassistant.components.kodi.config_flow.get_kodi_connection",
|
||||
return_value=MockConnection(),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=TEST_IMPORT,
|
||||
)
|
||||
|
||||
assert result["type"] == "abort"
|
||||
assert result["reason"] == "unknown"
|
||||
|
||||
@ -36,6 +36,16 @@ TEST_IMPORT = {
|
||||
}
|
||||
|
||||
|
||||
def get_kodi_connection(
|
||||
host, port, ws_port, username, password, ssl=False, timeout=5, session=None
|
||||
):
|
||||
"""Get Kodi connection."""
|
||||
if ws_port is None:
|
||||
return MockConnection()
|
||||
else:
|
||||
return MockWSConnection()
|
||||
|
||||
|
||||
class MockConnection:
|
||||
"""A mock kodi connection."""
|
||||
|
||||
@ -65,3 +75,34 @@ class MockConnection:
|
||||
def server(self):
|
||||
"""Mock server."""
|
||||
return None
|
||||
|
||||
|
||||
class MockWSConnection:
|
||||
"""A mock kodi websocket connection."""
|
||||
|
||||
def __init__(self, connected=True):
|
||||
"""Mock the websocket connection."""
|
||||
self._connected = connected
|
||||
|
||||
async def connect(self):
|
||||
"""Mock connect."""
|
||||
pass
|
||||
|
||||
@property
|
||||
def connected(self):
|
||||
"""Mock connected."""
|
||||
return self._connected
|
||||
|
||||
@property
|
||||
def can_subscribe(self):
|
||||
"""Mock can_subscribe."""
|
||||
return False
|
||||
|
||||
async def close(self):
|
||||
"""Mock close."""
|
||||
pass
|
||||
|
||||
@property
|
||||
def server(self):
|
||||
"""Mock server."""
|
||||
return None
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user