mirror of https://github.com/home-assistant/core
975 lines
35 KiB
Python
975 lines
35 KiB
Python
"""Test ONVIF config flow."""
|
|
|
|
import logging
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from homeassistant import config_entries
|
|
from homeassistant.components import dhcp
|
|
from homeassistant.components.onvif import DOMAIN, config_flow
|
|
from homeassistant.config_entries import SOURCE_DHCP
|
|
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_USERNAME
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.data_entry_flow import FlowResultType
|
|
from homeassistant.helpers import device_registry as dr
|
|
|
|
from . import (
|
|
HOST,
|
|
MAC,
|
|
NAME,
|
|
PASSWORD,
|
|
PORT,
|
|
URN,
|
|
USERNAME,
|
|
setup_mock_device,
|
|
setup_mock_onvif_camera,
|
|
setup_onvif_integration,
|
|
)
|
|
|
|
DISCOVERY = [
|
|
{
|
|
"EPR": URN,
|
|
config_flow.CONF_NAME: NAME,
|
|
config_flow.CONF_HOST: HOST,
|
|
config_flow.CONF_PORT: PORT,
|
|
"MAC": MAC,
|
|
"HARDWARE": "IPC model",
|
|
},
|
|
{
|
|
"EPR": "urn:uuid:987654321",
|
|
config_flow.CONF_NAME: "TestCamera2",
|
|
config_flow.CONF_HOST: "5.6.7.8",
|
|
config_flow.CONF_PORT: PORT,
|
|
"MAC": "ff:ee:dd:cc:bb:aa",
|
|
},
|
|
]
|
|
DHCP_DISCOVERY = dhcp.DhcpServiceInfo(
|
|
hostname="any", ip="5.6.7.8", macaddress=MAC.lower().replace(":", "")
|
|
)
|
|
DHCP_DISCOVERY_SAME_IP = dhcp.DhcpServiceInfo(
|
|
hostname="any", ip="1.2.3.4", macaddress=MAC.lower().replace(":", "")
|
|
)
|
|
|
|
|
|
def setup_mock_discovery(
|
|
mock_discovery,
|
|
with_name=False,
|
|
with_mac=False,
|
|
two_devices=False,
|
|
with_hardware=True,
|
|
no_devices=False,
|
|
):
|
|
"""Prepare mock discovery result."""
|
|
services = []
|
|
for item in DISCOVERY:
|
|
if no_devices:
|
|
continue
|
|
service = MagicMock()
|
|
service.getXAddrs = MagicMock(
|
|
return_value=[
|
|
f"http://{item[config_flow.CONF_HOST]}:{item[config_flow.CONF_PORT]}/onvif/device_service"
|
|
]
|
|
)
|
|
service.getEPR = MagicMock(return_value=item["EPR"])
|
|
scopes = []
|
|
if with_name:
|
|
scope = MagicMock()
|
|
scope.getValue = MagicMock(
|
|
return_value=f"onvif://www.onvif.org/name/{item[config_flow.CONF_NAME]}"
|
|
)
|
|
scopes.append(scope)
|
|
if with_mac:
|
|
scope = MagicMock()
|
|
scope.getValue = MagicMock(
|
|
return_value=f"onvif://www.onvif.org/mac/{item['MAC']}"
|
|
)
|
|
scopes.append(scope)
|
|
if with_hardware and "HARDWARE" in item:
|
|
scope = MagicMock()
|
|
scope.getValue = MagicMock(
|
|
return_value=f"onvif://www.onvif.org/hardware/{item['HARDWARE']}"
|
|
)
|
|
scopes.append(scope)
|
|
service.getScopes = MagicMock(return_value=scopes)
|
|
services.append(service)
|
|
|
|
mock_ws_discovery = MagicMock()
|
|
mock_ws_discovery.searchServices = MagicMock(return_value=services)
|
|
mock_discovery.return_value = mock_ws_discovery
|
|
|
|
|
|
async def test_flow_discovered_devices(hass: HomeAssistant) -> None:
|
|
"""Test that config flow works for discovered devices."""
|
|
logging.getLogger("homeassistant.components.onvif").setLevel(logging.DEBUG)
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
config_flow.DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "user"
|
|
|
|
with (
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.get_device"
|
|
) as mock_onvif_camera,
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.WSDiscovery"
|
|
) as mock_discovery,
|
|
patch("homeassistant.components.onvif.ONVIFDevice") as mock_device,
|
|
):
|
|
setup_mock_onvif_camera(mock_onvif_camera)
|
|
setup_mock_discovery(mock_discovery)
|
|
setup_mock_device(mock_device)
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={"auto": True}
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "device"
|
|
container = result["data_schema"].schema[config_flow.CONF_HOST].container
|
|
assert len(container) == 3
|
|
assert container == {
|
|
"Manually configure ONVIF device": "Manually configure ONVIF device",
|
|
"1.2.3.4": "urn:uuid:123456789 (1.2.3.4) [IPC model]",
|
|
"5.6.7.8": "urn:uuid:987654321 (5.6.7.8)",
|
|
}
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={config_flow.CONF_HOST: HOST}
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "configure"
|
|
|
|
with patch(
|
|
"homeassistant.components.onvif.async_setup_entry", return_value=True
|
|
) as mock_setup_entry:
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
user_input={
|
|
config_flow.CONF_USERNAME: USERNAME,
|
|
config_flow.CONF_PASSWORD: PASSWORD,
|
|
},
|
|
)
|
|
|
|
await hass.async_block_till_done()
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == f"{URN} - {MAC}"
|
|
assert result["data"] == {
|
|
config_flow.CONF_NAME: URN,
|
|
config_flow.CONF_HOST: HOST,
|
|
config_flow.CONF_PORT: PORT,
|
|
config_flow.CONF_USERNAME: USERNAME,
|
|
config_flow.CONF_PASSWORD: PASSWORD,
|
|
}
|
|
|
|
|
|
async def test_flow_discovered_devices_ignore_configured_manual_input(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test that config flow discovery ignores configured devices."""
|
|
logging.getLogger("homeassistant.components.onvif").setLevel(logging.DEBUG)
|
|
await setup_onvif_integration(hass)
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
config_flow.DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "user"
|
|
|
|
with (
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.get_device"
|
|
) as mock_onvif_camera,
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.WSDiscovery"
|
|
) as mock_discovery,
|
|
patch("homeassistant.components.onvif.ONVIFDevice") as mock_device,
|
|
):
|
|
setup_mock_onvif_camera(mock_onvif_camera)
|
|
setup_mock_discovery(mock_discovery, with_mac=True)
|
|
setup_mock_device(mock_device)
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={"auto": True}
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "device"
|
|
assert len(result["data_schema"].schema[config_flow.CONF_HOST].container) == 2
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
user_input={config_flow.CONF_HOST: config_flow.CONF_MANUAL_INPUT},
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "configure"
|
|
|
|
|
|
async def test_flow_discovered_no_device(hass: HomeAssistant) -> None:
|
|
"""Test that config flow discovery no device."""
|
|
await setup_onvif_integration(hass)
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
config_flow.DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "user"
|
|
|
|
with (
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.get_device"
|
|
) as mock_onvif_camera,
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.WSDiscovery"
|
|
) as mock_discovery,
|
|
patch("homeassistant.components.onvif.ONVIFDevice") as mock_device,
|
|
):
|
|
setup_mock_onvif_camera(mock_onvif_camera)
|
|
setup_mock_discovery(mock_discovery, no_devices=True)
|
|
setup_mock_device(mock_device)
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={"auto": True}
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "configure"
|
|
|
|
|
|
async def test_flow_discovery_ignore_existing_and_abort(hass: HomeAssistant) -> None:
|
|
"""Test that config flow discovery ignores setup devices."""
|
|
logging.getLogger("homeassistant.components.onvif").setLevel(logging.DEBUG)
|
|
await setup_onvif_integration(hass)
|
|
await setup_onvif_integration(
|
|
hass,
|
|
config={
|
|
config_flow.CONF_NAME: DISCOVERY[1]["EPR"],
|
|
config_flow.CONF_HOST: DISCOVERY[1][config_flow.CONF_HOST],
|
|
config_flow.CONF_PORT: DISCOVERY[1][config_flow.CONF_PORT],
|
|
config_flow.CONF_USERNAME: "",
|
|
config_flow.CONF_PASSWORD: "",
|
|
},
|
|
unique_id=DISCOVERY[1]["MAC"],
|
|
entry_id="2",
|
|
)
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
config_flow.DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "user"
|
|
|
|
with (
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.get_device"
|
|
) as mock_onvif_camera,
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.WSDiscovery"
|
|
) as mock_discovery,
|
|
patch("homeassistant.components.onvif.ONVIFDevice") as mock_device,
|
|
):
|
|
setup_mock_onvif_camera(mock_onvif_camera)
|
|
setup_mock_discovery(mock_discovery, with_name=True, with_mac=True)
|
|
setup_mock_device(mock_device)
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={"auto": True}
|
|
)
|
|
|
|
# It should skip to manual entry if the only devices are already configured
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "configure"
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
user_input={
|
|
config_flow.CONF_NAME: NAME,
|
|
config_flow.CONF_HOST: HOST,
|
|
config_flow.CONF_PORT: PORT,
|
|
config_flow.CONF_USERNAME: USERNAME,
|
|
config_flow.CONF_PASSWORD: PASSWORD,
|
|
},
|
|
)
|
|
|
|
# It should abort if already configured and entered manually
|
|
assert result["type"] is FlowResultType.ABORT
|
|
|
|
|
|
async def test_flow_manual_entry(hass: HomeAssistant) -> None:
|
|
"""Test that config flow works for discovered devices."""
|
|
logging.getLogger("homeassistant.components.onvif").setLevel(logging.DEBUG)
|
|
result = await hass.config_entries.flow.async_init(
|
|
config_flow.DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "user"
|
|
|
|
with (
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.get_device"
|
|
) as mock_onvif_camera,
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.WSDiscovery"
|
|
) as mock_discovery,
|
|
patch("homeassistant.components.onvif.ONVIFDevice") as mock_device,
|
|
):
|
|
setup_mock_onvif_camera(mock_onvif_camera, two_profiles=True)
|
|
# no discovery
|
|
mock_discovery.return_value = []
|
|
setup_mock_device(mock_device)
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
user_input={"auto": False},
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "configure"
|
|
|
|
with patch(
|
|
"homeassistant.components.onvif.async_setup_entry", return_value=True
|
|
) as mock_setup_entry:
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
user_input={
|
|
config_flow.CONF_NAME: NAME,
|
|
config_flow.CONF_HOST: HOST,
|
|
config_flow.CONF_PORT: PORT,
|
|
config_flow.CONF_USERNAME: USERNAME,
|
|
config_flow.CONF_PASSWORD: PASSWORD,
|
|
},
|
|
)
|
|
|
|
await hass.async_block_till_done()
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == f"{NAME} - {MAC}"
|
|
assert result["data"] == {
|
|
config_flow.CONF_NAME: NAME,
|
|
config_flow.CONF_HOST: HOST,
|
|
config_flow.CONF_PORT: PORT,
|
|
config_flow.CONF_USERNAME: USERNAME,
|
|
config_flow.CONF_PASSWORD: PASSWORD,
|
|
}
|
|
|
|
|
|
async def test_flow_manual_entry_no_profiles(hass: HomeAssistant) -> None:
|
|
"""Test that config flow when no profiles are returned."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
config_flow.DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "user"
|
|
|
|
with (
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.get_device"
|
|
) as mock_onvif_camera,
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.WSDiscovery"
|
|
) as mock_discovery,
|
|
patch("homeassistant.components.onvif.ONVIFDevice") as mock_device,
|
|
):
|
|
setup_mock_onvif_camera(mock_onvif_camera, no_profiles=True)
|
|
# no discovery
|
|
mock_discovery.return_value = []
|
|
setup_mock_device(mock_device)
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
user_input={"auto": False},
|
|
)
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
user_input={
|
|
config_flow.CONF_NAME: NAME,
|
|
config_flow.CONF_HOST: HOST,
|
|
config_flow.CONF_PORT: PORT,
|
|
config_flow.CONF_USERNAME: USERNAME,
|
|
config_flow.CONF_PASSWORD: PASSWORD,
|
|
},
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "no_h264"
|
|
|
|
|
|
async def test_flow_manual_entry_no_mac(hass: HomeAssistant) -> None:
|
|
"""Test that config flow when no mac address is returned."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
config_flow.DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "user"
|
|
|
|
with (
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.get_device"
|
|
) as mock_onvif_camera,
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.WSDiscovery"
|
|
) as mock_discovery,
|
|
patch("homeassistant.components.onvif.ONVIFDevice") as mock_device,
|
|
):
|
|
setup_mock_onvif_camera(
|
|
mock_onvif_camera, with_serial=False, with_interfaces=False
|
|
)
|
|
# no discovery
|
|
mock_discovery.return_value = []
|
|
setup_mock_device(mock_device)
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
user_input={"auto": False},
|
|
)
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
user_input={
|
|
config_flow.CONF_NAME: NAME,
|
|
config_flow.CONF_HOST: HOST,
|
|
config_flow.CONF_PORT: PORT,
|
|
config_flow.CONF_USERNAME: USERNAME,
|
|
config_flow.CONF_PASSWORD: PASSWORD,
|
|
},
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "no_mac"
|
|
|
|
|
|
async def test_flow_manual_entry_fails(hass: HomeAssistant) -> None:
|
|
"""Test that we get a good error when manual entry fails."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
config_flow.DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "user"
|
|
|
|
with (
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.get_device"
|
|
) as mock_onvif_camera,
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.WSDiscovery"
|
|
) as mock_discovery,
|
|
patch("homeassistant.components.onvif.ONVIFDevice") as mock_device,
|
|
):
|
|
setup_mock_onvif_camera(
|
|
mock_onvif_camera, two_profiles=True, profiles_transient_failure=True
|
|
)
|
|
# no discovery
|
|
mock_discovery.return_value = []
|
|
setup_mock_device(mock_device)
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
user_input={"auto": False},
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "configure"
|
|
|
|
with patch(
|
|
"homeassistant.components.onvif.async_setup_entry", return_value=True
|
|
) as mock_setup_entry:
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
user_input={
|
|
config_flow.CONF_NAME: NAME,
|
|
config_flow.CONF_HOST: HOST,
|
|
config_flow.CONF_PORT: PORT,
|
|
config_flow.CONF_USERNAME: USERNAME,
|
|
config_flow.CONF_PASSWORD: PASSWORD,
|
|
},
|
|
)
|
|
|
|
await hass.async_block_till_done()
|
|
assert len(mock_setup_entry.mock_calls) == 0
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "configure"
|
|
assert result["errors"] == {"base": "onvif_error"}
|
|
assert result["description_placeholders"] == {"error": "camera not ready"}
|
|
setup_mock_onvif_camera(
|
|
mock_onvif_camera, two_profiles=True, update_xaddrs_fail=True
|
|
)
|
|
|
|
with patch(
|
|
"homeassistant.components.onvif.async_setup_entry", return_value=True
|
|
) as mock_setup_entry:
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
user_input={
|
|
config_flow.CONF_NAME: NAME,
|
|
config_flow.CONF_HOST: HOST,
|
|
config_flow.CONF_PORT: PORT,
|
|
config_flow.CONF_USERNAME: USERNAME,
|
|
config_flow.CONF_PASSWORD: PASSWORD,
|
|
},
|
|
)
|
|
|
|
await hass.async_block_till_done()
|
|
assert len(mock_setup_entry.mock_calls) == 0
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "configure"
|
|
assert result["errors"] == {"base": "onvif_error"}
|
|
assert result["description_placeholders"] == {
|
|
"error": "Unknown error: camera not ready"
|
|
}
|
|
setup_mock_onvif_camera(mock_onvif_camera, two_profiles=True)
|
|
|
|
with patch(
|
|
"homeassistant.components.onvif.async_setup_entry", return_value=True
|
|
) as mock_setup_entry:
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
user_input={
|
|
config_flow.CONF_NAME: NAME,
|
|
config_flow.CONF_HOST: HOST,
|
|
config_flow.CONF_PORT: PORT,
|
|
config_flow.CONF_USERNAME: USERNAME,
|
|
config_flow.CONF_PASSWORD: PASSWORD,
|
|
},
|
|
)
|
|
|
|
await hass.async_block_till_done()
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
assert result["title"] == f"{NAME} - {MAC}"
|
|
assert result["data"] == {
|
|
config_flow.CONF_NAME: NAME,
|
|
config_flow.CONF_HOST: HOST,
|
|
config_flow.CONF_PORT: PORT,
|
|
config_flow.CONF_USERNAME: USERNAME,
|
|
config_flow.CONF_PASSWORD: PASSWORD,
|
|
}
|
|
|
|
|
|
async def test_flow_manual_entry_wrong_password(hass: HomeAssistant) -> None:
|
|
"""Test that we get a an auth error with the wrong password."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
config_flow.DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "user"
|
|
|
|
with (
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.get_device"
|
|
) as mock_onvif_camera,
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.WSDiscovery"
|
|
) as mock_discovery,
|
|
patch("homeassistant.components.onvif.ONVIFDevice") as mock_device,
|
|
):
|
|
setup_mock_onvif_camera(mock_onvif_camera, two_profiles=True, auth_fail=True)
|
|
# no discovery
|
|
mock_discovery.return_value = []
|
|
setup_mock_device(mock_device)
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
user_input={"auto": False},
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "configure"
|
|
|
|
with patch(
|
|
"homeassistant.components.onvif.async_setup_entry", return_value=True
|
|
) as mock_setup_entry:
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
user_input={
|
|
config_flow.CONF_NAME: NAME,
|
|
config_flow.CONF_HOST: HOST,
|
|
config_flow.CONF_PORT: PORT,
|
|
config_flow.CONF_USERNAME: USERNAME,
|
|
config_flow.CONF_PASSWORD: PASSWORD,
|
|
},
|
|
)
|
|
|
|
await hass.async_block_till_done()
|
|
assert len(mock_setup_entry.mock_calls) == 0
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "configure"
|
|
assert result["errors"] == {"password": "auth_failed"}
|
|
assert result["description_placeholders"] == {"error": "Authority failure"}
|
|
setup_mock_onvif_camera(mock_onvif_camera, two_profiles=True)
|
|
|
|
with patch(
|
|
"homeassistant.components.onvif.async_setup_entry", return_value=True
|
|
) as mock_setup_entry:
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
user_input={
|
|
config_flow.CONF_NAME: NAME,
|
|
config_flow.CONF_HOST: HOST,
|
|
config_flow.CONF_PORT: PORT,
|
|
config_flow.CONF_USERNAME: USERNAME,
|
|
config_flow.CONF_PASSWORD: PASSWORD,
|
|
},
|
|
)
|
|
|
|
await hass.async_block_till_done()
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
assert result["title"] == f"{NAME} - {MAC}"
|
|
assert result["data"] == {
|
|
config_flow.CONF_NAME: NAME,
|
|
config_flow.CONF_HOST: HOST,
|
|
config_flow.CONF_PORT: PORT,
|
|
config_flow.CONF_USERNAME: USERNAME,
|
|
config_flow.CONF_PASSWORD: PASSWORD,
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize("option_value", [True, False])
|
|
async def test_option_flow(hass: HomeAssistant, option_value: bool) -> None:
|
|
"""Test config flow options."""
|
|
entry, _, _ = await setup_onvif_integration(hass)
|
|
|
|
result = await hass.config_entries.options.async_init(
|
|
entry.entry_id, context={"show_advanced_options": True}
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "onvif_devices"
|
|
|
|
result = await hass.config_entries.options.async_configure(
|
|
result["flow_id"],
|
|
user_input={
|
|
config_flow.CONF_EXTRA_ARGUMENTS: "",
|
|
config_flow.CONF_RTSP_TRANSPORT: list(config_flow.RTSP_TRANSPORTS)[1],
|
|
config_flow.CONF_USE_WALLCLOCK_AS_TIMESTAMPS: option_value,
|
|
config_flow.CONF_ENABLE_WEBHOOKS: option_value,
|
|
},
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["data"] == {
|
|
config_flow.CONF_EXTRA_ARGUMENTS: "",
|
|
config_flow.CONF_RTSP_TRANSPORT: list(config_flow.RTSP_TRANSPORTS)[1],
|
|
config_flow.CONF_USE_WALLCLOCK_AS_TIMESTAMPS: option_value,
|
|
config_flow.CONF_ENABLE_WEBHOOKS: option_value,
|
|
}
|
|
|
|
|
|
async def test_discovered_by_dhcp_updates_host(
|
|
hass: HomeAssistant, device_registry: dr.DeviceRegistry
|
|
) -> None:
|
|
"""Test dhcp updates existing host."""
|
|
config_entry, _camera, device = await setup_onvif_integration(hass)
|
|
device.profiles = device.async_get_profiles()
|
|
devices = dr.async_entries_for_config_entry(device_registry, config_entry.entry_id)
|
|
assert len(devices) == 1
|
|
device = devices[0]
|
|
assert device.model == "TestModel"
|
|
assert device.connections == {(dr.CONNECTION_NETWORK_MAC, MAC)}
|
|
assert config_entry.data[CONF_HOST] == "1.2.3.4"
|
|
await hass.config_entries.async_unload(config_entry.entry_id)
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_DHCP}, data=DHCP_DISCOVERY
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "already_configured"
|
|
assert config_entry.data[CONF_HOST] == DHCP_DISCOVERY.ip
|
|
|
|
|
|
async def test_discovered_by_dhcp_does_nothing_if_host_is_the_same(
|
|
hass: HomeAssistant, device_registry: dr.DeviceRegistry
|
|
) -> None:
|
|
"""Test dhcp update does nothing if host is the same."""
|
|
config_entry, _camera, device = await setup_onvif_integration(hass)
|
|
device.profiles = device.async_get_profiles()
|
|
devices = dr.async_entries_for_config_entry(device_registry, config_entry.entry_id)
|
|
assert len(devices) == 1
|
|
device = devices[0]
|
|
assert device.model == "TestModel"
|
|
assert device.connections == {(dr.CONNECTION_NETWORK_MAC, MAC)}
|
|
assert config_entry.data[CONF_HOST] == DHCP_DISCOVERY_SAME_IP.ip
|
|
await hass.config_entries.async_unload(config_entry.entry_id)
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_DHCP}, data=DHCP_DISCOVERY_SAME_IP
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "already_configured"
|
|
assert config_entry.data[CONF_HOST] == DHCP_DISCOVERY_SAME_IP.ip
|
|
|
|
|
|
async def test_discovered_by_dhcp_does_not_update_if_already_loaded(
|
|
hass: HomeAssistant, device_registry: dr.DeviceRegistry
|
|
) -> None:
|
|
"""Test dhcp does not update existing host if its already loaded."""
|
|
config_entry, _camera, device = await setup_onvif_integration(hass)
|
|
device.profiles = device.async_get_profiles()
|
|
devices = dr.async_entries_for_config_entry(device_registry, config_entry.entry_id)
|
|
assert len(devices) == 1
|
|
device = devices[0]
|
|
assert device.model == "TestModel"
|
|
assert device.connections == {(dr.CONNECTION_NETWORK_MAC, MAC)}
|
|
assert config_entry.data[CONF_HOST] == "1.2.3.4"
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_DHCP}, data=DHCP_DISCOVERY
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "already_configured"
|
|
assert config_entry.data[CONF_HOST] != DHCP_DISCOVERY.ip
|
|
|
|
|
|
async def test_discovered_by_dhcp_does_not_update_if_no_matching_entry(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test dhcp does not update existing host if there are no matching entries."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_DHCP}, data=DHCP_DISCOVERY
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "no_devices_found"
|
|
|
|
|
|
def _get_schema_default(schema, key_name):
|
|
"""Iterate schema to find a key."""
|
|
for schema_key in schema:
|
|
if schema_key == key_name:
|
|
return schema_key.default()
|
|
raise KeyError(f"{key_name} not found in schema")
|
|
|
|
|
|
async def test_form_reauth(hass: HomeAssistant) -> None:
|
|
"""Test reauthenticate."""
|
|
entry, _, _ = await setup_onvif_integration(hass)
|
|
|
|
result = await entry.start_reauth_flow(hass)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "reauth_confirm"
|
|
assert (
|
|
_get_schema_default(result["data_schema"].schema, CONF_USERNAME)
|
|
== entry.data[CONF_USERNAME]
|
|
)
|
|
|
|
with (
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.get_device"
|
|
) as mock_onvif_camera,
|
|
patch("homeassistant.components.onvif.ONVIFDevice") as mock_device,
|
|
patch(
|
|
"homeassistant.components.onvif.async_setup_entry",
|
|
return_value=True,
|
|
) as mock_setup_entry,
|
|
):
|
|
setup_mock_onvif_camera(mock_onvif_camera, auth_failure=True)
|
|
setup_mock_device(mock_device)
|
|
|
|
result2 = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{
|
|
config_flow.CONF_USERNAME: "new-test-username",
|
|
config_flow.CONF_PASSWORD: "new-test-password",
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result2["type"] is FlowResultType.FORM
|
|
assert result2["step_id"] == "reauth_confirm"
|
|
assert result2["errors"] == {config_flow.CONF_PASSWORD: "auth_failed"}
|
|
assert result2["description_placeholders"] == {
|
|
CONF_NAME: "Mock Title",
|
|
"error": "not authorized (subcodes:NotAuthorized)",
|
|
}
|
|
|
|
with (
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.get_device"
|
|
) as mock_onvif_camera,
|
|
patch("homeassistant.components.onvif.ONVIFDevice") as mock_device,
|
|
patch(
|
|
"homeassistant.components.onvif.async_setup_entry",
|
|
return_value=True,
|
|
) as mock_setup_entry,
|
|
):
|
|
setup_mock_onvif_camera(mock_onvif_camera)
|
|
setup_mock_device(mock_device)
|
|
|
|
result3 = await hass.config_entries.flow.async_configure(
|
|
result2["flow_id"],
|
|
{
|
|
config_flow.CONF_USERNAME: "new-test-username",
|
|
config_flow.CONF_PASSWORD: "new-test-password",
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result3["type"] is FlowResultType.ABORT
|
|
assert result3["reason"] == "reauth_successful"
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
assert entry.data[config_flow.CONF_USERNAME] == "new-test-username"
|
|
assert entry.data[config_flow.CONF_PASSWORD] == "new-test-password"
|
|
|
|
|
|
async def test_flow_manual_entry_updates_existing_user_password(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test that the existing username and password can be updated via manual entry."""
|
|
entry, _, _ = await setup_onvif_integration(hass)
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
config_flow.DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "user"
|
|
|
|
with (
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.get_device"
|
|
) as mock_onvif_camera,
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.WSDiscovery"
|
|
) as mock_discovery,
|
|
patch("homeassistant.components.onvif.ONVIFDevice") as mock_device,
|
|
):
|
|
setup_mock_onvif_camera(mock_onvif_camera, two_profiles=True)
|
|
# no discovery
|
|
mock_discovery.return_value = []
|
|
setup_mock_device(mock_device)
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
user_input={"auto": False},
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "configure"
|
|
|
|
with patch(
|
|
"homeassistant.components.onvif.async_setup_entry", return_value=True
|
|
) as mock_setup_entry:
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
user_input={
|
|
config_flow.CONF_NAME: NAME,
|
|
config_flow.CONF_HOST: HOST,
|
|
config_flow.CONF_PORT: PORT,
|
|
config_flow.CONF_USERNAME: USERNAME,
|
|
config_flow.CONF_PASSWORD: "new_password",
|
|
},
|
|
)
|
|
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "already_configured"
|
|
assert entry.data[config_flow.CONF_USERNAME] == USERNAME
|
|
assert entry.data[config_flow.CONF_PASSWORD] == "new_password"
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
async def test_flow_manual_entry_wrong_port(hass: HomeAssistant) -> None:
|
|
"""Test that we get a useful error with the wrong port."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
config_flow.DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "user"
|
|
|
|
with (
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.get_device"
|
|
) as mock_onvif_camera,
|
|
patch(
|
|
"homeassistant.components.onvif.config_flow.WSDiscovery"
|
|
) as mock_discovery,
|
|
patch("homeassistant.components.onvif.ONVIFDevice") as mock_device,
|
|
):
|
|
setup_mock_onvif_camera(mock_onvif_camera, wrong_port=True)
|
|
# no discovery
|
|
mock_discovery.return_value = []
|
|
setup_mock_device(mock_device)
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
user_input={"auto": False},
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "configure"
|
|
|
|
with patch(
|
|
"homeassistant.components.onvif.async_setup_entry", return_value=True
|
|
) as mock_setup_entry:
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
user_input={
|
|
config_flow.CONF_NAME: NAME,
|
|
config_flow.CONF_HOST: HOST,
|
|
config_flow.CONF_PORT: PORT,
|
|
config_flow.CONF_USERNAME: USERNAME,
|
|
config_flow.CONF_PASSWORD: PASSWORD,
|
|
},
|
|
)
|
|
|
|
await hass.async_block_till_done()
|
|
assert len(mock_setup_entry.mock_calls) == 0
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "configure"
|
|
assert result["errors"] == {"port": "no_onvif_service"}
|
|
assert result["description_placeholders"] == {}
|
|
setup_mock_onvif_camera(mock_onvif_camera, two_profiles=True)
|
|
|
|
with patch(
|
|
"homeassistant.components.onvif.async_setup_entry", return_value=True
|
|
) as mock_setup_entry:
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
user_input={
|
|
config_flow.CONF_NAME: NAME,
|
|
config_flow.CONF_HOST: HOST,
|
|
config_flow.CONF_PORT: PORT,
|
|
config_flow.CONF_USERNAME: USERNAME,
|
|
config_flow.CONF_PASSWORD: PASSWORD,
|
|
},
|
|
)
|
|
|
|
await hass.async_block_till_done()
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
assert result["title"] == f"{NAME} - {MAC}"
|
|
assert result["data"] == {
|
|
config_flow.CONF_NAME: NAME,
|
|
config_flow.CONF_HOST: HOST,
|
|
config_flow.CONF_PORT: PORT,
|
|
config_flow.CONF_USERNAME: USERNAME,
|
|
config_flow.CONF_PASSWORD: PASSWORD,
|
|
}
|