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

Add Lojack integration (#162047)

Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: Joostlek <joostlek@outlook.com>
This commit is contained in:
Devin Slick
2026-03-16 16:09:10 -05:00
committed by GitHub
parent 3580fab26e
commit 2042f2e2bd
21 changed files with 1050 additions and 0 deletions

2
CODEOWNERS generated
View File

@@ -974,6 +974,8 @@ build.json @home-assistant/supervisor
/tests/components/logbook/ @home-assistant/core
/homeassistant/components/logger/ @home-assistant/core
/tests/components/logger/ @home-assistant/core
/homeassistant/components/lojack/ @devinslick
/tests/components/lojack/ @devinslick
/homeassistant/components/london_underground/ @jpbede
/tests/components/london_underground/ @jpbede
/homeassistant/components/lookin/ @ANMalko @bdraco

View File

@@ -0,0 +1,78 @@
"""The LoJack integration for Home Assistant."""
from __future__ import annotations
from dataclasses import dataclass, field
from lojack_api import ApiError, AuthenticationError, LoJackClient, Vehicle
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .coordinator import LoJackCoordinator
PLATFORMS: list[Platform] = [Platform.DEVICE_TRACKER]
@dataclass
class LoJackData:
"""Runtime data for a LoJack config entry."""
client: LoJackClient
coordinators: list[LoJackCoordinator] = field(default_factory=list)
type LoJackConfigEntry = ConfigEntry[LoJackData]
async def async_setup_entry(hass: HomeAssistant, entry: LoJackConfigEntry) -> bool:
"""Set up LoJack from a config entry."""
session = async_get_clientsession(hass)
try:
client = await LoJackClient.create(
entry.data[CONF_USERNAME],
entry.data[CONF_PASSWORD],
session=session,
)
except AuthenticationError as err:
raise ConfigEntryAuthFailed(f"Authentication failed: {err}") from err
except ApiError as err:
raise ConfigEntryNotReady(f"API error during setup: {err}") from err
try:
vehicles = await client.list_devices()
except AuthenticationError as err:
await client.close()
raise ConfigEntryAuthFailed(f"Authentication failed: {err}") from err
except ApiError as err:
await client.close()
raise ConfigEntryNotReady(f"API error during setup: {err}") from err
data = LoJackData(client=client)
entry.runtime_data = data
try:
for vehicle in vehicles or []:
if isinstance(vehicle, Vehicle):
coordinator = LoJackCoordinator(hass, client, entry, vehicle)
await coordinator.async_config_entry_first_refresh()
data.coordinators.append(coordinator)
except Exception:
await client.close()
raise
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: LoJackConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
await entry.runtime_data.client.close()
return unload_ok

View File

@@ -0,0 +1,111 @@
"""Config flow for LoJack integration."""
from __future__ import annotations
from collections.abc import Mapping
import logging
from typing import Any
from lojack_api import ApiError, AuthenticationError, LoJackClient
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
}
)
class LoJackConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for LoJack."""
VERSION = 1
MINOR_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:
try:
async with await LoJackClient.create(
user_input[CONF_USERNAME],
user_input[CONF_PASSWORD],
session=async_get_clientsession(self.hass),
) as client:
user_id = client.user_id
except AuthenticationError:
errors["base"] = "invalid_auth"
except ApiError:
errors["base"] = "cannot_connect"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
if not user_id:
errors["base"] = "unknown"
else:
await self.async_set_unique_id(user_id)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=f"LoJack ({user_input[CONF_USERNAME]})",
data=user_input,
)
return self.async_show_form(
step_id="user",
data_schema=STEP_USER_DATA_SCHEMA,
errors=errors,
)
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Handle reauthentication."""
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle reauthentication confirmation."""
errors: dict[str, str] = {}
reauth_entry = self._get_reauth_entry()
if user_input is not None:
try:
async with await LoJackClient.create(
reauth_entry.data[CONF_USERNAME],
user_input[CONF_PASSWORD],
session=async_get_clientsession(self.hass),
):
pass
except AuthenticationError:
errors["base"] = "invalid_auth"
except ApiError:
errors["base"] = "cannot_connect"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
return self.async_update_reload_and_abort(
reauth_entry,
data_updates={CONF_PASSWORD: user_input[CONF_PASSWORD]},
)
return self.async_show_form(
step_id="reauth_confirm",
data_schema=vol.Schema({vol.Required(CONF_PASSWORD): str}),
description_placeholders={CONF_USERNAME: reauth_entry.data[CONF_USERNAME]},
errors=errors,
)

View File

@@ -0,0 +1,13 @@
"""Constants for the LoJack integration."""
from __future__ import annotations
import logging
from typing import Final
DOMAIN: Final = "lojack"
LOGGER = logging.getLogger(__package__)
# Default polling interval (in minutes)
DEFAULT_UPDATE_INTERVAL: Final = 5

View File

@@ -0,0 +1,68 @@
"""Data update coordinator for the LoJack integration."""
from __future__ import annotations
from datetime import timedelta
from typing import TYPE_CHECKING
from lojack_api import ApiError, AuthenticationError, LoJackClient
from lojack_api.device import Vehicle
from lojack_api.models import Location
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 DEFAULT_UPDATE_INTERVAL, DOMAIN, LOGGER
if TYPE_CHECKING:
from . import LoJackConfigEntry
def get_device_name(vehicle: Vehicle) -> str:
"""Get a human-readable name for a vehicle."""
parts = [
str(vehicle.year) if vehicle.year else None,
vehicle.make,
vehicle.model,
]
name = " ".join(p for p in parts if p)
return name or vehicle.name or "Vehicle"
class LoJackCoordinator(DataUpdateCoordinator[Location]):
"""Class to manage fetching LoJack data for a single vehicle."""
config_entry: LoJackConfigEntry
def __init__(
self,
hass: HomeAssistant,
client: LoJackClient,
entry: ConfigEntry,
vehicle: Vehicle,
) -> None:
"""Initialize the coordinator."""
self.client = client
self.vehicle = vehicle
super().__init__(
hass,
LOGGER,
name=f"{DOMAIN}_{vehicle.id}",
update_interval=timedelta(minutes=DEFAULT_UPDATE_INTERVAL),
config_entry=entry,
)
async def _async_update_data(self) -> Location:
"""Fetch location data for this vehicle."""
try:
location = await self.vehicle.get_location(force=True)
except AuthenticationError as err:
raise ConfigEntryAuthFailed(f"Authentication failed: {err}") from err
except ApiError as err:
raise UpdateFailed(f"Error fetching data: {err}") from err
if location is None:
raise UpdateFailed("No location data available")
return location

View File

@@ -0,0 +1,78 @@
"""Device tracker platform for LoJack integration."""
from __future__ import annotations
from homeassistant.components.device_tracker import SourceType, TrackerEntity
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import LoJackConfigEntry
from .const import DOMAIN
from .coordinator import LoJackCoordinator, get_device_name
PARALLEL_UPDATES = 0
async def async_setup_entry(
hass: HomeAssistant,
entry: LoJackConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up LoJack device tracker from a config entry."""
async_add_entities(
LoJackDeviceTracker(coordinator)
for coordinator in entry.runtime_data.coordinators
)
class LoJackDeviceTracker(CoordinatorEntity[LoJackCoordinator], TrackerEntity):
"""Representation of a LoJack device tracker."""
_attr_has_entity_name = True
_attr_name = None # Main entity of the device, uses device name directly
def __init__(self, coordinator: LoJackCoordinator) -> None:
"""Initialize the device tracker."""
super().__init__(coordinator)
self._attr_unique_id = coordinator.vehicle.id
@property
def device_info(self) -> DeviceInfo:
"""Return the device info."""
return DeviceInfo(
identifiers={(DOMAIN, self.coordinator.vehicle.id)},
name=get_device_name(self.coordinator.vehicle),
manufacturer="Spireon LoJack",
model=self.coordinator.vehicle.model,
serial_number=self.coordinator.vehicle.vin,
)
@property
def source_type(self) -> SourceType:
"""Return the source type of the device."""
return SourceType.GPS
@property
def latitude(self) -> float | None:
"""Return the latitude of the device."""
return self.coordinator.data.latitude
@property
def longitude(self) -> float | None:
"""Return the longitude of the device."""
return self.coordinator.data.longitude
@property
def location_accuracy(self) -> int:
"""Return the location accuracy of the device."""
if self.coordinator.data.accuracy is not None:
return int(self.coordinator.data.accuracy)
return 0
@property
def battery_level(self) -> int | None:
"""Return the battery level of the device (if applicable)."""
# LoJack devices report vehicle battery voltage, not percentage
return None

