test: scaffold pytest-homeassistant-custom-component test harness
Adds HA component tests covering the localthings integration layer (previously only the samsung_appliance library was tested): - pyproject.toml: asyncio_mode=auto for async HA test patterns - tests/localthings/conftest.py: hass_config_dir override, mock fixtures for probe and coordinator session, fridge fixture loader - test_config_flow.py: first-device form, second-device cred reuse, happy path, cannot_connect error, duplicate-device abort (5 tests) - test_coordinator.py: discovery on first refresh, device_info populated, idle interval default, UpdateFailed on persistent poll error (4 tests) - test_sensor.py: sensors registered, sensor state from coordinator (2 tests) Total: 70 library + 11 component = 81 tests
This commit is contained in:
1 parent
ee779c955b
commit
0bb27a9204
6 files changed
+321
No files matched your search
@@ -0,0 +1,2 @@
|
||||
[tool.pytest.ini_options]
|
||||
asyncio_mode = "auto"
|
||||
Whitespace-only changes.
@@ -0,0 +1,98 @@
|
||||
"""Shared fixtures for localthings component tests."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
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_CERT_PEM, CONF_HOST, CONF_KEY_PEM, CONF_PORT, DOMAIN,
|
||||
)
|
||||
|
||||
FIXTURES = Path(__file__).resolve().parent.parent / 'fixtures'
|
||||
|
||||
MOCK_HOST = '10.0.0.254'
|
||||
MOCK_PORT = 49154
|
||||
MOCK_SERIAL = 'TEST-SERIAL-001'
|
||||
MOCK_CERT_PEM = '-----BEGIN CERTIFICATE-----\nTEST\n-----END CERTIFICATE-----'
|
||||
MOCK_KEY_PEM = '-----BEGIN PRIVATE KEY-----\nTEST\n-----END PRIVATE KEY-----'
|
||||
|
||||
ENTRY_DATA = {
|
||||
CONF_HOST: MOCK_HOST,
|
||||
CONF_PORT: MOCK_PORT,
|
||||
CONF_CERT_PEM: MOCK_CERT_PEM,
|
||||
CONF_KEY_PEM: MOCK_KEY_PEM,
|
||||
}
|
||||
|
||||
|
||||
def _load_fridge_resources() -> dict:
|
||||
from samsung_appliance.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 without a real DTLS connection."""
|
||||
with patch(
|
||||
'custom_components.localthings.config_flow._probe_and_validate',
|
||||
return_value={'port': MOCK_PORT, 'serial': MOCK_SERIAL},
|
||||
) 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
|
||||
|
||||
|
||||
@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
|
||||
@@ -0,0 +1,100 @@
|
||||
"""Tests for the localthings config flow."""
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
from pytest_homeassistant_custom_component.common import MockConfigEntry
|
||||
|
||||
from custom_components.localthings.const import (
|
||||
CONF_CERT_PEM, CONF_HOST, CONF_KEY_PEM, CONF_PORT, DOMAIN,
|
||||
)
|
||||
|
||||
from .conftest import (
|
||||
ENTRY_DATA, MOCK_CERT_PEM, MOCK_HOST, MOCK_KEY_PEM, MOCK_PORT, MOCK_SERIAL,
|
||||
)
|
||||
|
||||
|
||||
async def test_form_first_device(hass: HomeAssistant) -> None:
|
||||
"""First device: form asks for host, cert, and key."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={'source': 'user'}
|
||||
)
|
||||
assert result['type'] == FlowResultType.FORM
|
||||
assert result['step_id'] == 'user'
|
||||
assert CONF_CERT_PEM in result['data_schema'].schema
|
||||
assert CONF_KEY_PEM in result['data_schema'].schema
|
||||
|
||||
|
||||
async def test_form_second_device_reuses_creds(hass: HomeAssistant) -> None:
|
||||
"""Second device: form only asks for host; cert/key schema fields absent."""
|
||||
existing = MockConfigEntry(domain=DOMAIN, data=ENTRY_DATA)
|
||||
existing.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={'source': 'user'}
|
||||
)
|
||||
assert result['type'] == FlowResultType.FORM
|
||||
assert CONF_CERT_PEM not in result['data_schema'].schema
|
||||
assert CONF_KEY_PEM not in result['data_schema'].schema
|
||||
|
||||
|
||||
async def test_successful_setup(hass: HomeAssistant, mock_probe) -> None:
|
||||
"""Happy path: valid IP connects, entry created with discovered port."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={'source': 'user'}
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result['flow_id'],
|
||||
{CONF_HOST: MOCK_HOST, CONF_CERT_PEM: MOCK_CERT_PEM, CONF_KEY_PEM: MOCK_KEY_PEM},
|
||||
)
|
||||
assert result['type'] == FlowResultType.CREATE_ENTRY
|
||||
assert result['data'][CONF_HOST] == MOCK_HOST
|
||||
assert result['data'][CONF_PORT] == MOCK_PORT
|
||||
assert result['data'][CONF_CERT_PEM] == MOCK_CERT_PEM
|
||||
|
||||
|
||||
async def test_cannot_connect(hass: HomeAssistant) -> None:
|
||||
"""Failed probe: form re-shown with cannot_connect error."""
|
||||
from custom_components.localthings.config_flow import CannotConnect
|
||||
|
||||
with patch(
|
||||
'custom_components.localthings.config_flow._probe_and_validate',
|
||||
side_effect=CannotConnect('no port'),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={'source': 'user'}
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result['flow_id'],
|
||||
{CONF_HOST: MOCK_HOST, CONF_CERT_PEM: MOCK_CERT_PEM, CONF_KEY_PEM: MOCK_KEY_PEM},
|
||||
)
|
||||
assert result['type'] == FlowResultType.FORM
|
||||
assert result['errors']['base'] == 'cannot_connect'
|
||||
|
||||
|
||||
async def test_duplicate_device_aborted(hass: HomeAssistant, mock_probe) -> None:
|
||||
"""Second add of same serial: flow aborts.
|
||||
|
||||
When a device already exists the form only asks for host (cert/key are
|
||||
reused), so we only submit CONF_HOST in the second configure call.
|
||||
"""
|
||||
existing = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data=ENTRY_DATA,
|
||||
unique_id=f'localthings_{MOCK_SERIAL}',
|
||||
)
|
||||
existing.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={'source': 'user'}
|
||||
)
|
||||
# Second-device form only has CONF_HOST; cert/key are reused from existing.
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result['flow_id'],
|
||||
{CONF_HOST: MOCK_HOST},
|
||||
)
|
||||
assert result['type'] == FlowResultType.ABORT
|
||||
assert result['reason'] == 'already_configured'
|
||||
@@ -0,0 +1,74 @@
|
||||
"""Tests for the LocalThingsCoordinator."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from custom_components.localthings.const import (
|
||||
ACTIVE_INTERVAL_S, DOMAIN, IDLE_INTERVAL_S,
|
||||
)
|
||||
from custom_components.localthings.coordinator import LocalThingsCoordinator
|
||||
|
||||
from .conftest import ENTRY_DATA, MOCK_SERIAL
|
||||
|
||||
|
||||
async def test_first_refresh_runs_discovery(
|
||||
hass: HomeAssistant, mock_entry, mock_coordinator_session
|
||||
) -> None:
|
||||
"""After first refresh, coordinator.bound is populated."""
|
||||
await hass.config_entries.async_setup(mock_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
|
||||
assert coordinator.bound
|
||||
assert coordinator.data
|
||||
|
||||
|
||||
async def test_device_info_populated(
|
||||
hass: HomeAssistant, mock_entry, mock_coordinator_session
|
||||
) -> None:
|
||||
"""device_info is set from discovered resources after first refresh."""
|
||||
await hass.config_entries.async_setup(mock_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
|
||||
assert coordinator.device_info is not None
|
||||
assert any(
|
||||
DOMAIN in str(i)
|
||||
for i in coordinator.device_info.get('identifiers', set())
|
||||
)
|
||||
|
||||
|
||||
async def test_idle_interval_by_default(
|
||||
hass: HomeAssistant, mock_entry, mock_coordinator_session
|
||||
) -> None:
|
||||
"""Interval defaults to idle when no active_when fires."""
|
||||
await hass.config_entries.async_setup(mock_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
|
||||
assert coordinator.update_interval == timedelta(seconds=IDLE_INTERVAL_S)
|
||||
|
||||
|
||||
async def test_update_failed_on_persistent_poll_error(
|
||||
hass: HomeAssistant, mock_entry
|
||||
) -> None:
|
||||
"""ConfigEntryNotReady raised when poll fails even after reconnect."""
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
|
||||
with (
|
||||
patch('custom_components.localthings.coordinator.LocalThingsCoordinator._connect_session'),
|
||||
patch(
|
||||
'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once',
|
||||
side_effect=RuntimeError('connection lost'),
|
||||
),
|
||||
patch('custom_components.localthings.coordinator.LocalThingsCoordinator._close_session'),
|
||||
):
|
||||
result = await hass.config_entries.async_setup(mock_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# Setup fails — entry should not be in hass.data
|
||||
assert not result or mock_entry.entry_id not in hass.data.get(DOMAIN, {})
|
||||
@@ -0,0 +1,47 @@
|
||||
"""Tests for the sensor platform."""
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
|
||||
from custom_components.localthings.const import DOMAIN
|
||||
|
||||
|
||||
async def test_sensors_registered(
|
||||
hass: HomeAssistant, mock_entry, mock_coordinator_session
|
||||
) -> None:
|
||||
"""Sensors are registered for all SensorDesc entities in the fridge fixture."""
|
||||
await hass.config_entries.async_setup(mock_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
ent_reg = er.async_get(hass)
|
||||
entries = er.async_entries_for_config_entry(ent_reg, mock_entry.entry_id)
|
||||
sensor_entries = [e for e in entries if e.domain == 'sensor']
|
||||
assert sensor_entries, "Expected at least one sensor entity"
|
||||
|
||||
|
||||
async def test_sensor_state_from_coordinator(
|
||||
hass: HomeAssistant, mock_entry, mock_coordinator_session
|
||||
) -> None:
|
||||
"""A temperature sensor reads its value from coordinator.data.
|
||||
|
||||
Entity IDs are auto-generated by HA and don't embed capability path
|
||||
segments, so we filter by the mdi:thermometer icon that the fridge
|
||||
capability assigns to current-temperature sensors.
|
||||
"""
|
||||
await hass.config_entries.async_setup(mock_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
states = hass.states.async_all('sensor')
|
||||
temp_states = [
|
||||
s for s in states
|
||||
if s.attributes.get('icon') == 'mdi:thermometer'
|
||||
]
|
||||
assert temp_states, (
|
||||
f"No thermometer sensor found. Sensors: {[s.entity_id for s in states]}"
|
||||
)
|
||||
state = temp_states[0]
|
||||
assert state.state not in ('unknown', 'unavailable'), (
|
||||
f"Sensor state is {state.state}"
|
||||
)
|
||||
Reference in new issue
Block a user