Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Allow selecting camera in Trafikverket Camera (#105927) 

* Allow selecting camera in Trafikverket Camera

* Final config flow

* Add tests

* Fix load_int

* naming
pull/108231/head
G Johansson 2024-01-17 11:54:13 +01:00 committed by GitHub
parent e811cf1ae8
commit bdda38f274
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 187 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

91
homeassistant/components/trafikverket_camera/config_flow.py
View File

@ -4,12 +4,7 @@ from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from pytrafikverket.exceptions import (
InvalidAuthentication,
MultipleCamerasFound,
NoCameraFound,
UnknownError,
)
from pytrafikverket.exceptions import InvalidAuthentication, NoCameraFound, UnknownError
from pytrafikverket.trafikverket_camera import CameraInfo, TrafikverketCamera
import voluptuous as vol
@ -17,7 +12,13 @@ from homeassistant import config_entries
from homeassistant.const import CONF_API_KEY, CONF_ID, CONF_LOCATION
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.selector import TextSelector
from homeassistant.helpers.selector import (
SelectOptionDict,
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
TextSelector,
)
from .const import DOMAIN
@ -28,34 +29,28 @@ class TVCameraConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 3
entry: config_entries.ConfigEntry | None
cameras: list[CameraInfo]
api_key: str
async def validate_input(
self, sensor_api: str, location: str
) -> tuple[dict[str, str], str | None, str | None]:
) -> tuple[dict[str, str], list[CameraInfo] | None]:
"""Validate input from user input."""
errors: dict[str, str] = {}
camera_info: CameraInfo | None = None
camera_location: str | None = None
camera_id: str | None = None
cameras: list[CameraInfo] | None = None
web_session = async_get_clientsession(self.hass)
camera_api = TrafikverketCamera(web_session, sensor_api)
try:
camera_info = await camera_api.async_get_camera(location)
cameras = await camera_api.async_get_cameras(location)
except NoCameraFound:
errors["location"] = "invalid_location"
except MultipleCamerasFound:
errors["location"] = "more_locations"
except InvalidAuthentication:
errors["base"] = "invalid_auth"
except UnknownError:
errors["base"] = "cannot_connect"
if camera_info:
camera_id = camera_info.camera_id
camera_location = camera_info.camera_name or "Trafikverket Camera"
return (errors, camera_location, camera_id)
return (errors, cameras)
async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> FlowResult:
"""Handle re-authentication with Trafikverket."""
@ -73,7 +68,7 @@ class TVCameraConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
api_key = user_input[CONF_API_KEY]
assert self.entry is not None
errors, _, _ = await self.validate_input(api_key, self.entry.data[CONF_ID])
errors, _ = await self.validate_input(api_key, self.entry.data[CONF_ID])
if not errors:
self.hass.config_entries.async_update_entry(
@ -106,17 +101,18 @@ class TVCameraConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
api_key = user_input[CONF_API_KEY]
location = user_input[CONF_LOCATION]
errors, camera_location, camera_id = await self.validate_input(
api_key, location
)
errors, cameras = await self.validate_input(api_key, location)
if not errors:
assert camera_location
await self.async_set_unique_id(f"{DOMAIN}-{camera_id}")
if not errors and cameras:
if len(cameras) > 1:
self.cameras = cameras
self.api_key = api_key
return await self.async_step_multiple_cameras()
await self.async_set_unique_id(f"{DOMAIN}-{cameras[0].camera_id}")
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=camera_location,
data={CONF_API_KEY: api_key, CONF_ID: camera_id},
title=cameras[0].camera_name or "Trafikverket Camera",
data={CONF_API_KEY: api_key, CONF_ID: cameras[0].camera_id},
)
return self.async_show_form(
@ -129,3 +125,42 @@ class TVCameraConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
),
errors=errors,
)
async def async_step_multiple_cameras(
self, user_input: dict[str, str] | None = None
) -> FlowResult:
"""Handle when multiple cameras."""
if user_input:
errors, cameras = await self.validate_input(
self.api_key, user_input[CONF_ID]
)
if not errors and cameras:
await self.async_set_unique_id(f"{DOMAIN}-{cameras[0].camera_id}")
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=cameras[0].camera_name or "Trafikverket Camera",
data={CONF_API_KEY: self.api_key, CONF_ID: cameras[0].camera_id},
)
camera_choices = [
SelectOptionDict(
value=f"{camera_info.camera_id}",
label=f"{camera_info.camera_id} - {camera_info.camera_name} - {camera_info.location}",
)
for camera_info in self.cameras
]
return self.async_show_form(
step_id="multiple_cameras",
data_schema=vol.Schema(
{
vol.Required(CONF_ID): SelectSelector(
SelectSelectorConfig(
options=camera_choices, mode=SelectSelectorMode.LIST
)
),
}
),
)

