localthings/tests/localthings/conftest.py

"""Shared fixtures for localthings component tests."""
from __future__ import annotations

import json
from pathlib import Path
from typing import Any
from unittest.mock import patch

import cbor2
import pytest
from pytest_homeassistant_custom_component.common import MockConfigEntry

# Point HA's config dir at the repo root so the loader mounts custom_components/
# from the project into sys.path — otherwise IntegrationNotFound is raised.
PROJECT_ROOT = str(Path(__file__).resolve().parents[2])


@pytest.fixture
def hass_config_dir() -> str:
    """Override to let HA's loader find custom_components/localthings."""
    return PROJECT_ROOT


@pytest.fixture(autouse=True)
def auto_enable_custom_integrations(enable_custom_integrations):
    """Automatically enable custom integrations for all localthings tests.

    pytest-homeassistant-custom-component caches an empty custom-component
    dict during HA startup. This autouse fixture calls the upstream
    enable_custom_integrations fixture which pops that cache entry, forcing
    a re-discovery that finds custom_components/localthings.
    """
    return enable_custom_integrations


from custom_components.localthings.const import (
    CONF_CA_CERT_PEM, CONF_CA_KEY_PEM,
    CONF_HOST, CONF_LEAF_CERT_PEM, CONF_LEAF_KEY_PEM, CONF_PORT, DOMAIN,
)
from custom_components.localthings.coordinator import LocalThingsCoordinator


@pytest.fixture(autouse=True)
def _fast_coordinator_timers():
    """Shrink the coordinator's real-time delays for every localthings test.

    `_run_subpolls` and `_attempt_observe_mode` are, by design, blocking/
    real-time operations in production (a real sub-poll cadence and a real
    CoAP OBSERVE grace period). Tests that drive them through a full
    `hass.config_entries.async_setup(...)` + `hass.async_block_till_done()`
    would otherwise burn tens of real seconds per test waiting them out.

    This patches only the *class* attributes coordinator.py exposes for
    this purpose (`_SUBPOLL_STEP_S`, `_OBSERVE_GRACE_PERIOD_S`) — the
    production defaults they're computed from (`SUMMARY_INTERVAL_S`,
    `GRACE_PERIOD_S`) are untouched. Tests that need to exercise the real
    grace-period race (e.g. test_observe.py's own `ObserveManager` tests)
    still pass `grace_period_s=` explicitly and are unaffected.
    """
    with (
        patch.object(LocalThingsCoordinator, '_SUBPOLL_STEP_S', 0.001),
        patch.object(LocalThingsCoordinator, '_OBSERVE_GRACE_PERIOD_S', 0.02),
        patch.object(LocalThingsCoordinator, '_RECONNECT_PAUSE_S', 0.01),
    ):
        yield

FIXTURES = Path(__file__).resolve().parent.parent / 'fixtures'

MOCK_HOST = '10.0.0.254'
MOCK_PORT = 49154
MOCK_SERIAL = 'TEST-SERIAL-001'
MOCK_CA_CERT_PEM = '-----BEGIN CERTIFICATE-----\nTEST-CA\n-----END CERTIFICATE-----'
MOCK_CA_KEY_PEM = '-----BEGIN PRIVATE KEY-----\nTEST-CA-KEY\n-----END PRIVATE KEY-----'
MOCK_LEAF_CERT_PEM = '-----BEGIN CERTIFICATE-----\nTEST-LEAF\n-----END CERTIFICATE-----'
MOCK_LEAF_KEY_PEM = '-----BEGIN PRIVATE KEY-----\nTEST-LEAF-KEY\n-----END PRIVATE KEY-----'

ENTRY_DATA = {
    CONF_HOST:          MOCK_HOST,
    CONF_PORT:          MOCK_PORT,
    CONF_CA_CERT_PEM:   MOCK_CA_CERT_PEM,
    CONF_CA_KEY_PEM:    MOCK_CA_KEY_PEM,
    CONF_LEAF_CERT_PEM: MOCK_LEAF_CERT_PEM,
    CONF_LEAF_KEY_PEM:  MOCK_LEAF_KEY_PEM,
}


def _load_fridge_resources() -> dict:
    from custom_components.localthings.registry.batch import parse_device0_batch
    data = json.loads((FIXTURES / 'refrigerator_device.json').read_text())
    return parse_device0_batch(data['device0'])


