Add status sensors to paperless (#145591)

* Add first status sensor and coordinator

* New snapshot

* Add comment

* Add test for forbidden status endpoint

* Changed comment

* Fixed translation

* Minor changes and code optimization

* Add common translation; minor tweaks

* Moved translation from common to integration
This commit is contained in:
Florian von Garrel 2025-05-26 17:24:23 +02:00 committed by GitHub
parent b7ce0f63a9
commit c14d17f88c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 1106 additions and 116 deletions

View File

@ -1,9 +1,30 @@
"""The Paperless-ngx integration.""" """The Paperless-ngx integration."""
from homeassistant.const import Platform from pypaperless import Paperless
from homeassistant.core import HomeAssistant from pypaperless.exceptions import (
InitializationError,
PaperlessConnectionError,
PaperlessForbiddenError,
PaperlessInactiveOrDeletedError,
PaperlessInvalidTokenError,
)
from .coordinator import PaperlessConfigEntry, PaperlessCoordinator from homeassistant.const import CONF_API_KEY, CONF_URL, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import (
ConfigEntryAuthFailed,
ConfigEntryError,
ConfigEntryNotReady,
)
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN, LOGGER
from .coordinator import (
PaperlessConfigEntry,
PaperlessData,
PaperlessStatisticCoordinator,
PaperlessStatusCoordinator,
)
PLATFORMS: list[Platform] = [Platform.SENSOR] PLATFORMS: list[Platform] = [Platform.SENSOR]
@ -11,10 +32,28 @@ PLATFORMS: list[Platform] = [Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: PaperlessConfigEntry) -> bool: async def async_setup_entry(hass: HomeAssistant, entry: PaperlessConfigEntry) -> bool:
"""Set up Paperless-ngx from a config entry.""" """Set up Paperless-ngx from a config entry."""
coordinator = PaperlessCoordinator(hass, entry) api = await _get_paperless_api(hass, entry)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator statistics_coordinator = PaperlessStatisticCoordinator(hass, entry, api)
status_coordinator = PaperlessStatusCoordinator(hass, entry, api)
await statistics_coordinator.async_config_entry_first_refresh()
try:
await status_coordinator.async_config_entry_first_refresh()
except ConfigEntryNotReady as err:
# Catch the error so the integration doesn't fail just because status coordinator fails.
LOGGER.warning("Could not initialize status coordinator: %s", err)
entry.runtime_data = PaperlessData(
status=status_coordinator,
statistics=statistics_coordinator,
)
entry.runtime_data = PaperlessData(
status=status_coordinator,
statistics=statistics_coordinator,
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
@ -24,3 +63,47 @@ async def async_setup_entry(hass: HomeAssistant, entry: PaperlessConfigEntry) ->
async def async_unload_entry(hass: HomeAssistant, entry: PaperlessConfigEntry) -> bool: async def async_unload_entry(hass: HomeAssistant, entry: PaperlessConfigEntry) -> bool:
"""Unload paperless-ngx config entry.""" """Unload paperless-ngx config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS) return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def _get_paperless_api(
hass: HomeAssistant,
entry: PaperlessConfigEntry,
) -> Paperless:
"""Create and initialize paperless-ngx API."""
api = Paperless(
entry.data[CONF_URL],
entry.data[CONF_API_KEY],
session=async_get_clientsession(hass),
)
try:
await api.initialize()
await api.statistics() # test permissions on api
except PaperlessConnectionError as err:
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="cannot_connect",
) from err
except PaperlessInvalidTokenError as err:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN,
translation_key="invalid_api_key",
) from err
except PaperlessInactiveOrDeletedError as err:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN,
translation_key="user_inactive_or_deleted",
) from err
except PaperlessForbiddenError as err:
raise ConfigEntryError(
translation_domain=DOMAIN,
translation_key="forbidden",
) from err
except InitializationError as err:
raise ConfigEntryError(
translation_domain=DOMAIN,
translation_key="cannot_connect",
) from err
else:
return api

View File

@ -2,38 +2,45 @@
from __future__ import annotations from __future__ import annotations
from abc import abstractmethod
from dataclasses import dataclass
from datetime import timedelta from datetime import timedelta
from typing import TypeVar
from pypaperless import Paperless from pypaperless import Paperless
from pypaperless.exceptions import ( from pypaperless.exceptions import (
InitializationError,
PaperlessConnectionError, PaperlessConnectionError,
PaperlessForbiddenError, PaperlessForbiddenError,
PaperlessInactiveOrDeletedError, PaperlessInactiveOrDeletedError,
PaperlessInvalidTokenError, PaperlessInvalidTokenError,
) )
from pypaperless.models import Statistic from pypaperless.models import Statistic, Status
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_API_KEY, CONF_URL
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ( from homeassistant.exceptions import ConfigEntryAuthFailed
ConfigEntryAuthFailed,
ConfigEntryError,
ConfigEntryNotReady,
)
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN, LOGGER from .const import DOMAIN, LOGGER
type PaperlessConfigEntry = ConfigEntry[PaperlessCoordinator] type PaperlessConfigEntry = ConfigEntry[PaperlessData]
UPDATE_INTERVAL = 120 TData = TypeVar("TData")
UPDATE_INTERVAL_STATISTICS = timedelta(seconds=120)
UPDATE_INTERVAL_STATUS = timedelta(seconds=300)
class PaperlessCoordinator(DataUpdateCoordinator[Statistic]): @dataclass
"""Coordinator to manage Paperless-ngx statistic updates.""" class PaperlessData:
"""Data for the Paperless-ngx integration."""
statistics: PaperlessStatisticCoordinator
status: PaperlessStatusCoordinator
class PaperlessCoordinator(DataUpdateCoordinator[TData]):
"""Coordinator to manage fetching Paperless-ngx API."""
config_entry: PaperlessConfigEntry config_entry: PaperlessConfigEntry
@ -41,28 +48,27 @@ class PaperlessCoordinator(DataUpdateCoordinator[Statistic]):
self, self,
hass: HomeAssistant, hass: HomeAssistant,
entry: PaperlessConfigEntry, entry: PaperlessConfigEntry,
api: Paperless,
name: str,
update_interval: timedelta,
) -> None: ) -> None:
"""Initialize my coordinator.""" """Initialize Paperless-ngx statistics coordinator."""
self.api = api
super().__init__( super().__init__(
hass, hass,
LOGGER, LOGGER,
config_entry=entry, config_entry=entry,
name="Paperless-ngx Coordinator", name=name,
update_interval=timedelta(seconds=UPDATE_INTERVAL), update_interval=update_interval,
) )
self.api = Paperless( async def _async_update_data(self) -> TData:
entry.data[CONF_URL], """Update data via internal method."""
entry.data[CONF_API_KEY],
session=async_get_clientsession(self.hass),
)
async def _async_setup(self) -> None:
try: try:
await self.api.initialize() return await self._async_update_data_internal()
await self.api.statistics() # test permissions on api
except PaperlessConnectionError as err: except PaperlessConnectionError as err:
raise ConfigEntryNotReady( raise UpdateFailed(
translation_domain=DOMAIN, translation_domain=DOMAIN,
translation_key="cannot_connect", translation_key="cannot_connect",
) from err ) from err
@ -77,37 +83,57 @@ class PaperlessCoordinator(DataUpdateCoordinator[Statistic]):
translation_key="user_inactive_or_deleted", translation_key="user_inactive_or_deleted",
) from err ) from err
except PaperlessForbiddenError as err: except PaperlessForbiddenError as err:
raise ConfigEntryError( raise UpdateFailed(
translation_domain=DOMAIN, translation_domain=DOMAIN,
translation_key="forbidden", translation_key="forbidden",
) from err ) from err
except InitializationError as err:
raise ConfigEntryError(
translation_domain=DOMAIN,
translation_key="cannot_connect",
) from err
async def _async_update_data(self) -> Statistic: @abstractmethod
"""Fetch data from API endpoint.""" async def _async_update_data_internal(self) -> TData:
try: """Update data via paperless-ngx API."""
return await self.api.statistics()
except PaperlessConnectionError as err:
raise UpdateFailed( class PaperlessStatisticCoordinator(PaperlessCoordinator[Statistic]):
translation_domain=DOMAIN, """Coordinator to manage Paperless-ngx statistic updates."""
translation_key="cannot_connect",
) from err def __init__(
except PaperlessForbiddenError as err: self,
raise UpdateFailed( hass: HomeAssistant,
translation_domain=DOMAIN, entry: PaperlessConfigEntry,
translation_key="forbidden", api: Paperless,
) from err ) -> None:
except PaperlessInvalidTokenError as err: """Initialize Paperless-ngx status coordinator."""
raise ConfigEntryAuthFailed( super().__init__(
translation_domain=DOMAIN, hass,
translation_key="invalid_api_key", entry,
) from err api,
except PaperlessInactiveOrDeletedError as err: name="Statistics Coordinator",
raise ConfigEntryAuthFailed( update_interval=UPDATE_INTERVAL_STATISTICS,
translation_domain=DOMAIN, )
translation_key="user_inactive_or_deleted",
) from err async def _async_update_data_internal(self) -> Statistic:
"""Fetch statistics data from API endpoint."""
return await self.api.statistics()
class PaperlessStatusCoordinator(PaperlessCoordinator[Status]):
"""Coordinator to manage Paperless-ngx status updates."""
def __init__(
self,
hass: HomeAssistant,
entry: PaperlessConfigEntry,
api: Paperless,
) -> None:
"""Initialize Paperless-ngx status coordinator."""
super().__init__(
hass,
entry,
api,
name="Status Coordinator",
update_interval=UPDATE_INTERVAL_STATUS,
)
async def _async_update_data_internal(self) -> Status:
"""Fetch status data from API endpoint."""
return await self.api.status()

View File

@ -15,4 +15,9 @@ async def async_get_config_entry_diagnostics(
entry: PaperlessConfigEntry, entry: PaperlessConfigEntry,
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Return diagnostics for a config entry.""" """Return diagnostics for a config entry."""
return {"data": asdict(entry.runtime_data.data)} return {
"data": {
"statistics": asdict(entry.runtime_data.statistics.data),
"status": asdict(entry.runtime_data.status.data),
},
}

View File

@ -2,6 +2,8 @@
from __future__ import annotations from __future__ import annotations
from typing import Generic, TypeVar
from homeassistant.components.sensor import EntityDescription from homeassistant.components.sensor import EntityDescription
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.helpers.update_coordinator import CoordinatorEntity
@ -9,15 +11,17 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN from .const import DOMAIN
from .coordinator import PaperlessCoordinator from .coordinator import PaperlessCoordinator
TCoordinator = TypeVar("TCoordinator", bound=PaperlessCoordinator)
class PaperlessEntity(CoordinatorEntity[PaperlessCoordinator]):
class PaperlessEntity(CoordinatorEntity[TCoordinator], Generic[TCoordinator]):
"""Defines a base Paperless-ngx entity.""" """Defines a base Paperless-ngx entity."""
_attr_has_entity_name = True _attr_has_entity_name = True
def __init__( def __init__(
self, self,
coordinator: PaperlessCoordinator, coordinator: TCoordinator,
description: EntityDescription, description: EntityDescription,
) -> None: ) -> None:
"""Initialize the Paperless-ngx entity.""" """Initialize the Paperless-ngx entity."""

View File

@ -16,8 +16,65 @@
"correspondent_count": { "correspondent_count": {
"default": "mdi:account-group" "default": "mdi:account-group"
}, },
"document_type_count": { "storage_total": {
"default": "mdi:format-list-bulleted-type" "default": "mdi:harddisk"
},
"storage_available": {
"default": "mdi:harddisk"
},
"database_status": {
"default": "mdi:check-circle",
"state": {
"ok": "mdi:check-circle",
"warning": "mdi:alert",
"error": "mdi:alert-circle",
"unknown": "mdi:help-circle"
}
},
"index_status": {
"default": "mdi:check-circle",
"state": {
"ok": "mdi:check-circle",
"warning": "mdi:alert",
"error": "mdi:alert-circle",
"unknown": "mdi:help-circle"
}
},
"classifier_status": {
"default": "mdi:check-circle",
"state": {
"ok": "mdi:check-circle",
"warning": "mdi:alert",
"error": "mdi:alert-circle",
"unknown": "mdi:help-circle"
}
},
"celery_status": {
"default": "mdi:check-circle",
"state": {
"ok": "mdi:check-circle",
"warning": "mdi:alert",
"error": "mdi:alert-circle",
"unknown": "mdi:help-circle"
}
},
"redis_status": {
"default": "mdi:check-circle",
"state": {
"ok": "mdi:check-circle",
"warning": "mdi:alert",
"error": "mdi:alert-circle",
"unknown": "mdi:help-circle"
}
},
"sanity_check_status": {
"default": "mdi:check-circle",
"state": {
"ok": "mdi:check-circle",
"warning": "mdi:alert",
"error": "mdi:alert-circle",
"unknown": "mdi:help-circle"
}
} }
} }
} }