8
homeassistant/components/trafikverket_camera/strings.json
View File

@ -17,7 +17,13 @@
"location": "[%key:common::config_flow::data::location%]"
},
"data_description": {
"location": "Equal or part of name, description or camera id"
"location": "Equal or part of name, description or camera id. Be as specific as possible to avoid getting multiple cameras as result"
}
},
"multiple_cameras": {
"description": "Result came back with multiple cameras",
"data": {
"id": "Choose camera"
}
}
}

59
tests/components/trafikverket_camera/conftest.py
View File

@ -70,6 +70,65 @@ def fixture_get_camera() -> CameraInfo:
)
@pytest.fixture(name="get_camera2")
def fixture_get_camera2() -> CameraInfo:
"""Construct Camera Mock 2."""
return CameraInfo(
camera_name="Test Camera2",
camera_id="5678",
active=True,
deleted=False,
description="Test Camera for testing2",
direction="180",
fullsizephoto=True,
location="Test location2",
modified=datetime(2022, 4, 4, 4, 4, 4, tzinfo=dt_util.UTC),
phototime=datetime(2022, 4, 4, 4, 4, 4, tzinfo=dt_util.UTC),
photourl="https://www.testurl.com/test_photo2.jpg",
status="Running",
camera_type="Road",
)
@pytest.fixture(name="get_cameras")
def fixture_get_cameras() -> CameraInfo:
"""Construct Camera Mock with multiple cameras."""
return [
CameraInfo(
camera_name="Test Camera",
camera_id="1234",
active=True,
deleted=False,
description="Test Camera for testing",
direction="180",
fullsizephoto=True,
location="Test location",
modified=datetime(2022, 4, 4, 4, 4, 4, tzinfo=dt_util.UTC),
phototime=datetime(2022, 4, 4, 4, 4, 4, tzinfo=dt_util.UTC),
photourl="https://www.testurl.com/test_photo.jpg",
status="Running",
camera_type="Road",
),
CameraInfo(
camera_name="Test Camera2",
camera_id="5678",
active=True,
deleted=False,
description="Test Camera for testing2",
direction="180",
fullsizephoto=True,
location="Test location2",
modified=datetime(2022, 4, 4, 4, 4, 4, tzinfo=dt_util.UTC),
phototime=datetime(2022, 4, 4, 4, 4, 4, tzinfo=dt_util.UTC),
photourl="https://www.testurl.com/test_photo2.jpg",
status="Running",
camera_type="Road",
),
]
@pytest.fixture(name="get_camera_no_location")
def fixture_get_camera_no_location() -> CameraInfo:
"""Construct Camera Mock."""

82
tests/components/trafikverket_camera/test_config_flow.py
View File