@pytest.fixture
def fridge_resources():
    return _load_fridge_resources()


@pytest.fixture
def mock_probe():
    """Patch _probe_and_validate to succeed (recognized type) without a real DTLS connection."""
    with patch(
        'custom_components.localthings.config_flow._probe_and_validate',
        return_value={
            'port': MOCK_PORT,
            'serial': MOCK_SERIAL,
            'leaf_cert_pem': MOCK_LEAF_CERT_PEM,
            'leaf_key_pem': MOCK_LEAF_KEY_PEM,
            'one_ui_version': '7.0 Refrigerator',
            'device_type_recognized': True,
        },
    ) as m:
        yield m


@pytest.fixture
def mock_probe_unknown_type():
    """Patch _probe_and_validate to succeed, but with an unrecognized device type."""
    with patch(
        'custom_components.localthings.config_flow._probe_and_validate',
        return_value={
            'port': MOCK_PORT,
            'serial': MOCK_SERIAL,
            'leaf_cert_pem': MOCK_LEAF_CERT_PEM,
            'leaf_key_pem': MOCK_LEAF_KEY_PEM,
            'one_ui_version': '9.0 Space Heater',
            'device_type_recognized': False,
        },
    ) as m:
        yield m


@pytest.fixture
def mock_coordinator_session(fridge_resources):
    """Patch coordinator's blocking session methods so no real DTLS is needed."""
    with (
        patch('custom_components.localthings.coordinator.LocalThingsCoordinator._connect_session'),
        patch(
            'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once',
            return_value=fridge_resources,
        ),
        patch('custom_components.localthings.coordinator.LocalThingsCoordinator._close_session'),
    ):
        yield


class FakeObserveSession:
    """Stand-in for DtlsCoapSession that supports subscribe()/on_notification
    for coordinator-level observe tests, without a real DTLS connection."""

    def __init__(self, on_notification=None):
        self.on_notification = on_notification
        self.subscribed: list[str] = []
        self.fail_hrefs: set[str] = set()
        self.closed = False
        # When set to a rep dict, subscribe() immediately delivers that rep
        # as an OBSERVE notification for the href — what a real device does
        # when it answers a subscription with the current representation.
        # `try_enter_observe_mode` clears its notified set before it
        # subscribes, so this is the only way to deliver a notify that
        # reliably counts: it lands synchronously via on_notification,
        # after that clear and before the post-sleep fraction check — a
        # notify raced in from another thread can be wiped by the clear
        # (or arrive after the check) depending on scheduling. Set to
        # None to model a device that answers subscriptions but never
        # notifies.
        self.notify_on_subscribe: dict[str, Any] | None = None

    def subscribe(self, path_segs):
        href = '/' + '/'.join(path_segs)
        if href in self.fail_hrefs:
            raise ConnectionError("subscribe failed")
        self.subscribed.append(href)
        if self.notify_on_subscribe is not None and self.on_notification is not None:
            self.on_notification(href, cbor2.dumps(self.notify_on_subscribe))
        return b'\x01'

    def refresh_observes(self, paths):
        return None

    def close(self):
        self.closed = True


@pytest.fixture
def mock_coordinator_observe_session(fridge_resources):
    """Like mock_coordinator_session, but _connect_session installs a
    FakeObserveSession on coordinator._session so subscribe()/on_notification
    wiring can be exercised."""
    fake = FakeObserveSession()

    def _connect(self):
        fake.on_notification = self._observe.on_notification
        self._session = fake

    with (
        patch(
            'custom_components.localthings.coordinator.LocalThingsCoordinator._connect_session',
            _connect,
        ),
        patch(
            'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once',
            return_value=fridge_resources,
        ),
        patch('custom_components.localthings.coordinator.LocalThingsCoordinator._close_session'),
    ):
        yield fake


@pytest.fixture
def mock_entry(hass):
    """A MockConfigEntry added to hass, ready for async_setup."""
    entry = MockConfigEntry(
        domain=DOMAIN,
        data=ENTRY_DATA,
        unique_id=f'localthings_{MOCK_SERIAL}',
    )
    entry.add_to_hass(hass)
    return entry