home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-08-31 18:31:35 +02:00
Code Issues Packages Projects Releases Wiki Activity

Use fixtures instead of helper functions for APCUPSD tests (#151172)

Browse Source 
Co-authored-by: Franck Nijhof <frenck@frenck.nl>
This commit is contained in:
Yuxin Wang
2025-08-28 10:11:31 -04:00
committed by GitHub
parent e94a7b2ec1
commit ffcd5167b5
9 changed files with 381 additions and 382 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
homeassistant/components/apcupsd/__init__.py
View File

@@ -9,7 +9,7 @@ from homeassistant.core import HomeAssistant
from .coordinator import APCUPSdConfigEntry, APCUPSdCoordinator
PLATFORMS: Final = (Platform.BINARY_SENSOR, Platform.SENSOR)
PLATFORMS: Final = [Platform.BINARY_SENSOR, Platform.SENSOR]
async def async_setup_entry(

37
tests/components/apcupsd/__init__.py
View File

@@ -4,15 +4,8 @@ from __future__ import annotations
from collections import OrderedDict
from typing import Final
from unittest.mock import patch
from homeassistant.components.apcupsd.const import DOMAIN
from homeassistant.components.apcupsd.coordinator import APCUPSdData
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
CONF_DATA: Final = {CONF_HOST: "test", CONF_PORT: 1234}
@@ -79,33 +72,3 @@ MOCK_MINIMAL_STATUS: Final = OrderedDict(
("END APC", "1970-01-01 00:00:00 0000"),
]
)
async def async_init_integration(
hass: HomeAssistant,
*,
host: str = "test",
status: dict[str, str] | None = None,
entry_id: str = "mocked-config-entry-id",
) -> MockConfigEntry:
"""Set up the APC UPS Daemon integration in HomeAssistant."""
if status is None:
status = MOCK_STATUS
entry = MockConfigEntry(
entry_id=entry_id,
version=1,
domain=DOMAIN,
title="APCUPSd",
data=CONF_DATA | {CONF_HOST: host},
unique_id=APCUPSdData(status).serial_no,
source=SOURCE_USER,
)
entry.add_to_hass(hass)
with patch("aioapcaccess.request_status", return_value=status):
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
return entry

68
tests/components/apcupsd/conftest.py
View File

@@ -1,10 +1,21 @@
"""Common fixtures for the APC UPS Daemon (APCUPSD) tests."""
from collections.abc import Generator
from collections.abc import AsyncGenerator, Generator
from unittest.mock import AsyncMock, patch
import pytest
from homeassistant.components.apcupsd import PLATFORMS
from homeassistant.components.apcupsd.const import DOMAIN
from homeassistant.components.apcupsd.coordinator import APCUPSdData
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from . import CONF_DATA, MOCK_STATUS
from tests.common import MockConfigEntry
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
@@ -13,3 +24,58 @@ def mock_setup_entry() -> Generator[AsyncMock]:
"homeassistant.components.apcupsd.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture
async def mock_request_status(
request: pytest.FixtureRequest,
) -> AsyncGenerator[AsyncMock]:
"""Return a mocked aioapcaccess.request_status function."""
mocked_status = getattr(request, "param", None) or MOCK_STATUS
with patch("aioapcaccess.request_status") as mock_request_status:
mock_request_status.return_value = mocked_status
yield mock_request_status
@pytest.fixture
def mock_config_entry(
request: pytest.FixtureRequest,
mock_request_status: AsyncMock,
) -> MockConfigEntry:
"""Mock setting up a config entry."""
entry_id = getattr(request, "param", None)
return MockConfigEntry(
entry_id=entry_id,
version=1,
domain=DOMAIN,
title="APC UPS Daemon",
data=CONF_DATA,
unique_id=APCUPSdData(mock_request_status.return_value).serial_no,
source=SOURCE_USER,
)
@pytest.fixture
def platforms() -> list[Platform]:
"""Fixture to specify platforms to test."""
return PLATFORMS
@pytest.fixture
async def init_integration(
request: pytest.FixtureRequest,
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_request_status: AsyncMock,
platforms: list[Platform],
) -> MockConfigEntry:
"""Set up APC UPS Daemon integration for testing."""
mock_config_entry.add_to_hass(hass)
with patch("homeassistant.components.apcupsd.PLATFORMS", platforms):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
return mock_config_entry

8
tests/components/apcupsd/snapshots/test_init.ambr
View File

@@ -1,5 +1,5 @@
# serializer version: 1
# name: test_async_setup_entry[status0][device_MyUPS_XXXXXXXXXXXX]
# name: test_async_setup_entry[mock_request_status0-mocked-config-entry-id][device_MyUPS_XXXXXXXXXXXX]
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
@@ -30,7 +30,7 @@
'via_device_id': None,
})
# ---
# name: test_async_setup_entry[status1][device_APC UPS_XXXX]
# name: test_async_setup_entry[mock_request_status1-mocked-config-entry-id][device_APC UPS_XXXX]
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
@@ -61,7 +61,7 @@
'via_device_id': None,
})
# ---
# name: test_async_setup_entry[status2][device_APC UPS_<no serial>]
# name: test_async_setup_entry[mock_request_status2-mocked-config-entry-id][device_APC UPS_<no serial>]
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
@@ -92,7 +92,7 @@
'via_device_id': None,
})
# ---
# name: test_async_setup_entry[status3][device_APC UPS_Blank]
# name: test_async_setup_entry[mock_request_status3-mocked-config-entry-id][device_APC UPS_Blank]
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,

