How to use aiounifi - 10 common examples
To help you get started, we’ve selected a few aiounifi examples, based on popular ways it is used in public projects.
Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
home-assistant / home-assistant / tests / components / unifi / test_config_flow.py View on Github 
async def test_flow_fails_controller_unavailable(hass, aioclient_mock):
"""Test config flow."""
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, context={"source": "user"}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
with patch("aiounifi.Controller.login", side_effect=aiounifi.errors.RequestError):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "username",
CONF_PASSWORD: "password",
CONF_PORT: 1234,
CONF_VERIFY_SSL: True,
},
)
assert result["type"] == "form"
assert result["errors"] == {"base": "service_unavailable"}async def test_flow_fails_user_credentials_faulty(hass, aioclient_mock):
"""Test config flow."""
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, context={"source": "user"}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
with patch("aiounifi.Controller.login", side_effect=aiounifi.errors.Unauthorized):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "username",
CONF_PASSWORD: "password",
CONF_PORT: 1234,
CONF_VERIFY_SSL: True,
},
)
assert result["type"] == "form"
assert result["errors"] == {"base": "faulty_credentials"}async def test_get_controller_unknown_error(hass):
"""Check that get_controller can handle unkown errors."""
result = None
with patch("aiounifi.Controller.login", side_effect=aiounifi.AiounifiException):
try:
result = await unifi.controller.get_controller(hass, **CONTROLLER_DATA)
except unifi.errors.AuthenticationRequired:
pass
assert result is Noneasync def test_failed_update_failed_login(hass):
"""Running update can handle a failed login."""
controller = await setup_unifi_integration(
hass,
ENTRY_CONFIG,
options={},
sites=SITES,
clients_response=[],
devices_response=[],
clients_all_response=[],
)
with patch.object(
controller.api.clients, "update", side_effect=aiounifi.LoginRequired
), patch.object(controller.api, "login", side_effect=aiounifi.AiounifiException):
await controller.async_update()
await hass.async_block_till_done()
assert controller.available is Falseasync def test_failed_update_successful_login(hass):
"""Running update can login when requested."""
controller = await setup_unifi_integration(
hass,
ENTRY_CONFIG,
options={},
sites=SITES,
clients_response=[],
devices_response=[],
clients_all_response=[],
)
with patch.object(
controller.api.clients, "update", side_effect=aiounifi.LoginRequired
), patch.object(controller.api, "login", return_value=Mock(True)):
await controller.async_update()
await hass.async_block_till_done()
assert controller.available is Trueasync def test_failed_update_failed_login(hass):
"""Running update can handle a failed login."""
controller = await setup_unifi_integration(
hass,
ENTRY_CONFIG,
options={},
sites=SITES,
clients_response=[],
devices_response=[],
clients_all_response=[],
)
with patch.object(
controller.api.clients, "update", side_effect=aiounifi.LoginRequired
), patch.object(controller.api, "login", side_effect=aiounifi.AiounifiException):
await controller.async_update()
await hass.async_block_till_done()
assert controller.available is Falseasync def test_get_controller_controller_unavailable(hass):
"""Check that get_controller can handle controller being unavailable."""
result = None
with patch("aiounifi.Controller.login", side_effect=aiounifi.RequestError):
try:
result = await unifi.controller.get_controller(hass, **CONTROLLER_DATA)
except unifi.errors.CannotConnect:
pass
assert result is Noneasync def test_get_controller_login_failed(hass):
"""Check that get_controller can handle a failed login."""
result = None
with patch("aiounifi.Controller.login", side_effect=aiounifi.Unauthorized):
try:
result = await unifi.controller.get_controller(hass, **CONTROLLER_DATA)
except unifi.errors.AuthenticationRequired:
pass
assert result is Noneawait self.api.devices.update()
if self.option_block_clients:
await self.api.clients_all.update()
except aiounifi.LoginRequired:
try:
with async_timeout.timeout(5):
await self.api.login()
except (asyncio.TimeoutError, aiounifi.AiounifiException):
failed = True
if self.available:
LOGGER.error("Unable to reach controller %s", self.host)
self.available = False
except (asyncio.TimeoutError, aiounifi.AiounifiException):
failed = True
if self.available:
LOGGER.error("Unable to reach controller %s", self.host)
self.available = False
if not failed and not self.available:
LOGGER.info("Reconnected to controller %s", self.host)
self.available = True
self.update_wireless_clients()
async_dispatcher_send(self.hass, self.signal_update)async def async_update(self):
"""Update UniFi controller information."""
failed = False
try:
with async_timeout.timeout(10):
await self.api.clients.update()
await self.api.devices.update()
if self.option_block_clients:
await self.api.clients_all.update()
except aiounifi.LoginRequired:
try:
with async_timeout.timeout(5):
await self.api.login()
except (asyncio.TimeoutError, aiounifi.AiounifiException):
failed = True
if self.available:
LOGGER.error("Unable to reach controller %s", self.host)
self.available = False
except (asyncio.TimeoutError, aiounifi.AiounifiException):
failed = True
if self.available:
LOGGER.error("Unable to reach controller %s", self.host)
self.available = Falseaiounifi
Python library for communicating with UniFi Network Controller API
