evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Generic ipcam configflow2 followup (#73511)

Browse source 
* Address code review comments

* Add type hints

* Remvoe unused strings

* Remove persistent notification setup

* Patch async_configre

* Fix pylint warning

* Address review comments

* Clean types

* Code review: defer local var assignment

Co-authored-by: Dave T <davet2001@users.noreply.github.com>
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Dave T 2022-06-17 06:07:21 +01:00 • committed by GitHub
parent f276523ef3
commit cdd5a5f68b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 89 additions and 64 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
homeassistant/components/generic/camera.py
View file

@ -1,7 +1,9 @@
"""Support for IP Cameras."""
from __future__ import annotations
from collections.abc import Mapping
import logging
from typing import Any
import httpx
import voluptuous as vol
@ -115,12 +117,12 @@ async def async_setup_entry(
)
def generate_auth(device_info) -> httpx.Auth | None:
def generate_auth(device_info: Mapping[str, Any]) -> httpx.Auth | None:
"""Generate httpx.Auth object from credentials."""
username = device_info.get(CONF_USERNAME)
password = device_info.get(CONF_PASSWORD)
username: str | None = device_info.get(CONF_USERNAME)
password: str | None = device_info.get(CONF_PASSWORD)
authentication = device_info.get(CONF_AUTHENTICATION)
if username:
if username and password:
if authentication == HTTP_DIGEST_AUTHENTICATION:
return httpx.DigestAuth(username=username, password=password)
return httpx.BasicAuth(username=username, password=password)
@ -130,7 +132,15 @@ def generate_auth(device_info) -> httpx.Auth | None:
class GenericCamera(Camera):
"""A generic implementation of an IP camera."""
def __init__(self, hass, device_info, identifier, title):
_last_image: bytes | None
def __init__(
self,
hass: HomeAssistant,
device_info: Mapping[str, Any],
identifier: str,
title: str,
) -> None:
"""Initialize a generic camera."""
super().__init__()
self.hass = hass
@ -143,10 +153,10 @@ class GenericCamera(Camera):
and self._still_image_url
):
self._still_image_url = cv.template(self._still_image_url)
if self._still_image_url not in [None, ""]:
if self._still_image_url:
self._still_image_url.hass = hass
self._stream_source = device_info.get(CONF_STREAM_SOURCE)
if self._stream_source not in (None, ""):
if self._stream_source:
if not isinstance(self._stream_source, template_helper.Template):
self._stream_source = cv.template(self._stream_source)
self._stream_source.hass = hass
@ -207,7 +217,7 @@ class GenericCamera(Camera):
"""Return the name of this device."""
return self._name
async def stream_source(self):
async def stream_source(self) -> str | None:
"""Return the source of the stream."""
if self._stream_source is None:
return None

44
homeassistant/components/generic/config_flow.py
View file