View File

@ -4,62 +4,73 @@ from __future__ import annotations
from collections.abc import Callable from collections.abc import Callable
from dataclasses import dataclass from dataclasses import dataclass
from typing import Generic
from pypaperless.models import Statistic from pypaperless.models import Statistic, Status
from pypaperless.models.common import StatusType
from homeassistant.components.sensor import ( from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity, SensorEntity,
SensorEntityDescription, SensorEntityDescription,
SensorStateClass, SensorStateClass,
) )
from homeassistant.const import EntityCategory, UnitOfInformation
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.util.unit_conversion import InformationConverter
from .coordinator import PaperlessConfigEntry from .coordinator import (
from .entity import PaperlessEntity PaperlessConfigEntry,
PaperlessStatisticCoordinator,
PaperlessStatusCoordinator,
TData,
)
from .entity import PaperlessEntity, TCoordinator
PARALLEL_UPDATES = 0 PARALLEL_UPDATES = 0
@dataclass(frozen=True, kw_only=True) @dataclass(frozen=True, kw_only=True)
class PaperlessEntityDescription(SensorEntityDescription): class PaperlessEntityDescription(SensorEntityDescription, Generic[TData]):
"""Describes Paperless-ngx sensor entity.""" """Describes Paperless-ngx sensor entity."""
value_fn: Callable[[Statistic], int | None] value_fn: Callable[[TData], StateType]
SENSOR_DESCRIPTIONS: tuple[PaperlessEntityDescription, ...] = ( SENSOR_STATISTICS: tuple[PaperlessEntityDescription, ...] = (
PaperlessEntityDescription( PaperlessEntityDescription[Statistic](
key="documents_total", key="documents_total",
translation_key="documents_total", translation_key="documents_total",
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: data.documents_total, value_fn=lambda data: data.documents_total,
), ),
PaperlessEntityDescription( PaperlessEntityDescription[Statistic](
key="documents_inbox", key="documents_inbox",
translation_key="documents_inbox", translation_key="documents_inbox",
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: data.documents_inbox, value_fn=lambda data: data.documents_inbox,
), ),
PaperlessEntityDescription( PaperlessEntityDescription[Statistic](
key="characters_count", key="characters_count",
translation_key="characters_count", translation_key="characters_count",
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: data.character_count, value_fn=lambda data: data.character_count,
), ),
PaperlessEntityDescription( PaperlessEntityDescription[Statistic](
key="tag_count", key="tag_count",
translation_key="tag_count", translation_key="tag_count",
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: data.tag_count, value_fn=lambda data: data.tag_count,
), ),
PaperlessEntityDescription( PaperlessEntityDescription[Statistic](
key="correspondent_count", key="correspondent_count",
translation_key="correspondent_count", translation_key="correspondent_count",
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: data.correspondent_count, value_fn=lambda data: data.correspondent_count,
), ),
PaperlessEntityDescription( PaperlessEntityDescription[Statistic](
key="document_type_count", key="document_type_count",
translation_key="document_type_count", translation_key="document_type_count",
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
@ -67,6 +78,157 @@ SENSOR_DESCRIPTIONS: tuple[PaperlessEntityDescription, ...] = (
), ),
) )
SENSOR_STATUS: tuple[PaperlessEntityDescription, ...] = (
PaperlessEntityDescription[Status](
key="storage_total",
translation_key="storage_total",
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.DATA_SIZE,
native_unit_of_measurement=UnitOfInformation.GIGABYTES,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=(
lambda data: round(
InformationConverter().convert(
data.storage.total,
UnitOfInformation.BYTES,
UnitOfInformation.GIGABYTES,
),
2,
)
if data.storage is not None and data.storage.total is not None
else None
),
),
PaperlessEntityDescription[Status](
key="storage_available",
translation_key="storage_available",
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.DATA_SIZE,
native_unit_of_measurement=UnitOfInformation.GIGABYTES,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=(
lambda data: round(
InformationConverter().convert(
data.storage.available,
UnitOfInformation.BYTES,
UnitOfInformation.GIGABYTES,
),
2,
)
if data.storage is not None and data.storage.available is not None
else None
),
),
PaperlessEntityDescription[Status](
key="database_status",
translation_key="database_status",
device_class=SensorDeviceClass.ENUM,
entity_category=EntityCategory.DIAGNOSTIC,
options=[
item.value.lower() for item in StatusType if item != StatusType.UNKNOWN
],
value_fn=(
lambda data: data.database.status.value.lower()
if (
data.database is not None
and data.database.status is not None
and data.database.status != StatusType.UNKNOWN
)
else None
),
),
PaperlessEntityDescription[Status](
key="index_status",
translation_key="index_status",
device_class=SensorDeviceClass.ENUM,
entity_category=EntityCategory.DIAGNOSTIC,
options=[
item.value.lower() for item in StatusType if item != StatusType.UNKNOWN
],
value_fn=(
lambda data: data.tasks.index_status.value.lower()
if (
data.tasks is not None
and data.tasks.index_status is not None
and data.tasks.index_status != StatusType.UNKNOWN
)
else None
),
),
PaperlessEntityDescription[Status](
key="classifier_status",
translation_key="classifier_status",
device_class=SensorDeviceClass.ENUM,
entity_category=EntityCategory.DIAGNOSTIC,
options=[
item.value.lower() for item in StatusType if item != StatusType.UNKNOWN
],
value_fn=(
lambda data: data.tasks.classifier_status.value.lower()
if (
data.tasks is not None
and data.tasks.classifier_status is not None
and data.tasks.classifier_status != StatusType.UNKNOWN
)
else None
),
),
PaperlessEntityDescription[Status](
key="celery_status",
translation_key="celery_status",
device_class=SensorDeviceClass.ENUM,
entity_category=EntityCategory.DIAGNOSTIC,
options=[
item.value.lower() for item in StatusType if item != StatusType.UNKNOWN
],
value_fn=(
lambda data: data.tasks.celery_status.value.lower()
if (
data.tasks is not None
and data.tasks.celery_status is not None
and data.tasks.celery_status != StatusType.UNKNOWN
)
else None
),
),
PaperlessEntityDescription[Status](
key="redis_status",
translation_key="redis_status",
device_class=SensorDeviceClass.ENUM,
entity_category=EntityCategory.DIAGNOSTIC,
options=[
item.value.lower() for item in StatusType if item != StatusType.UNKNOWN
],
value_fn=(
lambda data: data.tasks.redis_status.value.lower()
if (
data.tasks is not None
and data.tasks.redis_status is not None
and data.tasks.redis_status != StatusType.UNKNOWN
)
else None
),
),
PaperlessEntityDescription[Status](
key="sanity_check_status",
translation_key="sanity_check_status",
device_class=SensorDeviceClass.ENUM,
entity_category=EntityCategory.DIAGNOSTIC,
options=[
item.value.lower() for item in StatusType if item != StatusType.UNKNOWN
],
value_fn=(
lambda data: data.tasks.sanity_check_status.value.lower()
if (
data.tasks is not None
and data.tasks.sanity_check_status is not None
and data.tasks.sanity_check_status != StatusType.UNKNOWN
)
else None
),
),
)
async def async_setup_entry( async def async_setup_entry(
hass: HomeAssistant, hass: HomeAssistant,
@ -74,21 +236,34 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback, async_add_entities: AddConfigEntryEntitiesCallback,
) -> None: ) -> None:
"""Set up Paperless-ngx sensors.""" """Set up Paperless-ngx sensors."""
async_add_entities(
PaperlessSensor( entities: list[PaperlessSensor] = []
coordinator=entry.runtime_data,
description=sensor_description, entities += [
PaperlessSensor[PaperlessStatisticCoordinator](
coordinator=entry.runtime_data.statistics,
description=description,
) )
for sensor_description in SENSOR_DESCRIPTIONS for description in SENSOR_STATISTICS
) ]
entities += [
PaperlessSensor[PaperlessStatusCoordinator](
coordinator=entry.runtime_data.status,
description=description,
)
for description in SENSOR_STATUS
]
async_add_entities(entities)
class PaperlessSensor(PaperlessEntity, SensorEntity): class PaperlessSensor(PaperlessEntity[TCoordinator], SensorEntity):
"""Defines a Paperless-ngx sensor entity.""" """Defines a Paperless-ngx sensor entity."""
entity_description: PaperlessEntityDescription entity_description: PaperlessEntityDescription
@property @property
def native_value(self) -> int | None: def native_value(self) -> StateType:
"""Return the current value of the sensor.""" """Return the current value of the sensor."""
return self.entity_description.value_fn(self.coordinator.data) return self.entity_description.value_fn(self.coordinator.data)

View File

@ -71,6 +71,60 @@
"document_type_count": { "document_type_count": {
"name": "Document types", "name": "Document types",
"unit_of_measurement": "document types" "unit_of_measurement": "document types"
},
"storage_total": {
"name": "Total storage"
},
"storage_available": {
"name": "Available storage"
},
"database_status": {
"name": "Status database",
"state": {
"ok": "OK",
"warning": "Warning",
"error": "[%key:common::state::error%]"
}
},
"index_status": {
"name": "Status index",
"state": {
"ok": "[%key:component::paperless_ngx::entity::sensor::database_status::state::ok%]",
"warning": "[%key:component::paperless_ngx::entity::sensor::database_status::state::warning%]",
"error": "[%key:common::state::error%]"
}
},
"classifier_status": {
"name": "Status classifier",
"state": {
"ok": "[%key:component::paperless_ngx::entity::sensor::database_status::state::ok%]",
"warning": "[%key:component::paperless_ngx::entity::sensor::database_status::state::warning%]",
"error": "[%key:common::state::error%]"
}
},
"celery_status": {
"name": "Status celery",
"state": {
"ok": "[%key:component::paperless_ngx::entity::sensor::database_status::state::ok%]",
"warning": "[%key:component::paperless_ngx::entity::sensor::database_status::state::warning%]",
"error": "[%key:common::state::error%]"
}
},
"redis_status": {
"name": "Status redis",
"state": {
"ok": "[%key:component::paperless_ngx::entity::sensor::database_status::state::ok%]",
"warning": "[%key:component::paperless_ngx::entity::sensor::database_status::state::warning%]",
"error": "[%key:common::state::error%]"
}
},
"sanity_check_status": {
"name": "Status sanity",
"state": {
"ok": "[%key:component::paperless_ngx::entity::sensor::database_status::state::ok%]",
"warning": "[%key:component::paperless_ngx::entity::sensor::database_status::state::warning%]",
"error": "[%key:common::state::error%]"
}
} }
} }
}, },