View File

@@ -0,0 +1,12 @@
{
"domain": "lojack",
"name": "LoJack",
"codeowners": ["@devinslick"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/lojack",
"integration_type": "hub",
"iot_class": "cloud_polling",
"loggers": ["lojack_api"],
"quality_scale": "silver",
"requirements": ["lojack-api==0.7.1"]
}

View File

@@ -0,0 +1,81 @@
rules:
# Bronze
action-setup:
status: exempt
comment: This integration does not provide actions.
appropriate-polling: done
brands: done
common-modules: done
config-flow: done
config-flow-test-coverage: done
dependency-transparency: done
docs-actions:
status: exempt
comment: This integration does not provide actions.
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: This integration does not provide actions.
config-entry-unloading: done
docs-configuration-parameters:
status: exempt
comment: This integration does not provide an options flow.
docs-installation-parameters:
status: done
comment: Documented in https://github.com/home-assistant/home-assistant.io/pull/43463
entity-unavailable: done
integration-owner: done
log-when-unavailable: done
parallel-updates: done
reauthentication-flow: done
test-coverage: done
# Gold
devices: done
diagnostics: todo
discovery:
status: exempt
comment: This is a cloud polling integration with no local discovery mechanism since the devices are not on a local network.
discovery-update-info:
status: exempt
comment: This is a cloud polling integration with no local discovery mechanism.
docs-data-update: todo
docs-examples: todo
docs-known-limitations: todo
docs-supported-devices: todo
docs-supported-functions: todo
docs-troubleshooting: todo
docs-use-cases: todo
dynamic-devices:
status: exempt
comment: Vehicles are tied to the user account. Changes require integration reload.
entity-category: done
entity-device-class: done
entity-disabled-by-default:
status: exempt
comment: The device tracker entity is the primary entity and should be enabled by default.
entity-translations: done
exception-translations: todo
icon-translations: todo
reconfiguration-flow: todo
repair-issues:
status: exempt
comment: No user-actionable repair scenarios identified for this integration.
stale-devices:
status: exempt
comment: Vehicles removed from the LoJack account stop appearing in API responses and become unavailable.
# Platinum
async-dependency: done
inject-websession: done
strict-typing: todo

View File

@@ -0,0 +1,38 @@
{
"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%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"initiate_flow": {
"user": "[%key:common::config_flow::initiate_flow::account%]"
},
"step": {
"reauth_confirm": {
"data": {
"password": "[%key:common::config_flow::data::password%]"
},
"data_description": {
"password": "[%key:component::lojack::config::step::user::data_description::password%]"
},
"description": "Re-enter the password for {username}."
},
"user": {
"data": {
"password": "[%key:common::config_flow::data::password%]",
"username": "[%key:common::config_flow::data::username%]"
},
"data_description": {
"password": "Your LoJack/Spireon account password",
"username": "Your LoJack/Spireon account email address"
},
"description": "Enter your LoJack/Spireon account credentials."
}
}
}
}