@ -1,11 +1,11 @@
"""Config flow for generic (IP Camera)."""
from __future__ import annotations
from collections.abc import Mapping
import contextlib
from errno import EHOSTUNREACH, EIO
import io
import logging
from types import MappingProxyType
from typing import Any
import PIL
@ -32,6 +32,7 @@ from homeassistant.const import (
HTTP_BASIC_AUTHENTICATION,
HTTP_DIGEST_AUTHENTICATION,
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResult
from homeassistant.exceptions import TemplateError
from homeassistant.helpers import config_validation as cv, template as template_helper
@ -64,7 +65,7 @@ SUPPORTED_IMAGE_TYPES = {"png", "jpeg", "gif", "svg+xml", "webp"}
def build_schema(
user_input: dict[str, Any] | MappingProxyType[str, Any],
user_input: Mapping[str, Any],
is_options_flow: bool = False,
show_advanced_options=False,
):
@ -119,7 +120,7 @@ def build_schema(
return vol.Schema(spec)
def get_image_type(image):
def get_image_type(image: bytes) -> str | None:
"""Get the format of downloaded bytes that could be an image."""
fmt = None
imagefile = io.BytesIO(image)
@ -135,7 +136,9 @@ def get_image_type(image):
return fmt
async def async_test_still(hass, info) -> tuple[dict[str, str], str | None]:
async def async_test_still(
hass: HomeAssistant, info: Mapping[str, Any]
) -> tuple[dict[str, str], str | None]:
"""Verify that the still image is valid before we create an entity."""
fmt = None
if not (url := info.get(CONF_STILL_IMAGE_URL)):
@ -147,7 +150,7 @@ async def async_test_still(hass, info) -> tuple[dict[str, str], str | None]:
except TemplateError as err:
_LOGGER.warning("Problem rendering template %s: %s", url, err)
return {CONF_STILL_IMAGE_URL: "template_error"}, None
verify_ssl = info.get(CONF_VERIFY_SSL)
verify_ssl = info[CONF_VERIFY_SSL]
auth = generate_auth(info)
try:
async_client = get_async_client(hass, verify_ssl=verify_ssl)
@ -177,7 +180,9 @@ async def async_test_still(hass, info) -> tuple[dict[str, str], str | None]:
return {}, f"image/{fmt}"
def slug(hass, template) -> str | None:
def slug(
hass: HomeAssistant, template: str | template_helper.Template | None
) -> str | None:
"""Convert a camera url into a string suitable for a camera name."""
if not template:
return None
@ -193,7 +198,9 @@ def slug(hass, template) -> str | None:
return None
async def async_test_stream(hass, info) -> dict[str, str]:
async def async_test_stream(
hass: HomeAssistant, info: Mapping[str, Any]
) -> dict[str, str]:
"""Verify that the stream is valid before we create an entity."""
if not (stream_source := info.get(CONF_STREAM_SOURCE)):
return {}
@ -240,7 +247,7 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
VERSION = 1
def __init__(self):
def __init__(self) -> None:
"""Initialize Generic ConfigFlow."""
self.cached_user_input: dict[str, Any] = {}
self.cached_title = ""
@ -252,7 +259,7 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
"""Get the options flow for this handler."""
return GenericOptionsFlowHandler(config_entry)
def check_for_existing(self, options):
def check_for_existing(self, options: dict[str, Any]) -> bool:
"""Check whether an existing entry is using the same URLs."""
return any(
entry.options.get(CONF_STILL_IMAGE_URL) == options.get(CONF_STILL_IMAGE_URL)
@ -273,14 +280,16 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
):
errors["base"] = "no_still_image_or_stream_url"
else:
errors, still_format = await async_test_still(self.hass, user_input)
errors = errors | await async_test_stream(self.hass, user_input)
still_url = user_input.get(CONF_STILL_IMAGE_URL)
stream_url = user_input.get(CONF_STREAM_SOURCE)
name = slug(hass, still_url) or slug(hass, stream_url) or DEFAULT_NAME
errors, still_format = await async_test_still(hass, user_input)
errors = errors | await async_test_stream(hass, user_input)
if not errors:
user_input[CONF_CONTENT_TYPE] = still_format
user_input[CONF_LIMIT_REFETCH_TO_URL_CHANGE] = False
still_url = user_input.get(CONF_STILL_IMAGE_URL)
stream_url = user_input.get(CONF_STREAM_SOURCE)
name = (
slug(hass, still_url) or slug(hass, stream_url) or DEFAULT_NAME
)
if still_url is None:
# If user didn't specify a still image URL,
# The automatically generated still image that stream generates
@ -299,7 +308,7 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
errors=errors,
)
async def async_step_import(self, import_config) -> FlowResult:
async def async_step_import(self, import_config: dict[str, Any]) -> FlowResult:
"""Handle config import from yaml."""
# abort if we've already got this one.
if self.check_for_existing(import_config):
@ -311,6 +320,7 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
CONF_NAME,
slug(self.hass, still_url) or slug(self.hass, stream_url) or DEFAULT_NAME,
)
if CONF_LIMIT_REFETCH_TO_URL_CHANGE not in import_config:
import_config[CONF_LIMIT_REFETCH_TO_URL_CHANGE] = False
still_format = import_config.get(CONF_CONTENT_TYPE, "image/jpeg")
@ -336,9 +346,9 @@ class GenericOptionsFlowHandler(OptionsFlow):
if user_input is not None:
errors, still_format = await async_test_still(
self.hass, self.config_entry.options | user_input
hass, self.config_entry.options | user_input
)
errors = errors | await async_test_stream(self.hass, user_input)
errors = errors | await async_test_stream(hass, user_input)
still_url = user_input.get(CONF_STILL_IMAGE_URL)
stream_url = user_input.get(CONF_STREAM_SOURCE)
if not errors:

