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_get_gateway_fails_cannot_connect(hass):
"""Failed call."""
with patch(
"pydeconz.DeconzSession.async_load_parameters",
side_effect=pydeconz.errors.RequestError,
), pytest.raises(deconz.errors.CannotConnect):
assert (
await deconz.gateway.get_gateway(hass, ENTRY_CONFIG, Mock(), Mock())
is False
)
async def test_request_fails_client_error() -> None:
"""Test a successful call of request."""
session = CoroutineMock(side_effect=aiohttp.ClientError)
with pytest.raises(errors.RequestError) as e_info:
await utils.async_request(session, "url")
assert str(e_info.value) == "Error requesting data from url: "
config[CONF_HOST],
config[CONF_PORT],
config[CONF_API_KEY],
async_add_device=async_add_device_callback,
connection_status=async_connection_status_callback,
)
try:
with async_timeout.timeout(10):
await deconz.initialize()
return deconz
except errors.Unauthorized:
_LOGGER.warning("Invalid key for deCONZ at %s", config[CONF_HOST])
raise AuthenticationRequired
except (asyncio.TimeoutError, errors.RequestError):
_LOGGER.error("Error connecting to deCONZ gateway at %s", config[CONF_HOST])
raise CannotConnect
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
"""The caller has no rights to access the requested URI."""
class ResourceNotFound(pydeconzException):
"""The requested resource (light, group, ...) was not found."""
class BridgeBusy(pydeconzException):
"""The Bridge is busy, too many requests (more than 20)."""
ERRORS = {
1: Unauthorized, # Unauthorized user
2: BadRequest, # Body contains invalid JSON
3: ResourceNotFound, # Resource not available
4: RequestError, # Method not available for resource
5: BadRequest, # Missing parameters in body
6: RequestError, # Parameter not available
7: RequestError, # Invalid value for parameter
8: RequestError, # Parameter is not modifiable
901: BridgeBusy, # May occur when sending too fast
}
def raise_error(error):
if error:
cls = ERRORS.get(error["type"], pydeconzException)
raise cls("{} {}".format(error["address"], error["description"]))
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
async def async_step_link(self, user_input=None):
"""Attempt to link with the deCONZ bridge."""
errors = {}
if user_input is not None:
session = aiohttp_client.async_get_clientsession(self.hass)
try:
with async_timeout.timeout(10):
api_key = await async_get_api_key(session, **self.deconz_config)
except (ResponseError, RequestError, asyncio.TimeoutError):
errors["base"] = "no_key"
else:
self.deconz_config[CONF_API_KEY] = api_key
return await self._create_entry()
return self.async_show_form(step_id="link", errors=errors)