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

Add Trafikverket Camera integration ()

pull/98946/head
G Johansson 2023-08-24 10:39:22 +02:00 committed by GitHub
parent 7926c5cea9
commit 147351be6e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
23 changed files with 1083 additions and 0 deletions

1
.strict-typing
View File

@ -326,6 +326,7 @@ homeassistant.components.tplink.*
homeassistant.components.tplink_omada.*
homeassistant.components.tractive.*
homeassistant.components.tradfri.*
homeassistant.components.trafikverket_camera.*
homeassistant.components.trafikverket_ferry.*
homeassistant.components.trafikverket_train.*
homeassistant.components.trafikverket_weatherstation.*

2
CODEOWNERS
View File

@ -1300,6 +1300,8 @@ build.json @home-assistant/supervisor
/tests/components/trace/ @home-assistant/core
/homeassistant/components/tractive/ @Danielhiversen @zhulik @bieniu
/tests/components/tractive/ @Danielhiversen @zhulik @bieniu
/homeassistant/components/trafikverket_camera/ @gjohansson-ST
/tests/components/trafikverket_camera/ @gjohansson-ST
/homeassistant/components/trafikverket_ferry/ @gjohansson-ST
/tests/components/trafikverket_ferry/ @gjohansson-ST
/homeassistant/components/trafikverket_train/ @endor-force @gjohansson-ST

1
homeassistant/brands/trafikverket.json
View File

