Add property-based testing: Hypothesis (Python) and Bombadil (TypeScript)

.eslintrc.cjs

@@ -30,6 +30,15 @@ module.exports = {
                 'import/no-extraneous-dependencies': 'off', // Test deps (vitest, etc.) are devDependencies
                 'sort-keys': 'off'                          // Test objects are ordered for readability, not alphabetically
             }
+        },
+        {
+            files: ['ts/pbt/**/*.ts'],
+            parserOptions: {
+                project: ['./ts/pbt/tsconfig.json']
+            },
+            rules: {
+                'import/no-extraneous-dependencies': 'off' // bombadil is a devDependency
+            }
         }
     ],
     parser: '@typescript-eslint/parser',

.github/workflows/bombadil.yml

@@ -0,0 +1,37 @@
+---
+name: Bombadil property tests
+
+# GHA requires bare `on:` which yamllint sees as boolean
+# yamllint disable-line rule:truthy
+on:
+  schedule:
+    - cron: "17 8 * * *"
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  bombadil:
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    steps:
+      - uses: actions/checkout@v7
+      - name: Set up Node
+        uses: actions/setup-node@v7
+        with:
+          node-version: lts/jod
+          cache: yarn
+      - name: Install dependencies
+        run: yarn install --frozen-lockfile
+      - name: Run Bombadil exploration
+        env:
+          BOMBADIL_HEADLESS: "1"
+          BOMBADIL_TIME_LIMIT: 5m
+        run: yarn test:pbt
+      - name: Upload trace on violation
+        if: failure()
+        uses: actions/upload-artifact@v4
+        with:
+          name: bombadil-output
+          path: pbt-output/

.github/workflows/python-checks.yml

@@ -91,6 +91,8 @@ jobs:
         run: uv pip install --system -r requirements_dev.txt pytest-cov pytest-github-actions-annotate-failures
       - name: Run tests and generate coverage report
         run: pytest ./tests/ --cov=custom_components/lock_code_manager/ --cov-report=xml --junitxml=junit.xml
+        env:
+          HYPOTHESIS_PROFILE: ci
       - name: Upload coverage to Codecov
         if: matrix.python-version == needs.setup.outputs.target-python
         uses: codecov/codecov-action@v7

.gitignore

@@ -7,6 +7,7 @@ venv
 .codex
 .coverage
 coverage/
+.hypothesis/
 .pytest_cache
 .mypy_cache
 .ruff_cache
@@ -22,3 +23,4 @@ docs/
 .github/hooks/
 .opencode/
 junit.xml
+pbt-output/

AGENTS.md

@@ -316,6 +316,29 @@ tests/providers/
 - Provider `async_setup` must be idempotent — tests verify it can be called multiple times.
 - Don't use `MagicMock()` for coordinators when the real coordinator can be used.
 
+## Property-based testing
+
+Two complementary layers exist alongside the example-based suites:
+
+- **Python (Hypothesis)** — `tests/properties/` runs as part of the normal
+  `pytest tests/` invocation. The default `dev` profile uses 15 examples to
+  keep the suite fast; CI sets `HYPOTHESIS_PROFILE=ci` (200 examples). Run
+  the deep profile locally with `HYPOTHESIS_PROFILE=ci pytest tests/properties/`.
+  `test_credential_machine.py` is a stateful machine driving the real
+  BaseLock write orchestration against `MockLCMLock`.
+- **TypeScript (Bombadil)** — `ts/pbt/` contains a browser harness that mounts
+  the built cards with a scripted mock `hass` plus a chaos panel, and a
+  Bombadil spec (`ts/pbt/spec.ts`) with temporal-logic properties (no PIN leaks
+  from masked cards, chip counts match the model, pushed data eventually
+  renders). Run locally with `yarn test:pbt` (env knobs:
+  `BOMBADIL_TIME_LIMIT`, `BOMBADIL_HEADLESS=1`). CI runs it nightly and on
+  manual dispatch (`bombadil.yml`), never on pull requests. Inspect
+  violations with `yarn bombadil browser inspect pbt-output`.
+
+Counterexample triage: a found counterexample is a deliverable. Either fix
+the code, or — only for documented contract edges — narrow the strategy
+with a comment citing the contract. Never silence one by rerunning.
+
 ## Adding Lock Provider Support
 
 1. Create new file in `providers/` (e.g., `my_provider.py`)

custom_components/lock_code_manager/providers/_util.py

@@ -98,7 +98,8 @@ def parse_slot_num(value: object) -> int | None:
     Convert a slot identifier to an int, or return None if not convertible.
 
     Mirrors ``int(value)`` while collapsing the ``TypeError``/``ValueError``
