diff --git a/majordom_integration_sdk/__init__.py b/majordom_integration_sdk/__init__.py index c19b8b5..ab51dc3 100644 --- a/majordom_integration_sdk/__init__.py +++ b/majordom_integration_sdk/__init__.py @@ -8,12 +8,14 @@ - ``majordom_integration_sdk.discovery`` — zeroconf / SSDP / BLE discovery services - ``majordom_integration_sdk.testing`` — shared test doubles for integration test suites - ``majordom_integration_sdk.dev`` — standalone runner for developing an integration without the Hub +- ``majordom_integration_sdk.errors`` — errors an integration raises for the Hub to act on The names an integration author reaches for most are re-exported here for convenience, e.g. ``from majordom_integration_sdk import AbstractController, Device``. """ from majordom_integration_sdk.controller import AbstractController, ControllerOutput +from majordom_integration_sdk.errors import MissingCredentialsError from majordom_integration_sdk.repository import ( DeviceRepositoryMemory, DeviceRepositoryProtocol, @@ -38,4 +40,5 @@ "ProvidedCredentials", "Event", "DeviceParameterChange", + "MissingCredentialsError", ] diff --git a/majordom_integration_sdk/dev/cli.py b/majordom_integration_sdk/dev/cli.py index b987952..5b3242b 100644 --- a/majordom_integration_sdk/dev/cli.py +++ b/majordom_integration_sdk/dev/cli.py @@ -275,7 +275,7 @@ def discoveries(): if not found: _print("(no discoveries yet)") for id, d in found.items(): - expects = ", ".join(c.value for c in d.expected_credentials_options) + expects = ", ".join(o.type.value for o in d.expected_credentials_options) _print(f"{id} {d.device_name} transport={d.transport} expects=[{expects}]") @app.command() diff --git a/majordom_integration_sdk/errors.py b/majordom_integration_sdk/errors.py new file mode 100644 index 0000000..af756b1 --- /dev/null +++ b/majordom_integration_sdk/errors.py @@ -0,0 +1,33 @@ +"""Errors an integration raises for the Hub (and through it the app) to act on.""" + +import re +from typing import ClassVar, Final, Self + +_TYPE = re.compile(r"[a-z][a-z0-9]*(?:_[a-z0-9]+)*") +_PREFIX: Final = "Missing credentials: " + + +class MissingCredentialsError(LookupError): + """The Hub lacks credentials of ``type`` to pair a device, e.g. a Hub on Ethernet has no Wi-Fi to give a node. + + Raised before the device is touched, with ``str(error)`` as the discovery's ``last_error``, so the app can ask the + user for them and pair again. ``type`` is a lower_snake_case string kept open for integrations: the known ones are + the class constants below; an integration documents any custom one it raises. The app maps it to an enum. + """ + + WIFI: ClassVar[str] = "wifi" + THREAD: ClassVar[str] = "thread" + + def __init__(self, type: str = WIFI) -> None: + if not _TYPE.fullmatch(type): + raise ValueError(f"Credentials type must be lower_snake_case, got {type!r}") + self.type = type + super().__init__(f"{_PREFIX}{type}") + + @classmethod + def parse(cls, last_error: str | None) -> Self | None: + """The error a ``last_error`` was set from, or None if it was set from anything else.""" + if last_error is None or not last_error.startswith(_PREFIX): + return None + type_ = last_error.removeprefix(_PREFIX) + return cls(type_) if _TYPE.fullmatch(type_) else None diff --git a/majordom_integration_sdk/schemas/__init__.py b/majordom_integration_sdk/schemas/__init__.py index 5430bd1..825c4a5 100644 --- a/majordom_integration_sdk/schemas/__init__.py +++ b/majordom_integration_sdk/schemas/__init__.py @@ -9,6 +9,7 @@ from .base import Base, NonEmptyStr, StrEnum, StrIdentifiable, UUIdentifable from .command import DeviceCommand from .device import ( + CredentialsOption, CredentialsType, CredentialsValue, Device, @@ -44,6 +45,7 @@ "Event", "DeviceParameterChange", # device + "CredentialsOption", "CredentialsType", "CredentialsValue", "Device", diff --git a/majordom_integration_sdk/schemas/device.py b/majordom_integration_sdk/schemas/device.py index 8ede008..f685775 100644 --- a/majordom_integration_sdk/schemas/device.py +++ b/majordom_integration_sdk/schemas/device.py @@ -2,9 +2,10 @@ from datetime import datetime from enum import Enum +from typing import Any from uuid import UUID -from pydantic import Field, SerializeAsAny +from pydantic import Field, SerializeAsAny, field_validator from .base import Base, NonEmptyStr from .parameter import Parameter, ParameterState @@ -18,15 +19,43 @@ class CredentialsType(str, Enum): code = "code" # e.g. 1234-123-1234 (matter) or 123-45-678 (homekit) # TODO: rename to pin qr = "qr" # raw qr data; TODO: decode to see what data is inside some protocols secret = "secret" # AES key like in esphome + # The user physically confirms on the device (ESPHome Improv authorizer, a Matter/HomeKit + # setup button). A confirmation step that carries no value — unlike `none`, which means + # pairing needs nothing from the user at all. + button = "button" none = "none" # like yeelight classic LAN - def with_mask(self, code_mask: str) -> CredentialsType: + def with_mask(self, mask: str, hint: str | None = None) -> CredentialsOption: """ + Returns a CredentialsOption of this type with the mask (and optionally a hint). mask format: D as digit placeholder, other symbols remain unchanged e.g. dashes, for example "DDD-DD-DDD" for "123-45-678" """ - self.code_mask = code_mask - return self + return CredentialsOption(type=self, mask=mask, hint=hint) + + def with_hint(self, hint: str) -> CredentialsOption: + """ + Returns a CredentialsOption of this type with the hint: a short UI prompt, + for example "Press the button on the device" + """ + return CredentialsOption(type=self, hint=hint) + + +class CredentialsOption(Base): + """One way a discovered device can be paired, with what the UI needs to ask for it.""" + + type: CredentialsType + # D is a digit placeholder, other symbols stay as they are: "DDD-DD-DDD" for "123-45-678" + mask: str | None = None + # a short UI prompt, e.g. "Press the button on the device" or "Printed on the label" + hint: str | None = None + + def __eq__(self, other: object) -> bool: + # Compat: integrations check `credentials.type in discovery.expected_credentials_options`; + # a bare type (or its str value) matches the option of that type. + if isinstance(other, str): + return self.type == other + return super().__eq__(other) # FUTURE: @@ -56,7 +85,7 @@ class Discovery(Base): # technical id: UUID integration: NonEmptyStr - expected_credentials_options: list[CredentialsType] # TODO: pass code mask and other things + expected_credentials_options: list[CredentialsOption] expiration: datetime | None = None # for UX transport: NonEmptyStr @@ -71,6 +100,19 @@ class Discovery(Base): # TODO: handle devices with multiple transports or multiple integrations, show options, add priority, allow choosing + @field_validator("expected_credentials_options", mode="before") + @classmethod + def _accept_bare_types(cls, options: Any) -> Any: + # stored discoveries and older integrations pass bare types: ["code"] / [CredentialsType.code] + if not isinstance(options, list): + return options + return [{"type": option} if isinstance(option, str) else option for option in options] + + def accepts(self, credentials: ProvidedCredentials | None) -> bool: + """Whether `credentials` match one of the advertised options; no credentials count as `none`.""" + provided = credentials.type if credentials else CredentialsType.none + return any(option.type == provided for option in self.expected_credentials_options) + # Device diff --git a/tests/test_credentials.py b/tests/test_credentials.py new file mode 100644 index 0000000..583c9d3 --- /dev/null +++ b/tests/test_credentials.py @@ -0,0 +1,124 @@ +"""Pairing credentials: what a discovery expects (CredentialsOption) vs. what a pairing supplies.""" + +import warnings +from uuid import uuid4 + +import pytest + +from majordom_integration_sdk.schemas.device import ( + CredentialsOption, + CredentialsType, + Discovery, + ProvidedCredentials, +) + + +def _discovery(options: list) -> Discovery: + return Discovery( + id=uuid4(), + integration="test", + expected_credentials_options=options, + transport="test", + device_manufacturer=None, + device_name="Bulb", + device_category=None, + device_icon=None, + ) + + +def test_the_mask_is_serialized(): + discovery = _discovery([CredentialsOption(type=CredentialsType.code, mask="DDDD-DDD-DDDD")]) + dumped = discovery.model_dump(mode="json")["expected_credentials_options"] + assert dumped == [{"type": "code", "mask": "DDDD-DDD-DDDD", "hint": None}] + assert Discovery.model_validate_json(discovery.model_dump_json()) == discovery + + +@pytest.mark.parametrize("legacy", [[CredentialsType.code, CredentialsType.qr], ["code", "qr"]]) +def test_bare_types_are_accepted_for_stored_discoveries_and_older_integrations(legacy: list): + discovery = _discovery(legacy) + assert discovery.expected_credentials_options == [ + CredentialsOption(type=CredentialsType.code), + CredentialsOption(type=CredentialsType.qr), + ] + + +def test_a_stored_legacy_discovery_loads_from_json(): + stored = _discovery([CredentialsOption(type=CredentialsType.secret)]).model_dump(mode="json") + stored["expected_credentials_options"] = ["secret"] + loaded = Discovery.model_validate(stored) + assert loaded.expected_credentials_options == [CredentialsOption(type=CredentialsType.secret)] + + +def test_with_mask_no_longer_mutates_the_shared_enum_member(): + with warnings.catch_warnings(): + warnings.simplefilter("error") + option = CredentialsType.code.with_mask("DDD-DD-DDD") + assert option == CredentialsOption(type=CredentialsType.code, mask="DDD-DD-DDD") + assert not hasattr(CredentialsType.code, "code_mask") + # the mask now survives the trip to the Hub + assert _discovery([option]).model_dump(mode="json")["expected_credentials_options"][0]["mask"] == "DDD-DD-DDD" + + +def test_with_hint_returns_an_option_with_the_hint(): + with warnings.catch_warnings(): + warnings.simplefilter("error") + option = CredentialsType.button.with_hint("Press the button") + assert option == CredentialsOption(type=CredentialsType.button, hint="Press the button") + assert not hasattr(CredentialsType.button, "hint") + + +def test_with_mask_can_carry_a_hint(): + option = CredentialsType.code.with_mask("DDD-DD-DDD", hint="On the box") + assert option == CredentialsOption(type=CredentialsType.code, mask="DDD-DD-DDD", hint="On the box") + dumped = _discovery([option]).model_dump(mode="json")["expected_credentials_options"] + assert dumped == [{"type": "code", "mask": "DDD-DD-DDD", "hint": "On the box"}] + + +def test_each_call_returns_a_new_option_and_leaves_the_type_untouched(): + assert CredentialsType.code.with_hint("x") is not CredentialsType.code.with_hint("x") + assert CredentialsType.code.with_mask("DDDD") is not CredentialsType.code.with_mask("DDDD") + assert CredentialsType.code == "code" + + +def test_option_is_a_plain_data_model_without_builders(): + option = CredentialsOption(type=CredentialsType.code) + assert not hasattr(option, "with_mask") + assert not hasattr(option, "with_hint") + + +def test_existing_membership_checks_against_bare_types_keep_working(): + # integrations do `credentials.type not in discovery.expected_credentials_options` + discovery = _discovery([CredentialsOption(type=CredentialsType.code, mask="DDD-DD-DDD")]) + assert CredentialsType.code in discovery.expected_credentials_options + assert CredentialsType.qr not in discovery.expected_credentials_options + + +@pytest.mark.parametrize( + ("credentials", "accepted"), + [ + (ProvidedCredentials(type=CredentialsType.code, value="123-45-678"), True), + (ProvidedCredentials(type=CredentialsType.button), True), + (ProvidedCredentials(type=CredentialsType.secret, value="key"), False), + (None, False), + ], +) +def test_accepts_matches_the_provided_type_against_the_options(credentials: ProvidedCredentials | None, accepted: bool): + discovery = _discovery([CredentialsType.button, CredentialsOption(type=CredentialsType.code, mask="DDD-DD-DDD")]) + assert discovery.accepts(credentials) is accepted + + +def test_no_credentials_are_accepted_only_where_none_is_an_option(): + assert _discovery([CredentialsType.none]).accepts(None) + assert _discovery([CredentialsType.none]).accepts(ProvidedCredentials(type=CredentialsType.none)) + + +def test_button_is_a_confirmation_without_a_value(): + credentials = ProvidedCredentials(type=CredentialsType.button) + assert credentials.value is None + assert ProvidedCredentials.model_validate({"type": "button"}) == credentials + assert CredentialsType.button != CredentialsType.none + + +def test_a_discovery_can_offer_button_or_qr(): + discovery = _discovery([CredentialsType.button, CredentialsType.qr]) + assert [o.type for o in discovery.expected_credentials_options] == [CredentialsType.button, CredentialsType.qr] diff --git a/tests/test_errors.py b/tests/test_errors.py new file mode 100644 index 0000000..6fbae73 --- /dev/null +++ b/tests/test_errors.py @@ -0,0 +1,43 @@ +"""MissingCredentialsError: what the Hub lacks to pair a device, readable back from a last_error.""" + +import pytest + +from majordom_integration_sdk import MissingCredentialsError + + +def test_the_type_defaults_to_wifi(): + assert MissingCredentialsError().type == MissingCredentialsError.WIFI == "wifi" + + +def test_the_known_types(): + assert MissingCredentialsError.THREAD == "thread" + + +def test_the_message_carries_the_type(): + assert str(MissingCredentialsError("thread")) == "Missing credentials: thread" + + +def test_a_custom_type_is_accepted(): + assert MissingCredentialsError("zigbee_network_key").type == "zigbee_network_key" + + +@pytest.mark.parametrize("type_", ["", "WiFi", "wi-fi", "thread network", "_wifi", "wifi_", "1wifi"]) +def test_the_type_is_lower_snake_case(type_: str): + with pytest.raises(ValueError): + MissingCredentialsError(type_) + + +def test_the_hub_already_handles_it_as_a_lookup_error(): + assert issubclass(MissingCredentialsError, LookupError) + + +@pytest.mark.parametrize("type_", ["wifi", "thread", "zigbee_network_key"]) +def test_the_type_is_read_back_from_a_last_error(type_: str): + parsed = MissingCredentialsError.parse(str(MissingCredentialsError(type_))) + assert parsed is not None + assert parsed.type == type_ + + +@pytest.mark.parametrize("last_error", [None, "", "Device is no longer connected", "Missing credentials: Wi-Fi"]) +def test_other_errors_are_not_read_as_missing_credentials(last_error: str | None): + assert MissingCredentialsError.parse(last_error) is None