1
0
mirror of https://github.com/home-assistant/core.git synced 2026-04-02 00:20:30 +01:00

Add autoskope integration (#146772)

Co-authored-by: Franck Nijhof <git@frenck.dev>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
mcisk
2026-03-13 19:19:00 +01:00
committed by GitHub
parent 356de12bce
commit b6c7b2952e
20 changed files with 1138 additions and 0 deletions

2
CODEOWNERS generated
View File

@@ -186,6 +186,8 @@ build.json @home-assistant/supervisor
/tests/components/auth/ @home-assistant/core
/homeassistant/components/automation/ @home-assistant/core
/tests/components/automation/ @home-assistant/core
/homeassistant/components/autoskope/ @mcisk
/tests/components/autoskope/ @mcisk
/homeassistant/components/avea/ @pattyland
/homeassistant/components/awair/ @ahayworth @ricohageman
/tests/components/awair/ @ahayworth @ricohageman

View File

@@ -0,0 +1,53 @@
"""The Autoskope integration."""
from __future__ import annotations
import aiohttp
from autoskope_client.api import AutoskopeApi
from autoskope_client.models import CannotConnect, InvalidAuth
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_create_clientsession
from .const import DEFAULT_HOST
from .coordinator import AutoskopeConfigEntry, AutoskopeDataUpdateCoordinator
PLATFORMS: list[Platform] = [Platform.DEVICE_TRACKER]
async def async_setup_entry(hass: HomeAssistant, entry: AutoskopeConfigEntry) -> bool:
"""Set up Autoskope from a config entry."""
session = async_create_clientsession(hass, cookie_jar=aiohttp.CookieJar())
api = AutoskopeApi(
host=entry.data.get(CONF_HOST, DEFAULT_HOST),
username=entry.data[CONF_USERNAME],
password=entry.data[CONF_PASSWORD],
session=session,
)
try:
await api.connect()
except InvalidAuth as err:
# Raise ConfigEntryError until reauth flow is implemented (then ConfigEntryAuthFailed)
raise ConfigEntryError(
"Authentication failed, please check credentials"
) from err
except CannotConnect as err:
raise ConfigEntryNotReady("Could not connect to Autoskope API") from err
coordinator = AutoskopeDataUpdateCoordinator(hass, api, entry)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: AutoskopeConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

View File

@@ -0,0 +1,89 @@
"""Config flow for the Autoskope integration."""
from __future__ import annotations
from typing import Any
from autoskope_client.api import AutoskopeApi
from autoskope_client.models import CannotConnect, InvalidAuth
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.data_entry_flow import section
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.selector import (
TextSelector,
TextSelectorConfig,
TextSelectorType,
)
from .const import DEFAULT_HOST, DOMAIN, SECTION_ADVANCED_SETTINGS
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): TextSelector(
TextSelectorConfig(type=TextSelectorType.PASSWORD)
),
vol.Required(SECTION_ADVANCED_SETTINGS): section(
vol.Schema(
{
vol.Required(CONF_HOST, default=DEFAULT_HOST): TextSelector(
TextSelectorConfig(type=TextSelectorType.URL)
),
}
),
{"collapsed": True},
),
}
)
class AutoskopeConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Autoskope."""
VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
errors: dict[str, str] = {}
if user_input is not None:
username = user_input[CONF_USERNAME].lower()
host = user_input[SECTION_ADVANCED_SETTINGS][CONF_HOST].lower()
try:
cv.url(host)
except vol.Invalid:
errors["base"] = "invalid_url"
if not errors:
await self.async_set_unique_id(f"{username}@{host}")
self._abort_if_unique_id_configured()
try:
async with AutoskopeApi(
host=host,
username=username,
password=user_input[CONF_PASSWORD],
):
pass
except CannotConnect:
errors["base"] = "cannot_connect"
except InvalidAuth:
errors["base"] = "invalid_auth"
else:
return self.async_create_entry(
title=f"Autoskope ({username})",
data={
CONF_USERNAME: username,
CONF_PASSWORD: user_input[CONF_PASSWORD],
CONF_HOST: host,
},
)
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
)

View File

@@ -0,0 +1,9 @@
"""Constants for the Autoskope integration."""
from datetime import timedelta
DOMAIN = "autoskope"
DEFAULT_HOST = "https://portal.autoskope.de"
SECTION_ADVANCED_SETTINGS = "advanced_settings"
UPDATE_INTERVAL = timedelta(seconds=60)

View File

@@ -0,0 +1,60 @@
"""Data update coordinator for the Autoskope integration."""
from __future__ import annotations
import logging
from autoskope_client.api import AutoskopeApi
from autoskope_client.models import CannotConnect, InvalidAuth, Vehicle
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN, UPDATE_INTERVAL
_LOGGER = logging.getLogger(__name__)
type AutoskopeConfigEntry = ConfigEntry[AutoskopeDataUpdateCoordinator]
class AutoskopeDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Vehicle]]):
"""Class to manage fetching Autoskope data."""
config_entry: AutoskopeConfigEntry
def __init__(
self, hass: HomeAssistant, api: AutoskopeApi, entry: AutoskopeConfigEntry
) -> None:
"""Initialize the coordinator."""
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=UPDATE_INTERVAL,
config_entry=entry,
)
self.api = api
async def _async_update_data(self) -> dict[str, Vehicle]:
"""Fetch data from API endpoint."""
try:
vehicles = await self.api.get_vehicles()
return {vehicle.id: vehicle for vehicle in vehicles}
except InvalidAuth:
# Attempt to re-authenticate using stored credentials
try:
await self.api.authenticate()
# Retry the request after successful re-authentication
vehicles = await self.api.get_vehicles()
return {vehicle.id: vehicle for vehicle in vehicles}
except InvalidAuth as reauth_err:
raise ConfigEntryAuthFailed(
f"Authentication failed: {reauth_err}"
) from reauth_err
except CannotConnect as err:
raise UpdateFailed(f"Error communicating with API: {err}") from err

View File

@@ -0,0 +1,145 @@
"""Support for Autoskope device tracking."""
from __future__ import annotations
from autoskope_client.constants import MANUFACTURER
from autoskope_client.models import Vehicle
from homeassistant.components.device_tracker import SourceType, TrackerEntity
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import AutoskopeConfigEntry, AutoskopeDataUpdateCoordinator
PARALLEL_UPDATES = 0
async def async_setup_entry(
hass: HomeAssistant,
entry: AutoskopeConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Autoskope device tracker entities."""
coordinator: AutoskopeDataUpdateCoordinator = entry.runtime_data
tracked_vehicles: set[str] = set()
@callback
def update_entities() -> None:
"""Update entities based on coordinator data."""
current_vehicles = set(coordinator.data.keys())
vehicles_to_add = current_vehicles - tracked_vehicles
if vehicles_to_add:
new_entities = [
AutoskopeDeviceTracker(coordinator, vehicle_id)
for vehicle_id in vehicles_to_add
]
tracked_vehicles.update(vehicles_to_add)
async_add_entities(new_entities)
entry.async_on_unload(coordinator.async_add_listener(update_entities))
update_entities()
class AutoskopeDeviceTracker(
CoordinatorEntity[AutoskopeDataUpdateCoordinator], TrackerEntity
):
"""Representation of an Autoskope tracked device."""
_attr_has_entity_name = True
_attr_name: str | None = None
def __init__(
self, coordinator: AutoskopeDataUpdateCoordinator, vehicle_id: str
) -> None:
"""Initialize the TrackerEntity."""
super().__init__(coordinator)
self._vehicle_id = vehicle_id
self._attr_unique_id = vehicle_id
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
if (
self._vehicle_id in self.coordinator.data
and (device_entry := self.device_entry) is not None
and device_entry.name != self._vehicle_data.name
):
device_registry = dr.async_get(self.hass)
device_registry.async_update_device(
device_entry.id, name=self._vehicle_data.name
)
super()._handle_coordinator_update()
@property
def device_info(self) -> DeviceInfo:
"""Return device info for the vehicle."""
vehicle = self.coordinator.data[self._vehicle_id]
return DeviceInfo(
identifiers={(DOMAIN, str(vehicle.id))},
name=vehicle.name,
manufacturer=MANUFACTURER,
model=vehicle.model,
serial_number=vehicle.imei,
)
@property
def available(self) -> bool:
"""Return if entity is available."""
return (
super().available
and self.coordinator.data is not None
and self._vehicle_id in self.coordinator.data
)
@property
def _vehicle_data(self) -> Vehicle:
"""Return the vehicle data for the current entity."""
return self.coordinator.data[self._vehicle_id]
@property
def latitude(self) -> float | None:
"""Return latitude value of the device."""
if (vehicle := self._vehicle_data) and vehicle.position:
return float(vehicle.position.latitude)
return None
@property
def longitude(self) -> float | None:
"""Return longitude value of the device."""
if (vehicle := self._vehicle_data) and vehicle.position:
return float(vehicle.position.longitude)
return None
@property
def source_type(self) -> SourceType:
"""Return the source type of the device."""
return SourceType.GPS
@property
def location_accuracy(self) -> float:
"""Return the location accuracy of the device in meters."""
if (vehicle := self._vehicle_data) and vehicle.gps_quality:
if vehicle.gps_quality > 0:
# HDOP to estimated accuracy in meters
# HDOP of 1-2 = good (5-10m), 2-5 = moderate (10-25m), >5 = poor (>25m)
return float(max(5, int(vehicle.gps_quality * 5.0)))
return 0.0
@property
def icon(self) -> str:
"""Return the icon based on the vehicle's activity."""
if self._vehicle_id not in self.coordinator.data:
return "mdi:car-clock"
vehicle = self._vehicle_data
if vehicle.position:
if vehicle.position.park_mode:
return "mdi:car-brake-parking"
if vehicle.position.speed > 5: # Moving threshold: 5 km/h
return "mdi:car-arrow-right"
return "mdi:car"
return "mdi:car-clock"