@ -2,6 +2,7 @@
"domain": "trafikverket",
"name": "Trafikverket",
"integrations": [
"trafikverket_camera",
"trafikverket_ferry",
"trafikverket_train",
"trafikverket_weatherstation"

29
homeassistant/components/trafikverket_camera/__init__.py Normal file
View File

@ -0,0 +1,29 @@
"""The trafikverket_camera component."""
from __future__ import annotations
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from .const import DOMAIN, PLATFORMS
from .coordinator import TVDataUpdateCoordinator
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Trafikverket Camera from a config entry."""
coordinator = TVDataUpdateCoordinator(hass, entry)
await coordinator.async_config_entry_first_refresh()
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload Trafikverket Camera config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok

84
homeassistant/components/trafikverket_camera/camera.py Normal file
View File

@ -0,0 +1,84 @@
"""Camera for the Trafikverket Camera integration."""
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from homeassistant.components.camera import Camera
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_LOCATION
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import ATTR_DESCRIPTION, ATTR_TYPE, DOMAIN
from .coordinator import TVDataUpdateCoordinator
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up a Trafikverket Camera."""
coordinator: TVDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
async_add_entities(
[
TVCamera(
coordinator,
entry.title,
entry.entry_id,
)
],
)
class TVCamera(CoordinatorEntity[TVDataUpdateCoordinator], Camera):
"""Implement Trafikverket camera."""
_attr_has_entity_name = True
_attr_name = None
_attr_translation_key = "tv_camera"
coordinator: TVDataUpdateCoordinator
def __init__(
self,
coordinator: TVDataUpdateCoordinator,
name: str,
entry_id: str,
) -> None:
"""Initialize the camera."""
super().__init__(coordinator)
Camera.__init__(self)
self._attr_unique_id = entry_id
self._attr_device_info = DeviceInfo(
entry_type=DeviceEntryType.SERVICE,
identifiers={(DOMAIN, entry_id)},
manufacturer="Trafikverket",
model="v1.0",
name=name,
configuration_url="https://api.trafikinfo.trafikverket.se/",
)
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return camera picture."""
return self.coordinator.data.image
@property
def is_on(self) -> bool:
"""Return camera on."""
return self.coordinator.data.data.active is True
@property
def extra_state_attributes(self) -> Mapping[str, Any] | None:
"""Return additional attributes."""
return {
ATTR_DESCRIPTION: self.coordinator.data.data.description,
ATTR_LOCATION: self.coordinator.data.data.location,
ATTR_TYPE: self.coordinator.data.data.camera_type,
}

122
homeassistant/components/trafikverket_camera/config_flow.py Normal file
View File

@ -0,0 +1,122 @@
"""Adds config flow for Trafikverket Camera integration."""
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from pytrafikverket.exceptions import (
InvalidAuthentication,
MultipleCamerasFound,
NoCameraFound,
UnknownError,
)
from pytrafikverket.trafikverket_camera import TrafikverketCamera
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_API_KEY
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
from .const import CONF_LOCATION, DOMAIN
class TVCameraConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Trafikverket Camera integration."""
VERSION = 1
entry: config_entries.ConfigEntry | None
async def validate_input(self, sensor_api: str, location: str) -> dict[str, str]:
"""Validate input from user input."""
errors: dict[str, str] = {}
web_session = async_get_clientsession(self.hass)
camera_api = TrafikverketCamera(web_session, sensor_api)
try:
await camera_api.async_get_camera(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"
return errors
async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> FlowResult:
"""Handle re-authentication with Trafikverket."""
self.entry = self.hass.config_entries.async_get_entry(self.context["entry_id"])
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Confirm re-authentication with Trafikverket."""
errors = {}
if user_input:
api_key = user_input[CONF_API_KEY]
assert self.entry is not None
errors = await self.validate_input(api_key, self.entry.data[CONF_LOCATION])
if not errors:
self.hass.config_entries.async_update_entry(
self.entry,
data={
**self.entry.data,
CONF_API_KEY: api_key,
},
)
await self.hass.config_entries.async_reload(self.entry.entry_id)
return self.async_abort(reason="reauth_successful")
return self.async_show_form(
step_id="reauth_confirm",
data_schema=vol.Schema(
{
vol.Required(CONF_API_KEY): cv.string,
}
),
errors=errors,
)
async def async_step_user(
self, user_input: dict[str, str] | None = None
) -> FlowResult:
"""Handle the initial step."""
errors = {}
if user_input:
api_key = user_input[CONF_API_KEY]
location = user_input[CONF_LOCATION]
errors = await self.validate_input(api_key, location)
if not errors:
await self.async_set_unique_id(f"{DOMAIN}-{location}")
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=user_input[CONF_LOCATION],
data={
CONF_API_KEY: api_key,
CONF_LOCATION: location,
},
)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_API_KEY): cv.string,
vol.Required(CONF_LOCATION): cv.string,
}
),
errors=errors,
)

10
homeassistant/components/trafikverket_camera/const.py Normal file
View File

@ -0,0 +1,10 @@
"""Adds constants for Trafikverket Camera integration."""
from homeassistant.const import Platform
DOMAIN = "trafikverket_camera"
CONF_LOCATION = "location"
PLATFORMS = [Platform.CAMERA]
ATTRIBUTION = "Data provided by Trafikverket"
ATTR_DESCRIPTION = "description"
ATTR_TYPE = "type"

76
homeassistant/components/trafikverket_camera/coordinator.py Normal file
View File

@ -0,0 +1,76 @@
"""DataUpdateCoordinator for the Trafikverket Camera integration."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import timedelta
from io import BytesIO
import logging
from pytrafikverket.exceptions import (
InvalidAuthentication,
MultipleCamerasFound,
NoCameraFound,
UnknownError,
)
from pytrafikverket.trafikverket_camera import CameraInfo, TrafikverketCamera
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_API_KEY
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import CONF_LOCATION, DOMAIN
_LOGGER = logging.getLogger(__name__)
TIME_BETWEEN_UPDATES = timedelta(minutes=5)
@dataclass
class CameraData:
"""Dataclass for Camera data."""
data: CameraInfo
image: bytes | None
class TVDataUpdateCoordinator(DataUpdateCoordinator[CameraData]):
"""A Trafikverket Data Update Coordinator."""
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Initialize the Trafikverket coordinator."""
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=TIME_BETWEEN_UPDATES,
)
self.session = async_get_clientsession(hass)
self._camera_api = TrafikverketCamera(self.session, entry.data[CONF_API_KEY])
self._location = entry.data[CONF_LOCATION]
async def _async_update_data(self) -> CameraData:
"""Fetch data from Trafikverket."""
camera_data: CameraInfo
image: bytes | None = None
try:
camera_data = await self._camera_api.async_get_camera(self._location)
except (NoCameraFound, MultipleCamerasFound, UnknownError) as error:
raise UpdateFailed from error
except InvalidAuthentication as error:
raise ConfigEntryAuthFailed from error
if camera_data.photourl is None:
return CameraData(data=camera_data, image=None)
image_url = camera_data.photourl
if camera_data.fullsizephoto:
image_url = f"{camera_data.photourl}?type=fullsize"
async with self.session.get(image_url, timeout=10) as get_image:
if get_image.status not in range(200, 299):
raise UpdateFailed("Could not retrieve image")
image = BytesIO(await get_image.read()).getvalue()
return CameraData(data=camera_data, image=image)

10
homeassistant/components/trafikverket_camera/manifest.json Normal file
View File

@ -0,0 +1,10 @@
{
"domain": "trafikverket_camera",
"name": "Trafikverket Camera",
"codeowners": ["@gjohansson-ST"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/trafikverket_camera",
"iot_class": "cloud_polling",
"loggers": ["pytrafikverket"],
"requirements": ["pytrafikverket==0.3.5"]
}

13
homeassistant/components/trafikverket_camera/recorder.py Normal file
View File

@ -0,0 +1,13 @@
"""Integration platform for recorder."""
from __future__ import annotations
from homeassistant.const import ATTR_LOCATION
from homeassistant.core import HomeAssistant, callback
from .const import ATTR_DESCRIPTION
@callback
def exclude_attributes(hass: HomeAssistant) -> set[str]:
"""Exclude description and location from being recorded in the database."""
return {ATTR_DESCRIPTION, ATTR_LOCATION}

51
homeassistant/components/trafikverket_camera/strings.json Normal file
View File

@ -0,0 +1,51 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"invalid_location": "Could not find a camera location with the specified name",
"more_locations": "Found multiple camera locations with the specified name"
},
"step": {
"user": {
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]",
"location": "[%key:common::config_flow::data::location%]"
}
}
}
},
"entity": {
"camera": {
"tv_camera": {
"state_attributes": {
"description": {
"name": "Description"
},
"direction": {
"name": "Direction"
},
"full_size_photo": {
"name": "Full size photo"
},
"location": {
"name": "[%key:common::config_flow::data::location%]"
},
"photo_url": {
"name": "Photo url"
},
"status": {
"name": "Status"
},
"type": {
"name": "Camera type"
}
}
}
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -480,6 +480,7 @@ FLOWS = {
"traccar",
"tractive",
"tradfri",
"trafikverket_camera",
"trafikverket_ferry",
"trafikverket_train",
"trafikverket_weatherstation",

6
homeassistant/generated/integrations.json
View File

@ -5878,6 +5878,12 @@
"trafikverket": {
"name": "Trafikverket",
"integrations": {
"trafikverket_camera": {
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling",
"name": "Trafikverket Camera"
},
"trafikverket_ferry": {
"integration_type": "hub",
"config_flow": true,

10
mypy.ini
View File

@ -3023,6 +3023,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.trafikverket_camera.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.trafikverket_ferry.*]
check_untyped_defs = true
disallow_incomplete_defs = true

1
requirements_all.txt
View File

@ -2191,6 +2191,7 @@ pytraccar==1.0.0
# homeassistant.components.tradfri
pytradfri[async]==9.0.1
# homeassistant.components.trafikverket_camera
# homeassistant.components.trafikverket_ferry
# homeassistant.components.trafikverket_train
# homeassistant.components.trafikverket_weatherstation

1
requirements_test_all.txt
View File

@ -1608,6 +1608,7 @@ pytraccar==1.0.0
# homeassistant.components.tradfri
pytradfri[async]==9.0.1
# homeassistant.components.trafikverket_camera
# homeassistant.components.trafikverket_ferry
# homeassistant.components.trafikverket_train
# homeassistant.components.trafikverket_weatherstation

10
tests/components/trafikverket_camera/__init__.py Normal file
View File

@ -0,0 +1,10 @@
"""Tests for the Trafikverket Camera integration."""
from __future__ import annotations
from homeassistant.components.trafikverket_camera.const import CONF_LOCATION
from homeassistant.const import CONF_API_KEY
ENTRY_CONFIG = {
CONF_API_KEY: "1234567890",
CONF_LOCATION: "Test location",
}

69
tests/components/trafikverket_camera/conftest.py Normal file
View File

@ -0,0 +1,69 @@
"""Fixtures for Trafikverket Camera integration tests."""
from __future__ import annotations
from datetime import datetime
from unittest.mock import patch
import pytest
from pytrafikverket.trafikverket_camera import CameraInfo
from homeassistant.components.trafikverket_camera.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.core import HomeAssistant
from homeassistant.util import dt as dt_util
from . import ENTRY_CONFIG
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
@pytest.fixture(name="load_int")
async def load_integration_from_entry(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, get_camera: CameraInfo
) -> MockConfigEntry:
"""Set up the Trafikverket Ferry integration in Home Assistant."""
aioclient_mock.get(
"https://www.testurl.com/test_photo.jpg?type=fullsize", content=b"0123456789"
)
config_entry = MockConfigEntry(
domain=DOMAIN,
source=SOURCE_USER,
data=ENTRY_CONFIG,
entry_id="1",
unique_id="123",
title="Test location",
)
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.trafikverket_camera.coordinator.TrafikverketCamera.async_get_camera",
return_value=get_camera,
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
return config_entry
@pytest.fixture(name="get_camera")
def fixture_get_camera() -> CameraInfo:
"""Construct Camera Mock."""
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",
)

75
tests/components/trafikverket_camera/test_camera.py Normal file
View File

@ -0,0 +1,75 @@
"""The test for the Trafikverket camera platform."""
from __future__ import annotations
from datetime import timedelta
from unittest.mock import patch
import pytest
from pytrafikverket.trafikverket_camera import CameraInfo
from homeassistant.components.camera import async_get_image
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.util import dt as dt_util
from tests.common import async_fire_time_changed
from tests.test_util.aiohttp import AiohttpClientMocker
async def test_camera(
hass: HomeAssistant,
load_int: ConfigEntry,
monkeypatch: pytest.MonkeyPatch,
aioclient_mock: AiohttpClientMocker,
get_camera: CameraInfo,
) -> None:
"""Test the Trafikverket Camera sensor."""
state1 = hass.states.get("camera.test_location")
assert state1.state == "idle"
assert state1.attributes["description"] == "Test Camera for testing"
assert state1.attributes["location"] == "Test location"
assert state1.attributes["type"] == "Road"
with patch(
"homeassistant.components.trafikverket_camera.coordinator.TrafikverketCamera.async_get_camera",
return_value=get_camera,
):
aioclient_mock.get(
"https://www.testurl.com/test_photo.jpg?type=fullsize",
content=b"9876543210",
)
async_fire_time_changed(
hass,
dt_util.utcnow() + timedelta(minutes=6),
)
await hass.async_block_till_done()
state1 = hass.states.get("camera.test_location")
assert state1.state == "idle"
assert state1.attributes != {}
assert await async_get_image(hass, "camera.test_location")
monkeypatch.setattr(
get_camera,
"photourl",
None,
)
aioclient_mock.get(
"https://www.testurl.com/test_photo.jpg?type=fullsize",
status=404,
)
with patch(
"homeassistant.components.trafikverket_camera.coordinator.TrafikverketCamera.async_get_camera",
return_value=get_camera,
):
async_fire_time_changed(
hass,
dt_util.utcnow() + timedelta(minutes=6),
)
await hass.async_block_till_done()
with pytest.raises(HomeAssistantError):
await async_get_image(hass, "camera.test_location")

234
tests/components/trafikverket_camera/test_config_flow.py Normal file
View File

@ -0,0 +1,234 @@
"""Test the Trafikverket Camera config flow."""
from __future__ import annotations
from unittest.mock import patch
import pytest
from pytrafikverket.exceptions import (
InvalidAuthentication,
MultipleCamerasFound,
NoCameraFound,
UnknownError,
)
from homeassistant import config_entries
from homeassistant.components.trafikverket_camera.const import CONF_LOCATION, DOMAIN
from homeassistant.const import CONF_API_KEY
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
async def test_form(hass: HomeAssistant) -> None:
"""Test we get the form."""
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_camera",
), patch(
"homeassistant.components.trafikverket_camera.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_API_KEY: "1234567890",
CONF_LOCATION: "Test location",
},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "Test location"
assert result2["data"] == {
"api_key": "1234567890",
"location": "Test location",
}
assert len(mock_setup_entry.mock_calls) == 1
assert result2["result"].unique_id == "trafikverket_camera-Test location"
@pytest.mark.parametrize(
("side_effect", "error_key", "base_error"),
[
(
InvalidAuthentication,
"base",
"invalid_auth",
),
(
NoCameraFound,
"location",
"invalid_location",
),
(
MultipleCamerasFound,
"location",
"more_locations",
),
(
UnknownError,
"base",
"cannot_connect",
),
],
)
async def test_flow_fails(
hass: HomeAssistant, side_effect: Exception, error_key: str, base_error: str
) -> None:
"""Test config flow errors."""
result4 = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result4["type"] == FlowResultType.FORM
assert result4["step_id"] == config_entries.SOURCE_USER
with patch(
"homeassistant.components.trafikverket_camera.config_flow.TrafikverketCamera.async_get_camera",
side_effect=side_effect,
):
result4 = await hass.config_entries.flow.async_configure(
result4["flow_id"],
user_input={
CONF_API_KEY: "1234567890",
CONF_LOCATION: "incorrect",
},
)
assert result4["errors"] == {error_key: base_error}
async def test_reauth_flow(hass: HomeAssistant) -> None:
"""Test a reauthentication flow."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_API_KEY: "1234567890",
CONF_LOCATION: "Test location",
},
unique_id="1234",
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": config_entries.SOURCE_REAUTH,
"unique_id": entry.unique_id,
"entry_id": entry.entry_id,
},
data=entry.data,
)
assert result["step_id"] == "reauth_confirm"
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {}
with patch(
"homeassistant.components.trafikverket_camera.config_flow.TrafikverketCamera.async_get_camera",
), patch(
"homeassistant.components.trafikverket_camera.async_setup_entry",
return_value=True,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_API_KEY: "1234567891"},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.ABORT
assert result2["reason"] == "reauth_successful"
assert entry.data == {
"api_key": "1234567891",
"location": "Test location",
}
@pytest.mark.parametrize(
("side_effect", "error_key", "p_error"),
[
(
InvalidAuthentication,
"base",
"invalid_auth",
),
(
NoCameraFound,
"location",
"invalid_location",
),
(
MultipleCamerasFound,
"location",
"more_locations",
),
(
UnknownError,
"base",
"cannot_connect",
),
],
)
async def test_reauth_flow_error(
hass: HomeAssistant, side_effect: Exception, error_key: str, p_error: str
) -> None:
"""Test a reauthentication flow with error."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_API_KEY: "1234567890",
CONF_LOCATION: "Test location",
},
unique_id="1234",
)
entry.add_to_hass(hass)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": config_entries.SOURCE_REAUTH,
"unique_id": entry.unique_id,
"entry_id": entry.entry_id,
},
data=entry.data,
)
with patch(
"homeassistant.components.trafikverket_camera.config_flow.TrafikverketCamera.async_get_camera",
side_effect=side_effect,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_API_KEY: "1234567890"},
)
await hass.async_block_till_done()
assert result2["step_id"] == "reauth_confirm"
assert result2["type"] == FlowResultType.FORM
assert result2["errors"] == {error_key: p_error}
with patch(
"homeassistant.components.trafikverket_camera.config_flow.TrafikverketCamera.async_get_camera",
), patch(
"homeassistant.components.trafikverket_camera.async_setup_entry",
return_value=True,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_API_KEY: "1234567891"},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.ABORT
assert result2["reason"] == "reauth_successful"
assert entry.data == {
"api_key": "1234567891",
"location": "Test location",
}

151
tests/components/trafikverket_camera/test_coordinator.py Normal file
View File

@ -0,0 +1,151 @@
"""The test for the Trafikverket Camera coordinator."""
from __future__ import annotations
from unittest.mock import patch
import pytest
from pytrafikverket.exceptions import (
InvalidAuthentication,
MultipleCamerasFound,
NoCameraFound,
UnknownError,
)
from homeassistant import config_entries
from homeassistant.components.trafikverket_camera.const import DOMAIN
from homeassistant.components.trafikverket_camera.coordinator import CameraData
from homeassistant.config_entries import SOURCE_USER
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import UpdateFailed
from . import ENTRY_CONFIG
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
async def test_coordinator(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
get_camera: CameraData,
) -> None:
"""Test the Trafikverket Camera coordinator."""
aioclient_mock.get(
"https://www.testurl.com/test_photo.jpg?type=fullsize", content=b"0123456789"
)
entry = MockConfigEntry(
domain=DOMAIN,
source=SOURCE_USER,
data=ENTRY_CONFIG,
entry_id="1",
unique_id="123",
title="Test location",
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.trafikverket_camera.coordinator.TrafikverketCamera.async_get_camera",
return_value=get_camera,
) as mock_data:
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
mock_data.assert_called_once()
state1 = hass.states.get("camera.test_location")
assert state1.state == "idle"
@pytest.mark.parametrize(
("sideeffect", "p_error", "entry_state"),
[
(
InvalidAuthentication,
ConfigEntryAuthFailed,
config_entries.ConfigEntryState.SETUP_ERROR,
),
(
NoCameraFound,
UpdateFailed,
config_entries.ConfigEntryState.SETUP_RETRY,
),
(
MultipleCamerasFound,
UpdateFailed,
config_entries.ConfigEntryState.SETUP_RETRY,
),
(
UnknownError,
UpdateFailed,
config_entries.ConfigEntryState.SETUP_RETRY,
),
],
)
async def test_coordinator_failed_update(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
get_camera: CameraData,
sideeffect: str,
p_error: Exception,
entry_state: str,
) -> None:
"""Test the Trafikverket Camera coordinator."""
aioclient_mock.get(
"https://www.testurl.com/test_photo.jpg?type=fullsize", content=b"0123456789"
)
entry = MockConfigEntry(
domain=DOMAIN,
source=SOURCE_USER,
data=ENTRY_CONFIG,
entry_id="1",
unique_id="123",
title="Test location",
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.trafikverket_camera.coordinator.TrafikverketCamera.async_get_camera",
side_effect=sideeffect,
) as mock_data:
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
mock_data.assert_called_once()
state = hass.states.get("camera.test_location")
assert state is None
assert entry.state == entry_state
async def test_coordinator_failed_get_image(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
get_camera: CameraData,
) -> None:
"""Test the Trafikverket Camera coordinator."""
aioclient_mock.get(
"https://www.testurl.com/test_photo.jpg?type=fullsize", status=404
)
entry = MockConfigEntry(
domain=DOMAIN,
source=SOURCE_USER,
data=ENTRY_CONFIG,
entry_id="1",
unique_id="123",
title="Test location",
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.trafikverket_camera.coordinator.TrafikverketCamera.async_get_camera",
return_value=get_camera,
) as mock_data:
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
mock_data.assert_called_once()
state = hass.states.get("camera.test_location")
assert state is None
assert entry.state is config_entries.ConfigEntryState.SETUP_RETRY

80
tests/components/trafikverket_camera/test_init.py Normal file
View File

@ -0,0 +1,80 @@
"""Test for Trafikverket Ferry component Init."""
from __future__ import annotations
from unittest.mock import patch
from pytrafikverket.trafikverket_camera import CameraInfo
from homeassistant import config_entries
from homeassistant.components.trafikverket_camera.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.core import HomeAssistant
from . import ENTRY_CONFIG
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
async def test_setup_entry(
hass: HomeAssistant,
get_camera: CameraInfo,
aioclient_mock: AiohttpClientMocker,
) -> None:
"""Test setup entry."""
aioclient_mock.get(
"https://www.testurl.com/test_photo.jpg?type=fullsize", content=b"0123456789"
)
entry = MockConfigEntry(
domain=DOMAIN,
source=SOURCE_USER,
data=ENTRY_CONFIG,
entry_id="1",
unique_id="123",
title="Test location",
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.trafikverket_camera.coordinator.TrafikverketCamera.async_get_camera",
return_value=get_camera,
) as mock_tvt_camera:
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is config_entries.ConfigEntryState.LOADED
assert len(mock_tvt_camera.mock_calls) == 1
async def test_unload_entry(
hass: HomeAssistant,
get_camera: CameraInfo,
aioclient_mock: AiohttpClientMocker,
) -> None:
"""Test unload an entry."""
aioclient_mock.get(
"https://www.testurl.com/test_photo.jpg?type=fullsize", content=b"0123456789"
)
entry = MockConfigEntry(
domain=DOMAIN,
source=SOURCE_USER,
data=ENTRY_CONFIG,
entry_id="1",
unique_id="321",
title="Test location",
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.trafikverket_camera.coordinator.TrafikverketCamera.async_get_camera",
return_value=get_camera,
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is config_entries.ConfigEntryState.LOADED
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is config_entries.ConfigEntryState.NOT_LOADED

46
tests/components/trafikverket_camera/test_recorder.py Normal file
View File

@ -0,0 +1,46 @@
"""The tests for Trafikcerket Camera recorder."""
from __future__ import annotations
import pytest
from pytrafikverket.trafikverket_camera import CameraInfo
from homeassistant.components.recorder import Recorder
from homeassistant.components.recorder.history import get_significant_states
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.util import dt as dt_util
from tests.components.recorder.common import async_wait_recording_done
from tests.test_util.aiohttp import AiohttpClientMocker
async def test_exclude_attributes(
recorder_mock: Recorder,
hass: HomeAssistant,
load_int: ConfigEntry,
monkeypatch: pytest.MonkeyPatch,
aioclient_mock: AiohttpClientMocker,
get_camera: CameraInfo,
) -> None:
"""Test camera has description and location excluded from recording."""
state1 = hass.states.get("camera.test_location")
assert state1.state == "idle"
assert state1.attributes["description"] == "Test Camera for testing"
assert state1.attributes["location"] == "Test location"
assert state1.attributes["type"] == "Road"
await async_wait_recording_done(hass)
states = await hass.async_add_executor_job(
get_significant_states,
hass,
dt_util.now(),
None,
hass.states.async_entity_ids(),
)
assert len(states) == 1
assert states.get("camera.test_location")
for entity_states in states.values():
for state in entity_states:
assert "location" not in state.attributes
assert "description" not in state.attributes
assert "type" in state.attributes