[bot] Add CredentialsOption and the button credentials type

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",
 ]

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()

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

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",

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
 

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]

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