View File

@@ -0,0 +1,11 @@
{
"domain": "autoskope",
"name": "Autoskope",
"codeowners": ["@mcisk"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/autoskope",
"integration_type": "hub",
"iot_class": "cloud_polling",
"quality_scale": "bronze",
"requirements": ["autoskope_client==1.4.1"]
}

View File

@@ -0,0 +1,88 @@
# + in comment indicates requirement for quality scale
# - in comment indicates issue to be fixed, not impacting quality scale
rules:
# Bronze
action-setup:
status: exempt
comment: |
Integration does not provide custom services.
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: |
Integration does not provide custom services.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup: done
entity-unique-id: done
has-entity-name: done
runtime-data: done
test-before-configure: done
test-before-setup: done
unique-config-entry: done
# Silver
action-exceptions:
status: exempt
comment: |
Integration does not provide custom services.
config-entry-unloading: done
docs-configuration-parameters: done
docs-installation-parameters: done
entity-unavailable: done
integration-owner: done
log-when-unavailable: todo
parallel-updates: done
reauthentication-flow:
status: todo
comment: |
Reauthentication flow removed for initial PR, will be added in follow-up.
test-coverage: done
# Gold
devices: done
diagnostics: todo
discovery-update-info:
status: exempt
comment: |
Integration does not use discovery. Autoskope devices use NB-IoT/LTE-M (via IoT SIMs) and LoRaWAN.
discovery:
status: exempt
comment: |
Integration does not use discovery. Autoskope devices use NB-IoT/LTE-M (via IoT SIMs) and LoRaWAN.
docs-data-update: done
docs-examples: done
docs-known-limitations: done
docs-supported-devices: done
docs-supported-functions: done
docs-troubleshooting: done
docs-use-cases: done
dynamic-devices: done
entity-category: done
entity-device-class: done
entity-disabled-by-default:
status: exempt
comment: |
Only one entity type (device_tracker) is created, making this not applicable.
entity-translations: done
exception-translations: done
icon-translations: done
reconfiguration-flow:
status: todo
comment: |
Reconfiguration flow removed for initial PR, will be added in follow-up.
repair-issues: todo
stale-devices: done
# Platinum
async-dependency: done
inject-websession: done
strict-typing:
status: todo
comment: |
Integration needs to be added to .strict-typing file for full compliance.

View File

@@ -0,0 +1,52 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"invalid_url": "Invalid URL",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"step": {
"user": {
"data": {
"password": "[%key:common::config_flow::data::password%]",
"username": "[%key:common::config_flow::data::username%]"
},
"data_description": {
"password": "The password for your Autoskope account.",
"username": "The username for your Autoskope account."
},
"description": "Enter your Autoskope credentials.",
"sections": {
"advanced_settings": {
"data": {
"host": "API endpoint"
},
"data_description": {
"host": "The URL of your Autoskope API endpoint. Only change this if you use a white-label portal."
},
"name": "Advanced settings"
}
},
"title": "Connect to Autoskope"
}
}
},
"issues": {
"cannot_connect": {
"description": "Home Assistant could not connect to the Autoskope API at {host}. Please check the connection details and ensure the API endpoint is reachable.\n\nError: {error}",
"title": "Failed to connect to Autoskope"
},
"invalid_auth": {
"description": "Authentication with Autoskope failed for user {username}. Please re-authenticate the integration with the correct password.",
"title": "Invalid Autoskope authentication"
},
"low_battery": {
"description": "The battery voltage for vehicle {vehicle_name} ({vehicle_id}) is low ({value}V). Consider checking or replacing the battery.",
"title": "Low vehicle battery ({vehicle_name})"
}
}
}