63
tests/components/apcupsd/test_binary_sensor.py
View File

@@ -1,6 +1,6 @@
"""Test binary sensors of APCUPSd integration."""
from unittest.mock import patch
from unittest.mock import AsyncMock
import pytest
from syrupy.assertion import SnapshotAssertion
@@ -10,47 +10,60 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.util import slugify
from . import MOCK_STATUS, async_init_integration
from . import MOCK_STATUS
from tests.common import snapshot_platform
from tests.common import MockConfigEntry, snapshot_platform
pytestmark = pytest.mark.usefixtures(
"entity_registry_enabled_by_default", "init_integration"
)
@pytest.fixture
def platforms() -> list[Platform]:
"""Overridden fixture to specify platforms to test."""
return [Platform.BINARY_SENSOR]
async def test_binary_sensor(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test states of binary sensors."""
with patch("homeassistant.components.apcupsd.PLATFORMS", [Platform.BINARY_SENSOR]):
config_entry = await async_init_integration(hass, status=MOCK_STATUS)
await snapshot_platform(hass, entity_registry, snapshot, config_entry.entry_id)
"""Test states of binary sensor entities."""
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
async def test_no_binary_sensor(hass: HomeAssistant) -> None:
@pytest.mark.parametrize(
"mock_request_status",
[{k: v for k, v in MOCK_STATUS.items() if k != "STATFLAG"}],
indirect=True,
)
async def test_no_binary_sensor(
hass: HomeAssistant,
mock_request_status: AsyncMock,
) -> None:
"""Test binary sensor when STATFLAG is not available."""
status = MOCK_STATUS.copy()
status.pop("STATFLAG")
await async_init_integration(hass, status=status)
device_slug = slugify(MOCK_STATUS["UPSNAME"])
device_slug = slugify(mock_request_status.return_value["UPSNAME"])
state = hass.states.get(f"binary_sensor.{device_slug}_online_status")
assert state is None
@pytest.mark.parametrize(
("override", "expected"),
("mock_request_status", "expected"),
[
("0x008", "on"),
("0x02040010 Status Flag", "off"),
(MOCK_STATUS | {"STATFLAG": "0x008"}, "on"),
(MOCK_STATUS | {"STATFLAG": "0x02040010 Status Flag"}, "off"),
],
indirect=["mock_request_status"],
)
async def test_statflag(hass: HomeAssistant, override: str, expected: str) -> None:
async def test_statflag(
hass: HomeAssistant,
mock_request_status: AsyncMock,
expected: str,
) -> None:
"""Test binary sensor for different STATFLAG values."""
status = MOCK_STATUS.copy()
status["STATFLAG"] = override
await async_init_integration(hass, status=status)
device_slug = slugify(MOCK_STATUS["UPSNAME"])
assert (
hass.states.get(f"binary_sensor.{device_slug}_online_status").state == expected
)
device_slug = slugify(mock_request_status.return_value["UPSNAME"])
state = hass.states.get(f"binary_sensor.{device_slug}_online_status")
assert state.state == expected

168
tests/components/apcupsd/test_config_flow.py
View File

@@ -3,7 +3,7 @@
from __future__ import annotations
import asyncio
from unittest.mock import AsyncMock, patch
from unittest.mock import AsyncMock
import pytest
@@ -23,41 +23,29 @@ from tests.common import MockConfigEntry
[OSError(), asyncio.IncompleteReadError(partial=b"", expected=100), TimeoutError()],
)
async def test_config_flow_cannot_connect(
hass: HomeAssistant, exception: Exception
hass: HomeAssistant,
exception: Exception,
mock_request_status: AsyncMock,
) -> None:
"""Test config flow setup with a connection error."""
with patch(
"homeassistant.components.apcupsd.coordinator.aioapcaccess.request_status"
) as mock_request_status:
mock_request_status.side_effect = exception
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data=CONF_DATA,
DOMAIN, context={"source": SOURCE_USER}, data=CONF_DATA
)
assert result["type"] is FlowResultType.FORM
assert result["errors"]["base"] == "cannot_connect"
async def test_config_flow_duplicate_host_port(
hass: HomeAssistant, mock_setup_entry: AsyncMock
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_config_entry: MockConfigEntry,
mock_request_status: AsyncMock,
) -> None:
"""Test duplicate config flow setup with the same host / port."""
# First add an existing config entry to hass.
mock_entry = MockConfigEntry(
version=1,
domain=DOMAIN,
title="APCUPSd",
data=CONF_DATA,
unique_id=MOCK_STATUS["SERIALNO"],
source=SOURCE_USER,
)
mock_entry.add_to_hass(hass)
mock_config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.apcupsd.coordinator.aioapcaccess.request_status"
) as mock_request_status:
# Assign the same host and port, which we should reject since the entry already exists.
mock_request_status.return_value = MOCK_STATUS
result = await hass.config_entries.flow.async_init(
@@ -81,23 +69,14 @@ async def test_config_flow_duplicate_host_port(
async def test_config_flow_duplicate_serial_number(
hass: HomeAssistant, mock_setup_entry: AsyncMock
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_config_entry: MockConfigEntry,
mock_request_status: AsyncMock,
) -> None:
"""Test duplicate config flow setup with different host but the same serial number."""
# First add an existing config entry to hass.
mock_entry = MockConfigEntry(
version=1,
domain=DOMAIN,
title="APCUPSd",
data=CONF_DATA,
unique_id=MOCK_STATUS["SERIALNO"],
source=SOURCE_USER,
)
mock_entry.add_to_hass(hass)
mock_config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.apcupsd.coordinator.aioapcaccess.request_status"
) as mock_request_status:
# Assign the different host and port, but we should still reject the creation since the
# serial number is the same as the existing entry.
mock_request_status.return_value = MOCK_STATUS
@@ -121,12 +100,12 @@ async def test_config_flow_duplicate_serial_number(
assert result["data"] == another_host
async def test_flow_works(hass: HomeAssistant, mock_setup_entry: AsyncMock) -> None:
async def test_flow_works(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_request_status: AsyncMock,
) -> None:
"""Test successful creation of config entries via user configuration."""
with patch(
"homeassistant.components.apcupsd.coordinator.aioapcaccess.request_status",
return_value=MOCK_STATUS,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: SOURCE_USER},
@@ -146,33 +125,28 @@ async def test_flow_works(hass: HomeAssistant, mock_setup_entry: AsyncMock) -> N
@pytest.mark.parametrize(
("extra_status", "expected_title"),
("mock_request_status", "expected_title"),
[
({"UPSNAME": "Friendly Name"}, "Friendly Name"),
({"MODEL": "MODEL X"}, "MODEL X"),
({"SERIALNO": "ZZZZ"}, "ZZZZ"),
# Some models report "Blank" as serial number, which we should treat it as not reported.
({"SERIALNO": "Blank"}, "APC UPS"),
({}, "APC UPS"),
(MOCK_MINIMAL_STATUS | {"UPSNAME": "Friendly Name"}, "Friendly Name"),
(MOCK_MINIMAL_STATUS | {"MODEL": "MODEL X"}, "MODEL X"),
(MOCK_MINIMAL_STATUS | {"SERIALNO": "ZZZZ"}, "ZZZZ"),
# Some models report "Blank" as the serial number, which we should treat it as not reported.
(MOCK_MINIMAL_STATUS | {"SERIALNO": "Blank"}, "APC UPS"),
(MOCK_MINIMAL_STATUS | {}, "APC UPS"),
],
indirect=["mock_request_status"],
)
async def test_flow_minimal_status(
hass: HomeAssistant,
extra_status: dict[str, str],
expected_title: str,
mock_setup_entry: AsyncMock,
mock_request_status: AsyncMock,
) -> None:
"""Test successful creation of config entries via user configuration when minimal status is reported.
We test different combinations of minimal statuses, where the title of the
integration will vary.
"""
with patch(
"homeassistant.components.apcupsd.coordinator.aioapcaccess.request_status"
) as mock_request_status:
status = MOCK_MINIMAL_STATUS | extra_status
mock_request_status.return_value = status
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_USER}, data=CONF_DATA
)
@@ -184,30 +158,21 @@ async def test_flow_minimal_status(
async def test_reconfigure_flow_works(
hass: HomeAssistant, mock_setup_entry: AsyncMock
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_config_entry: MockConfigEntry,
mock_request_status: AsyncMock,
) -> None:
"""Test successful reconfiguration of an existing entry."""
mock_entry = MockConfigEntry(
version=1,
domain=DOMAIN,
title="APCUPSd",
data=CONF_DATA,
unique_id=MOCK_STATUS["SERIALNO"],
source=SOURCE_USER,
)
mock_entry.add_to_hass(hass)
mock_config_entry.add_to_hass(hass)
result = await mock_entry.start_reconfigure_flow(hass)
result = await mock_config_entry.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
# New configuration data with different host/port.
new_conf_data = {CONF_HOST: "new_host", CONF_PORT: 4321}
with patch(
"homeassistant.components.apcupsd.coordinator.aioapcaccess.request_status",
return_value=MOCK_STATUS,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=new_conf_data
)
@@ -218,34 +183,26 @@ async def test_reconfigure_flow_works(
assert result["reason"] == "reconfigure_successful"
# Check that the entry was updated with the new configuration.
assert mock_entry.data[CONF_HOST] == new_conf_data[CONF_HOST]
assert mock_entry.data[CONF_PORT] == new_conf_data[CONF_PORT]
assert mock_config_entry.data[CONF_HOST] == new_conf_data[CONF_HOST]
assert mock_config_entry.data[CONF_PORT] == new_conf_data[CONF_PORT]
async def test_reconfigure_flow_cannot_connect(
hass: HomeAssistant, mock_setup_entry: AsyncMock
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_config_entry: MockConfigEntry,
mock_request_status: AsyncMock,
) -> None:
"""Test reconfiguration with connection error and recovery."""
mock_entry = MockConfigEntry(
version=1,
domain=DOMAIN,
title="APCUPSd",
data=CONF_DATA,
unique_id=MOCK_STATUS["SERIALNO"],
source=SOURCE_USER,
)
mock_entry.add_to_hass(hass)
mock_config_entry.add_to_hass(hass)
result = await mock_entry.start_reconfigure_flow(hass)
result = await mock_config_entry.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
# New configuration data with different host/port.
new_conf_data = {CONF_HOST: "new_host", CONF_PORT: 4321}
with patch(
"homeassistant.components.apcupsd.coordinator.aioapcaccess.request_status",
side_effect=OSError(),
):
mock_request_status.side_effect = OSError()
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=new_conf_data
)
@@ -254,17 +211,14 @@ async def test_reconfigure_flow_cannot_connect(
assert result["errors"]["base"] == "cannot_connect"
# Test recovery by fixing the connection issue.
with patch(
"homeassistant.components.apcupsd.coordinator.aioapcaccess.request_status",
return_value=MOCK_STATUS,
):
mock_request_status.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=new_conf_data
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
assert mock_entry.data == new_conf_data
assert mock_config_entry.data == new_conf_data
@pytest.mark.parametrize(
@@ -276,34 +230,26 @@ async def test_reconfigure_flow_cannot_connect(
],
)
async def test_reconfigure_flow_wrong_device(
hass: HomeAssistant, unique_id_before: str | None, unique_id_after: str | None
hass: HomeAssistant,
unique_id_before: str | None,
unique_id_after: str,
mock_config_entry: MockConfigEntry,
mock_request_status: AsyncMock,
) -> None:
"""Test reconfiguration with a different device (wrong serial number)."""
mock_entry = MockConfigEntry(
version=1,
domain=DOMAIN,
title="APCUPSd",
data=CONF_DATA,
unique_id=unique_id_before,
source=SOURCE_USER,
mock_config_entry.add_to_hass(hass)
hass.config_entries.async_update_entry(
mock_config_entry, unique_id=unique_id_before
)
mock_entry.add_to_hass(hass)
result = await mock_entry.start_reconfigure_flow(hass)
result = await mock_config_entry.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
# New configuration data with different host/port.
new_conf_data = {CONF_HOST: "new_host", CONF_PORT: 4321}
# Make a copy of the status and modify the serial number if needed.
mock_status = {k: v for k, v in MOCK_STATUS.items() if k != "SERIALNO"}
mock_status["SERIALNO"] = unique_id_after
with patch(
"homeassistant.components.apcupsd.coordinator.aioapcaccess.request_status",
return_value=mock_status,
):
mock_request_status.return_value = MOCK_STATUS | {"SERIALNO": unique_id_after}
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=new_conf_data
result["flow_id"], user_input={CONF_HOST: "new_host", CONF_PORT: 4321}
)
assert result["type"] is FlowResultType.ABORT

6
tests/components/apcupsd/test_diagnostics.py
View File

@@ -4,8 +4,7 @@ from syrupy.assertion import SnapshotAssertion
from homeassistant.core import HomeAssistant
from . import async_init_integration
from tests.common import MockConfigEntry
from tests.components.diagnostics import get_diagnostics_for_config_entry
from tests.typing import ClientSessionGenerator
@@ -14,7 +13,8 @@ async def test_diagnostics(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
snapshot: SnapshotAssertion,
init_integration: MockConfigEntry,
) -> None:
"""Test diagnostics."""
entry = await async_init_integration(hass)
entry = init_integration
assert await get_diagnostics_for_config_entry(hass, hass_client, entry) == snapshot

68
tests/components/apcupsd/test_init.py
View File

@@ -1,28 +1,28 @@
"""Test init of APCUPSd integration."""
import asyncio
from collections import OrderedDict
from unittest.mock import patch
from unittest.mock import AsyncMock
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.apcupsd.const import DOMAIN
from homeassistant.components.apcupsd.coordinator import UPDATE_INTERVAL
from homeassistant.config_entries import SOURCE_USER, ConfigEntryState
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.entity_platform import async_get_platforms
from homeassistant.util import slugify, utcnow
from . import CONF_DATA, MOCK_MINIMAL_STATUS, MOCK_STATUS, async_init_integration
from . import MOCK_MINIMAL_STATUS, MOCK_STATUS
from tests.common import MockConfigEntry, async_fire_time_changed
@pytest.mark.parametrize("mock_config_entry", ["mocked-config-entry-id"], indirect=True)
@pytest.mark.parametrize(
"status",
"mock_request_status",
[
# Contains "SERIALNO" and "UPSNAME" fields.
# We should create devices for the entities and prefix their IDs with "MyUPS".
@@ -32,23 +32,26 @@ from tests.common import MockConfigEntry, async_fire_time_changed
MOCK_MINIMAL_STATUS | {"SERIALNO": "XXXX"},
# Does not contain either "SERIALNO" field or "UPSNAME" field.
# Our integration should work fine without it by falling back to config entry ID as unique
# ID and "APC UPS" as default name.
# ID and "APC UPS" as the default name.
MOCK_MINIMAL_STATUS,
# Some models report "Blank" as SERIALNO, but we should treat it as not reported.
MOCK_MINIMAL_STATUS | {"SERIALNO": "Blank"},
],
indirect=True,
)
async def test_async_setup_entry(
hass: HomeAssistant,
status: OrderedDict,
device_registry: dr.DeviceRegistry,
snapshot: SnapshotAssertion,
init_integration: MockConfigEntry,
mock_request_status: AsyncMock,
) -> None:
"""Test a successful setup entry."""
config_entry = await async_init_integration(hass, status=status)
device_entry = device_registry.async_get_device(
identifiers={(DOMAIN, config_entry.unique_id or config_entry.entry_id)}
)
status = mock_request_status.return_value
entry = init_integration
identifiers = {(DOMAIN, entry.unique_id or entry.entry_id)}
device_entry = device_registry.async_get_device(identifiers=identifiers)
name = f"device_{device_entry.name}_{status.get('SERIALNO', '<no serial>')}"
assert device_entry == snapshot(name=name)
@@ -61,28 +64,26 @@ async def test_async_setup_entry(
"error",
[OSError(), asyncio.IncompleteReadError(partial=b"", expected=0)],
)
async def test_connection_error(hass: HomeAssistant, error: Exception) -> None:
async def test_connection_error(
hass: HomeAssistant,
error: Exception,
mock_config_entry: MockConfigEntry,
mock_request_status: AsyncMock,
) -> None:
"""Test connection error during integration setup."""
entry = MockConfigEntry(
version=1,
domain=DOMAIN,
title="APCUPSd",
data=CONF_DATA,
source=SOURCE_USER,
)
mock_config_entry.add_to_hass(hass)
mock_request_status.side_effect = error
entry.add_to_hass(hass)
with patch("aioapcaccess.request_status", side_effect=error):
await hass.config_entries.async_setup(entry.entry_id)
assert entry.state is ConfigEntryState.SETUP_RETRY
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_unload_remove_entry(hass: HomeAssistant) -> None:
async def test_unload_remove_entry(
hass: HomeAssistant,
init_integration: MockConfigEntry,
) -> None:
"""Test successful unload and removal of an entry."""
entry = await async_init_integration(
hass, host="test1", status=MOCK_STATUS, entry_id="entry-id-1"
)
entry = init_integration
assert entry.state is ConfigEntryState.LOADED
# Unload the entry.
@@ -96,17 +97,18 @@ async def test_unload_remove_entry(hass: HomeAssistant) -> None:
assert len(hass.config_entries.async_entries(DOMAIN)) == 0
async def test_availability(hass: HomeAssistant) -> None:
async def test_availability(
hass: HomeAssistant,
mock_request_status: AsyncMock,
init_integration: MockConfigEntry,
) -> None:
"""Ensure that we mark the entity's availability properly when network is down / back up."""
await async_init_integration(hass)
device_slug = slugify(MOCK_STATUS["UPSNAME"])
device_slug = slugify(mock_request_status.return_value["UPSNAME"])
state = hass.states.get(f"sensor.{device_slug}_load")
assert state
assert state.state != STATE_UNAVAILABLE
assert pytest.approx(float(state.state)) == 14.0
with patch("aioapcaccess.request_status") as mock_request_status:
# Mock a network error and then trigger an auto-polling event.
mock_request_status.side_effect = OSError()
future = utcnow() + UPDATE_INTERVAL

59
tests/components/apcupsd/test_sensor.py
View File

@@ -1,7 +1,7 @@
"""Test sensors of APCUPSd integration."""
from datetime import timedelta
from unittest.mock import patch
from unittest.mock import AsyncMock
import pytest
from syrupy.assertion import SnapshotAssertion
@@ -19,35 +19,43 @@ from homeassistant.setup import async_setup_component
from homeassistant.util import slugify
from homeassistant.util.dt import utcnow
from . import MOCK_MINIMAL_STATUS, MOCK_STATUS, async_init_integration
from . import MOCK_MINIMAL_STATUS, MOCK_STATUS
from tests.common import async_fire_time_changed, snapshot_platform
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
pytestmark = pytest.mark.usefixtures(
"entity_registry_enabled_by_default", "init_integration"
)
@pytest.fixture
def platforms() -> list[Platform]:
"""Overridden fixture to specify platforms to test."""
return [Platform.SENSOR]
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_sensor(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test states of sensor."""
with patch("homeassistant.components.apcupsd.PLATFORMS", [Platform.SENSOR]):
config_entry = await async_init_integration(hass, status=MOCK_STATUS)
await snapshot_platform(hass, entity_registry, snapshot, config_entry.entry_id)
"""Test states of sensor entities."""
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
async def test_state_update(hass: HomeAssistant) -> None:
async def test_state_update(
hass: HomeAssistant,
mock_request_status: AsyncMock,
) -> None:
"""Ensure the sensor state changes after updating the data."""
await async_init_integration(hass)
device_slug = slugify(MOCK_STATUS["UPSNAME"])
device_slug = slugify(mock_request_status.return_value["UPSNAME"])
state = hass.states.get(f"sensor.{device_slug}_load")
assert state
assert state.state != STATE_UNAVAILABLE
assert state.state == "14.0"
new_status = MOCK_STATUS | {"LOADPCT": "15.0 Percent"}
with patch("aioapcaccess.request_status", return_value=new_status):
mock_request_status.return_value = MOCK_STATUS | {"LOADPCT": "15.0 Percent"}
future = utcnow() + timedelta(minutes=2)
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
@@ -58,14 +66,15 @@ async def test_state_update(hass: HomeAssistant) -> None:
assert state.state == "15.0"
async def test_manual_update_entity(hass: HomeAssistant) -> None:
async def test_manual_update_entity(
hass: HomeAssistant,
mock_request_status: AsyncMock,
) -> None:
"""Test multiple simultaneous manual update entity via service homeassistant/update_entity.
We should only do network call once for the multiple simultaneous update entity services.
"""
await async_init_integration(hass)
device_slug = slugify(MOCK_STATUS["UPSNAME"])
device_slug = slugify(mock_request_status.return_value["UPSNAME"])
# Assert the initial state of sensor.ups_load.
state = hass.states.get(f"sensor.{device_slug}_load")
assert state
@@ -75,13 +84,13 @@ async def test_manual_update_entity(hass: HomeAssistant) -> None:
# Setup HASS for calling the update_entity service.
await async_setup_component(hass, "homeassistant", {})
with patch("aioapcaccess.request_status") as mock_request_status:
mock_request_status.return_value = MOCK_STATUS | {
"LOADPCT": "15.0 Percent",
"BCHARGE": "99.0 Percent",
}
# Now, we fast-forward the time to pass the debouncer cooldown, but put it
# before the normal update interval to see if the manual update works.
request_call_count_before = mock_request_status.call_count
future = utcnow() + timedelta(seconds=REQUEST_REFRESH_COOLDOWN)
async_fire_time_changed(hass, future)
await hass.services.async_call(
@@ -97,7 +106,7 @@ async def test_manual_update_entity(hass: HomeAssistant) -> None:
)
# Even if we requested updates for two entities, our integration should smartly
# group the API calls to just one.
assert mock_request_status.call_count == 1
assert mock_request_status.call_count == request_call_count_before + 1
# The new state should be effective.
state = hass.states.get(f"sensor.{device_slug}_load")
@@ -106,10 +115,12 @@ async def test_manual_update_entity(hass: HomeAssistant) -> None:
assert state.state == "15.0"
async def test_sensor_unknown(hass: HomeAssistant) -> None:
@pytest.mark.parametrize("mock_request_status", [MOCK_MINIMAL_STATUS], indirect=True)
async def test_sensor_unknown(
hass: HomeAssistant,
mock_request_status: AsyncMock,
) -> None:
"""Test if our integration can properly mark certain sensors as unknown when it becomes so."""
await async_init_integration(hass, status=MOCK_MINIMAL_STATUS)
ups_mode_id = "sensor.apc_ups_mode"
last_self_test_id = "sensor.apc_ups_last_self_test"
@@ -121,7 +132,6 @@ async def test_sensor_unknown(hass: HomeAssistant) -> None:
# Simulate an event (a self test) such that "LASTSTEST" field is being reported, the state of
# the sensor should be properly updated with the corresponding value.
with patch("aioapcaccess.request_status") as mock_request_status:
mock_request_status.return_value = MOCK_MINIMAL_STATUS | {
"LASTSTEST": "1970-01-01 00:00:00 0000"
}
@@ -131,7 +141,6 @@ async def test_sensor_unknown(hass: HomeAssistant) -> None:
assert hass.states.get(last_self_test_id).state == "1970-01-01 00:00:00 0000"
# Simulate another event (e.g., daemon restart) such that "LASTSTEST" is no longer reported.
with patch("aioapcaccess.request_status") as mock_request_status:
mock_request_status.return_value = MOCK_MINIMAL_STATUS
future = utcnow() + timedelta(minutes=2)
async_fire_time_changed(hass, future)