View File

@ -4,7 +4,7 @@ from collections.abc import Generator
import json import json
from unittest.mock import AsyncMock, MagicMock, patch from unittest.mock import AsyncMock, MagicMock, patch
from pypaperless.models import Statistic from pypaperless.models import Statistic, Status
import pytest import pytest
from homeassistant.components.paperless_ngx.const import DOMAIN from homeassistant.components.paperless_ngx.const import DOMAIN
@ -16,6 +16,12 @@ from .const import USER_INPUT_ONE
from tests.common import MockConfigEntry, load_fixture from tests.common import MockConfigEntry, load_fixture
@pytest.fixture
def mock_status_data() -> Generator[MagicMock]:
"""Return test status data."""
return json.loads(load_fixture("test_data_status.json", DOMAIN))
@pytest.fixture @pytest.fixture
def mock_statistic_data() -> Generator[MagicMock]: def mock_statistic_data() -> Generator[MagicMock]:
"""Return test statistic data.""" """Return test statistic data."""
@ -29,7 +35,9 @@ def mock_statistic_data_update() -> Generator[MagicMock]:
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def mock_paperless(mock_statistic_data: MagicMock) -> Generator[AsyncMock]: def mock_paperless(
mock_statistic_data: MagicMock, mock_status_data: MagicMock
) -> Generator[AsyncMock]:
"""Mock the pypaperless.Paperless client.""" """Mock the pypaperless.Paperless client."""
with ( with (
patch( patch(
@ -40,6 +48,10 @@ def mock_paperless(mock_statistic_data: MagicMock) -> Generator[AsyncMock]:
"homeassistant.components.paperless_ngx.config_flow.Paperless", "homeassistant.components.paperless_ngx.config_flow.Paperless",
new=paperless_mock, new=paperless_mock,
), ),
patch(
"homeassistant.components.paperless_ngx.Paperless",
new=paperless_mock,
),
): ):
paperless = paperless_mock.return_value paperless = paperless_mock.return_value
@ -51,6 +63,11 @@ def mock_paperless(mock_statistic_data: MagicMock) -> Generator[AsyncMock]:
paperless, data=mock_statistic_data, fetched=True paperless, data=mock_statistic_data, fetched=True
) )
) )
paperless.status = AsyncMock(
return_value=Status.create_with_data(
paperless, data=mock_status_data, fetched=True
)
)
yield paperless yield paperless