View File

@@ -82,6 +82,7 @@ FLOWS = {
"aurora_abb_powerone",
"aussie_broadband",
"autarco",
"autoskope",
"awair",
"aws_s3",
"axis",

View File

@@ -647,6 +647,12 @@
"config_flow": true,
"iot_class": "cloud_polling"
},
"autoskope": {
"name": "Autoskope",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling"
},
"avion": {
"name": "Avi-on",
"integration_type": "hub",

3
requirements_all.txt generated
View File

@@ -579,6 +579,9 @@ autarco==3.2.0
# homeassistant.components.husqvarna_automower_ble
automower-ble==0.2.8
# homeassistant.components.autoskope
autoskope_client==1.4.1
# homeassistant.components.generic
# homeassistant.components.stream
av==16.0.1

View File

@@ -537,6 +537,9 @@ autarco==3.2.0
# homeassistant.components.husqvarna_automower_ble
automower-ble==0.2.8
# homeassistant.components.autoskope
autoskope_client==1.4.1
# homeassistant.components.generic
# homeassistant.components.stream
av==16.0.1

View File

@@ -0,0 +1,12 @@
"""Tests for Autoskope integration."""
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
"""Set up the Autoskope integration."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()

View File

@@ -0,0 +1,69 @@
"""Test fixtures for Autoskope integration."""
from collections.abc import Generator
from json import loads
from unittest.mock import AsyncMock, patch
from autoskope_client.models import Vehicle
import pytest
from homeassistant.components.autoskope.const import DEFAULT_HOST, DOMAIN
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from tests.common import MockConfigEntry, load_fixture
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return a mock config entry."""
return MockConfigEntry(
domain=DOMAIN,
title="Autoskope (test_user)",
data={
CONF_USERNAME: "test_user",
CONF_PASSWORD: "test_password",
CONF_HOST: DEFAULT_HOST,
},
unique_id=f"test_user@{DEFAULT_HOST}",
entry_id="01AUTOSKOPE_TEST_ENTRY",
)
@pytest.fixture
def mock_vehicles() -> list[Vehicle]:
"""Return a list of mock vehicles from fixture data."""
data = loads(load_fixture("vehicles.json", DOMAIN))
return [
Vehicle.from_api(vehicle, data["positions"]) for vehicle in data["vehicles"]
]
@pytest.fixture
def mock_autoskope_client(mock_vehicles: list[Vehicle]) -> Generator[AsyncMock]:
"""Mock the Autoskope API client."""
with (
patch(
"homeassistant.components.autoskope.AutoskopeApi",
autospec=True,
) as mock_client,
patch(
"homeassistant.components.autoskope.config_flow.AutoskopeApi",
new=mock_client,
),
):
client = mock_client.return_value
client.connect.return_value = None
client.get_vehicles.return_value = mock_vehicles
client.__aenter__.return_value = client
client.__aexit__.return_value = None
yield client
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.autoskope.async_setup_entry",
return_value=True,
) as mock_setup:
yield mock_setup