@ -4,12 +4,7 @@ from __future__ import annotations
from unittest.mock import patch
import pytest
from pytrafikverket.exceptions import (
InvalidAuthentication,
MultipleCamerasFound,
NoCameraFound,
UnknownError,
)
from pytrafikverket.exceptions import InvalidAuthentication, NoCameraFound, UnknownError
from pytrafikverket.trafikverket_camera import CameraInfo
from homeassistant import config_entries
@ -31,8 +26,8 @@ async def test_form(hass: HomeAssistant, get_camera: CameraInfo) -> None:
assert result["errors"] == {}
with patch(
"homeassistant.components.trafikverket_camera.config_flow.TrafikverketCamera.async_get_camera",
return_value=get_camera,
"homeassistant.components.trafikverket_camera.config_flow.TrafikverketCamera.async_get_cameras",
return_value=[get_camera],
), patch(
"homeassistant.components.trafikverket_camera.async_setup_entry",
return_value=True,
@ -56,6 +51,55 @@ async def test_form(hass: HomeAssistant, get_camera: CameraInfo) -> None:
assert result2["result"].unique_id == "trafikverket_camera-1234"
async def test_form_multiple_cameras(
hass: HomeAssistant, get_cameras: list[CameraInfo], get_camera2: CameraInfo
) -> None:
"""Test we get the form with multiple cameras."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {}
with patch(
"homeassistant.components.trafikverket_camera.config_flow.TrafikverketCamera.async_get_cameras",
return_value=get_cameras,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_API_KEY: "1234567890",
CONF_LOCATION: "Test loc",
},
)
await hass.async_block_till_done()
with patch(
"homeassistant.components.trafikverket_camera.config_flow.TrafikverketCamera.async_get_cameras",
return_value=[get_camera2],
), patch(
"homeassistant.components.trafikverket_camera.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_ID: "5678",
},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "Test Camera2"
assert result["data"] == {
"api_key": "1234567890",
"id": "5678",
}
assert len(mock_setup_entry.mock_calls) == 1
assert result["result"].unique_id == "trafikverket_camera-5678"
async def test_form_no_location_data(
hass: HomeAssistant, get_camera_no_location: CameraInfo
) -> None:
@ -68,8 +112,8 @@ async def test_form_no_location_data(
assert result["errors"] == {}
with patch(
"homeassistant.components.trafikverket_camera.config_flow.TrafikverketCamera.async_get_camera",
return_value=get_camera_no_location,
"homeassistant.components.trafikverket_camera.config_flow.TrafikverketCamera.async_get_cameras",
return_value=[get_camera_no_location],
), patch(
"homeassistant.components.trafikverket_camera.async_setup_entry",
return_value=True,
@ -106,11 +150,6 @@ async def test_form_no_location_data(
"location",
"invalid_location",
),
(
MultipleCamerasFound,
"location",
"more_locations",
),
(
UnknownError,
"base",
@ -130,7 +169,7 @@ async def test_flow_fails(
assert result4["step_id"] == config_entries.SOURCE_USER
with patch(
"homeassistant.components.trafikverket_camera.config_flow.TrafikverketCamera.async_get_camera",
"homeassistant.components.trafikverket_camera.config_flow.TrafikverketCamera.async_get_cameras",
side_effect=side_effect,
):
result4 = await hass.config_entries.flow.async_configure(
@ -171,7 +210,7 @@ async def test_reauth_flow(hass: HomeAssistant) -> None:
assert result["errors"] == {}
with patch(
"homeassistant.components.trafikverket_camera.config_flow.TrafikverketCamera.async_get_camera",
"homeassistant.components.trafikverket_camera.config_flow.TrafikverketCamera.async_get_cameras",
), patch(
"homeassistant.components.trafikverket_camera.async_setup_entry",
return_value=True,
@ -203,11 +242,6 @@ async def test_reauth_flow(hass: HomeAssistant) -> None:
"location",
"invalid_location",
),
(
MultipleCamerasFound,
"location",
"more_locations",
),
(
UnknownError,
"base",
@ -242,7 +276,7 @@ async def test_reauth_flow_error(
)
with patch(
"homeassistant.components.trafikverket_camera.config_flow.TrafikverketCamera.async_get_camera",
"homeassistant.components.trafikverket_camera.config_flow.TrafikverketCamera.async_get_cameras",
side_effect=side_effect,
):
result2 = await hass.config_entries.flow.async_configure(
@ -256,7 +290,7 @@ async def test_reauth_flow_error(
assert result2["errors"] == {error_key: p_error}
with patch(
"homeassistant.components.trafikverket_camera.config_flow.TrafikverketCamera.async_get_camera",
"homeassistant.components.trafikverket_camera.config_flow.TrafikverketCamera.async_get_cameras",
), patch(
"homeassistant.components.trafikverket_camera.async_setup_entry",
return_value=True,