-    that providers otherwise catch when a lock reports a non-numeric slot key.
+    that providers otherwise catch when a lock reports a non-numeric slot key
+    (and ``OverflowError``, which ``int`` raises for infinite floats).
     JSON booleans are rejected rather than coerced (``int(True)`` is 1, so a
     malformed ``true`` would otherwise silently address slot 1).
     Call sites remain responsible for their own logging and skip/return flow.
@@ -107,5 +108,5 @@ def parse_slot_num(value: object) -> int | None:
         return None
     try:
         return int(value)  # type: ignore[call-overload]
-    except TypeError, ValueError:
+    except TypeError, ValueError, OverflowError:
         return None

package.json

@@ -18,7 +18,8 @@
     "lint:fix": "yarn lint --fix",
     "test": "vitest run",
     "test:watch": "vitest",
-    "test:coverage": "vitest run --coverage"
+    "test:coverage": "vitest run --coverage",
+    "test:pbt": "yarn build && node ts/pbt/run.mjs"
   },
   "dependencies": {
     "@mdi/js": "^7.4.47",
@@ -27,6 +28,7 @@
     "lit-html": "^3.3.3"
   },
   "devDependencies": {
+    "@antithesishq/bombadil": "0.6.1",
     "@babel/core": "^7.29.7",
     "@babel/preset-env": "^7.29.7",
     "@rollup/plugin-babel": "^7.1.0",

pyproject.toml

@@ -136,7 +136,7 @@ overgeneral-exceptions = ["BaseException", "Exception", "HomeAssistantError"]
 asyncio_mode = "auto"
 asyncio_default_fixture_loop_scope = "function"
 testpaths = ["tests"]
-norecursedirs = [".git", "testing_config"]
+norecursedirs = [".git", "testing_config", ".hypothesis"]
 
 [tool.mypy]
 ignore_missing_imports = true

requirements_test.txt

@@ -1,3 +1,4 @@
+hypothesis>=6.100
 pylint-strict-informational>=0.1
 pytest>=9.0.3
 pytest-homeassistant-custom-component==0.13.348

tests/conftest.py

@@ -4,9 +4,11 @@
 
 from collections.abc import Generator
 from datetime import timedelta
+import os
 from typing import Any
 from unittest.mock import patch
 
+from hypothesis import settings as hypothesis_settings
 import pytest
 from pytest_homeassistant_custom_component.common import (
     MockConfigEntry,
@@ -37,6 +39,16 @@
 
 pytest_plugins = ["pytest_homeassistant_custom_component"]
 
+# Hypothesis profiles: "dev" keeps the full-suite run within its
+# seconds budget; CI opts into deeper exploration via HYPOTHESIS_PROFILE=ci.
+# deadline=None in both: per-example deadlines flake under the HA test
+# harness's timing noise and CI runner variance.
+hypothesis_settings.register_profile("dev", max_examples=15, deadline=None)
+hypothesis_settings.register_profile(
+    "ci", max_examples=200, deadline=None, print_blob=True
+)
+hypothesis_settings.load_profile(os.environ.get("HYPOTHESIS_PROFILE", "dev"))
+
 TEST_DOMAIN = "test"
 
 

tests/properties/__init__.py

@@ -0,0 +1 @@
+"""Property-based tests (Hypothesis)."""

tests/properties/test_credential_machine.py

@@ -0,0 +1,176 @@
+"""Stateful property test for credential write orchestration.
+
+Drives the real BaseLock write path (rate limiting, duplicate detection,
+connection checks, WriteResult handling) and LockUsercodeUpdateCoordinator
+against MockLCMLock, comparing everything to a plain-dict oracle.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+from hypothesis import strategies as st
+from hypothesis.stateful import RuleBasedStateMachine, invariant, rule
+import pytest
+from pytest_homeassistant_custom_component.common import (
+    MockConfigEntry,
+    async_test_home_assistant,
+)
+
+from homeassistant.helpers import device_registry as dr, entity_registry as er
+
+from custom_components.lock_code_manager.const import DOMAIN
+from custom_components.lock_code_manager.domain.coordinator import (
+    LockUsercodeUpdateCoordinator,
+)
+from custom_components.lock_code_manager.domain.exceptions import (
+    DuplicateCodeError,
+    LockDisconnected,
+)
+
+from ..common import MockLCMLock
+
+PINS = st.text(alphabet="0123456789", min_size=4, max_size=8)
+SLOTS = st.integers(min_value=1, max_value=5)
+
+
+class CredentialMachine(RuleBasedStateMachine):
+    """Random interleavings of writes, deletes, faults, and refreshes."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(self.loop)
+        self._hass_cm = async_test_home_assistant(self.loop)
+        self.hass = self.loop.run_until_complete(self._hass_cm.__aenter__())
+
+        self.config_entry = MockConfigEntry(domain=DOMAIN)
+        self.config_entry.add_to_hass(self.hass)
+        ent_reg = er.async_get(self.hass)
+        dev_reg = dr.async_get(self.hass)
+        lock_entity = ent_reg.async_get_or_create(
+            "lock", "test", "pbt_lock", config_entry=self.config_entry
+        )
+        self.lock = MockLCMLock(self.hass, dev_reg, ent_reg, None, lock_entity)
+        self.lock.codes = {}
+        # Rate-limit delay between operations would dominate machine runtime.
+        self.lock._min_operation_delay = 0
+        # Machine correctness also assumes an example never spans the
+        # coordinator's 10-second request-refresh cooldown: past it, the
+        # deferred debounced refresh could freshen coordinator.data between a
+        # rule's _would_duplicate read and its assertion. Examples run in
+        # milliseconds, leaving orders of magnitude of margin.
+        self.coordinator = LockUsercodeUpdateCoordinator(
+            self.hass, self.lock, self.config_entry
+        )
+        self.lock.coordinator = self.coordinator
+        self._run(self.coordinator.async_refresh())
+
+        # Oracle: lock-truth we expect, slot -> pin.
+        self.expected: dict[int, str] = {}
+
+    def _run(self, coro):
+        return self.loop.run_until_complete(coro)
+
+    def _would_duplicate(self, slot: int, pin: str) -> bool:
+        # Mirror of the source _check_duplicate_code reads: coordinator.data,
+        # which may lag lock.codes until the next refresh.
+        return any(
+            other_slot != slot and credential.matches(pin)
+            for other_slot, credential in self.coordinator.data.items()
+        )
+
+    @rule(slot=SLOTS, pin=PINS)
+    def set_credential(self, slot: int, pin: str) -> None:
+        if self._would_duplicate(slot, pin):
+            before = dict(self.lock.codes)
+            with pytest.raises(DuplicateCodeError):
+                self._run(
+                    self.lock.async_internal_set_usercode(
+                        slot, pin, name=f"PBT user {slot}"
+                    )
+                )
+            assert self.lock.codes == before
+        else:
+            already_set = self.lock.codes.get(slot) == pin
+            self._run(
+                self.lock.async_internal_set_usercode(
+                    slot, pin, name=f"PBT user {slot}"
+                )
+            )
+            self.expected[slot] = pin
+            if not already_set:
+                # Names must reach the provider verbatim (tagging is a
+                # name-keyed-provider concern, not BaseLock's).
+                assert self.lock.service_calls["set_usercode"][-1] == (
+                    slot,
+                    pin,
+                    f"PBT user {slot}",
+                )
+
+    @rule(slot=SLOTS)
+    def delete_credential(self, slot: int) -> None:
+        self._run(self.lock.async_internal_clear_usercode(slot))
+        self.expected.pop(slot, None)
+
+    @rule(slot=SLOTS)
+    def set_same_pin_is_no_change(self, slot: int) -> None:
+        if slot not in self.expected:
+            return
+        pin = self.expected[slot]
+        if self._would_duplicate(slot, pin):
+            # An external change may have copied this PIN onto another slot;
+            # the duplicate guard fires before the no-change shortcut.
+            return
+        calls_before = len(self.lock.service_calls["set_usercode"])
+        self._run(
+            self.lock.async_internal_set_usercode(slot, pin, name=f"PBT user {slot}")
+        )
+        assert len(self.lock.service_calls["set_usercode"]) == calls_before
+
+    @rule(slot=SLOTS, pin=PINS)
+    def external_change(self, slot: int, pin: str) -> None:
+        self.lock.codes[slot] = pin
+        self.expected[slot] = pin
+
+    @rule()
+    def refresh_converges_coordinator(self) -> None:
+        self._run(self.coordinator.async_refresh())
+        observed = {
+            slot: credential.readable_pin
+            for slot, credential in self.coordinator.data.items()
+            if credential.is_present
+        }
+        assert observed == self.expected
+
+    @rule(slot=SLOTS, pin=PINS)
+    def write_while_disconnected_fails_loud(self, slot: int, pin: str) -> None:
+        self.lock.set_connected(False)
+        before = dict(self.lock.codes)
+        try:
+            with pytest.raises(LockDisconnected):
+                self._run(
+                    self.lock.async_internal_set_usercode(
+                        slot, pin, name=f"PBT user {slot}"
+                    )
+                )
+            assert self.lock.codes == before
+        finally:
+            self.lock.set_connected(True)
+
+    @invariant()
+    def lock_state_matches_oracle(self) -> None:
+        assert self.lock.codes == self.expected
+
+    def teardown(self) -> None:
+        async def _shutdown() -> None:
+            await self.coordinator.async_shutdown()
+            await self.hass.async_stop(force=True)
+            await self._hass_cm.__aexit__(None, None, None)
+
+        self.loop.run_until_complete(_shutdown())
+        self.loop.close()
+        asyncio.set_event_loop(None)
+
+
+TestCredentialMachine = CredentialMachine.TestCase