3
homeassistant/components/generic/strings.json
View file

@ -15,8 +15,7 @@
"stream_not_permitted": "Operation not permitted while trying to connect to stream. Wrong RTSP transport protocol?"
},
"abort": {
"single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]",
"no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]"
"single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]"
},
"step": {
"user": {

7
homeassistant/components/generic/translations/en.json
View file

@ -1,7 +1,6 @@
{
"config": {
"abort": {
"no_devices_found": "No devices found on the network",
"single_instance_allowed": "Already configured. Only a single configuration possible."
},
"error": {
@ -12,9 +11,7 @@
"stream_http_not_found": "HTTP 404 Not found while trying to connect to stream",
"stream_io_error": "Input/Output error while trying to connect to stream. Wrong RTSP transport protocol?",
"stream_no_route_to_host": "Could not find host while trying to connect to stream",
"stream_no_video": "Stream has no video",
"stream_not_permitted": "Operation not permitted while trying to connect to stream. Wrong RTSP transport protocol?",
"stream_unauthorised": "Authorisation failed while trying to connect to stream",
"template_error": "Error rendering template. Review log for more info.",
"timeout": "Timeout while loading URL",
"unable_still_load": "Unable to load valid image from still image URL (e.g. invalid host, URL or authentication failure). Review log for more info.",
@ -51,13 +48,9 @@
"already_exists": "A camera with these URL settings already exists.",
"invalid_still_image": "URL did not return a valid still image",
"no_still_image_or_stream_url": "You must specify at least a still image or stream URL",
"stream_file_not_found": "File not found while trying to connect to stream (is ffmpeg installed?)",
"stream_http_not_found": "HTTP 404 Not found while trying to connect to stream",
"stream_io_error": "Input/Output error while trying to connect to stream. Wrong RTSP transport protocol?",
"stream_no_route_to_host": "Could not find host while trying to connect to stream",
"stream_no_video": "Stream has no video",
"stream_not_permitted": "Operation not permitted while trying to connect to stream. Wrong RTSP transport protocol?",
"stream_unauthorised": "Authorisation failed while trying to connect to stream",
"template_error": "Error rendering template. Review log for more info.",
"timeout": "Timeout while loading URL",
"unable_still_load": "Unable to load valid image from still image URL (e.g. invalid host, URL or authentication failure). Review log for more info.",

3
tests/components/generic/conftest.py
View file

@ -7,7 +7,7 @@ from PIL import Image
import pytest
import respx
from homeassistant import config_entries, setup
from homeassistant import config_entries
from homeassistant.components.generic.const import DOMAIN
from tests.common import MockConfigEntry
@ -79,7 +79,6 @@ def mock_create_stream():
async def user_flow(hass):
"""Initiate a user flow."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)

70
tests/components/generic/test_config_flow.py
View file

@ -8,7 +8,7 @@ import httpx
import pytest
import respx
from homeassistant import config_entries, data_entry_flow, setup
from homeassistant import config_entries, data_entry_flow
from homeassistant.components.camera import async_get_image
from homeassistant.components.generic.const import (
CONF_CONTENT_TYPE,
@ -60,7 +60,9 @@ TESTDATA_YAML = {
async def test_form(hass, fakeimg_png, user_flow, mock_create_stream):
"""Test the form with a normal set of settings."""
with mock_create_stream as mock_setup:
with mock_create_stream as mock_setup, patch(
"homeassistant.components.generic.async_setup_entry", return_value=True
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
user_flow["flow_id"],
TESTDATA,
@ -81,12 +83,12 @@ async def test_form(hass, fakeimg_png, user_flow, mock_create_stream):
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
@respx.mock
async def test_form_only_stillimage(hass, fakeimg_png, user_flow):
"""Test we complete ok if the user wants still images only."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
@ -95,10 +97,12 @@ async def test_form_only_stillimage(hass, fakeimg_png, user_flow):
data = TESTDATA.copy()
data.pop(CONF_STREAM_SOURCE)
result2 = await hass.config_entries.flow.async_configure(
user_flow["flow_id"],
data,
)
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()
assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result2["title"] == "127_0_0_1"
assert result2["options"] == {
@ -112,7 +116,6 @@ async def test_form_only_stillimage(hass, fakeimg_png, user_flow):
CONF_VERIFY_SSL: False,
}
await hass.async_block_till_done()
assert respx.calls.call_count == 1
@ -121,10 +124,12 @@ async def test_form_only_stillimage_gif(hass, fakeimg_gif, user_flow):
"""Test we complete ok if the user wants a gif."""
data = TESTDATA.copy()
data.pop(CONF_STREAM_SOURCE)
result2 = await hass.config_entries.flow.async_configure(
user_flow["flow_id"],
data,
)
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()
assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result2["options"][CONF_CONTENT_TYPE] == "image/gif"
@ -136,10 +141,12 @@ async def test_form_only_svg_whitespace(hass, fakeimgbytes_svg, user_flow):
respx.get("http://127.0.0.1/testurl/1").respond(stream=fakeimgbytes_wspace_svg)
data = TESTDATA.copy()
data.pop(CONF_STREAM_SOURCE)
result2 = await hass.config_entries.flow.async_configure(
user_flow["flow_id"],
data,
)
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()
assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
@ -161,10 +168,12 @@ async def test_form_only_still_sample(hass, user_flow, image_file):
respx.get("http://127.0.0.1/testurl/1").respond(stream=image.read())
data = TESTDATA.copy()
data.pop(CONF_STREAM_SOURCE)
result2 = await hass.config_entries.flow.async_configure(
user_flow["flow_id"],
data,
)
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()
assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
@ -203,10 +212,12 @@ async def test_still_template(
data = TESTDATA.copy()
data.pop(CONF_STREAM_SOURCE)
data[CONF_STILL_IMAGE_URL] = template
result2 = await hass.config_entries.flow.async_configure(
user_flow["flow_id"],
data,
)
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()
assert result2["type"] == expected_result
@ -216,7 +227,9 @@ async def test_form_rtsp_mode(hass, fakeimg_png, user_flow, mock_create_stream):
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:
with mock_create_stream as mock_setup, patch(
"homeassistant.components.generic.async_setup_entry", return_value=True
):
result2 = await hass.config_entries.flow.async_configure(
user_flow["flow_id"], data
)
@ -242,7 +255,6 @@ async def test_form_rtsp_mode(hass, fakeimg_png, user_flow, mock_create_stream):
async def test_form_only_stream(hass, fakeimgbytes_jpg, mock_create_stream):
"""Test we complete ok if the user wants stream only."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
@ -254,6 +266,7 @@ async def test_form_only_stream(hass, fakeimgbytes_jpg, mock_create_stream):
result["flow_id"],
data,
)
await hass.async_block_till_done()
assert result3["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result3["title"] == "127_0_0_1"
@ -454,7 +467,6 @@ async def test_options_template_error(hass, fakeimgbytes_png, mock_create_stream
"""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)
await setup.async_setup_component(hass, "persistent_notification", {})
mock_entry = MockConfigEntry(
title="Test Camera",
@ -649,7 +661,9 @@ async def test_use_wallclock_as_timestamps_option(
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
with mock_create_stream:
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},