View File

@ -0,0 +1,36 @@
{
"pngx_version": "2.15.3",
"server_os": "Linux-6.6.74-haos-raspi-aarch64-with-glibc2.36",
"install_type": "docker",
"storage": {
"total": 62101651456,
"available": 25376927744
},
"database": {
"type": "sqlite",
"url": "/config/data/db.sqlite3",
"status": "OK",
"error": null,
"migration_status": {
"latest_migration": "paperless_mail.0029_mailrule_pdf_layout",
"unapplied_migrations": []
}
},
"tasks": {
"redis_url": "redis://localhost:6379",
"redis_status": "OK",
"redis_error": null,
"celery_status": "OK",
"celery_url": "celery@ca5234a0-paperless-ngx",
"celery_error": null,
"index_status": "OK",
"index_last_modified": "2025-05-25T00:00:27.053090+02:00",
"index_error": null,
"classifier_status": "OK",
"classifier_last_trained": "2025-05-25T15:05:15.824671Z",
"classifier_error": null,
"sanity_check_status": "OK",
"sanity_check_last_run": "2025-05-24T22:30:21.005536Z",
"sanity_check_error": null
}
}

View File

@ -2,28 +2,85 @@
# name: test_config_entry_diagnostics # name: test_config_entry_diagnostics
dict({ dict({
'data': dict({ 'data': dict({
'character_count': 99999, 'statistics': dict({
'correspondent_count': 99, 'character_count': 99999,
'current_asn': 99, 'correspondent_count': 99,
'document_file_type_counts': list([ 'current_asn': 99,
dict({ 'document_file_type_counts': list([
'mime_type': 'application/pdf', dict({
'mime_type_count': 998, 'mime_type': 'application/pdf',
'mime_type_count': 998,
}),
dict({
'mime_type': 'image/png',
'mime_type_count': 1,
}),
]),
'document_type_count': 99,
'documents_inbox': 9,
'documents_total': 999,
'inbox_tag': 9,
'inbox_tags': list([
9,
]),
'storage_path_count': 9,
'tag_count': 99,
}),
'status': dict({
'database': dict({
'error': None,
'migration_status': dict({
'latest_migration': 'paperless_mail.0029_mailrule_pdf_layout',
'unapplied_migrations': list([
]),
}),
'status': dict({
'__type': "<enum 'StatusType'>",
'repr': "<StatusType.OK: 'OK'>",
}),
'type': 'sqlite',
'url': '/config/data/db.sqlite3',
}), }),
dict({ 'install_type': 'docker',
'mime_type': 'image/png', 'pngx_version': '2.15.3',
'mime_type_count': 1, 'server_os': 'Linux-6.6.74-haos-raspi-aarch64-with-glibc2.36',
'storage': dict({
'available': 25376927744,
'total': 62101651456,
}), }),
]), 'tasks': dict({
'document_type_count': 99, 'celery_error': None,
'documents_inbox': 9, 'celery_status': dict({
'documents_total': 999, '__type': "<enum 'StatusType'>",
'inbox_tag': 9, 'repr': "<StatusType.OK: 'OK'>",
'inbox_tags': list([ }),
9, 'celery_url': 'celery@ca5234a0-paperless-ngx',
]), 'classifier_error': None,
'storage_path_count': 9, 'classifier_last_trained': '2025-05-25T15:05:15.824671+00:00',
'tag_count': 99, 'classifier_status': dict({
'__type': "<enum 'StatusType'>",
'repr': "<StatusType.OK: 'OK'>",
}),
'index_error': None,
'index_last_modified': '2025-05-25T00:00:27.053090+02:00',
'index_status': dict({
'__type': "<enum 'StatusType'>",
'repr': "<StatusType.OK: 'OK'>",
}),
'redis_error': None,
'redis_status': dict({
'__type': "<enum 'StatusType'>",
'repr': "<StatusType.OK: 'OK'>",
}),
'redis_url': 'redis://localhost:6379',
'sanity_check_error': None,
'sanity_check_last_run': '2025-05-24T22:30:21.005536+00:00',
'sanity_check_status': dict({
'__type': "<enum 'StatusType'>",
'repr': "<StatusType.OK: 'OK'>",
}),
}),
}),
}), }),
}) })
# --- # ---

