mirror of
https://github.com/home-assistant/core.git
synced 2025-04-23 16:57:53 +00:00
Refactor config flow tests in generic camera (#134385)
Co-authored-by: Dave T <17680170+davet2001@users.noreply.github.com> Co-authored-by: Allen Porter <allen.porter@gmail.com>
This commit is contained in:
Dave T
GitHub
parent
6fd9476bb9
commit
9b55faa879
@ -4,7 +4,7 @@ from __future__ import annotations
|
||||
|
||||
from collections.abc import Generator
|
||||
from io import BytesIO
|
||||
from unittest.mock import AsyncMock, MagicMock, Mock, _patch, patch
|
||||
from unittest.mock import AsyncMock, MagicMock, Mock, patch
|
||||
|
||||
from PIL import Image
|
||||
import pytest
|
||||
@ -54,11 +54,15 @@ def fakeimgbytes_gif() -> bytes:
|
||||
@pytest.fixture
|
||||
def fakeimg_png(fakeimgbytes_png: bytes) -> Generator[None]:
|
||||
"""Set up respx to respond to test url with fake image bytes."""
|
||||
respx.get("http://127.0.0.1/testurl/1", name="fake_img").respond(
|
||||
respx.get("http://127.0.0.1/testurl/1", name="fake_img1").respond(
|
||||
stream=fakeimgbytes_png
|
||||
)
|
||||
respx.get("http://127.0.0.1/testurl/2", name="fake_img2").respond(
|
||||
stream=fakeimgbytes_png
|
||||
)
|
||||
yield
|
||||
respx.pop("fake_img")
|
||||
respx.pop("fake_img1")
|
||||
respx.pop("fake_img2")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@ -71,8 +75,8 @@ def fakeimg_gif(fakeimgbytes_gif: bytes) -> Generator[None]:
|
||||
respx.pop("fake_img")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_create_stream(hass: HomeAssistant) -> _patch[MagicMock]:
|
||||
@pytest.fixture(name="mock_create_stream")
|
||||
def mock_create_stream(hass: HomeAssistant) -> Generator[AsyncMock]:
|
||||
"""Mock create stream."""
|
||||
mock_stream = MagicMock()
|
||||
mock_stream.hass = hass
|
||||
@ -83,14 +87,25 @@ def mock_create_stream(hass: HomeAssistant) -> _patch[MagicMock]:
|
||||
mock_stream.start = AsyncMock()
|
||||
mock_stream.stop = AsyncMock()
|
||||
mock_stream.endpoint_url.return_value = "http://127.0.0.1/nothing"
|
||||
return patch(
|
||||
with patch(
|
||||
"homeassistant.components.generic.config_flow.create_stream",
|
||||
return_value=mock_stream,
|
||||
)
|
||||
) as mock_create_stream:
|
||||
yield mock_create_stream
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def user_flow(hass: HomeAssistant) -> ConfigFlowResult:
|
||||
def mock_setup_entry() -> Generator[AsyncMock]:
|
||||
"""Mock setup entry."""
|
||||
with patch(
|
||||
"homeassistant.components.generic.async_setup_entry",
|
||||
return_value=True,
|
||||
) as mock_setup_entry:
|
||||
yield mock_setup_entry
|
||||
|
||||
|
||||
@pytest.fixture(name="user_flow")
|
||||
async def user_flow_fixture(hass: HomeAssistant) -> ConfigFlowResult:
|
||||
"""Initiate a user flow."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
@ -126,8 +141,8 @@ def config_entry_fixture(hass: HomeAssistant) -> MockConfigEntry:
|
||||
return entry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def setup_entry(
|
||||
@pytest.fixture(name="setup_entry")
|
||||
async def setup_entry_fixture(
|
||||
hass: HomeAssistant, config_entry: MockConfigEntry
|
||||
) -> MockConfigEntry:
|
||||
"""Set up a config entry ready to be used in tests."""
|
||||
|
||||
@ -2,6 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Generator
|
||||
import contextlib
|
||||
import errno
|
||||
from http import HTTPStatus
|
||||
@ -9,12 +10,10 @@ import os.path
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock, MagicMock, PropertyMock, _patch, patch
|
||||
|
||||
from freezegun.api import FrozenDateTimeFactory
|
||||
import httpx
|
||||
import pytest
|
||||
import respx
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components.camera import async_get_image
|
||||
from homeassistant.components.generic.config_flow import slug
|
||||
from homeassistant.components.generic.const import (
|
||||
@ -30,7 +29,7 @@ from homeassistant.components.stream import (
|
||||
CONF_RTSP_TRANSPORT,
|
||||
CONF_USE_WALLCLOCK_AS_TIMESTAMPS,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntryState, ConfigFlowResult
|
||||
from homeassistant.config_entries import ConfigFlowResult
|
||||
from homeassistant.const import (
|
||||
CONF_AUTHENTICATION,
|
||||
CONF_NAME,
|
||||
@ -44,7 +43,7 @@ from homeassistant.data_entry_flow import FlowResultType
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
|
||||
from tests.common import MockConfigEntry, async_fire_time_changed
|
||||
from tests.common import MockConfigEntry
|
||||
from tests.typing import ClientSessionGenerator, WebSocketGenerator
|
||||
|
||||
TESTDATA = {
|
||||
@ -69,52 +68,47 @@ TESTDATA_YAML = {
|
||||
|
||||
|
||||
@respx.mock
|
||||
@pytest.mark.usefixtures("fakeimg_png")
|
||||
async def test_form(
|
||||
hass: HomeAssistant,
|
||||
fakeimgbytes_png: bytes,
|
||||
hass_client: ClientSessionGenerator,
|
||||
user_flow: ConfigFlowResult,
|
||||
mock_create_stream: _patch[MagicMock],
|
||||
mock_setup_entry: _patch[MagicMock],
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
) -> None:
|
||||
"""Test the form with a normal set of settings."""
|
||||
|
||||
respx.get("http://127.0.0.1/testurl/1").respond(stream=fakeimgbytes_png)
|
||||
with (
|
||||
mock_create_stream as mock_setup,
|
||||
patch(
|
||||
"homeassistant.components.generic.async_setup_entry", return_value=True
|
||||
) as mock_setup_entry,
|
||||
):
|
||||
result1 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
TESTDATA,
|
||||
)
|
||||
assert result1["type"] is FlowResultType.FORM
|
||||
assert result1["step_id"] == "user_confirm"
|
||||
result1 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
TESTDATA,
|
||||
)
|
||||
assert result1["type"] is FlowResultType.FORM
|
||||
assert result1["step_id"] == "user_confirm"
|
||||
|
||||
# HA should now be serving a WS connection for a preview stream.
|
||||
ws_client = await hass_ws_client()
|
||||
flow_id = user_flow["flow_id"]
|
||||
await ws_client.send_json_auto_id(
|
||||
{
|
||||
"type": "generic_camera/start_preview",
|
||||
"flow_id": flow_id,
|
||||
},
|
||||
)
|
||||
json = await ws_client.receive_json()
|
||||
# HA should now be serving a WS connection for a preview stream.
|
||||
ws_client = await hass_ws_client()
|
||||
flow_id = user_flow["flow_id"]
|
||||
await ws_client.send_json_auto_id(
|
||||
{
|
||||
"type": "generic_camera/start_preview",
|
||||
"flow_id": flow_id,
|
||||
},
|
||||
)
|
||||
json = await ws_client.receive_json()
|
||||
|
||||
client = await hass_client()
|
||||
still_preview_url = json["event"]["attributes"]["still_url"]
|
||||
# Check the preview image works.
|
||||
resp = await client.get(still_preview_url)
|
||||
assert resp.status == HTTPStatus.OK
|
||||
assert await resp.read() == fakeimgbytes_png
|
||||
client = await hass_client()
|
||||
still_preview_url = json["event"]["attributes"]["still_url"]
|
||||
# Check the preview image works.
|
||||
resp = await client.get(still_preview_url)
|
||||
assert resp.status == HTTPStatus.OK
|
||||
assert await resp.read() == fakeimgbytes_png
|
||||
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result1["flow_id"],
|
||||
user_input={CONF_CONFIRMED_OK: True},
|
||||
)
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result1["flow_id"],
|
||||
user_input={CONF_CONFIRMED_OK: True},
|
||||
)
|
||||
assert result2["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result2["title"] == "127_0_0_1"
|
||||
assert result2["options"] == {
|
||||
@ -131,36 +125,29 @@ async def test_form(
|
||||
# Check that the preview image is disabled after.
|
||||
resp = await client.get(still_preview_url)
|
||||
assert resp.status == HTTPStatus.NOT_FOUND
|
||||
assert len(mock_setup.mock_calls) == 1
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
@respx.mock
|
||||
@pytest.mark.usefixtures("fakeimg_png")
|
||||
async def test_form_only_stillimage(
|
||||
hass: HomeAssistant, user_flow: ConfigFlowResult
|
||||
hass: HomeAssistant,
|
||||
user_flow: ConfigFlowResult,
|
||||
mock_setup_entry: _patch[MagicMock],
|
||||
) -> None:
|
||||
"""Test we complete ok if the user wants still images only."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {}
|
||||
|
||||
data = TESTDATA.copy()
|
||||
data.pop(CONF_STREAM_SOURCE)
|
||||
with patch("homeassistant.components.generic.async_setup_entry", return_value=True):
|
||||
result1 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
data,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert result1["type"] is FlowResultType.FORM
|
||||
assert result1["step_id"] == "user_confirm"
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result1["flow_id"],
|
||||
user_input={CONF_CONFIRMED_OK: True},
|
||||
)
|
||||
result1 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
data,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert result1["type"] is FlowResultType.FORM
|
||||
assert result1["step_id"] == "user_confirm"
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result1["flow_id"],
|
||||
user_input={CONF_CONFIRMED_OK: True},
|
||||
)
|
||||
assert result2["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result2["title"] == "127_0_0_1"
|
||||
assert result2["options"] == {
|
||||
@ -177,19 +164,17 @@ async def test_form_only_stillimage(
|
||||
|
||||
|
||||
@respx.mock
|
||||
@pytest.mark.usefixtures("fakeimg_png")
|
||||
async def test_form_reject_preview(
|
||||
hass: HomeAssistant,
|
||||
fakeimgbytes_png: bytes,
|
||||
mock_create_stream: _patch[MagicMock],
|
||||
user_flow: ConfigFlowResult,
|
||||
) -> None:
|
||||
"""Test we go back to the config screen if the user rejects the preview."""
|
||||
respx.get("http://127.0.0.1/testurl/1").respond(stream=fakeimgbytes_png)
|
||||
with mock_create_stream:
|
||||
result1 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
TESTDATA,
|
||||
)
|
||||
result1 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
TESTDATA,
|
||||
)
|
||||
assert result1["type"] is FlowResultType.FORM
|
||||
assert result1["step_id"] == "user_confirm"
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
@ -215,7 +200,6 @@ async def test_form_still_preview_cam_off(
|
||||
"homeassistant.components.generic.camera.GenericCamera.is_on",
|
||||
new_callable=PropertyMock(return_value=False),
|
||||
),
|
||||
mock_create_stream,
|
||||
):
|
||||
result1 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
@ -246,47 +230,50 @@ async def test_form_still_preview_cam_off(
|
||||
@respx.mock
|
||||
@pytest.mark.usefixtures("fakeimg_gif")
|
||||
async def test_form_only_stillimage_gif(
|
||||
hass: HomeAssistant, user_flow: ConfigFlowResult
|
||||
hass: HomeAssistant,
|
||||
user_flow: ConfigFlowResult,
|
||||
mock_setup_entry: _patch[MagicMock],
|
||||
) -> None:
|
||||
"""Test we complete ok if the user wants a gif."""
|
||||
data = TESTDATA.copy()
|
||||
data.pop(CONF_STREAM_SOURCE)
|
||||
with patch("homeassistant.components.generic.async_setup_entry", return_value=True):
|
||||
result1 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
data,
|
||||
)
|
||||
assert result1["type"] is FlowResultType.FORM
|
||||
assert result1["step_id"] == "user_confirm"
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result1["flow_id"],
|
||||
user_input={CONF_CONFIRMED_OK: True},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
result1 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
data,
|
||||
)
|
||||
assert result1["type"] is FlowResultType.FORM
|
||||
assert result1["step_id"] == "user_confirm"
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result1["flow_id"],
|
||||
user_input={CONF_CONFIRMED_OK: True},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert result2["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result2["options"][CONF_CONTENT_TYPE] == "image/gif"
|
||||
|
||||
|
||||
@respx.mock
|
||||
async def test_form_only_svg_whitespace(
|
||||
hass: HomeAssistant, fakeimgbytes_svg: bytes, user_flow: ConfigFlowResult
|
||||
hass: HomeAssistant,
|
||||
fakeimgbytes_svg: bytes,
|
||||
user_flow: ConfigFlowResult,
|
||||
mock_setup_entry: _patch[MagicMock],
|
||||
) -> None:
|
||||
"""Test we complete ok if svg starts with whitespace, issue #68889."""
|
||||
fakeimgbytes_wspace_svg = bytes(" \n ", encoding="utf-8") + fakeimgbytes_svg
|
||||
respx.get("http://127.0.0.1/testurl/1").respond(stream=fakeimgbytes_wspace_svg)
|
||||
data = TESTDATA.copy()
|
||||
data.pop(CONF_STREAM_SOURCE)
|
||||
with patch("homeassistant.components.generic.async_setup_entry", return_value=True):
|
||||
result1 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
data,
|
||||
)
|
||||
assert result1["type"] is FlowResultType.FORM
|
||||
assert result1["step_id"] == "user_confirm"
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result1["flow_id"],
|
||||
user_input={CONF_CONFIRMED_OK: True},
|
||||
)
|
||||
result1 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
data,
|
||||
)
|
||||
assert result1["type"] is FlowResultType.FORM
|
||||
assert result1["step_id"] == "user_confirm"
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result1["flow_id"],
|
||||
user_input={CONF_CONFIRMED_OK: True},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert result2["type"] is FlowResultType.CREATE_ENTRY
|
||||
|
||||
@ -303,7 +290,7 @@ async def test_form_only_svg_whitespace(
|
||||
],
|
||||
)
|
||||
async def test_form_only_still_sample(
|
||||
hass: HomeAssistant, user_flow: ConfigFlowResult, image_file
|
||||
hass: HomeAssistant, user_flow: ConfigFlowResult, image_file, mock_setup_entry
|
||||
) -> None:
|
||||
"""Test various sample images #69037."""
|
||||
image_path = os.path.join(os.path.dirname(__file__), image_file)
|
||||
@ -311,18 +298,17 @@ async def test_form_only_still_sample(
|
||||
respx.get("http://127.0.0.1/testurl/1").respond(stream=image_bytes)
|
||||
data = TESTDATA.copy()
|
||||
data.pop(CONF_STREAM_SOURCE)
|
||||
with patch("homeassistant.components.generic.async_setup_entry", return_value=True):
|
||||
result1 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
data,
|
||||
)
|
||||
assert result1["type"] is FlowResultType.FORM
|
||||
assert result1["step_id"] == "user_confirm"
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result1["flow_id"],
|
||||
user_input={CONF_CONFIRMED_OK: True},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
result1 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
data,
|
||||
)
|
||||
assert result1["type"] is FlowResultType.FORM
|
||||
assert result1["step_id"] == "user_confirm"
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result1["flow_id"],
|
||||
user_input={CONF_CONFIRMED_OK: True},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert result2["type"] is FlowResultType.CREATE_ENTRY
|
||||
|
||||
|
||||
@ -363,10 +349,11 @@ async def test_form_only_still_sample(
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_still_template(
|
||||
async def test_form_still_template(
|
||||
hass: HomeAssistant,
|
||||
user_flow: ConfigFlowResult,
|
||||
fakeimgbytes_png: bytes,
|
||||
mock_setup_entry: Generator[AsyncMock],
|
||||
template,
|
||||
url,
|
||||
expected_result,
|
||||
@ -380,12 +367,11 @@ async def test_still_template(
|
||||
data = TESTDATA.copy()
|
||||
data.pop(CONF_STREAM_SOURCE)
|
||||
data[CONF_STILL_IMAGE_URL] = template
|
||||
with patch("homeassistant.components.generic.async_setup_entry", return_value=True):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
data,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
data,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert result2["step_id"] == expected_result
|
||||
assert result2.get("errors") == expected_errors
|
||||
|
||||
@ -396,24 +382,19 @@ async def test_form_rtsp_mode(
|
||||
hass: HomeAssistant,
|
||||
user_flow: ConfigFlowResult,
|
||||
mock_create_stream: _patch[MagicMock],
|
||||
mock_setup_entry: _patch[MagicMock],
|
||||
) -> None:
|
||||
"""Test we complete ok if the user enters a stream url."""
|
||||
data = TESTDATA.copy()
|
||||
data[CONF_RTSP_TRANSPORT] = "tcp"
|
||||
data[CONF_STREAM_SOURCE] = "rtsp://127.0.0.1/testurl/2"
|
||||
with (
|
||||
mock_create_stream as mock_setup,
|
||||
patch("homeassistant.components.generic.async_setup_entry", return_value=True),
|
||||
):
|
||||
result1 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"], data
|
||||
)
|
||||
assert result1["type"] is FlowResultType.FORM
|
||||
assert result1["step_id"] == "user_confirm"
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result1["flow_id"],
|
||||
user_input={CONF_CONFIRMED_OK: True},
|
||||
)
|
||||
result1 = await hass.config_entries.flow.async_configure(user_flow["flow_id"], data)
|
||||
assert result1["type"] is FlowResultType.FORM
|
||||
assert result1["step_id"] == "user_confirm"
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result1["flow_id"],
|
||||
user_input={CONF_CONFIRMED_OK: True},
|
||||
)
|
||||
assert result2["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result2["title"] == "127_0_0_1"
|
||||
assert result2["options"] == {
|
||||
@ -428,8 +409,6 @@ async def test_form_rtsp_mode(
|
||||
CONF_VERIFY_SSL: False,
|
||||
}
|
||||
|
||||
assert len(mock_setup.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_form_only_stream(
|
||||
hass: HomeAssistant,
|
||||
@ -441,18 +420,16 @@ async def test_form_only_stream(
|
||||
data = TESTDATA.copy()
|
||||
data.pop(CONF_STILL_IMAGE_URL)
|
||||
data[CONF_STREAM_SOURCE] = "rtsp://user:pass@127.0.0.1/testurl/2"
|
||||
with mock_create_stream:
|
||||
result1 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
data,
|
||||
)
|
||||
result1 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
data,
|
||||
)
|
||||
|
||||
assert result1["type"] is FlowResultType.FORM
|
||||
with mock_create_stream:
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result1["flow_id"],
|
||||
user_input={CONF_CONFIRMED_OK: True},
|
||||
)
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result1["flow_id"],
|
||||
user_input={CONF_CONFIRMED_OK: True},
|
||||
)
|
||||
|
||||
assert result2["title"] == "127_0_0_1"
|
||||
assert result2["options"] == {
|
||||
@ -528,15 +505,11 @@ async def test_form_image_http_exceptions(
|
||||
mock_create_stream: _patch[MagicMock],
|
||||
) -> None:
|
||||
"""Test we handle image http exceptions."""
|
||||
respx.get("http://127.0.0.1/testurl/1").side_effect = [
|
||||
side_effect,
|
||||
]
|
||||
|
||||
with mock_create_stream:
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
TESTDATA,
|
||||
)
|
||||
respx.get("http://127.0.0.1/testurl/1").side_effect = [side_effect]
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
TESTDATA,
|
||||
)
|
||||
|
||||
assert result2["type"] is FlowResultType.FORM
|
||||
assert result2["errors"] == expected_message
|
||||
@ -550,11 +523,10 @@ async def test_form_stream_invalidimage(
|
||||
) -> None:
|
||||
"""Test we handle invalid image when a stream is specified."""
|
||||
respx.get("http://127.0.0.1/testurl/1").respond(stream=b"invalid")
|
||||
with mock_create_stream:
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
TESTDATA,
|
||||
)
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
TESTDATA,
|
||||
)
|
||||
|
||||
assert result2["type"] is FlowResultType.FORM
|
||||
assert result2["errors"] == {"still_image_url": "invalid_still_image"}
|
||||
@ -568,11 +540,10 @@ async def test_form_stream_invalidimage2(
|
||||
) -> None:
|
||||
"""Test we handle invalid image when a stream is specified."""
|
||||
respx.get("http://127.0.0.1/testurl/1").respond(content=None)
|
||||
with mock_create_stream:
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
TESTDATA,
|
||||
)
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
TESTDATA,
|
||||
)
|
||||
|
||||
assert result2["type"] is FlowResultType.FORM
|
||||
assert result2["errors"] == {"still_image_url": "unable_still_load_no_image"}
|
||||
@ -586,11 +557,10 @@ async def test_form_stream_invalidimage3(
|
||||
) -> None:
|
||||
"""Test we handle invalid image when a stream is specified."""
|
||||
respx.get("http://127.0.0.1/testurl/1").respond(content=bytes([0xFF]))
|
||||
with mock_create_stream:
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
TESTDATA,
|
||||
)
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
TESTDATA,
|
||||
)
|
||||
|
||||
assert result2["type"] is FlowResultType.FORM
|
||||
assert result2["errors"] == {"still_image_url": "invalid_still_image"}
|
||||
@ -599,23 +569,22 @@ async def test_form_stream_invalidimage3(
|
||||
@respx.mock
|
||||
@pytest.mark.usefixtures("fakeimg_png")
|
||||
async def test_form_stream_timeout(
|
||||
hass: HomeAssistant, user_flow: ConfigFlowResult
|
||||
hass: HomeAssistant,
|
||||
user_flow: ConfigFlowResult,
|
||||
mock_create_stream: _patch[MagicMock],
|
||||
) -> None:
|
||||
"""Test we handle invalid auth."""
|
||||
with patch(
|
||||
"homeassistant.components.generic.config_flow.create_stream"
|
||||
) as create_stream:
|
||||
create_stream.return_value.start = AsyncMock()
|
||||
create_stream.return_value.stop = AsyncMock()
|
||||
create_stream.return_value.hass = hass
|
||||
create_stream.return_value.add_provider.return_value.part_recv = AsyncMock()
|
||||
create_stream.return_value.add_provider.return_value.part_recv.return_value = (
|
||||
False
|
||||
)
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
TESTDATA,
|
||||
)
|
||||
mock_create_stream.return_value.start = AsyncMock()
|
||||
mock_create_stream.return_value.stop = AsyncMock()
|
||||
mock_create_stream.return_value.hass = hass
|
||||
mock_create_stream.return_value.add_provider.return_value.part_recv = AsyncMock()
|
||||
mock_create_stream.return_value.add_provider.return_value.part_recv.return_value = (
|
||||
False
|
||||
)
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
TESTDATA,
|
||||
)
|
||||
assert result2["type"] is FlowResultType.FORM
|
||||
assert result2["errors"] == {"stream_source": "timeout"}
|
||||
|
||||
@ -661,11 +630,11 @@ async def test_form_stream_other_error(hass: HomeAssistant, user_flow) -> None:
|
||||
|
||||
|
||||
@respx.mock
|
||||
@pytest.mark.usefixtures("fakeimg_png")
|
||||
async def test_form_stream_permission_error(
|
||||
hass: HomeAssistant, fakeimgbytes_png: bytes, user_flow: ConfigFlowResult
|
||||
hass: HomeAssistant, user_flow: ConfigFlowResult
|
||||
) -> None:
|
||||
"""Test we handle permission error."""
|
||||
respx.get("http://127.0.0.1/testurl/1").respond(stream=fakeimgbytes_png)
|
||||
with patch(
|
||||
"homeassistant.components.generic.config_flow.create_stream",
|
||||
side_effect=PermissionError(),
|
||||
@ -732,116 +701,73 @@ async def test_form_oserror(hass: HomeAssistant, user_flow: ConfigFlowResult) ->
|
||||
|
||||
|
||||
@respx.mock
|
||||
async def test_form_stream_preview_auto_timeout(
|
||||
hass: HomeAssistant,
|
||||
user_flow: ConfigFlowResult,
|
||||
mock_create_stream: _patch[MagicMock],
|
||||
freezer: FrozenDateTimeFactory,
|
||||
fakeimgbytes_png: bytes,
|
||||
) -> None:
|
||||
"""Test that the stream preview times out after 10mins."""
|
||||
respx.get("http://fred_flintstone:bambam@127.0.0.1/testurl/2").respond(
|
||||
stream=fakeimgbytes_png
|
||||
)
|
||||
data = TESTDATA.copy()
|
||||
data.pop(CONF_STILL_IMAGE_URL)
|
||||
|
||||
with mock_create_stream as mock_stream:
|
||||
result1 = await hass.config_entries.flow.async_configure(
|
||||
user_flow["flow_id"],
|
||||
data,
|
||||
)
|
||||
assert result1["type"] is FlowResultType.FORM
|
||||
assert result1["step_id"] == "user_confirm"
|
||||
|
||||
freezer.tick(600 + 12)
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
mock_str = mock_stream.return_value
|
||||
mock_str.start.assert_awaited_once()
|
||||
|
||||
|
||||
@respx.mock
|
||||
@pytest.mark.usefixtures("fakeimg_png")
|
||||
async def test_options_template_error(
|
||||
hass: HomeAssistant, fakeimgbytes_png: bytes, mock_create_stream: _patch[MagicMock]
|
||||
hass: HomeAssistant,
|
||||
mock_create_stream: _patch[MagicMock],
|
||||
config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Test the options flow with a template error."""
|
||||
respx.get("http://127.0.0.1/testurl/1").respond(stream=fakeimgbytes_png)
|
||||
respx.get("http://127.0.0.1/testurl/2").respond(stream=fakeimgbytes_png)
|
||||
|
||||
mock_entry = MockConfigEntry(
|
||||
title="Test Camera",
|
||||
domain=DOMAIN,
|
||||
data={},
|
||||
options=TESTDATA,
|
||||
)
|
||||
|
||||
mock_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
result = await hass.config_entries.options.async_init(mock_entry.entry_id)
|
||||
result = await hass.config_entries.options.async_init(config_entry.entry_id)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "init"
|
||||
|
||||
# try updating the still image url
|
||||
data = TESTDATA.copy()
|
||||
data[CONF_STILL_IMAGE_URL] = "http://127.0.0.1/testurl/2"
|
||||
with mock_create_stream:
|
||||
result2 = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"],
|
||||
user_input=data,
|
||||
)
|
||||
assert result2["type"] is FlowResultType.FORM
|
||||
assert result2["step_id"] == "user_confirm"
|
||||
result2 = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"],
|
||||
user_input=data,
|
||||
)
|
||||
assert result2["type"] is FlowResultType.FORM
|
||||
assert result2["step_id"] == "user_confirm"
|
||||
|
||||
result2a = await hass.config_entries.options.async_configure(
|
||||
result2["flow_id"], user_input={CONF_CONFIRMED_OK: True}
|
||||
)
|
||||
assert result2a["type"] is FlowResultType.CREATE_ENTRY
|
||||
result2a = await hass.config_entries.options.async_configure(
|
||||
result2["flow_id"], user_input={CONF_CONFIRMED_OK: True}
|
||||
)
|
||||
assert result2a["type"] is FlowResultType.CREATE_ENTRY
|
||||
|
||||
result3 = await hass.config_entries.options.async_init(mock_entry.entry_id)
|
||||
assert result3["type"] is FlowResultType.FORM
|
||||
assert result3["step_id"] == "init"
|
||||
result3 = await hass.config_entries.options.async_init(config_entry.entry_id)
|
||||
assert result3["type"] is FlowResultType.FORM
|
||||
assert result3["step_id"] == "init"
|
||||
|
||||
# verify that an invalid template reports the correct UI error.
|
||||
data[CONF_STILL_IMAGE_URL] = "http://127.0.0.1/testurl/{{1/0}}"
|
||||
result4 = await hass.config_entries.options.async_configure(
|
||||
result3["flow_id"],
|
||||
user_input=data,
|
||||
)
|
||||
assert result4.get("type") is FlowResultType.FORM
|
||||
assert result4["errors"] == {"still_image_url": "template_error"}
|
||||
# verify that an invalid template reports the correct UI error.
|
||||
data[CONF_STILL_IMAGE_URL] = "http://127.0.0.1/testurl/{{1/0}}"
|
||||
result4 = await hass.config_entries.options.async_configure(
|
||||
result3["flow_id"],
|
||||
user_input=data,
|
||||
)
|
||||
assert result4.get("type") is FlowResultType.FORM
|
||||
assert result4["errors"] == {"still_image_url": "template_error"}
|
||||
|
||||
# verify that an invalid template reports the correct UI error.
|
||||
data[CONF_STILL_IMAGE_URL] = "http://127.0.0.1/testurl/1"
|
||||
data[CONF_STREAM_SOURCE] = "http://127.0.0.2/testurl/{{1/0}}"
|
||||
result5 = await hass.config_entries.options.async_configure(
|
||||
result4["flow_id"],
|
||||
user_input=data,
|
||||
)
|
||||
# verify that an invalid template reports the correct UI error.
|
||||
data[CONF_STILL_IMAGE_URL] = "http://127.0.0.1/testurl/1"
|
||||
data[CONF_STREAM_SOURCE] = "http://127.0.0.2/testurl/{{1/0}}"
|
||||
result5 = await hass.config_entries.options.async_configure(
|
||||
result4["flow_id"],
|
||||
user_input=data,
|
||||
)
|
||||
|
||||
assert result5.get("type") is FlowResultType.FORM
|
||||
assert result5["errors"] == {"stream_source": "template_error"}
|
||||
assert result5.get("type") is FlowResultType.FORM
|
||||
assert result5["errors"] == {"stream_source": "template_error"}
|
||||
|
||||
# verify that an relative stream url is rejected.
|
||||
data[CONF_STILL_IMAGE_URL] = "http://127.0.0.1/testurl/1"
|
||||
data[CONF_STREAM_SOURCE] = "relative/stream.mjpeg"
|
||||
result6 = await hass.config_entries.options.async_configure(
|
||||
result5["flow_id"],
|
||||
user_input=data,
|
||||
)
|
||||
assert result6.get("type") is FlowResultType.FORM
|
||||
assert result6["errors"] == {"stream_source": "relative_url"}
|
||||
# verify that an relative stream url is rejected.
|
||||
data[CONF_STILL_IMAGE_URL] = "http://127.0.0.1/testurl/1"
|
||||
data[CONF_STREAM_SOURCE] = "relative/stream.mjpeg"
|
||||
result6 = await hass.config_entries.options.async_configure(
|
||||
result5["flow_id"],
|
||||
user_input=data,
|
||||
)
|
||||
assert result6.get("type") is FlowResultType.FORM
|
||||
assert result6["errors"] == {"stream_source": "relative_url"}
|
||||
|
||||
# verify that an malformed stream url is rejected.
|
||||
data[CONF_STILL_IMAGE_URL] = "http://127.0.0.1/testurl/1"
|
||||
data[CONF_STREAM_SOURCE] = "http://example.com:45:56"
|
||||
result7 = await hass.config_entries.options.async_configure(
|
||||
result6["flow_id"],
|
||||
user_input=data,
|
||||
)
|
||||
# verify that an malformed stream url is rejected.
|
||||
data[CONF_STILL_IMAGE_URL] = "http://127.0.0.1/testurl/1"
|
||||
data[CONF_STREAM_SOURCE] = "http://example.com:45:56"
|
||||
result7 = await hass.config_entries.options.async_configure(
|
||||
result6["flow_id"],
|
||||
user_input=data,
|
||||
)
|
||||
assert result7.get("type") is FlowResultType.FORM
|
||||
assert result7["errors"] == {"stream_source": "malformed_url"}
|
||||
|
||||
@ -861,11 +787,13 @@ async def test_slug(hass: HomeAssistant, caplog: pytest.LogCaptureFixture) -> No
|
||||
|
||||
|
||||
@respx.mock
|
||||
@pytest.mark.usefixtures("fakeimg_png")
|
||||
async def test_options_only_stream(
|
||||
hass: HomeAssistant, fakeimgbytes_png: bytes, mock_create_stream: _patch[MagicMock]
|
||||
hass: HomeAssistant,
|
||||
mock_setup_entry: _patch[MagicMock],
|
||||
mock_create_stream: _patch[MagicMock],
|
||||
) -> None:
|
||||
"""Test the options flow without a still_image_url."""
|
||||
respx.get("http://127.0.0.1/testurl/2").respond(stream=fakeimgbytes_png)
|
||||
data = TESTDATA.copy()
|
||||
data.pop(CONF_STILL_IMAGE_URL)
|
||||
|
||||
@ -883,11 +811,10 @@ async def test_options_only_stream(
|
||||
assert result["step_id"] == "init"
|
||||
|
||||
# try updating the config options
|
||||
with mock_create_stream:
|
||||
result2 = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"],
|
||||
user_input=data,
|
||||
)
|
||||
result2 = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"],
|
||||
user_input=data,
|
||||
)
|
||||
assert result2["type"] is FlowResultType.FORM
|
||||
assert result2["step_id"] == "user_confirm"
|
||||
|
||||
@ -900,6 +827,7 @@ async def test_options_only_stream(
|
||||
|
||||
async def test_options_still_and_stream_not_provided(
|
||||
hass: HomeAssistant,
|
||||
mock_setup_entry: _patch[MagicMock],
|
||||
) -> None:
|
||||
"""Test we show a suitable error if neither still or stream URL are provided."""
|
||||
data = TESTDATA.copy()
|
||||
@ -929,7 +857,7 @@ async def test_options_still_and_stream_not_provided(
|
||||
|
||||
@respx.mock
|
||||
@pytest.mark.usefixtures("fakeimg_png")
|
||||
async def test_form_options_permission_error(
|
||||
async def test_options_permission_error(
|
||||
hass: HomeAssistant, config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test we handle a PermissionError and pass the message through."""
|
||||
@ -947,43 +875,6 @@ async def test_form_options_permission_error(
|
||||
assert result2["errors"] == {"stream_source": "stream_not_permitted"}
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("fakeimg_png")
|
||||
async def test_unload_entry(hass: HomeAssistant) -> None:
|
||||
"""Test unloading the generic IP Camera entry."""
|
||||
mock_entry = MockConfigEntry(domain=DOMAIN, options=TESTDATA)
|
||||
mock_entry.add_to_hass(hass)
|
||||
|
||||
await hass.config_entries.async_setup(mock_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert mock_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
await hass.config_entries.async_unload(mock_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert mock_entry.state is ConfigEntryState.NOT_LOADED
|
||||
|
||||
|
||||
async def test_reload_on_title_change(hass: HomeAssistant) -> None:
|
||||
"""Test the integration gets reloaded when the title is updated."""
|
||||
|
||||
test_data = TESTDATA_OPTIONS
|
||||
test_data[CONF_CONTENT_TYPE] = "image/png"
|
||||
mock_entry = MockConfigEntry(
|
||||
domain=DOMAIN, unique_id="54321", options=test_data, title="My Title"
|
||||
)
|
||||
mock_entry.add_to_hass(hass)
|
||||
|
||||
await hass.config_entries.async_setup(mock_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert mock_entry.state is ConfigEntryState.LOADED
|
||||
assert hass.states.get("camera.my_title").attributes["friendly_name"] == "My Title"
|
||||
|
||||
hass.config_entries.async_update_entry(mock_entry, title="New Title")
|
||||
assert mock_entry.title == "New Title"
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert hass.states.get("camera.my_title").attributes["friendly_name"] == "New Title"
|
||||
|
||||
|
||||
async def test_migrate_existing_ids(
|
||||
hass: HomeAssistant, entity_registry: er.EntityRegistry
|
||||
) -> None:
|
||||
@ -1019,40 +910,26 @@ async def test_migrate_existing_ids(
|
||||
|
||||
@respx.mock
|
||||
@pytest.mark.usefixtures("fakeimg_png")
|
||||
async def test_use_wallclock_as_timestamps_option(
|
||||
async def test_options_use_wallclock_as_timestamps(
|
||||
hass: HomeAssistant,
|
||||
mock_create_stream: _patch[MagicMock],
|
||||
hass_client: ClientSessionGenerator,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
fakeimgbytes_png: bytes,
|
||||
config_entry: MockConfigEntry,
|
||||
mock_setup_entry: _patch[MagicMock],
|
||||
) -> None:
|
||||
"""Test the use_wallclock_as_timestamps option flow."""
|
||||
|
||||
respx.get("http://127.0.0.1/testurl/1").respond(stream=fakeimgbytes_png)
|
||||
mock_entry = MockConfigEntry(
|
||||
title="Test Camera",
|
||||
domain=DOMAIN,
|
||||
data={},
|
||||
options=TESTDATA,
|
||||
)
|
||||
|
||||
mock_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
result = await hass.config_entries.options.async_init(
|
||||
mock_entry.entry_id, context={"show_advanced_options": True}
|
||||
config_entry.entry_id, context={"show_advanced_options": True}
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "init"
|
||||
with (
|
||||
patch("homeassistant.components.generic.async_setup_entry", return_value=True),
|
||||
mock_create_stream,
|
||||
):
|
||||
result2 = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_USE_WALLCLOCK_AS_TIMESTAMPS: True, **TESTDATA},
|
||||
)
|
||||
result2 = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_USE_WALLCLOCK_AS_TIMESTAMPS: True, **TESTDATA},
|
||||
)
|
||||
assert result2["type"] is FlowResultType.FORM
|
||||
|
||||
ws_client = await hass_ws_client()
|
||||
@ -1079,14 +956,10 @@ async def test_use_wallclock_as_timestamps_option(
|
||||
)
|
||||
assert result3["type"] is FlowResultType.FORM
|
||||
assert result3["step_id"] == "init"
|
||||
with (
|
||||
patch("homeassistant.components.generic.async_setup_entry", return_value=True),
|
||||
mock_create_stream,
|
||||
):
|
||||
result4 = await hass.config_entries.options.async_configure(
|
||||
result3["flow_id"],
|
||||
user_input={CONF_USE_WALLCLOCK_AS_TIMESTAMPS: True, **TESTDATA},
|
||||
)
|
||||
result4 = await hass.config_entries.options.async_configure(
|
||||
result3["flow_id"],
|
||||
user_input={CONF_USE_WALLCLOCK_AS_TIMESTAMPS: True, **TESTDATA},
|
||||
)
|
||||
assert result4["type"] is FlowResultType.FORM
|
||||
assert result4["step_id"] == "user_confirm"
|
||||
result5 = await hass.config_entries.options.async_configure(
|
||||
|
||||
37
tests/components/generic/test_init.py
Normal file
37
tests/components/generic/test_init.py
Normal file
@ -0,0 +1,37 @@
|
||||
"""Define tests for the generic (IP camera) integration."""
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("fakeimg_png")
|
||||
async def test_unload_entry(hass: HomeAssistant, setup_entry: MockConfigEntry) -> None:
|
||||
"""Test unloading the generic IP Camera entry."""
|
||||
assert setup_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
await hass.config_entries.async_unload(setup_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert setup_entry.state is ConfigEntryState.NOT_LOADED
|
||||
|
||||
|
||||
async def test_reload_on_title_change(
|
||||
hass: HomeAssistant, setup_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test the integration gets reloaded when the title is updated."""
|
||||
assert setup_entry.state is ConfigEntryState.LOADED
|
||||
assert (
|
||||
hass.states.get("camera.test_camera").attributes["friendly_name"]
|
||||
== "Test Camera"
|
||||
)
|
||||
|
||||
hass.config_entries.async_update_entry(setup_entry, title="New Title")
|
||||
assert setup_entry.title == "New Title"
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert (
|
||||
hass.states.get("camera.test_camera").attributes["friendly_name"] == "New Title"
|
||||
)
|
||||
Loading…
x
Reference in New Issue
Block a user