github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

Add PrusaLink integration (#77429) 

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Paulus Schoutsen 2022-08-29 20:45:27 -04:00 committed by GitHub
parent 035cd16a95
commit 481205535c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
26 changed files with 1061 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.strict-typing
View File

@ -201,6 +201,7 @@ homeassistant.components.persistent_notification.*
homeassistant.components.pi_hole.*
homeassistant.components.powerwall.*
homeassistant.components.proximity.*
homeassistant.components.prusalink.*
homeassistant.components.pvoutput.*
homeassistant.components.pure_energie.*
homeassistant.components.qnap_qsw.*

2
CODEOWNERS
View File

@ -846,6 +846,8 @@ build.json @home-assistant/supervisor
/homeassistant/components/prosegur/ @dgomes
/tests/components/prosegur/ @dgomes
/homeassistant/components/proxmoxve/ @jhollowe @Corbeno
/homeassistant/components/prusalink/ @balloob
/tests/components/prusalink/ @balloob
/homeassistant/components/ps4/ @ktnrg45
/tests/components/ps4/ @ktnrg45
/homeassistant/components/pure_energie/ @klaasnicolaas

120
homeassistant/components/prusalink/__init__.py Normal file
View File

@ -0,0 +1,120 @@
"""The PrusaLink integration."""
from __future__ import annotations
from abc import abstractmethod
from datetime import timedelta
import logging
from typing import Generic, TypeVar
import async_timeout
from pyprusalink import InvalidAuth, JobInfo, PrinterInfo, PrusaLink, PrusaLinkError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
UpdateFailed,
)
from .const import DOMAIN
PLATFORMS: list[Platform] = [Platform.SENSOR, Platform.CAMERA]
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up PrusaLink from a config entry."""
api = PrusaLink(
async_get_clientsession(hass),
entry.data["host"],
entry.data["api_key"],
)
coordinators = {
"printer": PrinterUpdateCoordinator(hass, api),
"job": JobUpdateCoordinator(hass, api),
}
for coordinator in coordinators.values():
await coordinator.async_config_entry_first_refresh()
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinators
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
T = TypeVar("T", PrinterInfo, JobInfo)
class PrusaLinkUpdateCoordinator(DataUpdateCoordinator, Generic[T]):
"""Update coordinator for the printer."""
config_entry: ConfigEntry
def __init__(self, hass: HomeAssistant, api: PrusaLink) -> None:
"""Initialize the update coordinator."""
self.api = api
super().__init__(
hass, _LOGGER, name=DOMAIN, update_interval=timedelta(seconds=30)
)
async def _async_update_data(self) -> T:
"""Update the data."""
try:
with async_timeout.timeout(5):
return await self._fetch_data()
except InvalidAuth:
raise UpdateFailed("Invalid authentication") from None
except PrusaLinkError as err:
raise UpdateFailed(str(err)) from err
@abstractmethod
async def _fetch_data(self) -> T:
"""Fetch the actual data."""
raise NotImplementedError
class PrinterUpdateCoordinator(PrusaLinkUpdateCoordinator[PrinterInfo]):
"""Printer update coordinator."""
async def _fetch_data(self) -> PrinterInfo:
"""Fetch the printer data."""
return await self.api.get_printer()
class JobUpdateCoordinator(PrusaLinkUpdateCoordinator[JobInfo]):
"""Job update coordinator."""
async def _fetch_data(self) -> JobInfo:
"""Fetch the printer data."""
return await self.api.get_job()
class PrusaLinkEntity(CoordinatorEntity[PrusaLinkUpdateCoordinator]):
"""Defines a base PrusaLink entity."""
_attr_has_entity_name = True
@property
def device_info(self) -> DeviceInfo:
"""Return device information about this PrusaLink device."""
return DeviceInfo(
identifiers={(DOMAIN, self.coordinator.config_entry.entry_id)},
name=self.coordinator.config_entry.title,
manufacturer="Prusa",
configuration_url=self.coordinator.api.host,
)

54
homeassistant/components/prusalink/camera.py Normal file
View File

@ -0,0 +1,54 @@
"""Camera entity for PrusaLink."""
from __future__ import annotations
from homeassistant.components.camera import Camera
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import DOMAIN, JobUpdateCoordinator, PrusaLinkEntity
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up PrusaLink camera."""
coordinator: JobUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]["job"]
async_add_entities([PrusaLinkJobPreviewEntity(coordinator)])
class PrusaLinkJobPreviewEntity(PrusaLinkEntity, Camera):
"""Defines a PrusaLink camera."""
last_path = ""
last_image: bytes
_attr_name = "Job Preview"
def __init__(self, coordinator: JobUpdateCoordinator) -> None:
"""Initialize a PrusaLink camera entity."""
super().__init__(coordinator)
Camera.__init__(self)
self._attr_unique_id = f"{self.coordinator.config_entry.entry_id}_job_preview"
@property
def available(self) -> bool:
"""Get if camera is available."""
return super().available and self.coordinator.data.get("job") is not None
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a still image from the camera."""
if not self.available:
return None
path = self.coordinator.data["job"]["file"]["path"]
if self.last_path == path:
return self.last_image
self.last_image = await self.coordinator.api.get_large_thumbnail(path)
self.last_path = path
return self.last_image

106
homeassistant/components/prusalink/config_flow.py Normal file
View File

@ -0,0 +1,106 @@
"""Config flow for PrusaLink integration."""
from __future__ import annotations
import asyncio
import logging
from typing import Any
from aiohttp import ClientError
import async_timeout
from awesomeversion import AwesomeVersion, AwesomeVersionException
from pyprusalink import InvalidAuth, PrusaLink
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResult
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
def add_protocol(value: str) -> str:
"""Validate URL has a scheme."""
value = value.rstrip("/")
if value.startswith(("http://", "https://")):
return value
return f"http://{value}"
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required("host"): vol.All(str, add_protocol),
vol.Required("api_key"): str,
}
)
async def validate_input(hass: HomeAssistant, data: dict[str, str]) -> dict[str, str]:
"""Validate the user input allows us to connect.
Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
"""
api = PrusaLink(async_get_clientsession(hass), data["host"], data["api_key"])
try:
async with async_timeout.timeout(5):
version = await api.get_version()
except (asyncio.TimeoutError, ClientError) as err:
_LOGGER.error("Could not connect to PrusaLink: %s", err)
raise CannotConnect from err
try:
if AwesomeVersion(version["api"]) < AwesomeVersion("2.0.0"):
raise NotSupported
except AwesomeVersionException as err:
raise NotSupported from err
return {"title": version["hostname"]}
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for PrusaLink."""
VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
if user_input is None:
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA
)
errors = {}
try:
info = await validate_input(self.hass, user_input)
except CannotConnect:
errors["base"] = "cannot_connect"
except NotSupported:
errors["base"] = "not_supported"
except InvalidAuth:
errors["base"] = "invalid_auth"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
return self.async_create_entry(title=info["title"], data=user_input)
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
)
class CannotConnect(HomeAssistantError):
"""Error to indicate we cannot connect."""
class NotSupported(HomeAssistantError):
"""Error to indicate we cannot connect."""