View File

@ -1,4 +1,56 @@
# serializer version: 1 # serializer version: 1
# name: test_sensor_platform[sensor.paperless_ngx_available_storage-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.paperless_ngx_available_storage',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.DATA_SIZE: 'data_size'>,
'original_icon': None,
'original_name': 'Available storage',
'platform': 'paperless_ngx',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'storage_available',
'unique_id': '0KLG00V55WEVTJ0CJHM0GADNGH_storage_available',
'unit_of_measurement': <UnitOfInformation.GIGABYTES: 'GB'>,
})
# ---
# name: test_sensor_platform[sensor.paperless_ngx_available_storage-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'data_size',
'friendly_name': 'Paperless-ngx Available storage',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfInformation.GIGABYTES: 'GB'>,
}),
'context': <ANY>,
'entity_id': 'sensor.paperless_ngx_available_storage',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '25.38',
})
# ---
# name: test_sensor_platform[sensor.paperless_ngx_correspondents-entry] # name: test_sensor_platform[sensor.paperless_ngx_correspondents-entry]
EntityRegistryEntrySnapshot({ EntityRegistryEntrySnapshot({
'aliases': set({ 'aliases': set({
@ -152,6 +204,360 @@
'state': '9', 'state': '9',
}) })
# --- # ---
# name: test_sensor_platform[sensor.paperless_ngx_status_celery-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'ok',
'error',
'warning',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.paperless_ngx_status_celery',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'Status celery',
'platform': 'paperless_ngx',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'celery_status',
'unique_id': '0KLG00V55WEVTJ0CJHM0GADNGH_celery_status',
'unit_of_measurement': None,
})
# ---
# name: test_sensor_platform[sensor.paperless_ngx_status_celery-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'Paperless-ngx Status celery',
'options': list([
'ok',
'error',
'warning',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.paperless_ngx_status_celery',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'ok',
})
# ---
# name: test_sensor_platform[sensor.paperless_ngx_status_classifier-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'ok',
'error',
'warning',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.paperless_ngx_status_classifier',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'Status classifier',
'platform': 'paperless_ngx',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'classifier_status',
'unique_id': '0KLG00V55WEVTJ0CJHM0GADNGH_classifier_status',
'unit_of_measurement': None,
})
# ---
# name: test_sensor_platform[sensor.paperless_ngx_status_classifier-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'Paperless-ngx Status classifier',
'options': list([
'ok',
'error',
'warning',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.paperless_ngx_status_classifier',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'ok',
})
# ---
# name: test_sensor_platform[sensor.paperless_ngx_status_database-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'ok',
'error',
'warning',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.paperless_ngx_status_database',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'Status database',
'platform': 'paperless_ngx',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'database_status',
'unique_id': '0KLG00V55WEVTJ0CJHM0GADNGH_database_status',
'unit_of_measurement': None,
})
# ---
# name: test_sensor_platform[sensor.paperless_ngx_status_database-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'Paperless-ngx Status database',
'options': list([
'ok',
'error',
'warning',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.paperless_ngx_status_database',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'ok',
})
# ---
# name: test_sensor_platform[sensor.paperless_ngx_status_index-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'ok',
'error',
'warning',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.paperless_ngx_status_index',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'Status index',
'platform': 'paperless_ngx',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'index_status',
'unique_id': '0KLG00V55WEVTJ0CJHM0GADNGH_index_status',
'unit_of_measurement': None,
})
# ---
# name: test_sensor_platform[sensor.paperless_ngx_status_index-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'Paperless-ngx Status index',
'options': list([
'ok',
'error',
'warning',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.paperless_ngx_status_index',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'ok',
})
# ---
# name: test_sensor_platform[sensor.paperless_ngx_status_redis-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'ok',
'error',
'warning',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.paperless_ngx_status_redis',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'Status redis',
'platform': 'paperless_ngx',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'redis_status',
'unique_id': '0KLG00V55WEVTJ0CJHM0GADNGH_redis_status',
'unit_of_measurement': None,
})
# ---
# name: test_sensor_platform[sensor.paperless_ngx_status_redis-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'Paperless-ngx Status redis',
'options': list([
'ok',
'error',
'warning',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.paperless_ngx_status_redis',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'ok',
})
# ---
# name: test_sensor_platform[sensor.paperless_ngx_status_sanity-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'ok',
'error',
'warning',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.paperless_ngx_status_sanity',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'Status sanity',
'platform': 'paperless_ngx',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'sanity_check_status',
'unique_id': '0KLG00V55WEVTJ0CJHM0GADNGH_sanity_check_status',
'unit_of_measurement': None,
})
# ---
# name: test_sensor_platform[sensor.paperless_ngx_status_sanity-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'Paperless-ngx Status sanity',
'options': list([
'ok',
'error',
'warning',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.paperless_ngx_status_sanity',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'ok',
})
# ---
# name: test_sensor_platform[sensor.paperless_ngx_tags-entry] # name: test_sensor_platform[sensor.paperless_ngx_tags-entry]
EntityRegistryEntrySnapshot({ EntityRegistryEntrySnapshot({
'aliases': set({ 'aliases': set({
@ -305,3 +711,55 @@
'state': '999', 'state': '999',
}) })
# --- # ---
# name: test_sensor_platform[sensor.paperless_ngx_total_storage-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.paperless_ngx_total_storage',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.DATA_SIZE: 'data_size'>,
'original_icon': None,
'original_name': 'Total storage',
'platform': 'paperless_ngx',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'storage_total',
'unique_id': '0KLG00V55WEVTJ0CJHM0GADNGH_storage_total',
'unit_of_measurement': <UnitOfInformation.GIGABYTES: 'GB'>,
})
# ---
# name: test_sensor_platform[sensor.paperless_ngx_total_storage-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'data_size',
'friendly_name': 'Paperless-ngx Total storage',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfInformation.GIGABYTES: 'GB'>,
}),
'context': <ANY>,
'entity_id': 'sensor.paperless_ngx_total_storage',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '62.1',
})
# ---

