Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_link_no_api_key(hass):
"""Test config flow should abort if no API key was possible to retrieve."""
flow = config_flow.DeconzFlowHandler()
flow.hass = hass
flow.deconz_config = {config_flow.CONF_HOST: "1.2.3.4", config_flow.CONF_PORT: 80}
with patch(
"homeassistant.components.deconz.config_flow.async_get_api_key",
side_effect=pydeconz.errors.ResponseError,
):
result = await flow.async_step_link(user_input={})
assert result["type"] == "form"
assert result["step_id"] == "link"
assert result["errors"] == {"base": "no_key"}
async def test_request_fails_invalid_content() -> None:
"""Test a successful call of request."""
response = Mock()
response.content_type = "application/binary"
session = CoroutineMock(return_value=response)
with pytest.raises(errors.ResponseError) as e_info:
await utils.async_request(session, "url")
assert str(e_info.value) == "Invalid content type: application/binary"
async def async_request(session, url, **kwargs):
"""Do a web request and manage response."""
LOGGER.debug("Sending %s to %s", kwargs, url)
try:
res = await session(url, **kwargs)
if res.content_type != "application/json":
raise ResponseError("Invalid content type: {}".format(res.content_type))
response = await res.json()
LOGGER.debug("HTTP request response: %s", response)
_raise_on_error(response)
return response
except aiohttp.client_exceptions.ClientError as err:
raise RequestError(
"Error requesting data from {}: {}".format(url, err)
) from None
if user_input is not None:
for bridge in self.bridges:
if bridge[CONF_HOST] == user_input[CONF_HOST]:
self.deconz_config = bridge
return await self.async_step_link()
self.deconz_config = user_input
return await self.async_step_link()
session = aiohttp_client.async_get_clientsession(self.hass)
try:
with async_timeout.timeout(10):
self.bridges = await async_discovery(session)
except (asyncio.TimeoutError, ResponseError):
self.bridges = []
if len(self.bridges) == 1:
self.deconz_config = self.bridges[0]
return await self.async_step_link()
if len(self.bridges) > 1:
hosts = []
for bridge in self.bridges:
hosts.append(bridge[CONF_HOST])
return self.async_show_form(
step_id="init",
data_schema=vol.Schema({vol.Required(CONF_HOST): vol.In(hosts)}),
)
async def request(self, method, path="", json=None):
"""Make a request to the API."""
LOGGER.debug('Sending "%s" "%s" to "%s %s"', method, json, self.host, path)
url = f"http://{self.host}:{self.port}/api/{self.api_key}{path}"
try:
async with self.session.request(method, url, json=json) as res:
if res.content_type != "application/json":
raise ResponseError(
"Invalid content type: {}".format(res.content_type)
)
response = await res.json()
LOGGER.debug("HTTP request response: %s", pformat(response))
_raise_on_error(response)
return response
except client_exceptions.ClientError as err:
raise RequestError(
"Error requesting data from {}: {}".format(self.host, err)
) from None