Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion .coveragerc
Original file line number Diff line number Diff line change
Expand Up @@ -974,7 +974,6 @@ omit =
homeassistant/components/rejseplanen/sensor.py
homeassistant/components/remember_the_milk/__init__.py
homeassistant/components/remote_rpi_gpio/*
homeassistant/components/reolink/__init__.py
homeassistant/components/reolink/binary_sensor.py
homeassistant/components/reolink/camera.py
homeassistant/components/reolink/entity.py
Expand Down
92 changes: 92 additions & 0 deletions tests/components/reolink/conftest.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,92 @@
"""Setup the Reolink tests."""
from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from homeassistant.components.reolink import const
from homeassistant.components.reolink.config_flow import DEFAULT_PROTOCOL
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import format_mac

from tests.common import MockConfigEntry

TEST_HOST = "1.2.3.4"
TEST_HOST2 = "4.5.6.7"
TEST_USERNAME = "admin"
TEST_USERNAME2 = "username"
TEST_PASSWORD = "password"
TEST_PASSWORD2 = "new_password"
TEST_MAC = "ab:cd:ef:gh:ij:kl"
TEST_PORT = 1234
TEST_NVR_NAME = "test_reolink_name"
TEST_USE_HTTPS = True


@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock, None, None]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.reolink.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry


@pytest.fixture
def reolink_connect(mock_get_source_ip: None) -> Generator[MagicMock, None, None]:
"""Mock reolink connection."""
with patch(
"homeassistant.components.reolink.host.webhook.async_register",
return_value=True,
), patch(
"homeassistant.components.reolink.host.Host", autospec=True
) as host_mock_class:
host_mock = host_mock_class.return_value
host_mock.get_host_data.return_value = None
host_mock.get_states.return_value = None
host_mock.check_new_firmware.return_value = False
host_mock.unsubscribe.return_value = True
host_mock.logout.return_value = True
host_mock.mac_address = TEST_MAC
host_mock.onvif_enabled = True
host_mock.rtmp_enabled = True
host_mock.rtsp_enabled = True
host_mock.nvr_name = TEST_NVR_NAME
host_mock.port = TEST_PORT
host_mock.use_https = TEST_USE_HTTPS
host_mock.is_admin = True
host_mock.user_level = "admin"
host_mock.sw_version_update_required = False
host_mock.timeout = 60
host_mock.renewtimer = 600
yield host_mock


@pytest.fixture
def reolink_platforms(mock_get_source_ip: None) -> Generator[None, None, None]:
"""Mock reolink entry setup."""
with patch("homeassistant.components.reolink.PLATFORMS", return_value=[]):
yield


@pytest.fixture
def config_entry(hass: HomeAssistant) -> MockConfigEntry:
"""Add the reolink mock config entry to hass."""
config_entry = MockConfigEntry(
domain=const.DOMAIN,
unique_id=format_mac(TEST_MAC),
data={
CONF_HOST: TEST_HOST,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_PORT: TEST_PORT,
const.CONF_USE_HTTPS: TEST_USE_HTTPS,
},
options={
const.CONF_PROTOCOL: DEFAULT_PROTOCOL,
},
title=TEST_NVR_NAME,
)
config_entry.add_to_hass(hass)
return config_entry
234 changes: 71 additions & 163 deletions tests/components/reolink/test_config_flow.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
"""Test the Reolink config flow."""
import json
from unittest.mock import AsyncMock, Mock, patch
from unittest.mock import MagicMock

import pytest
from reolink_aio.exceptions import ApiError, CredentialsInvalidError, ReolinkError
Expand All @@ -9,61 +9,26 @@
from homeassistant.components import dhcp
from homeassistant.components.reolink import const
from homeassistant.components.reolink.config_flow import DEFAULT_PROTOCOL
from homeassistant.config import async_process_ha_core_config
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers import issue_registry as ir
from homeassistant.helpers.device_registry import format_mac

from .conftest import (
TEST_HOST,
TEST_HOST2,
TEST_MAC,
TEST_NVR_NAME,
TEST_PASSWORD,
TEST_PASSWORD2,
TEST_PORT,
TEST_USE_HTTPS,
TEST_USERNAME,
TEST_USERNAME2,
)

from tests.common import MockConfigEntry

TEST_HOST = "1.2.3.4"
TEST_HOST2 = "4.5.6.7"
TEST_USERNAME = "admin"
TEST_USERNAME2 = "username"
TEST_PASSWORD = "password"
TEST_PASSWORD2 = "new_password"
TEST_MAC = "ab:cd:ef:gh:ij:kl"
TEST_PORT = 1234
TEST_NVR_NAME = "test_reolink_name"
TEST_USE_HTTPS = True


def get_mock_info(error=None, user_level="admin"):
"""Return a mock gateway info instance."""
host_mock = Mock()
if error is None:
host_mock.get_host_data = AsyncMock(return_value=None)
else:
host_mock.get_host_data = AsyncMock(side_effect=error)
host_mock.check_new_firmware = AsyncMock(return_value=False)
host_mock.unsubscribe = AsyncMock(return_value=True)
host_mock.logout = AsyncMock(return_value=True)
host_mock.mac_address = TEST_MAC
host_mock.onvif_enabled = True
host_mock.rtmp_enabled = True
host_mock.rtsp_enabled = True
host_mock.nvr_name = TEST_NVR_NAME
host_mock.port = TEST_PORT
host_mock.use_https = TEST_USE_HTTPS
host_mock.is_admin = user_level == "admin"
host_mock.user_level = user_level
host_mock.timeout = 60
host_mock.renewtimer = 600
host_mock.get_states = AsyncMock(return_value=None)
return host_mock


@pytest.fixture(name="reolink_connect", autouse=True)
def reolink_connect_fixture(mock_get_source_ip):
"""Mock reolink connection and entry setup."""
with patch(
"homeassistant.components.reolink.host.webhook.async_register",
return_value=True,
), patch("homeassistant.components.reolink.PLATFORMS", return_value=[]), patch(
"homeassistant.components.reolink.host.Host", return_value=get_mock_info()
):
yield
pytestmark = pytest.mark.usefixtures("mock_setup_entry", "reolink_connect")


async def test_config_flow_manual_success(hass: HomeAssistant) -> None:
Expand Down Expand Up @@ -99,7 +64,9 @@ async def test_config_flow_manual_success(hass: HomeAssistant) -> None:
}


async def test_config_flow_errors(hass: HomeAssistant) -> None:
async def test_config_flow_errors(
hass: HomeAssistant, reolink_connect: MagicMock
) -> None:
"""Successful flow manually initialized by the user after some errors."""
result = await hass.config_entries.flow.async_init(
const.DOMAIN, context={"source": config_entries.SOURCE_USER}
Expand All @@ -109,81 +76,82 @@ async def test_config_flow_errors(hass: HomeAssistant) -> None:
assert result["step_id"] == "user"
assert result["errors"] == {}

host_mock = get_mock_info(error=ReolinkError("Test error"))
with patch("homeassistant.components.reolink.host.Host", return_value=host_mock):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_HOST: TEST_HOST,
},
)
reolink_connect.is_admin = False
reolink_connect.user_level = "guest"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_HOST: TEST_HOST,
},
)

assert result["type"] is data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {CONF_HOST: "cannot_connect"}
assert result["errors"] == {CONF_USERNAME: "not_admin"}

host_mock = get_mock_info(user_level="guest")
with patch("homeassistant.components.reolink.host.Host", return_value=host_mock):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_HOST: TEST_HOST,
},
)
reolink_connect.is_admin = True
reolink_connect.user_level = "admin"
reolink_connect.get_host_data.side_effect = ReolinkError("Test error")
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_HOST: TEST_HOST,
},
)

assert result["type"] is data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {CONF_USERNAME: "not_admin"}
assert result["errors"] == {CONF_HOST: "cannot_connect"}

host_mock = get_mock_info(error=json.JSONDecodeError("test_error", "test", 1))
with patch("homeassistant.components.reolink.host.Host", return_value=host_mock):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_HOST: TEST_HOST,
},
)
reolink_connect.get_host_data.side_effect = json.JSONDecodeError(
"test_error", "test", 1
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_HOST: TEST_HOST,
},
)

assert result["type"] is data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {CONF_HOST: "unknown"}

host_mock = get_mock_info(error=CredentialsInvalidError("Test error"))
with patch("homeassistant.components.reolink.host.Host", return_value=host_mock):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_HOST: TEST_HOST,
},
)
reolink_connect.get_host_data.side_effect = CredentialsInvalidError("Test error")
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_HOST: TEST_HOST,
},
)

assert result["type"] is data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {CONF_HOST: "invalid_auth"}

host_mock = get_mock_info(error=ApiError("Test error"))
with patch("homeassistant.components.reolink.host.Host", return_value=host_mock):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_HOST: TEST_HOST,
},
)
reolink_connect.get_host_data.side_effect = ApiError("Test error")
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_HOST: TEST_HOST,
},
)

assert result["type"] is data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {CONF_HOST: "api_error"}

reolink_connect.get_host_data.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
Expand Down Expand Up @@ -422,63 +390,3 @@ async def test_dhcp_abort_flow(hass: HomeAssistant) -> None:

assert result["type"] is data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "already_configured"


async def test_http_no_repair_issue(hass: HomeAssistant) -> None:
"""Test no repairs issue is raised when http local url is used."""
config_entry = MockConfigEntry(
domain=const.DOMAIN,
unique_id=format_mac(TEST_MAC),
data={
CONF_HOST: TEST_HOST,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_PORT: TEST_PORT,
const.CONF_USE_HTTPS: TEST_USE_HTTPS,
},
options={
const.CONF_PROTOCOL: DEFAULT_PROTOCOL,
},
title=TEST_NVR_NAME,
)
config_entry.add_to_hass(hass)

await async_process_ha_core_config(
hass, {"country": "GB", "internal_url": "http://test_homeassistant_address"}
)

assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()

issue_registry = ir.async_get(hass)
assert len(issue_registry.issues) == 0


async def test_https_repair_issue(hass: HomeAssistant) -> None:
"""Test repairs issue is raised when https local url is used."""
config_entry = MockConfigEntry(
domain=const.DOMAIN,
unique_id=format_mac(TEST_MAC),
data={
CONF_HOST: TEST_HOST,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_PORT: TEST_PORT,
const.CONF_USE_HTTPS: TEST_USE_HTTPS,
},
options={
const.CONF_PROTOCOL: DEFAULT_PROTOCOL,
},
title=TEST_NVR_NAME,
)
config_entry.add_to_hass(hass)

await async_process_ha_core_config(
hass, {"country": "GB", "internal_url": "https://test_homeassistant_address"}
)

assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()

issue_registry = ir.async_get(hass)
assert len(issue_registry.issues) == 1
Loading