View File

@@ -0,0 +1,33 @@
{
"vehicles": [
{
"id": "12345",
"name": "Test Vehicle",
"ex_pow": 12.5,
"bat_pow": 3.7,
"hdop": 1.2,
"support_infos": {
"imei": "123456789012345"
},
"model": "Autoskope"
}
],
"positions": {
"type": "FeatureCollection",
"features": [
{
"type": "Feature",
"geometry": {
"type": "Point",
"coordinates": [8.6821267, 50.1109221]
},
"properties": {
"carid": "12345",
"s": 0,
"dt": "2025-05-28T10:00:00Z",
"park": false
}
}
]
}
}

View File

@@ -0,0 +1,55 @@
# serializer version: 1
# name: test_all_entities[device_tracker.test_vehicle-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'device_tracker',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'device_tracker.test_vehicle',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': 'mdi:car',
'original_name': None,
'platform': 'autoskope',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '12345',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[device_tracker.test_vehicle-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Test Vehicle',
'gps_accuracy': 6.0,
'icon': 'mdi:car',
'latitude': 50.1109221,
'longitude': 8.6821267,
'source_type': <SourceType.GPS: 'gps'>,
}),
'context': <ANY>,
'entity_id': 'device_tracker.test_vehicle',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'not_home',
})
# ---