3
homeassistant/components/prusalink/const.py Normal file
View File

@ -0,0 +1,3 @@
"""Constants for the PrusaLink integration."""
DOMAIN = "prusalink"

14
homeassistant/components/prusalink/manifest.json Normal file
View File

@ -0,0 +1,14 @@
{
"domain": "prusalink",
"name": "PrusaLink",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/prusalink",
"requirements": ["pyprusalink==1.0.1"],
"dhcp": [
{
"macaddress": "109C70*"
}
],
"codeowners": ["@balloob"],
"iot_class": "local_polling"
}

173
homeassistant/components/prusalink/sensor.py Normal file
View File

@ -0,0 +1,173 @@
"""PrusaLink sensors."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Generic, TypeVar, cast
from pyprusalink import JobInfo, PrinterInfo
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import PERCENTAGE, TEMP_CELSIUS
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.util.dt import utcnow
from homeassistant.util.variance import ignore_variance
from . import DOMAIN, PrusaLinkEntity, PrusaLinkUpdateCoordinator
T = TypeVar("T", PrinterInfo, JobInfo)
@dataclass
class PrusaLinkSensorEntityDescriptionMixin(Generic[T]):
"""Mixin for required keys."""
value_fn: Callable[[T], datetime | StateType]
@dataclass
class PrusaLinkSensorEntityDescription(
SensorEntityDescription, PrusaLinkSensorEntityDescriptionMixin[T], Generic[T]
):
"""Describes PrusaLink sensor entity."""
available_fn: Callable[[T], bool] = lambda _: True
SENSORS: dict[str, tuple[PrusaLinkSensorEntityDescription, ...]] = {
"printer": (
PrusaLinkSensorEntityDescription[PrinterInfo](
key="printer.state",
icon="mdi:printer-3d",
value_fn=lambda data: (
"pausing"
if (flags := data["state"]["flags"])["pausing"]
else "cancelling"
if flags["cancelling"]
else "paused"
if flags["paused"]
else "printing"
if flags["printing"]
else "idle"
),
device_class="prusalink__printer_state",
),
PrusaLinkSensorEntityDescription[PrinterInfo](
key="printer.telemetry.temp-bed",
name="Heatbed",
native_unit_of_measurement=TEMP_CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: cast(float, data["telemetry"]["temp-bed"]),
entity_registry_enabled_default=False,
),
PrusaLinkSensorEntityDescription[PrinterInfo](
key="printer.telemetry.temp-nozzle",
name="Nozzle Temperature",
native_unit_of_measurement=TEMP_CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: cast(float, data["telemetry"]["temp-nozzle"]),
entity_registry_enabled_default=False,
),
),
"job": (
PrusaLinkSensorEntityDescription[JobInfo](
key="job.progress",
name="Progress",
native_unit_of_measurement=PERCENTAGE,
value_fn=lambda data: cast(float, data["progress"]["completion"]) * 100,
available_fn=lambda data: data.get("progress") is not None,
),
PrusaLinkSensorEntityDescription[JobInfo](
key="job.filename",
name="Filename",
icon="mdi:file-image-outline",
value_fn=lambda data: cast(str, data["job"]["file"]["display"]),
available_fn=lambda data: data.get("job") is not None,
),
PrusaLinkSensorEntityDescription[JobInfo](
key="job.start",
name="Print Start",
device_class=SensorDeviceClass.TIMESTAMP,
value_fn=ignore_variance(
lambda data: (
utcnow() - timedelta(seconds=data["progress"]["printTime"])
),
timedelta(minutes=2),
),
available_fn=lambda data: data.get("progress") is not None,
),
PrusaLinkSensorEntityDescription[JobInfo](
key="job.finish",
name="Print Finish",
device_class=SensorDeviceClass.TIMESTAMP,
value_fn=ignore_variance(
lambda data: (
utcnow() + timedelta(seconds=data["progress"]["printTimeLeft"])
),
timedelta(minutes=2),
),
available_fn=lambda data: data.get("progress") is not None,
),
),
}
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up PrusaLink sensor based on a config entry."""
coordinators: dict[str, PrusaLinkUpdateCoordinator] = hass.data[DOMAIN][
entry.entry_id
]
entities: list[PrusaLinkEntity] = []
for coordinator_type, sensors in SENSORS.items():
coordinator = coordinators[coordinator_type]
entities.extend(
PrusaLinkSensorEntity(coordinator, sensor_description)
for sensor_description in sensors
)
async_add_entities(entities)
class PrusaLinkSensorEntity(PrusaLinkEntity, SensorEntity):
"""Defines a PrusaLink sensor."""
entity_description: PrusaLinkSensorEntityDescription
def __init__(
self,
coordinator: PrusaLinkUpdateCoordinator,
description: PrusaLinkSensorEntityDescription,
) -> None:
"""Initialize a PrusaLink sensor entity."""
super().__init__(coordinator=coordinator)
self.entity_description = description
self._attr_unique_id = f"{coordinator.config_entry.entry_id}_{description.key}"
@property
def native_value(self) -> datetime | StateType:
"""Return the state of the sensor."""
return self.entity_description.value_fn(self.coordinator.data)
@property
def available(self) -> bool:
"""Return if sensor is available."""
return super().available and self.entity_description.available_fn(
self.coordinator.data
)