View File

@@ -399,6 +399,7 @@ FLOWS = {
"local_ip",
"local_todo",
"locative",
"lojack",
"london_underground",
"lookin",
"loqed",

View File

@@ -3828,6 +3828,12 @@
}
}
},
"lojack": {
"name": "LoJack",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling"
},
"london_air": {
"name": "London Air",
"integration_type": "hub",

3
requirements_all.txt generated
View File

@@ -1448,6 +1448,9 @@ livisi==0.0.25
# homeassistant.components.google_maps
locationsharinglib==5.0.1
# homeassistant.components.lojack
lojack-api==0.7.1
# homeassistant.components.london_underground
london-tube-status==0.5

View File

@@ -1267,6 +1267,9 @@ libsoundtouch==0.8
# homeassistant.components.livisi
livisi==0.0.25
# homeassistant.components.lojack
lojack-api==0.7.1
# homeassistant.components.london_underground
london-tube-status==0.5

View File

@@ -0,0 +1,12 @@
"""Tests for the LoJack integration."""
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
"""Set up the LoJack integration for testing."""
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,107 @@
"""Test fixtures for the LoJack integration."""
from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, create_autospec, patch
from lojack_api import LoJackClient
from lojack_api.device import Vehicle
from lojack_api.models import Location
import pytest
from homeassistant.components.lojack.const import DOMAIN
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from .const import (
TEST_ACCURACY,
TEST_ADDRESS,
TEST_DEVICE_ID,
TEST_DEVICE_NAME,
TEST_HEADING,
TEST_LATITUDE,
TEST_LONGITUDE,
TEST_MAKE,
TEST_MODEL,
TEST_PASSWORD,
TEST_TIMESTAMP,
TEST_USER_ID,
TEST_USERNAME,
TEST_VIN,
TEST_YEAR,
)
from tests.common import MockConfigEntry
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return a mock config entry."""
return MockConfigEntry(
domain=DOMAIN,
unique_id=TEST_USER_ID,
data={
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
},
title=f"LoJack ({TEST_USERNAME})",
)
@pytest.fixture
def mock_location() -> Location:
"""Return a mock LoJack location."""
return Location(
latitude=TEST_LATITUDE,
longitude=TEST_LONGITUDE,
accuracy=TEST_ACCURACY,
heading=TEST_HEADING,
address=TEST_ADDRESS,
timestamp=TEST_TIMESTAMP,
)
@pytest.fixture
def mock_device(mock_location: Location) -> MagicMock:
"""Return a mock LoJack device."""
device = create_autospec(Vehicle, instance=True)
device.id = TEST_DEVICE_ID
device.name = TEST_DEVICE_NAME
device.vin = TEST_VIN
device.make = TEST_MAKE
device.model = TEST_MODEL
device.year = TEST_YEAR
device.get_location = AsyncMock(return_value=mock_location)
return device
@pytest.fixture
def mock_lojack_client(
mock_device: MagicMock,
) -> Generator[MagicMock]:
"""Return a mock LoJack client."""
client = create_autospec(LoJackClient, instance=True)
client.user_id = TEST_USER_ID
client.list_devices = AsyncMock(return_value=[mock_device])
client.__aenter__ = AsyncMock(return_value=client)
client.__aexit__ = AsyncMock(return_value=False)
with (
patch(
"homeassistant.components.lojack.LoJackClient.create",
return_value=client,
),
patch(
"homeassistant.components.lojack.config_flow.LoJackClient.create",
return_value=client,
),
):
yield client
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Mock async_setup_entry."""
with patch(
"homeassistant.components.lojack.async_setup_entry",
return_value=True,
) as mock:
yield mock

View File

@@ -0,0 +1,20 @@
"""Constants for the LoJack integration tests."""
from datetime import datetime
TEST_USERNAME = "test@example.com"
TEST_PASSWORD = "testpassword123"
TEST_DEVICE_ID = "12345"
TEST_DEVICE_NAME = "My Car"
TEST_VIN = "1HGBH41JXMN109186"
TEST_MAKE = "Honda"
TEST_MODEL = "Accord"
TEST_YEAR = 2021
TEST_USER_ID = "user_abc123"
TEST_LATITUDE = 37.7749
TEST_LONGITUDE = -122.4194
TEST_ACCURACY = 10.5
TEST_HEADING = 180.0
TEST_ADDRESS = "123 Main St, San Francisco, CA 94102"
TEST_TIMESTAMP = datetime.fromisoformat("2020-02-02T14:00:00Z")

View File

@@ -0,0 +1,54 @@
# serializer version: 1
# name: test_all_entities[device_tracker.2021_honda_accord-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.2021_honda_accord',
'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': None,
'original_name': None,
'platform': 'lojack',
'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.2021_honda_accord-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': '2021 Honda Accord',
'gps_accuracy': 10,
'latitude': 37.7749,
'longitude': -122.4194,
'source_type': <SourceType.GPS: 'gps'>,
}),
'context': <ANY>,
'entity_id': 'device_tracker.2021_honda_accord',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'not_home',
})
# ---

