Added PAJ GPS integration (#165070)

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Josef Zweck <josef@zweck.dev>
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
Tomasz Dylewski
2026-05-07 17:04:19 +02:00
committed by GitHub
co-authored by Copilot Josef Zweck Joost Lekkerkerker
parent 776fd69e39
commit a82205fed7
24 changed files with 1015 additions and 0 deletions
+1
View File
@@ -423,6 +423,7 @@ homeassistant.components.otp.*
homeassistant.components.overkiz.*
homeassistant.components.overseerr.*
homeassistant.components.p1_monitor.*
homeassistant.components.paj_gps.*
homeassistant.components.panel_custom.*
homeassistant.components.paperless_ngx.*
homeassistant.components.peblar.*
Generated
+2
View File
@@ -1308,6 +1308,8 @@ CLAUDE.md @home-assistant/core
/tests/components/ovo_energy/ @timmo001
/homeassistant/components/p1_monitor/ @klaasnicolaas
/tests/components/p1_monitor/ @klaasnicolaas
/homeassistant/components/paj_gps/ @skipperro
/tests/components/paj_gps/ @skipperro
/homeassistant/components/palazzetti/ @dotvav
/tests/components/palazzetti/ @dotvav
/homeassistant/components/panel_custom/ @home-assistant/frontend
@@ -0,0 +1,28 @@
"""Integration for PAJ GPS trackers."""
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from .const import DOMAIN
from .coordinator import PajGpsConfigEntry, PajGpsCoordinator
PLATFORMS: list[Platform] = [Platform.DEVICE_TRACKER]
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
async def async_setup_entry(hass: HomeAssistant, entry: PajGpsConfigEntry) -> bool:
"""Set up platform from a ConfigEntry."""
pajgps_coordinator = PajGpsCoordinator(hass, entry)
await pajgps_coordinator.async_config_entry_first_refresh()
entry.runtime_data = pajgps_coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: PajGpsConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
@@ -0,0 +1,92 @@
"""Config flow for PAJ GPS Tracker integration."""
import logging
from typing import Any
from aiohttp import ClientError
from pajgps_api import PajGpsApi
from pajgps_api.models.auth import AuthResponse
from pajgps_api.pajgps_api_error import AuthenticationError, TokenRefreshError
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.selector import (
TextSelector,
TextSelectorConfig,
TextSelectorType,
)
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_EMAIL): TextSelector(
TextSelectorConfig(
type=TextSelectorType.EMAIL,
autocomplete="email",
)
),
vol.Required(CONF_PASSWORD): TextSelector(
TextSelectorConfig(
type=TextSelectorType.PASSWORD,
autocomplete="current-password",
)
),
}
)
class PajGPSConfigFlow(ConfigFlow, domain=DOMAIN):
"""Config flow for PAJ GPS Tracker."""
async def _validate_credentials(
self, email: str, password: str
) -> tuple[str | None, AuthResponse | None]:
"""Attempt a real login with the given credentials.
Returns (None, auth) on success, or (error_key, None) on failure.
"""
websession = async_get_clientsession(self.hass)
try:
api = PajGpsApi(email=email, password=password, websession=websession)
auth = await api.login()
except AuthenticationError, TokenRefreshError:
return "invalid_auth", None
except ClientError:
return "cannot_connect", None
except Exception:
_LOGGER.exception("Unexpected error validating PAJ GPS credentials")
return "unknown", None
return None, auth
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle a flow initiated by the user."""
errors: dict[str, str] = {}
if user_input is not None:
normalized_email = user_input[CONF_EMAIL].strip().lower()
user_input[CONF_EMAIL] = normalized_email
error, auth = await self._validate_credentials(
user_input[CONF_EMAIL], user_input[CONF_PASSWORD]
)
if error is None and auth is not None:
await self.async_set_unique_id(str(auth.userID))
self._abort_if_unique_id_configured()
return self.async_create_entry(title=normalized_email, data=user_input)
if error is not None:
errors["base"] = error
return self.async_show_form(
step_id="user",
data_schema=self.add_suggested_values_to_schema(
STEP_USER_DATA_SCHEMA, user_input
),
errors=errors,
)
@@ -0,0 +1,4 @@
"""Constants for the PajGPS integration."""
DOMAIN = "paj_gps"
UPDATE_INTERVAL = 30
@@ -0,0 +1,107 @@
"""DataUpdateCoordinator for the PAJ GPS integration."""
from dataclasses import dataclass
from datetime import timedelta
import logging
from pajgps_api import PajGpsApi
from pajgps_api.models.device import Device
from pajgps_api.models.trackpoint import TrackPoint
from pajgps_api.pajgps_api_error import (
AuthenticationError,
PajGpsApiError,
TokenRefreshError,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN, UPDATE_INTERVAL
_LOGGER = logging.getLogger(__name__)
type PajGpsConfigEntry = ConfigEntry[PajGpsCoordinator]
@dataclass
class PajGpsData:
"""Snapshot of all PAJ GPS data for one coordinator tick."""
devices: dict[int, Device]
positions: dict[int, TrackPoint]
class PajGpsCoordinator(DataUpdateCoordinator[PajGpsData]):
"""Coordinator for the PAJ GPS integration."""
config_entry: PajGpsConfigEntry
def __init__(
self,
hass: HomeAssistant,
config_entry: PajGpsConfigEntry,
) -> None:
"""Initialize the coordinator from config-entry data."""
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=timedelta(seconds=UPDATE_INTERVAL),
config_entry=config_entry,
)
self._email: str = config_entry.data[CONF_EMAIL]
self._user_id: int | None = None
self.api = PajGpsApi(
email=self._email,
password=config_entry.data[CONF_PASSWORD],
websession=async_get_clientsession(hass),
)
@property
def email(self) -> str:
"""Return the account email address for this coordinator."""
return self._email
@property
def user_id(self) -> int | None:
"""Return the user ID obtained from the login response."""
return self._user_id
async def _async_setup(self) -> None:
"""Perform initial and first data refresh."""
try:
auth = await self.api.login()
self._user_id = auth.userID
except (AuthenticationError, TokenRefreshError) as exc:
raise ConfigEntryAuthFailed from exc
except Exception as exc:
raise ConfigEntryNotReady from exc
async def _async_update_data(self) -> PajGpsData:
"""Fetch device list and positions."""
devices: dict[int, Device] = {}
try:
device_list = await self.api.get_devices()
devices = {
device.id: device for device in device_list if device.id is not None
}
except PajGpsApiError as exc:
raise UpdateFailed(f"Failed to fetch device list: {exc}") from exc
device_ids = list(devices.keys())
positions: dict[int, TrackPoint] = {}
if device_ids:
try:
track_points = await self.api.get_all_last_positions(device_ids)
except PajGpsApiError as exc:
raise UpdateFailed(f"Failed to fetch positions: {exc}") from exc
positions = {
tp.iddevice: tp for tp in track_points if tp.iddevice is not None
}
return PajGpsData(devices=devices, positions=positions)
@@ -0,0 +1,78 @@
"""Platform for GPS device tracker integration.
Reads position data from PajGpsCoordinator and exposes it as a TrackerEntity.
"""
import logging
from homeassistant.components.device_tracker import SourceType
from homeassistant.components.device_tracker.config_entry import TrackerEntity
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import PajGpsConfigEntry
from .coordinator import PajGpsCoordinator
from .entity import PajGpsEntity
_LOGGER = logging.getLogger(__name__)
PARALLEL_UPDATES = 0
async def async_setup_entry(
hass: HomeAssistant,
config_entry: PajGpsConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up PAJ GPS tracker entities from a config entry."""
coordinator = config_entry.runtime_data
known_device_ids: set[int] = set()
@callback
def _async_add_new_devices() -> None:
"""Add entities for any device IDs not yet tracked."""
current_ids = set(coordinator.data.devices.keys())
new_ids = current_ids - known_device_ids
if new_ids:
sorted_new_ids = sorted(new_ids)
async_add_entities(
PajGPSDeviceTracker(coordinator, device_id)
for device_id in sorted_new_ids
)
known_device_ids.update(sorted_new_ids)
_async_add_new_devices()
if not known_device_ids:
_LOGGER.warning("No PAJ GPS devices found to add as trackers")
config_entry.async_on_unload(coordinator.async_add_listener(_async_add_new_devices))
class PajGPSDeviceTracker(PajGpsEntity, TrackerEntity):
"""Tracker entity that reads position from the coordinator snapshot."""
_attr_name = None
_attr_icon = "mdi:map-marker"
def __init__(self, pajgps_coordinator: PajGpsCoordinator, device_id: int) -> None:
"""Initialize the GPS position tracker entity."""
super().__init__(pajgps_coordinator, device_id)
self._attr_unique_id = f"{pajgps_coordinator.user_id}_{device_id}"
@property
def latitude(self) -> float | None:
"""Return the latitude of the device."""
tp = self.coordinator.data.positions.get(self._device_id)
return float(tp.lat) if tp and tp.lat is not None else None
@property
def longitude(self) -> float | None:
"""Return the longitude of the device."""
tp = self.coordinator.data.positions.get(self._device_id)
return float(tp.lng) if tp and tp.lng is not None else None
@property
def source_type(self) -> SourceType:
"""Return the source type of the tracker."""
return SourceType.GPS
@@ -0,0 +1,39 @@
"""Base entity class for the PAJ GPS integration."""
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import Device, PajGpsCoordinator
class PajGpsEntity(CoordinatorEntity[PajGpsCoordinator]):
"""Base class for all PAJ GPS entities."""
_attr_has_entity_name = True
def __init__(self, coordinator: PajGpsCoordinator, device_id: int) -> None:
"""Initialize the entity and build DeviceInfo."""
super().__init__(coordinator)
self._device_id = device_id
model = None
device_models = self.device.device_models
if device_models and isinstance(device_models[0], dict):
model = device_models[0].get("model")
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, f"{coordinator.user_id}_{device_id}")},
name=self.device.name or f"PAJ GPS {device_id}",
manufacturer="PAJ GPS",
model=model,
)
@property
def available(self) -> bool:
"""Return False when the device has been removed from the account."""
return super().available and self._device_id in self.coordinator.data.devices
@property
def device(self) -> Device:
"""Return the device from coordinator data."""
return self.coordinator.data.devices[self._device_id]
@@ -0,0 +1,11 @@
{
"domain": "paj_gps",
"name": "PAJ GPS",
"codeowners": ["@skipperro"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/paj_gps",
"integration_type": "hub",
"iot_class": "cloud_polling",
"quality_scale": "bronze",
"requirements": ["pajgps-api==0.3.1"]
}
@@ -0,0 +1,84 @@
rules:
# Bronze
action-setup:
status: exempt
comment: |
This integration does not provide additional actions.
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: |
This integration does not provide additional actions.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup:
status: exempt
comment: |
Entities of this integration do not explicitly subscribe to events.
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 additional actions.
config-entry-unloading: done
docs-configuration-parameters: done
docs-installation-parameters: done
entity-unavailable: done
integration-owner: done
log-when-unavailable: done
parallel-updates: done
reauthentication-flow: todo
test-coverage:
status: todo
comment: |
Add setup/entity lifecycle tests (setup, unload, coordinator refresh, and device_tracker entity creation/availability) before marking test-coverage as done.
# Gold
devices: done
diagnostics: todo
discovery-update-info:
status: exempt
comment: |
This integration does not use device discovery. The API returns all devices directly.
discovery:
status: exempt
comment: |
This integration does not use device discovery. The API returns all devices directly.
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: |
There are no entities that would be considered less popular or noisy. All entities are enabled by default.
entity-translations: done
exception-translations: todo
icon-translations: todo
reconfiguration-flow: todo
repair-issues: done
stale-devices: done
# Platinum
async-dependency: done
inject-websession: done
strict-typing: done
@@ -0,0 +1,26 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
},
"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%]"
},
"step": {
"user": {
"data": {
"email": "[%key:common::config_flow::data::email%]",
"password": "[%key:common::config_flow::data::password%]"
},
"data_description": {
"email": "Email used to log in to Finder Portal (finder-portal.com)",
"password": "Password used to log in to Finder Portal (finder-portal.com)"
},
"description": "Set credentials for your PAJ GPS account (finder-portal.com).",
"title": "PAJ GPS configuration"
}
}
}
}
+1
View File
@@ -544,6 +544,7 @@ FLOWS = {
"ovo_energy",
"owntracks",
"p1_monitor",
"paj_gps",
"palazzetti",
"panasonic_viera",
"paperless_ngx",
@@ -5170,6 +5170,12 @@
"config_flow": true,
"iot_class": "local_polling"
},
"paj_gps": {
"name": "PAJ GPS",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling"
},
"palazzetti": {
"name": "Palazzetti",
"integration_type": "device",
Generated
+10
View File
@@ -3985,6 +3985,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.paj_gps.*]
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.panel_custom.*]
check_untyped_defs = true
disallow_incomplete_defs = true
+3
View File
@@ -1777,6 +1777,9 @@ p1monitor==3.2.0
# homeassistant.components.mqtt
paho-mqtt==2.1.0
# homeassistant.components.paj_gps
pajgps-api==0.3.1
# homeassistant.components.panasonic_bluray
panacotta==0.2
+3
View File
@@ -1554,6 +1554,9 @@ p1monitor==3.2.0
# homeassistant.components.mqtt
paho-mqtt==2.1.0
# homeassistant.components.paj_gps
pajgps-api==0.3.1
# homeassistant.components.panasonic_viera
panasonic-viera==0.4.4
+12
View File
@@ -0,0 +1,12 @@
"""Tests for the PAJ GPS integration."""
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
"""Set up the PAJ GPS 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()
+66
View File
@@ -0,0 +1,66 @@
"""Common fixtures for PAJ GPS integration tests."""
from __future__ import annotations
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
from pajgps_api.models.auth import AuthResponse
from pajgps_api.models.device import Device
from pajgps_api.models.trackpoint import TrackPoint
import pytest
from homeassistant.components.paj_gps.const import DOMAIN
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from tests.common import MockConfigEntry, load_json_object_fixture
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Prevent the PAJ GPS integration from setting up during config flow tests."""
with patch(
"homeassistant.components.paj_gps.async_setup_entry",
return_value=True,
) as mock:
yield mock
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return a mock config entry for PAJ GPS."""
return MockConfigEntry(
domain=DOMAIN,
title="test@example.com",
unique_id="42",
data={
CONF_EMAIL: "test@example.com",
CONF_PASSWORD: "secret",
},
)
@pytest.fixture
def mock_paj_gps_api() -> Generator[AsyncMock]:
"""Mock PajGpsApi for PAJ GPS integration tests."""
with (
patch(
"homeassistant.components.paj_gps.coordinator.PajGpsApi",
autospec=True,
) as mock_api_cls,
patch(
"homeassistant.components.paj_gps.config_flow.PajGpsApi",
new=mock_api_cls,
),
):
api = mock_api_cls.return_value
api.login.return_value = AuthResponse(
userID=42, token="test_token", refresh_token="test_refresh"
)
api.get_devices.return_value = [
Device(**load_json_object_fixture("device.json", DOMAIN))
]
api.get_all_last_positions.return_value = [
TrackPoint(**load_json_object_fixture("trackpoint.json", DOMAIN))
]
yield api
@@ -0,0 +1,15 @@
{
"id": 1,
"name": "Device 1",
"imei": "IMEI1",
"modellid": 100,
"alarmbewegung": 1,
"alarmakkuwarnung": 1,
"alarmsos": 1,
"alarmgeschwindigkeit": 1,
"alarmstromunterbrechung": 1,
"alarmzuendalarm": 1,
"alarm_fall_enabled": 1,
"alarm_volt": 1,
"device_models": []
}
@@ -0,0 +1,8 @@
{
"iddevice": 1,
"lat": 52.0,
"lng": 13.0,
"speed": 50,
"battery": 80,
"direction": 90
}
@@ -0,0 +1,58 @@
# serializer version: 1
# name: test_all_entities[device_tracker.device_1-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'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.device_1',
'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:map-marker',
'original_name': None,
'platform': 'paj_gps',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '42_1',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[device_tracker.device_1-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Device 1',
'gps_accuracy': 0,
'icon': 'mdi:map-marker',
'in_zones': list([
]),
'latitude': 52.0,
'longitude': 13.0,
'source_type': <SourceType.GPS: 'gps'>,
}),
'context': <ANY>,
'entity_id': 'device_tracker.device_1',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'not_home',
})
# ---
@@ -0,0 +1,108 @@
"""Tests for the PAJ GPS config flow."""
from __future__ import annotations
from unittest.mock import AsyncMock
from aiohttp import ClientError
from pajgps_api.pajgps_api_error import AuthenticationError, TokenRefreshError
import pytest
from homeassistant.components.paj_gps.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
VALID_USER_INPUT = {
CONF_EMAIL: "user@example.com",
CONF_PASSWORD: "s3cr3t",
}
@pytest.mark.parametrize(
("raw_email", "expected_email"),
[
("user@example.com", "user@example.com"),
(" USER@EXAMPLE.COM ", "user@example.com"),
],
)
async def test_full_user_flow(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_paj_gps_api: AsyncMock,
raw_email: str,
expected_email: str,
) -> None:
"""Full user flow must show a form then create an entry with normalized data."""
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"],
user_input={CONF_EMAIL: raw_email, CONF_PASSWORD: "s3cr3t"},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == expected_email
assert result["data"][CONF_EMAIL] == expected_email
assert result["data"][CONF_PASSWORD] == "s3cr3t"
assert result["result"].unique_id == "42"
async def test_duplicate_email_aborts(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_paj_gps_api: AsyncMock,
) -> None:
"""A flow for an already-configured account must abort with 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=VALID_USER_INPUT,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.parametrize(
("side_effect", "expected_error"),
[
(AuthenticationError("bad creds"), "invalid_auth"),
(TokenRefreshError("refresh failed"), "invalid_auth"),
(ClientError(), "cannot_connect"),
(ConnectionError("timeout"), "unknown"),
],
)
async def test_invalid_credentials_shows_form_error(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_paj_gps_api: AsyncMock,
side_effect: Exception,
expected_error: str,
) -> None:
"""Credential errors must re-show the form with the correct error key."""
mock_paj_gps_api.login.side_effect = side_effect
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=VALID_USER_INPUT,
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": expected_error}
mock_paj_gps_api.login.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=VALID_USER_INPUT,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
@@ -0,0 +1,164 @@
"""Tests for PAJ GPS device tracker."""
from __future__ import annotations
from datetime import timedelta
from unittest.mock import AsyncMock
from freezegun.api import FrozenDateTimeFactory
from pajgps_api.models.device import Device
from pajgps_api.models.trackpoint import TrackPoint
from pajgps_api.pajgps_api_error import PajGpsApiError
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.paj_gps.const import UPDATE_INTERVAL
from homeassistant.const import STATE_UNAVAILABLE
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
async def test_all_entities(
hass: HomeAssistant,
mock_paj_gps_api: AsyncMock,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test all device tracker entities against snapshot."""
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
async def test_entity_unavailable_on_coordinator_error(
hass: HomeAssistant,
mock_paj_gps_api: AsyncMock,
mock_config_entry: MockConfigEntry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test that entity becomes unavailable when the coordinator update fails."""
await setup_integration(hass, mock_config_entry)
state = hass.states.get("device_tracker.device_1")
assert state is not None
assert state.state != STATE_UNAVAILABLE
mock_paj_gps_api.get_devices.side_effect = PajGpsApiError("API error")
freezer.tick(timedelta(seconds=UPDATE_INTERVAL))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get("device_tracker.device_1").state == STATE_UNAVAILABLE
async def test_entity_recovers_after_coordinator_error(
hass: HomeAssistant,
mock_paj_gps_api: AsyncMock,
mock_config_entry: MockConfigEntry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test that the entity recovers after a transient coordinator error."""
await setup_integration(hass, mock_config_entry)
mock_paj_gps_api.get_devices.side_effect = PajGpsApiError("API error")
freezer.tick(timedelta(seconds=UPDATE_INTERVAL))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get("device_tracker.device_1").state == STATE_UNAVAILABLE
mock_paj_gps_api.get_devices.side_effect = None
freezer.tick(timedelta(seconds=UPDATE_INTERVAL))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get("device_tracker.device_1").state != STATE_UNAVAILABLE
async def test_device_removed_from_account(
hass: HomeAssistant,
mock_paj_gps_api: AsyncMock,
mock_config_entry: MockConfigEntry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test entity becomes unavailable when the device disappears from the account."""
await setup_integration(hass, mock_config_entry)
assert hass.states.get("device_tracker.device_1").state != STATE_UNAVAILABLE
mock_paj_gps_api.get_devices.return_value = []
mock_paj_gps_api.get_all_last_positions.return_value = []
freezer.tick(timedelta(seconds=UPDATE_INTERVAL))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get("device_tracker.device_1").state == STATE_UNAVAILABLE
async def test_new_device_added_dynamically(
hass: HomeAssistant,
mock_paj_gps_api: AsyncMock,
mock_config_entry: MockConfigEntry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test that a new entity is added when a new device appears in coordinator data."""
await setup_integration(hass, mock_config_entry)
assert hass.states.get("device_tracker.device_1") is not None
assert hass.states.get("device_tracker.device_2") is None
new_device = Device(id=2, name="Device 2", device_models=[])
new_trackpoint = TrackPoint(iddevice=2, lat=48.8566, lng=2.3522)
mock_paj_gps_api.get_devices.return_value = [
*mock_paj_gps_api.get_devices.return_value,
new_device,
]
mock_paj_gps_api.get_all_last_positions.return_value = [
*mock_paj_gps_api.get_all_last_positions.return_value,
new_trackpoint,
]
freezer.tick(timedelta(seconds=UPDATE_INTERVAL))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get("device_tracker.device_2") is not None
@pytest.mark.parametrize(
("lat", "lng", "expected_lat", "expected_lng"),
[
(None, 13.0, None, None),
(52.0, None, None, None),
(52.0, 13.0, 52.0, 13.0),
],
ids=["no_latitude", "no_longitude", "valid_position"],
)
async def test_position_none_when_coordinates_missing(
hass: HomeAssistant,
mock_paj_gps_api: AsyncMock,
mock_config_entry: MockConfigEntry,
lat: float | None,
lng: float | None,
expected_lat: float | None,
expected_lng: float | None,
) -> None:
"""Test that latitude/longitude are None when coordinates are missing."""
mock_paj_gps_api.get_all_last_positions.return_value = [
TrackPoint(iddevice=1, lat=lat, lng=lng)
]
await setup_integration(hass, mock_config_entry)
state = hass.states.get("device_tracker.device_1")
assert state is not None
assert state.attributes.get("latitude") == expected_lat
assert state.attributes.get("longitude") == expected_lng
+89
View File
@@ -0,0 +1,89 @@
"""Tests for the PAJ GPS base entity."""
from __future__ import annotations
from unittest.mock import AsyncMock
from pajgps_api.models.device import Device
from homeassistant.components.paj_gps.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from . import setup_integration
from tests.common import MockConfigEntry
async def test_entity_device_info(
hass: HomeAssistant,
mock_paj_gps_api: AsyncMock,
mock_config_entry: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test that device info is correctly set up."""
await setup_integration(hass, mock_config_entry)
device = device_registry.async_get_device(identifiers={(DOMAIN, "42_1")})
assert device is not None
assert device.name == "Device 1"
assert device.manufacturer == "PAJ GPS"
assert device.model is None
async def test_entity_device_info_with_model(
hass: HomeAssistant,
mock_paj_gps_api: AsyncMock,
mock_config_entry: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test that device model is populated when device_models is present."""
mock_paj_gps_api.get_devices.return_value = [
Device(
id=1,
name="Device 1",
device_models=[{"model": "ALLROUND Finder 4G"}],
)
]
await setup_integration(hass, mock_config_entry)
device = device_registry.async_get_device(identifiers={(DOMAIN, "42_1")})
assert device is not None
assert device.model == "ALLROUND Finder 4G"
async def test_entity_device_info_fallback_name(
hass: HomeAssistant,
mock_paj_gps_api: AsyncMock,
mock_config_entry: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test that device name falls back to 'PAJ GPS <id>' when name is absent."""
mock_paj_gps_api.get_devices.return_value = [
Device(id=1, name=None, device_models=[])
]
await setup_integration(hass, mock_config_entry)
device = device_registry.async_get_device(identifiers={(DOMAIN, "42_1")})
assert device is not None
assert device.name == "PAJ GPS 1"
async def test_entity_device_info_non_dict_device_models(
hass: HomeAssistant,
mock_paj_gps_api: AsyncMock,
mock_config_entry: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test that model is None when device_models entries are not dicts."""
mock_paj_gps_api.get_devices.return_value = [
Device(id=1, name="Device 1", device_models=[100])
]
await setup_integration(hass, mock_config_entry)
device = device_registry.async_get_device(identifiers={(DOMAIN, "42_1")})
assert device is not None
assert device.model is None