18
homeassistant/components/prusalink/strings.json Normal file
View File

@ -0,0 +1,18 @@
{
"config": {
"step": {
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"api_key": "[%key:common::config_flow::data::api_key%]"
}
}
},
"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%]",
"not_supported": "Only PrusaLink API v2 is supported"
}
}
}

11
homeassistant/components/prusalink/strings.sensor.json Normal file
View File

@ -0,0 +1,11 @@
{
"state": {
"prusalink__printer_state": {
"pausing": "Pausing",
"cancelling": "Cancelling",
"paused": "Paused",
"printing": "Printing",
"idle": "Idle"
}
}
}

17
homeassistant/components/prusalink/translations/en.json Normal file
View File

@ -0,0 +1,17 @@
{
"config": {
"error": {
"cannot_connect": "Failed to connect",
"invalid_auth": "Invalid authentication",
"unknown": "Unexpected error"
},
"step": {
"user": {
"data": {
"api_key": "API Key",
"host": "Host"
}
}
}
}
}

11
homeassistant/components/prusalink/translations/sensor.en.json Normal file
View File

@ -0,0 +1,11 @@
{
"state": {
"prusalink__printer_state": {
"cancelling": "Cancelling",
"idle": "Idle",
"paused": "Paused",
"pausing": "Pausing",
"printing": "Printing"
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -290,6 +290,7 @@ FLOWS = {
"profiler",
"progettihwsw",
"prosegur",
"prusalink",
"ps4",
"pure_energie",
"pushover",

1
homeassistant/generated/dhcp.py
View File

@ -80,6 +80,7 @@ DHCP: list[dict[str, str | bool]] = [
{'domain': 'oncue', 'hostname': 'kohlergen*', 'macaddress': '00146F*'},
{'domain': 'overkiz', 'hostname': 'gateway*', 'macaddress': 'F8811A*'},
{'domain': 'powerwall', 'hostname': '1118431-*'},
{'domain': 'prusalink', 'macaddress': '109C70*'},
{'domain': 'qnap_qsw', 'macaddress': '245EBE*'},
{'domain': 'rachio', 'hostname': 'rachio-*', 'macaddress': '009D6B*'},
{'domain': 'rachio', 'hostname': 'rachio-*', 'macaddress': 'F0038C*'},

49
homeassistant/util/variance.py Normal file
View File

@ -0,0 +1,49 @@
"""Util functions to help filter out similar results."""
from __future__ import annotations
from collections.abc import Callable
from datetime import datetime, timedelta
import functools
from typing import Any, TypeVar, overload
T = TypeVar("T", int, float, datetime)
@overload
def ignore_variance(
func: Callable[..., int], ignored_variance: int
) -> Callable[..., int]:
...
@overload
def ignore_variance(
func: Callable[..., float], ignored_variance: float
) -> Callable[..., float]:
...
@overload
def ignore_variance(
func: Callable[..., datetime], ignored_variance: timedelta
) -> Callable[..., datetime]:
...
def ignore_variance(func: Callable[..., T], ignored_variance: Any) -> Callable[..., T]:
"""Wrap a function that returns old result if new result does not vary enough."""
last_value: T | None = None
@functools.wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> T:
nonlocal last_value
value = func(*args, **kwargs)
if last_value is not None and abs(value - last_value) < ignored_variance:
return last_value
last_value = value
return value
return wrapper

10
mypy.ini
View File

@ -1769,6 +1769,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.prusalink.*]
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.pvoutput.*]
check_untyped_defs = true
disallow_incomplete_defs = true

3
requirements_all.txt
View File

@ -1780,6 +1780,9 @@ pyprof2calltree==1.4.5
# homeassistant.components.prosegur
pyprosegur==0.0.5
# homeassistant.components.prusalink
pyprusalink==1.0.1
# homeassistant.components.ps4
pyps4-2ndscreen==1.3.1

3
requirements_test_all.txt
View File

@ -1251,6 +1251,9 @@ pyprof2calltree==1.4.5
# homeassistant.components.prosegur
pyprosegur==0.0.5
# homeassistant.components.prusalink
pyprusalink==1.0.1
# homeassistant.components.ps4
pyps4-2ndscreen==1.3.1

2
script/scaffold/templates/config_flow/integration/__init__.py
View File

@ -21,7 +21,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
# TODO 3. Store an API object for your platforms to access
# hass.data[DOMAIN][entry.entry_id] = MyApi(...)
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True

1
tests/components/prusalink/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the PrusaLink integration."""

109
tests/components/prusalink/conftest.py Normal file
View File

@ -0,0 +1,109 @@
"""Fixtures for PrusaLink."""
from unittest.mock import patch
import pytest
from tests.common import MockConfigEntry
@pytest.fixture
def mock_config_entry(hass):
"""Mock a PrusaLink config entry."""
entry = MockConfigEntry(
domain="prusalink", data={"host": "http://example.com", "api_key": "abcdefgh"}
)
entry.add_to_hass(hass)
return entry
@pytest.fixture
def mock_version_api(hass):
"""Mock PrusaLink version API."""
resp = {
"api": "2.0.0",
"server": "2.1.2",
"text": "PrusaLink MINI",
"hostname": "PrusaMINI",
}
with patch("pyprusalink.PrusaLink.get_version", return_value=resp):
yield resp
@pytest.fixture
def mock_printer_api(hass):
"""Mock PrusaLink printer API."""
resp = {
"telemetry": {
"temp-bed": 41.9,
"temp-nozzle": 47.8,
"print-speed": 100,
"z-height": 1.8,
"material": "PLA",
},
"temperature": {
"tool0": {"actual": 47.8, "target": 0.0, "display": 0.0, "offset": 0},
"bed": {"actual": 41.9, "target": 0.0, "offset": 0},
},
"state": {
"text": "Operational",
"flags": {
"operational": True,
"paused": False,
"printing": False,
"cancelling": False,
"pausing": False,
"sdReady": False,
"error": False,
"closedOnError": False,
"ready": True,
"busy": False,
},
},
}
with patch("pyprusalink.PrusaLink.get_printer", return_value=resp):
yield resp
@pytest.fixture
def mock_job_api_idle(hass):
"""Mock PrusaLink job API having no job."""
with patch(
"pyprusalink.PrusaLink.get_job",
return_value={
"state": "Operational",
"job": None,
"progress": None,
},
):
yield
@pytest.fixture
def mock_job_api_active(hass):
"""Mock PrusaLink job API having no job."""
with patch(
"pyprusalink.PrusaLink.get_job",
return_value={
"state": "Printing",
"job": {
"estimatedPrintTime": 117007,
"file": {
"name": "TabletStand3.gcode",
"path": "/usb/TABLET~1.GCO",
"display": "TabletStand3.gcode",
},
},
"progress": {
"completion": 0.37,
"printTime": 43987,
"printTimeLeft": 73020,
},
},
):
yield
@pytest.fixture
def mock_api(mock_version_api, mock_printer_api, mock_job_api_idle):
"""Mock PrusaLink API."""

51
tests/components/prusalink/test_camera.py Normal file
View File

@ -0,0 +1,51 @@
"""Test Prusalink camera."""
from unittest.mock import patch
import pytest
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
@pytest.fixture(autouse=True)
def setup_camera_platform_only():
"""Only setup camera platform."""
with patch("homeassistant.components.prusalink.PLATFORMS", [Platform.CAMERA]):
yield
async def test_camera_no_job(
hass: HomeAssistant,
mock_config_entry,
mock_api,
) -> None:
"""Test sensors while no job active."""
assert await async_setup_component(hass, "prusalink", {})
state = hass.states.get("camera.mock_title_job_preview")
assert state is not None
assert state.state == "unavailable"
async def test_camera_active_job(
hass: HomeAssistant, mock_config_entry, mock_api, mock_job_api_active, hass_client
):
"""Test sensors while no job active."""
assert await async_setup_component(hass, "prusalink", {})
state = hass.states.get("camera.mock_title_job_preview")
assert state is not None
assert state.state == "idle"
client = await hass_client()
with patch("pyprusalink.PrusaLink.get_large_thumbnail", return_value=b"hello"):
resp = await client.get("/api/camera_proxy/camera.mock_title_job_preview")
assert resp.status == 200
assert await resp.read() == b"hello"
# Make sure we hit cached value.
with patch("pyprusalink.PrusaLink.get_large_thumbnail", side_effect=ValueError):
resp = await client.get("/api/camera_proxy/camera.mock_title_job_preview")
assert resp.status == 200
assert await resp.read() == b"hello"

125
tests/components/prusalink/test_config_flow.py Normal file
View File

@ -0,0 +1,125 @@
"""Test the PrusaLink config flow."""
import asyncio
from unittest.mock import patch
from homeassistant import config_entries
from homeassistant.components.prusalink.config_flow import InvalidAuth
from homeassistant.components.prusalink.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
async def test_form(hass: HomeAssistant, mock_version_api) -> None:
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] is None
with patch(
"homeassistant.components.prusalink.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "http://1.1.1.1/",
"api_key": "abcdefg",
},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "PrusaMINI"
assert result2["data"] == {
"host": "http://1.1.1.1",
"api_key": "abcdefg",
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_invalid_auth(hass: HomeAssistant) -> None:
"""Test we handle invalid auth."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.prusalink.config_flow.PrusaLink.get_version",
side_effect=InvalidAuth,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "1.1.1.1",
"api_key": "abcdefg",
},
)
assert result2["type"] == FlowResultType.FORM
assert result2["errors"] == {"base": "invalid_auth"}
async def test_form_unknown(hass: HomeAssistant) -> None:
"""Test we handle unknown error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.prusalink.config_flow.PrusaLink.get_version",
side_effect=ValueError,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "1.1.1.1",
"api_key": "abcdefg",
},
)
assert result2["type"] == FlowResultType.FORM
assert result2["errors"] == {"base": "unknown"}
async def test_form_invalid_version(hass: HomeAssistant, mock_version_api) -> None:
"""Test we handle invalid auth."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
mock_version_api["api"] = "1.2.0"
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "1.1.1.1",
"api_key": "abcdefg",
},
)
assert result2["type"] == FlowResultType.FORM
assert result2["errors"] == {"base": "not_supported"}
async def test_form_cannot_connect(hass: HomeAssistant) -> None:
"""Test we handle cannot connect error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.prusalink.config_flow.PrusaLink.get_version",
side_effect=asyncio.TimeoutError,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "1.1.1.1",
"api_key": "abcdefg",
},
)
assert result2["type"] == FlowResultType.FORM
assert result2["errors"] == {"base": "cannot_connect"}

23
tests/components/prusalink/test_init.py Normal file
View File

@ -0,0 +1,23 @@
"""Test setting up and unloading PrusaLink."""
from homeassistant.config_entries import ConfigEntry, ConfigEntryState
from homeassistant.core import HomeAssistant
async def test_sensors_no_job(
hass: HomeAssistant,
mock_config_entry: ConfigEntry,
mock_api,
):
"""Test sensors while no job active."""
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state == ConfigEntryState.LOADED
assert hass.states.async_entity_ids_count() > 0
assert await hass.config_entries.async_unload(mock_config_entry.entry_id)
assert mock_config_entry.state == ConfigEntryState.NOT_LOADED
for state in hass.states.async_all():
assert state.state == "unavailable"

114
tests/components/prusalink/test_sensor.py Normal file
View File

@ -0,0 +1,114 @@
"""Test Prusalink sensors."""
from datetime import datetime, timezone
from unittest.mock import PropertyMock, patch
import pytest
from homeassistant.components.sensor import (
ATTR_STATE_CLASS,
SensorDeviceClass,
SensorStateClass,
)
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_UNIT_OF_MEASUREMENT,
TEMP_CELSIUS,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
@pytest.fixture(autouse=True)
def setup_sensor_platform_only():
"""Only setup sensor platform."""
with patch(
"homeassistant.components.prusalink.PLATFORMS", [Platform.SENSOR]
), patch(
"homeassistant.helpers.entity.Entity.entity_registry_enabled_default",
PropertyMock(return_value=True),
):
yield
async def test_sensors_no_job(hass: HomeAssistant, mock_config_entry, mock_api):
"""Test sensors while no job active."""
assert await async_setup_component(hass, "prusalink", {})
state = hass.states.get("sensor.mock_title")
assert state is not None
assert state.state == "idle"
state = hass.states.get("sensor.mock_title_heatbed")
assert state is not None
assert state.state == "41.9"
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == TEMP_CELSIUS
assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TEMPERATURE
assert state.attributes[ATTR_STATE_CLASS] == SensorStateClass.MEASUREMENT
state = hass.states.get("sensor.mock_title_nozzle_temperature")
assert state is not None
assert state.state == "47.8"
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == TEMP_CELSIUS
assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TEMPERATURE
assert state.attributes[ATTR_STATE_CLASS] == SensorStateClass.MEASUREMENT
state = hass.states.get("sensor.mock_title_progress")
assert state is not None
assert state.state == "unavailable"
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == "%"
state = hass.states.get("sensor.mock_title_filename")
assert state is not None
assert state.state == "unavailable"
state = hass.states.get("sensor.mock_title_print_start")
assert state is not None
assert state.state == "unavailable"
assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP
state = hass.states.get("sensor.mock_title_print_finish")
assert state is not None
assert state.state == "unavailable"
assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP
async def test_sensors_active_job(
hass: HomeAssistant,
mock_config_entry,
mock_api,
mock_printer_api,
mock_job_api_active,
):
"""Test sensors while active job."""
mock_printer_api["state"]["flags"]["printing"] = True
with patch(
"homeassistant.components.prusalink.sensor.utcnow",
return_value=datetime(2022, 8, 27, 14, 0, 0, tzinfo=timezone.utc),
):
assert await async_setup_component(hass, "prusalink", {})
state = hass.states.get("sensor.mock_title")
assert state is not None
assert state.state == "printing"
state = hass.states.get("sensor.mock_title_progress")
assert state is not None
assert state.state == "37.0"
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == "%"
state = hass.states.get("sensor.mock_title_filename")
assert state is not None
assert state.state == "TabletStand3.gcode"
state = hass.states.get("sensor.mock_title_print_start")
assert state is not None
assert state.state == "2022-08-27T01:46:53+00:00"
assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP
state = hass.states.get("sensor.mock_title_print_finish")
assert state is not None
assert state.state == "2022-08-28T10:17:00+00:00"
assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP

40
tests/util/test_variance.py Normal file
View File

@ -0,0 +1,40 @@
"""Test variance method."""
from datetime import datetime, timedelta
import pytest
from homeassistant.util.variance import ignore_variance
@pytest.mark.parametrize(
"value_1, value_2, variance, expected",
[
(1, 1, 1, 1),
(1, 2, 2, 1),
(1, 2, 0, 2),
(2, 1, 0, 1),
(
datetime(2020, 1, 1, 0, 0),
datetime(2020, 1, 2, 0, 0),
timedelta(days=2),
datetime(2020, 1, 1, 0, 0),
),
(
datetime(2020, 1, 2, 0, 0),
datetime(2020, 1, 1, 0, 0),
timedelta(days=2),
datetime(2020, 1, 2, 0, 0),
),
(
datetime(2020, 1, 1, 0, 0),
datetime(2020, 1, 2, 0, 0),
timedelta(days=1),
datetime(2020, 1, 2, 0, 0),
),
],
)
def test_ignore_variance(value_1, value_2, variance, expected):
"""Test ignore_variance."""
with_ignore = ignore_variance(lambda x: x, variance)
assert with_ignore(value_1) == value_1
assert with_ignore(value_2) == expected