View File

@@ -0,0 +1,32 @@
# serializer version: 1
# name: test_device_info
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'config_entries_subentries': <ANY>,
'configuration_url': None,
'connections': set({
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'lojack',
'12345',
),
}),
'labels': set({
}),
'manufacturer': 'Spireon LoJack',
'model': 'Accord',
'model_id': None,
'name': '2021 Honda Accord',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': '1HGBH41JXMN109186',
'sw_version': None,
'via_device_id': None,
})
# ---

View File

@@ -0,0 +1,119 @@
"""Tests for the LoJack config flow."""
from unittest.mock import AsyncMock, MagicMock, patch
from lojack_api import ApiError, AuthenticationError
import pytest
from homeassistant.components.lojack.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .const import TEST_PASSWORD, TEST_USER_ID, TEST_USERNAME
from tests.common import MockConfigEntry
async def test_full_user_flow(
hass: HomeAssistant,
mock_lojack_client: MagicMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test the full user configuration flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == f"LoJack ({TEST_USERNAME})"
assert result["data"] == {
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
}
assert result["result"].unique_id == TEST_USER_ID
@pytest.mark.parametrize(
("side_effect", "expected_error"),
[
(AuthenticationError("Invalid credentials"), "invalid_auth"),
(ApiError("Connection failed"), "cannot_connect"),
(Exception("Unknown error"), "unknown"),
],
)
async def test_user_flow_errors(
hass: HomeAssistant,
mock_lojack_client: MagicMock,
mock_setup_entry: AsyncMock,
side_effect: Exception,
expected_error: str,
) -> None:
"""Test error handling and recovery in the user flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
with patch(
"homeassistant.components.lojack.config_flow.LoJackClient.create",
side_effect=side_effect,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": expected_error}
# Verify flow recovers after error
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_user_flow_already_configured(
hass: HomeAssistant,
mock_lojack_client: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test that duplicate accounts are rejected."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"

View File

@@ -0,0 +1,56 @@
"""Tests for the LoJack device tracker platform."""
from datetime import timedelta
from unittest.mock import AsyncMock, MagicMock
from freezegun.api import FrozenDateTimeFactory
from lojack_api import ApiError
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.lojack.const import DEFAULT_UPDATE_INTERVAL
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
ENTITY_ID = "device_tracker.2021_honda_accord"
async def test_all_entities(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_lojack_client: MagicMock,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test all device tracker entities are created."""
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
async def test_device_tracker_becomes_unavailable_on_api_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_lojack_client: MagicMock,
mock_device: MagicMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test device tracker becomes unavailable when coordinator update fails."""
await setup_integration(hass, mock_config_entry)
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state != "unavailable"
mock_device.get_location = AsyncMock(side_effect=ApiError("API unavailable"))
freezer.tick(timedelta(minutes=DEFAULT_UPDATE_INTERVAL))
async_fire_time_changed(hass)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state == "unavailable"

View File

@@ -0,0 +1,156 @@
"""Tests for the LoJack integration setup."""
from datetime import timedelta
from unittest.mock import AsyncMock, MagicMock, patch
from freezegun.api import FrozenDateTimeFactory
from lojack_api import ApiError, AuthenticationError
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.lojack.const import DEFAULT_UPDATE_INTERVAL, DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from . import setup_integration
from .const import TEST_DEVICE_ID
from tests.common import MockConfigEntry, async_fire_time_changed
async def test_setup_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_lojack_client: MagicMock,
) -> None:
"""Test successful setup of the integration."""
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.LOADED
assert hass.states.get("device_tracker.2021_honda_accord") is not None
@pytest.mark.parametrize(
("side_effect", "expected_state"),
[
(AuthenticationError("Invalid credentials"), ConfigEntryState.SETUP_ERROR),
(ApiError("Connection failed"), ConfigEntryState.SETUP_RETRY),
],
)
async def test_setup_entry_create_errors(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
side_effect: Exception,
expected_state: ConfigEntryState,
) -> None:
"""Test setup failure when LoJackClient.create raises an error."""
with patch(
"homeassistant.components.lojack.LoJackClient.create",
side_effect=side_effect,
):
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is expected_state
@pytest.mark.parametrize(
("side_effect", "expected_state"),
[
(AuthenticationError("Invalid credentials"), ConfigEntryState.SETUP_ERROR),
(ApiError("Connection failed"), ConfigEntryState.SETUP_RETRY),
],
)
async def test_setup_entry_list_devices_errors(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_lojack_client: MagicMock,
side_effect: Exception,
expected_state: ConfigEntryState,
) -> None:
"""Test setup failure when list_devices raises an error."""
mock_lojack_client.list_devices = AsyncMock(side_effect=side_effect)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is expected_state
async def test_setup_entry_no_vehicles(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_lojack_client: MagicMock,
) -> None:
"""Test integration loads successfully with no vehicles."""
mock_lojack_client.list_devices = AsyncMock(return_value=[])
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.LOADED
assert len(hass.states.async_entity_ids("device_tracker")) == 0
async def test_unload_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_lojack_client: MagicMock,
) -> None:
"""Test successful unload of the integration."""
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
async def test_coordinator_update_auth_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_lojack_client: MagicMock,
mock_device: MagicMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test entry stays loaded and reauth is triggered on auth error during polling."""
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.LOADED
mock_device.get_location = AsyncMock(
side_effect=AuthenticationError("Token expired")
)
freezer.tick(timedelta(minutes=DEFAULT_UPDATE_INTERVAL))
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Entry stays loaded; HA initiates a reauth flow
assert mock_config_entry.state is ConfigEntryState.LOADED
assert len(hass.config_entries.flow.async_progress()) == 1
flow = hass.config_entries.flow.async_progress()[0]
assert flow["flow_id"] is not None
assert flow["handler"] == DOMAIN
assert flow["context"]["source"] == "reauth"
async def test_device_info(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_lojack_client: MagicMock,
device_registry: dr.DeviceRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test device registry entry is created."""
await setup_integration(hass, mock_config_entry)
device_entry = device_registry.async_get_device(
identifiers={(DOMAIN, TEST_DEVICE_ID)}
)
assert device_entry is not None
assert device_entry == snapshot