View File

@@ -0,0 +1,167 @@
"""Test Autoskope config flow."""
from unittest.mock import AsyncMock
from autoskope_client.models import CannotConnect, InvalidAuth
import pytest
from homeassistant.components.autoskope.const import (
DEFAULT_HOST,
DOMAIN,
SECTION_ADVANCED_SETTINGS,
)
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
USER_INPUT = {
CONF_USERNAME: "test_user",
CONF_PASSWORD: "test_password",
SECTION_ADVANCED_SETTINGS: {
CONF_HOST: DEFAULT_HOST,
},
}
async def test_full_flow(
hass: HomeAssistant,
mock_autoskope_client: AsyncMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test full user config flow from form to entry creation."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Autoskope (test_user)"
assert result["data"] == {
CONF_USERNAME: "test_user",
CONF_PASSWORD: "test_password",
CONF_HOST: DEFAULT_HOST,
}
assert result["result"].unique_id == f"test_user@{DEFAULT_HOST}"
@pytest.mark.parametrize(
("exception", "error"),
[
(InvalidAuth("Invalid credentials"), "invalid_auth"),
(CannotConnect("Connection failed"), "cannot_connect"),
],
)
async def test_flow_errors(
hass: HomeAssistant,
mock_autoskope_client: AsyncMock,
mock_setup_entry: AsyncMock,
exception: Exception,
error: str,
) -> None:
"""Test config flow error handling with recovery."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
mock_autoskope_client.__aenter__.side_effect = exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT,
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": error}
# Recovery: clear the error and retry
mock_autoskope_client.__aenter__.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_flow_invalid_url(
hass: HomeAssistant,
mock_autoskope_client: AsyncMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test config flow rejects invalid URL with recovery."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test_user",
CONF_PASSWORD: "test_password",
SECTION_ADVANCED_SETTINGS: {
CONF_HOST: "not-a-valid-url",
},
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "invalid_url"}
# Recovery: provide a valid URL
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_already_configured(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_autoskope_client: AsyncMock,
) -> None:
"""Test aborting if already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_custom_host(
hass: HomeAssistant,
mock_autoskope_client: AsyncMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test config flow with a custom white-label host."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test_user",
CONF_PASSWORD: "test_password",
SECTION_ADVANCED_SETTINGS: {
CONF_HOST: "https://custom.autoskope.server",
},
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"][CONF_HOST] == "https://custom.autoskope.server"
assert result["result"].unique_id == "test_user@https://custom.autoskope.server"

View File

@@ -0,0 +1,232 @@
"""Test Autoskope device tracker."""
from unittest.mock import AsyncMock
from autoskope_client.models import CannotConnect, InvalidAuth, Vehicle, VehiclePosition
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.autoskope.const import DOMAIN, UPDATE_INTERVAL
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
async def test_all_entities(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_autoskope_client: AsyncMock,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test all entities with snapshot."""
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.parametrize(
("speed", "park_mode", "has_position", "expected_icon"),
[
(50, False, True, "mdi:car-arrow-right"),
(0, True, True, "mdi:car-brake-parking"),
(2, False, True, "mdi:car"),
(0, False, False, "mdi:car-clock"),
],
ids=["moving", "parked", "idle", "no_position"],
)
async def test_vehicle_icons(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_autoskope_client: AsyncMock,
speed: int,
park_mode: bool,
has_position: bool,
expected_icon: str,
) -> None:
"""Test device tracker icon for different vehicle states."""
position = (
VehiclePosition(
latitude=50.1109221,
longitude=8.6821267,
speed=speed,
timestamp="2025-05-28T10:00:00Z",
park_mode=park_mode,
)
if has_position
else None
)
mock_autoskope_client.get_vehicles.return_value = [
Vehicle(
id="12345",
name="Test Vehicle",
position=position,
external_voltage=12.5,
battery_voltage=3.7,
gps_quality=1.2,
imei="123456789012345",
model="Autoskope",
)
]
await setup_integration(hass, mock_config_entry)
state = hass.states.get("device_tracker.test_vehicle")
assert state is not None
assert state.attributes["icon"] == expected_icon
async def test_entity_unavailable_on_coordinator_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_autoskope_client: AsyncMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test entity becomes unavailable when coordinator update fails."""
await setup_integration(hass, mock_config_entry)
state = hass.states.get("device_tracker.test_vehicle")
assert state is not None
assert state.state != STATE_UNAVAILABLE
# Simulate connection error on next update
mock_autoskope_client.get_vehicles.side_effect = CannotConnect("Connection lost")
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
state = hass.states.get("device_tracker.test_vehicle")
assert state is not None
assert state.state == STATE_UNAVAILABLE
async def test_entity_recovers_after_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_autoskope_client: AsyncMock,
mock_vehicles: list[Vehicle],
freezer: FrozenDateTimeFactory,
) -> None:
"""Test entity recovers after a transient coordinator error."""
await setup_integration(hass, mock_config_entry)
# Simulate error
mock_autoskope_client.get_vehicles.side_effect = CannotConnect("Connection lost")
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get("device_tracker.test_vehicle").state == STATE_UNAVAILABLE
# Recover
mock_autoskope_client.get_vehicles.side_effect = None
mock_autoskope_client.get_vehicles.return_value = mock_vehicles
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
state = hass.states.get("device_tracker.test_vehicle")
assert state.state != STATE_UNAVAILABLE
async def test_reauth_success(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_autoskope_client: AsyncMock,
mock_vehicles: list[Vehicle],
freezer: FrozenDateTimeFactory,
) -> None:
"""Test entity stays available after successful re-authentication."""
await setup_integration(hass, mock_config_entry)
# First get_vehicles raises InvalidAuth, retry after authenticate succeeds
mock_autoskope_client.get_vehicles.side_effect = [
InvalidAuth("Token expired"),
mock_vehicles,
]
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
state = hass.states.get("device_tracker.test_vehicle")
assert state is not None
assert state.state != STATE_UNAVAILABLE
async def test_reauth_failure(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_autoskope_client: AsyncMock,
mock_vehicles: list[Vehicle],
freezer: FrozenDateTimeFactory,
) -> None:
"""Test entity becomes unavailable on permanent auth failure."""
await setup_integration(hass, mock_config_entry)
# get_vehicles raises InvalidAuth, and re-authentication also fails
mock_autoskope_client.get_vehicles.side_effect = InvalidAuth("Token expired")
mock_autoskope_client.authenticate.side_effect = InvalidAuth("Invalid credentials")
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
state = hass.states.get("device_tracker.test_vehicle")
assert state is not None
assert state.state == STATE_UNAVAILABLE
# Clean up side effects to prevent teardown errors
mock_autoskope_client.get_vehicles.side_effect = None
mock_autoskope_client.authenticate.side_effect = None
mock_autoskope_client.get_vehicles.return_value = mock_vehicles
async def test_vehicle_name_update(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_autoskope_client: AsyncMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test device name updates in device registry when vehicle is renamed."""
await setup_integration(hass, mock_config_entry)
device_registry = dr.async_get(hass)
device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "12345")})
assert device_entry is not None
assert device_entry.name == "Test Vehicle"
# Simulate vehicle rename on Autoskope side
mock_autoskope_client.get_vehicles.return_value = [
Vehicle(
id="12345",
name="Renamed Vehicle",
position=VehiclePosition(
latitude=50.1109221,
longitude=8.6821267,
speed=0,
timestamp="2025-05-28T10:00:00Z",
park_mode=True,
),
external_voltage=12.5,
battery_voltage=3.7,
gps_quality=1.2,
imei="123456789012345",
model="Autoskope",
)
]
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Device registry should reflect the new name
device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "12345")})
assert device_entry is not None
assert device_entry.name == "Renamed Vehicle"

View File

@@ -0,0 +1,48 @@
"""Test Autoskope integration setup."""
from unittest.mock import AsyncMock
from autoskope_client.models import CannotConnect, InvalidAuth
import pytest
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from . import setup_integration
from tests.common import MockConfigEntry
async def test_setup_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_autoskope_client: AsyncMock,
) -> None:
"""Test successful setup and unload of entry."""
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.LOADED
await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
@pytest.mark.parametrize(
("exception", "expected_state"),
[
(InvalidAuth("Invalid credentials"), ConfigEntryState.SETUP_ERROR),
(CannotConnect("Connection failed"), ConfigEntryState.SETUP_RETRY),
],
)
async def test_setup_entry_errors(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_autoskope_client: AsyncMock,
exception: Exception,
expected_state: ConfigEntryState,
) -> None:
"""Test setup with authentication and connection errors."""
mock_autoskope_client.connect.side_effect = exception
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is expected_state