View File

@ -34,10 +34,28 @@ async def test_load_unload_config_entry(
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
async def test_load_config_status_forbidden(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_paperless: AsyncMock,
) -> None:
"""Test loading and unloading the integration."""
mock_paperless.status.side_effect = PaperlessForbiddenError
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( @pytest.mark.parametrize(
("side_effect", "expected_state", "expected_error_key"), ("side_effect", "expected_state", "expected_error_key"),
[ [
(PaperlessConnectionError(), ConfigEntryState.SETUP_RETRY, None), (PaperlessConnectionError(), ConfigEntryState.SETUP_RETRY, "cannot_connect"),
(PaperlessInvalidTokenError(), ConfigEntryState.SETUP_ERROR, "invalid_api_key"), (PaperlessInvalidTokenError(), ConfigEntryState.SETUP_ERROR, "invalid_api_key"),
( (
PaperlessInactiveOrDeletedError(), PaperlessInactiveOrDeletedError(),

View File

@ -1,7 +1,5 @@
"""Tests for Paperless-ngx sensor platform.""" """Tests for Paperless-ngx sensor platform."""
from datetime import timedelta
from freezegun.api import FrozenDateTimeFactory from freezegun.api import FrozenDateTimeFactory
from pypaperless.exceptions import ( from pypaperless.exceptions import (
PaperlessConnectionError, PaperlessConnectionError,
@ -12,7 +10,9 @@ from pypaperless.exceptions import (
from pypaperless.models import Statistic from pypaperless.models import Statistic
import pytest import pytest
from homeassistant.components.paperless_ngx.coordinator import UPDATE_INTERVAL from homeassistant.components.paperless_ngx.coordinator import (
UPDATE_INTERVAL_STATISTICS,
)
from homeassistant.const import STATE_UNAVAILABLE, Platform from homeassistant.const import STATE_UNAVAILABLE, Platform
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er from homeassistant.helpers import entity_registry as er
@ -61,7 +61,7 @@ async def test_statistic_sensor_state(
) )
) )
freezer.tick(timedelta(seconds=UPDATE_INTERVAL)) freezer.tick(UPDATE_INTERVAL_STATISTICS)
async_fire_time_changed(hass) async_fire_time_changed(hass)
await hass.async_block_till_done() await hass.async_block_till_done()
@ -91,7 +91,7 @@ async def test__statistic_sensor_state_on_error(
# simulate error # simulate error
mock_paperless.statistics.side_effect = error_cls mock_paperless.statistics.side_effect = error_cls
freezer.tick(timedelta(seconds=UPDATE_INTERVAL)) freezer.tick(UPDATE_INTERVAL_STATISTICS)
async_fire_time_changed(hass) async_fire_time_changed(hass)
await hass.async_block_till_done() await hass.async_block_till_done()
@ -105,7 +105,7 @@ async def test__statistic_sensor_state_on_error(
) )
) )
freezer.tick(timedelta(seconds=UPDATE_INTERVAL)) freezer.tick(UPDATE_INTERVAL_STATISTICS)
async_fire_time_changed(hass) async_fire_time_changed(hass)
await hass.async_block_till_done() await hass.async_block_till_done()