jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-21 04:07:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Rewrite HomeWizard Energy tests (#103000)

Browse Source 
Co-authored-by: Duco Sebel <74970928+DCSBL@users.noreply.github.com>
This commit is contained in:
Franck Nijhof 2023-10-30 14:07:42 +01:00 committed by GitHub
parent b3743937de
commit 487dcf227e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
20 changed files with 5953 additions and 3198 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

109
tests/components/homewizard/conftest.py
View File

@ -14,58 +14,85 @@ from tests.common import MockConfigEntry, load_fixture
@pytest.fixture
def mock_config_entry_data():
"""Return the default mocked config entry data."""
return {
"product_name": "Product Name",
"product_type": "product_type",
"serial": "aabbccddeeff",
"name": "Product Name",
CONF_IP_ADDRESS: "1.2.3.4",
}
def device_fixture() -> str:
"""Return the device fixture for a specific device."""
return "device-HWE-P1.json"
@pytest.fixture
def data_fixture() -> str:
"""Return the data fixture for a specific device."""
return "data-HWE-P1.json"
@pytest.fixture
def state_fixture() -> str:
"""Return the state fixture for a specific device."""
return "state.json"
@pytest.fixture
def system_fixture() -> str:
"""Return the system fixture for a specific device."""
return "system.json"
@pytest.fixture
def mock_homewizardenergy(
device_fixture: str,
data_fixture: str,
state_fixture: str,
system_fixture: str,
) -> MagicMock:
"""Return a mock bridge."""
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
autospec=True,
) as homewizard, patch(
"homeassistant.components.homewizard.config_flow.HomeWizardEnergy",
new=homewizard,
):
client = homewizard.return_value
client.device.return_value = Device.from_dict(
json.loads(load_fixture(device_fixture, DOMAIN))
)
client.data.return_value = Data.from_dict(
json.loads(load_fixture(data_fixture, DOMAIN))
)
client.state.return_value = State.from_dict(
json.loads(load_fixture(state_fixture, DOMAIN))
)
client.system.return_value = System.from_dict(
json.loads(load_fixture(system_fixture, DOMAIN))
)
yield client
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock, None, None]:
"""Mock setting up a config entry."""
with patch(
"homeassistant.components.homewizard.async_setup_entry", return_value=True
) as mock_setup:
yield mock_setup
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return the default mocked config entry."""
return MockConfigEntry(
title="Product Name (aabbccddeeff)",
title="Device",
domain=DOMAIN,
data={CONF_IP_ADDRESS: "1.2.3.4"},
data={
"product_name": "Product name",
"product_type": "product_type",
"serial": "aabbccddeeff",
CONF_IP_ADDRESS: "127.0.0.1",
},
unique_id="aabbccddeeff",
)
@pytest.fixture
def mock_homewizardenergy():
"""Return a mocked all-feature device."""
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
) as device:
client = device.return_value
client.device = AsyncMock(
side_effect=lambda: Device.from_dict(
json.loads(load_fixture("homewizard/device.json"))
)
)
client.data = AsyncMock(
side_effect=lambda: Data.from_dict(
json.loads(load_fixture("homewizard/data.json"))
)
)
client.state = AsyncMock(
side_effect=lambda: State.from_dict(
json.loads(load_fixture("homewizard/state.json"))
)
)
client.system = AsyncMock(
side_effect=lambda: System.from_dict(
json.loads(load_fixture("homewizard/system.json"))
)
)
yield device
@pytest.fixture
async def init_integration(
hass: HomeAssistant,

45
tests/components/homewizard/fixtures/data-HWE-P1-unused-exports.json Normal file
View File

@ -0,0 +1,45 @@
{
"wifi_ssid": "My Wi-Fi",
"wifi_strength": 100,
"smr_version": 50,
"meter_model": "ISKRA 2M550T-101",
"unique_id": "00112233445566778899AABBCCDDEEFF",
"active_tariff": 2,
"total_power_import_kwh": 13779.338,
"total_power_import_t1_kwh": 10830.511,
"total_power_import_t2_kwh": 2948.827,
"total_power_import_t3_kwh": 2948.827,
"total_power_import_t4_kwh": 2948.827,
"total_power_export_kwh": 0,
"total_power_export_t1_kwh": 0,
"total_power_export_t2_kwh": 0,
"total_power_export_t3_kwh": 0,
"total_power_export_t4_kwh": 0,
"active_power_w": -123,
"active_power_l1_w": -123,
"active_power_l2_w": 456,
"active_power_l3_w": 123.456,
"active_voltage_l1_v": 230.111,
"active_voltage_l2_v": 230.222,
"active_voltage_l3_v": 230.333,
"active_current_l1_a": -4,
"active_current_l2_a": 2,
"active_current_l3_a": 0,
"active_frequency_hz": 50,
"voltage_sag_l1_count": 1,
"voltage_sag_l2_count": 2,
"voltage_sag_l3_count": 3,
"voltage_swell_l1_count": 4,
"voltage_swell_l2_count": 5,
"voltage_swell_l3_count": 6,
"any_power_fail_count": 4,
"long_power_fail_count": 5,
"total_gas_m3": 1122.333,
"gas_timestamp": 210314112233,
"gas_unique_id": "01FFEEDDCCBBAA99887766554433221100",
"active_power_average_w": 123.0,
"montly_power_peak_w": 1111.0,
"montly_power_peak_timestamp": 230101080010,
"active_liter_lpm": 12.345,
"total_liter_m3": 1234.567
}

4
tests/components/homewizard/fixtures/data.json → tests/components/homewizard/fixtures/data-HWE-P1.json
View File

@ -8,9 +8,13 @@
"total_power_import_kwh": 13779.338,
"total_power_import_t1_kwh": 10830.511,
"total_power_import_t2_kwh": 2948.827,
"total_power_import_t3_kwh": 2948.827,
"total_power_import_t4_kwh": 2948.827,
"total_power_export_kwh": 13086.777,
"total_power_export_t1_kwh": 4321.333,
"total_power_export_t2_kwh": 8765.444,
"total_power_export_t3_kwh": 8765.444,
"total_power_export_t4_kwh": 8765.444,
"active_power_w": -123,
"active_power_l1_w": -123,
"active_power_l2_w": 456,

7
tests/components/homewizard/fixtures/device-HWE-P1.json Normal file
View File

@ -0,0 +1,7 @@
{
"product_type": "HWE-P1",
"product_name": "P1 meter",
"serial": "3c39e7aabbcc",
"firmware_version": "4.19",
"api_version": "v1"
}

4
tests/components/homewizard/fixtures/device.json → tests/components/homewizard/fixtures/device-HWE-SKT.json
View File

@ -1,7 +1,7 @@
{
"product_type": "HWE-SKT",
"product_name": "P1 Meter",
"product_name": "Energy Socket",
"serial": "3c39e7aabbcc",
"firmware_version": "2.11",
"firmware_version": "3.03",
"api_version": "v1"
}

7
tests/components/homewizard/fixtures/device-sdm230.json Normal file
View File

@ -0,0 +1,7 @@
{
"product_type": "SDM230-WIFI",
"product_name": "kWh meter",
"serial": "3c39e7aabbcc",
"firmware_version": "3.06",
"api_version": "v1"
}

34
tests/components/homewizard/generator.py
View File

@ -1,34 +0,0 @@
"""Helper files for unit tests."""
from unittest.mock import AsyncMock
from homewizard_energy.models import Data, Device
def get_mock_device(
serial="aabbccddeeff",
host="1.2.3.4",
product_name="P1 meter",
product_type="HWE-P1",
firmware_version="1.00",
):
"""Return a mock bridge."""
mock_device = AsyncMock()
mock_device.host = host
mock_device.device = AsyncMock(
return_value=Device(
product_name=product_name,
product_type=product_type,
serial=serial,
api_version="V1",
firmware_version=firmware_version,
)
)
mock_device.data = AsyncMock(return_value=Data.from_dict({}))
mock_device.state = AsyncMock(return_value=None)
mock_device.system = AsyncMock(return_value=None)
mock_device.close = AsyncMock()
return mock_device

77
tests/components/homewizard/snapshots/test_button.ambr Normal file
View File

@ -0,0 +1,77 @@
# serializer version: 1
# name: test_identify_button
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'identify',
'friendly_name': 'Device Identify',
}),
'context': <ANY>,
'entity_id': 'button.device_identify',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_identify_button.1
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'button.device_identify',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'name': None,
'options': dict({
}),
'original_device_class': <ButtonDeviceClass.IDENTIFY: 'identify'>,
'original_icon': None,
'original_name': 'Identify',
'platform': 'homewizard',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'aabbccddeeff_identify',
'unit_of_measurement': None,
})
# ---
# name: test_identify_button.2
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'configuration_url': None,
'connections': set({
tuple(
'mac',
'3c:39:e7:aa:bb:cc',
),
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'homewizard',
'3c39e7aabbcc',
),
}),
'is_new': False,
'manufacturer': 'HomeWizard',
'model': 'HWE-P1',
'name': 'Device',
'name_by_user': None,
'serial_number': None,
'suggested_area': None,
'sw_version': '4.19',
'via_device_id': None,
})
# ---

153
tests/components/homewizard/snapshots/test_config_flow.ambr Normal file
View File

@ -0,0 +1,153 @@
# serializer version: 1
# name: test_discovery_flow_during_onboarding
FlowResultSnapshot({
'context': dict({
'source': 'zeroconf',
'unique_id': 'HWE-P1_aabbccddeeff',
}),
'data': dict({
'ip_address': '127.0.0.1',
}),
'description': None,
'description_placeholders': None,
'flow_id': <ANY>,
'handler': 'homewizard',
'options': dict({
}),
'result': ConfigEntrySnapshot({
'data': dict({
'ip_address': '127.0.0.1',
}),
'disabled_by': None,
'domain': 'homewizard',
'entry_id': <ANY>,
'options': dict({
}),
'pref_disable_new_entities': False,
'pref_disable_polling': False,
'source': 'zeroconf',
'title': 'P1 meter',
'unique_id': 'HWE-P1_aabbccddeeff',
'version': 1,
}),
'title': 'P1 meter',
'type': <FlowResultType.CREATE_ENTRY: 'create_entry'>,
'version': 1,
})
# ---
# name: test_discovery_flow_during_onboarding_disabled_api
FlowResultSnapshot({
'context': dict({
'confirm_only': True,
'source': 'zeroconf',
'title_placeholders': dict({
'name': 'P1 meter',
}),
'unique_id': 'HWE-P1_aabbccddeeff',
}),
'data': dict({
'ip_address': '127.0.0.1',
}),
'description': None,
'description_placeholders': None,
'flow_id': <ANY>,
'handler': 'homewizard',
'options': dict({
}),
'result': ConfigEntrySnapshot({
'data': dict({
'ip_address': '127.0.0.1',
}),
'disabled_by': None,
'domain': 'homewizard',
'entry_id': <ANY>,
'options': dict({
}),
'pref_disable_new_entities': False,
'pref_disable_polling': False,
'source': 'zeroconf',
'title': 'P1 meter',
'unique_id': 'HWE-P1_aabbccddeeff',
'version': 1,
}),
'title': 'P1 meter',
'type': <FlowResultType.CREATE_ENTRY: 'create_entry'>,
'version': 1,
})
# ---
# name: test_discovery_flow_works
FlowResultSnapshot({
'context': dict({
'confirm_only': True,
'source': 'zeroconf',
'title_placeholders': dict({
'name': 'Energy Socket (aabbccddeeff)',
}),
'unique_id': 'HWE-SKT_aabbccddeeff',
}),
'data': dict({
'ip_address': '127.0.0.1',
}),
'description': None,
'description_placeholders': None,
'flow_id': <ANY>,
'handler': 'homewizard',
'options': dict({
}),
'result': ConfigEntrySnapshot({
'data': dict({
'ip_address': '127.0.0.1',
}),
'disabled_by': None,
'domain': 'homewizard',
'entry_id': <ANY>,
'options': dict({
}),
'pref_disable_new_entities': False,
'pref_disable_polling': False,
'source': 'zeroconf',
'title': 'Energy Socket',
'unique_id': 'HWE-SKT_aabbccddeeff',
'version': 1,
}),
'title': 'Energy Socket',
'type': <FlowResultType.CREATE_ENTRY: 'create_entry'>,
'version': 1,
})
# ---
# name: test_manual_flow_works
FlowResultSnapshot({
'context': dict({
'source': 'user',
'unique_id': 'HWE-P1_3c39e7aabbcc',
}),
'data': dict({
'ip_address': '2.2.2.2',
}),
'description': None,
'description_placeholders': None,
'flow_id': <ANY>,
'handler': 'homewizard',
'options': dict({
}),
'result': ConfigEntrySnapshot({
'data': dict({
'ip_address': '2.2.2.2',
}),
'disabled_by': None,
'domain': 'homewizard',
'entry_id': <ANY>,
'options': dict({
}),
'pref_disable_new_entities': False,
'pref_disable_polling': False,
'source': 'user',
'title': 'P1 meter',
'unique_id': 'HWE-P1_3c39e7aabbcc',
'version': 1,
}),
'title': 'P1 meter',
'type': <FlowResultType.CREATE_ENTRY: 'create_entry'>,
'version': 1,
})
# ---

142
tests/components/homewizard/snapshots/test_diagnostics.ambr
View File

@ -69,3 +69,145 @@
}),
})
# ---
# name: test_diagnostics[device-HWE-P1.json]
dict({
'data': dict({
'data': dict({
'active_current_l1_a': -4,
'active_current_l2_a': 2,
'active_current_l3_a': 0,
'active_frequency_hz': 50,
'active_liter_lpm': 12.345,
'active_power_average_w': 123.0,
'active_power_l1_w': -123,
'active_power_l2_w': 456,
'active_power_l3_w': 123.456,
'active_power_w': -123,
'active_tariff': 2,
'active_voltage_l1_v': 230.111,
'active_voltage_l2_v': 230.222,
'active_voltage_l3_v': 230.333,
'any_power_fail_count': 4,
'external_devices': None,
'gas_timestamp': '2021-03-14T11:22:33',
'gas_unique_id': '**REDACTED**',
'long_power_fail_count': 5,
'meter_model': 'ISKRA 2M550T-101',
'monthly_power_peak_timestamp': '2023-01-01T08:00:10',
'monthly_power_peak_w': 1111.0,
'smr_version': 50,
'total_gas_m3': 1122.333,
'total_liter_m3': 1234.567,
'total_power_export_kwh': 13086.777,
'total_power_export_t1_kwh': 4321.333,
'total_power_export_t2_kwh': 8765.444,
'total_power_export_t3_kwh': 8765.444,
'total_power_export_t4_kwh': 8765.444,
'total_power_import_kwh': 13779.338,
'total_power_import_t1_kwh': 10830.511,
'total_power_import_t2_kwh': 2948.827,
'total_power_import_t3_kwh': 2948.827,
'total_power_import_t4_kwh': 2948.827,
'unique_meter_id': '**REDACTED**',
'voltage_sag_l1_count': 1,
'voltage_sag_l2_count': 2,
'voltage_sag_l3_count': 3,
'voltage_swell_l1_count': 4,
'voltage_swell_l2_count': 5,
'voltage_swell_l3_count': 6,
'wifi_ssid': '**REDACTED**',
'wifi_strength': 100,
}),
'device': dict({
'api_version': 'v1',
'firmware_version': '4.19',
'product_name': 'P1 meter',
'product_type': 'HWE-P1',
'serial': '**REDACTED**',
}),
'state': None,
'system': dict({
'cloud_enabled': True,
}),
}),
'entry': dict({
'ip_address': '**REDACTED**',
'product_name': 'Product name',
'product_type': 'product_type',
'serial': '**REDACTED**',
}),
})
# ---
# name: test_diagnostics[device-HWE-SKT.json]
dict({
'data': dict({
'data': dict({
'active_current_l1_a': -4,
'active_current_l2_a': 2,
'active_current_l3_a': 0,
'active_frequency_hz': 50,
'active_liter_lpm': 12.345,
'active_power_average_w': 123.0,
'active_power_l1_w': -123,
'active_power_l2_w': 456,
'active_power_l3_w': 123.456,
'active_power_w': -123,
'active_tariff': 2,
'active_voltage_l1_v': 230.111,
'active_voltage_l2_v': 230.222,
'active_voltage_l3_v': 230.333,
'any_power_fail_count': 4,
'external_devices': None,
'gas_timestamp': '2021-03-14T11:22:33',
'gas_unique_id': '**REDACTED**',
'long_power_fail_count': 5,
'meter_model': 'ISKRA 2M550T-101',
'monthly_power_peak_timestamp': '2023-01-01T08:00:10',
'monthly_power_peak_w': 1111.0,
'smr_version': 50,
'total_gas_m3': 1122.333,
'total_liter_m3': 1234.567,
'total_power_export_kwh': 13086.777,
'total_power_export_t1_kwh': 4321.333,
'total_power_export_t2_kwh': 8765.444,
'total_power_export_t3_kwh': 8765.444,
'total_power_export_t4_kwh': 8765.444,
'total_power_import_kwh': 13779.338,
'total_power_import_t1_kwh': 10830.511,
'total_power_import_t2_kwh': 2948.827,
'total_power_import_t3_kwh': 2948.827,
'total_power_import_t4_kwh': 2948.827,
'unique_meter_id': '**REDACTED**',
'voltage_sag_l1_count': 1,
'voltage_sag_l2_count': 2,
'voltage_sag_l3_count': 3,
'voltage_swell_l1_count': 4,
'voltage_swell_l2_count': 5,
'voltage_swell_l3_count': 6,
'wifi_ssid': '**REDACTED**',
'wifi_strength': 100,
}),
'device': dict({
'api_version': 'v1',
'firmware_version': '3.03',
'product_name': 'Energy Socket',
'product_type': 'HWE-SKT',
'serial': '**REDACTED**',
}),
'state': dict({
'brightness': 255,
'power_on': True,
'switch_lock': False,
}),
'system': dict({
'cloud_enabled': True,
}),
}),
'entry': dict({
'ip_address': '**REDACTED**',
'product_name': 'Product name',
'product_type': 'product_type',
'serial': '**REDACTED**',
}),
})
# ---

87
tests/components/homewizard/snapshots/test_number.ambr Normal file
View File

@ -0,0 +1,87 @@
# serializer version: 1
# name: test_number_entities[device-HWE-SKT.json]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Device Status light brightness',
'icon': 'mdi:lightbulb-on',
'max': 100.0,
'min': 0.0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1.0,
'unit_of_measurement': '%',
}),
'context': <ANY>,
'entity_id': 'number.device_status_light_brightness',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': '100',
})
# ---
# name: test_number_entities[device-HWE-SKT.json].1
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'max': 100.0,
'min': 0.0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1.0,
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'number',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'number.device_status_light_brightness',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': 'mdi:lightbulb-on',
'original_name': 'Status light brightness',
'platform': 'homewizard',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'status_light_brightness',
'unique_id': 'aabbccddeeff_status_light_brightness',
'unit_of_measurement': '%',
})
# ---
# name: test_number_entities[device-HWE-SKT.json].2
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'configuration_url': None,
'connections': set({
tuple(
'mac',
'3c:39:e7:aa:bb:cc',
),
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'homewizard',
'3c39e7aabbcc',
),
}),
'is_new': False,
'manufacturer': 'HomeWizard',
'model': 'HWE-SKT',
'name': 'Device',
'name_by_user': None,
'serial_number': None,
'suggested_area': None,
'sw_version': '3.03',
'via_device_id': None,
})
# ---

4546
tests/components/homewizard/snapshots/test_sensor.ambr Normal file
View File

File diff suppressed because it is too large Load Diff

229
tests/components/homewizard/snapshots/test_switch.ambr Normal file
View File

@ -0,0 +1,229 @@
# serializer version: 1
# name: test_switch_entities[switch.device-state_set-power_on-device-HWE-SKT.json]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'outlet',
'friendly_name': 'Device',
}),
'context': <ANY>,
'entity_id': 'switch.device',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_switch_entities[switch.device-state_set-power_on-device-HWE-SKT.json].1
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': None,
'entity_id': 'switch.device',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'name': None,
'options': dict({
}),
'original_device_class': <SwitchDeviceClass.OUTLET: 'outlet'>,
'original_icon': None,
'original_name': None,
'platform': 'homewizard',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'aabbccddeeff_power_on',
'unit_of_measurement': None,
})
# ---
# name: test_switch_entities[switch.device-state_set-power_on-device-HWE-SKT.json].2
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'configuration_url': None,
'connections': set({
tuple(
'mac',
'3c:39:e7:aa:bb:cc',
),
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'homewizard',
'3c39e7aabbcc',
),
}),
'is_new': False,
'manufacturer': 'HomeWizard',
'model': 'HWE-SKT',
'name': 'Device',
'name_by_user': None,
'serial_number': None,
'suggested_area': None,
'sw_version': '3.03',
'via_device_id': None,
})
# ---
# name: test_switch_entities[switch.device_cloud_connection-system_set-cloud_enabled-device-HWE-SKT.json]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Device Cloud connection',
'icon': 'mdi:cloud',
}),
'context': <ANY>,
'entity_id': 'switch.device_cloud_connection',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_switch_entities[switch.device_cloud_connection-system_set-cloud_enabled-device-HWE-SKT.json].1
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'switch.device_cloud_connection',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': 'mdi:cloud',
'original_name': 'Cloud connection',
'platform': 'homewizard',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'cloud_connection',
'unique_id': 'aabbccddeeff_cloud_connection',
'unit_of_measurement': None,
})
# ---
# name: test_switch_entities[switch.device_cloud_connection-system_set-cloud_enabled-device-HWE-SKT.json].2
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'configuration_url': None,
'connections': set({
tuple(
'mac',
'3c:39:e7:aa:bb:cc',
),
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'homewizard',
'3c39e7aabbcc',
),
}),
'is_new': False,
'manufacturer': 'HomeWizard',
'model': 'HWE-SKT',
'name': 'Device',
'name_by_user': None,
'serial_number': None,
'suggested_area': None,
'sw_version': '3.03',
'via_device_id': None,
})
# ---
# name: test_switch_entities[switch.device_switch_lock-state_set-switch_lock-device-HWE-SKT.json]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Device Switch lock',
'icon': 'mdi:lock-open',
}),
'context': <ANY>,
'entity_id': 'switch.device_switch_lock',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_switch_entities[switch.device_switch_lock-state_set-switch_lock-device-HWE-SKT.json].1
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'switch.device_switch_lock',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': 'mdi:lock-open',
'original_name': 'Switch lock',
'platform': 'homewizard',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'switch_lock',
'unique_id': 'aabbccddeeff_switch_lock',
'unit_of_measurement': None,
})
# ---
# name: test_switch_entities[switch.device_switch_lock-state_set-switch_lock-device-HWE-SKT.json].2
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'configuration_url': None,
'connections': set({
tuple(
'mac',
'3c:39:e7:aa:bb:cc',
),
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'homewizard',
'3c39e7aabbcc',
),
}),
'is_new': False,
'manufacturer': 'HomeWizard',
'model': 'HWE-SKT',
'name': 'Device',
'name_by_user': None,
'serial_number': None,
'suggested_area': None,
'sw_version': '3.03',
'via_device_id': None,
})
# ---

168
tests/components/homewizard/test_button.py
View File

@ -1,174 +1,86 @@
"""Test the identify button for HomeWizard."""
from unittest.mock import patch
from unittest.mock import MagicMock
from homewizard_energy.errors import DisabledError, RequestError
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components import button
from homeassistant.const import ATTR_FRIENDLY_NAME, STATE_UNKNOWN
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers import device_registry as dr, entity_registry as er
from .generator import get_mock_device
pytestmark = [
pytest.mark.usefixtures("init_integration"),
pytest.mark.freeze_time("2021-01-01 12:00:00"),
]
@pytest.mark.parametrize("device_fixture", ["device-sdm230.json"])
async def test_identify_button_entity_not_loaded_when_not_available(
hass: HomeAssistant, mock_config_entry_data, mock_config_entry
hass: HomeAssistant,
) -> None:
"""Does not load button when device has no support for it."""
api = get_mock_device(product_type="SDM230-WIFI")
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=api,
):
entry = mock_config_entry
entry.data = mock_config_entry_data
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get("button.product_name_aabbccddeeff_identify") is None
assert not hass.states.get("button.device_identify")
async def test_identify_button_is_loaded(
hass: HomeAssistant, mock_config_entry_data, mock_config_entry
async def test_identify_button(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
mock_homewizardenergy: MagicMock,
snapshot: SnapshotAssertion,
) -> None:
"""Loads button when device has support."""
assert (state := hass.states.get("button.device_identify"))
assert snapshot == state
api = get_mock_device(product_type="HWE-SKT", firmware_version="3.02")
assert (entity_entry := entity_registry.async_get(state.entity_id))
assert snapshot == entity_entry
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=api,
):
entry = mock_config_entry
entry.data = mock_config_entry_data
entry.add_to_hass(hass)
assert entity_entry.device_id
assert (device_entry := device_registry.async_get(entity_entry.device_id))
assert snapshot == device_entry
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get("button.product_name_aabbccddeeff_identify")
assert state
assert (
state.attributes.get(ATTR_FRIENDLY_NAME)
== "Product Name (aabbccddeeff) Identify"
)
entity_registry = er.async_get(hass)
entry = entity_registry.async_get("button.product_name_aabbccddeeff_identify")
assert entry
assert entry.unique_id == "aabbccddeeff_identify"
async def test_identify_press(
hass: HomeAssistant, mock_config_entry_data, mock_config_entry
) -> None:
"""Test button press is handled correctly."""
api = get_mock_device(product_type="HWE-SKT", firmware_version="3.02")
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=api,
):
entry = mock_config_entry
entry.data = mock_config_entry_data
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert (
hass.states.get("button.product_name_aabbccddeeff_identify").state
== STATE_UNKNOWN
)
assert api.identify.call_count == 0
assert len(mock_homewizardenergy.identify.mock_calls) == 0
await hass.services.async_call(
button.DOMAIN,
button.SERVICE_PRESS,
{"entity_id": "button.product_name_aabbccddeeff_identify"},
{ATTR_ENTITY_ID: state.entity_id},
blocking=True,
)
assert api.identify.call_count == 1
assert len(mock_homewizardenergy.identify.mock_calls) == 1
async def test_identify_press_catches_requesterror(
hass: HomeAssistant, mock_config_entry_data, mock_config_entry
) -> None:
"""Test button press is handled RequestError correctly."""
api = get_mock_device(product_type="HWE-SKT", firmware_version="3.02")
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=api,
):
entry = mock_config_entry
entry.data = mock_config_entry_data
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert (
hass.states.get("button.product_name_aabbccddeeff_identify").state
== STATE_UNKNOWN
)
assert (state := hass.states.get(state.entity_id))
assert state.state == "2021-01-01T12:00:00+00:00"
# Raise RequestError when identify is called
api.identify.side_effect = RequestError()
assert api.identify.call_count == 0
mock_homewizardenergy.identify.side_effect = RequestError()
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
button.DOMAIN,
button.SERVICE_PRESS,
{"entity_id": "button.product_name_aabbccddeeff_identify"},
{ATTR_ENTITY_ID: state.entity_id},
blocking=True,
)
assert api.identify.call_count == 1
assert len(mock_homewizardenergy.identify.mock_calls) == 2
async def test_identify_press_catches_disablederror(
hass: HomeAssistant, mock_config_entry_data, mock_config_entry
) -> None:
"""Test button press is handled DisabledError correctly."""
api = get_mock_device(product_type="HWE-SKT", firmware_version="3.02")
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=api,
):
entry = mock_config_entry
entry.data = mock_config_entry_data
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert (
hass.states.get("button.product_name_aabbccddeeff_identify").state
== STATE_UNKNOWN
)
assert (state := hass.states.get(state.entity_id))
assert state.state == "2021-01-01T12:00:00+00:00"
# Raise RequestError when identify is called
api.identify.side_effect = DisabledError()
assert api.identify.call_count == 0
mock_homewizardenergy.identify.side_effect = DisabledError()
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
button.DOMAIN,
button.SERVICE_PRESS,
{"entity_id": "button.product_name_aabbccddeeff_identify"},
{ATTR_ENTITY_ID: state.entity_id},
blocking=True,
)
assert api.identify.call_count == 1
assert len(mock_homewizardenergy.identify.mock_calls) == 3
assert (state := hass.states.get(state.entity_id))
assert state.state == "2021-01-01T12:00:00+00:00"

581
tests/components/homewizard/test_config_flow.py
View File

@ -1,8 +1,10 @@
"""Test the homewizard config flow."""
from ipaddress import ip_address
from unittest.mock import MagicMock, patch
from unittest.mock import AsyncMock, MagicMock
from homewizard_energy.errors import DisabledError, RequestError, UnsupportedError
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant import config_entries
from homeassistant.components import zeroconf
@ -11,371 +13,308 @@ from homeassistant.const import CONF_IP_ADDRESS
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .generator import get_mock_device
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
@pytest.mark.usefixtures("mock_setup_entry")
async def test_manual_flow_works(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
hass: HomeAssistant,
mock_homewizardenergy: MagicMock,
mock_setup_entry: AsyncMock,
snapshot: SnapshotAssertion,
) -> None:
"""Test config flow accepts user configuration."""
device = get_mock_device()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
with patch(
"homeassistant.components.homewizard.config_flow.HomeWizardEnergy",
return_value=device,
), patch(
"homeassistant.components.homewizard.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_IP_ADDRESS: "2.2.2.2"}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_IP_ADDRESS: "2.2.2.2"}
)
assert result["type"] == "create_entry"
assert result["title"] == "P1 meter"
assert result["data"][CONF_IP_ADDRESS] == "2.2.2.2"
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result == snapshot
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert len(device.close.mock_calls) == len(device.device.mock_calls)
assert len(mock_homewizardenergy.close.mock_calls) == 1
assert len(mock_homewizardenergy.device.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.usefixtures("mock_homewizardenergy", "mock_setup_entry")
async def test_discovery_flow_works(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
hass: HomeAssistant,
snapshot: SnapshotAssertion,
) -> None:
"""Test discovery setup flow works."""
service_info = zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("192.168.43.183"),
ip_addresses=[ip_address("192.168.43.183")],
port=80,
hostname="p1meter-ddeeff.local.",
type="",
name="",
properties={
"api_enabled": "1",
"path": "/api/v1",
"product_name": "Energy Socket",
"product_type": "HWE-SKT",
"serial": "aabbccddeeff",
},
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("127.0.0.1"),
ip_addresses=[ip_address("127.0.0.1")],
port=80,
hostname="p1meter-ddeeff.local.",
type="",
name="",
properties={
"api_enabled": "1",
"path": "/api/v1",
"product_name": "Energy Socket",
"product_type": "HWE-SKT",
"serial": "aabbccddeeff",
},
),
)
with patch(
"homeassistant.components.homewizard.config_flow.HomeWizardEnergy",
return_value=get_mock_device(),
):
flow = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=service_info,
)
with patch(
"homeassistant.components.homewizard.async_setup_entry",
return_value=True,
), patch(
"homeassistant.components.homewizard.config_flow.HomeWizardEnergy",
return_value=get_mock_device(),
):
result = await hass.config_entries.flow.async_configure(
flow["flow_id"], user_input=None
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "discovery_confirm"
with patch(
"homeassistant.components.homewizard.async_setup_entry",
return_value=True,
), patch(
"homeassistant.components.homewizard.config_flow.HomeWizardEnergy",
return_value=get_mock_device(),
):
result = await hass.config_entries.flow.async_configure(
flow["flow_id"], user_input={"ip_address": "192.168.43.183"}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=None
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "discovery_confirm"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={"ip_address": "127.0.0.1"}
)
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "Energy Socket"
assert result["data"][CONF_IP_ADDRESS] == "192.168.43.183"
assert result["result"]
assert result["result"].unique_id == "HWE-SKT_aabbccddeeff"
assert result == snapshot
@pytest.mark.usefixtures("mock_homewizardenergy")
async def test_discovery_flow_during_onboarding(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, mock_onboarding: MagicMock
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_onboarding: MagicMock,
snapshot: SnapshotAssertion,
) -> None:
"""Test discovery setup flow during onboarding."""
with patch(
"homeassistant.components.homewizard.async_setup_entry",
return_value=True,
) as mock_setup_entry, patch(
"homeassistant.components.homewizard.config_flow.HomeWizardEnergy",
return_value=get_mock_device(),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("192.168.43.183"),
ip_addresses=[ip_address("192.168.43.183")],
port=80,
hostname="p1meter-ddeeff.local.",
type="mock_type",
name="mock_name",
properties={
"api_enabled": "1",
"path": "/api/v1",
"product_name": "P1 meter",
"product_type": "HWE-P1",
"serial": "aabbccddeeff",
},
),
)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("127.0.0.1"),
ip_addresses=[ip_address("127.0.0.1")],
port=80,
hostname="p1meter-ddeeff.local.",
type="mock_type",
name="mock_name",
properties={
"api_enabled": "1",
"path": "/api/v1",
"product_name": "P1 meter",
"product_type": "HWE-P1",
"serial": "aabbccddeeff",
},
),
)
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "P1 meter"
assert result["data"][CONF_IP_ADDRESS] == "192.168.43.183"
assert result["result"]
assert result["result"].unique_id == "HWE-P1_aabbccddeeff"
assert result == snapshot
assert len(mock_setup_entry.mock_calls) == 1
assert len(mock_onboarding.mock_calls) == 1
async def test_discovery_flow_during_onboarding_disabled_api(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, mock_onboarding: MagicMock
hass: HomeAssistant,
mock_homewizardenergy: MagicMock,
mock_setup_entry: AsyncMock,
mock_onboarding: MagicMock,
snapshot: SnapshotAssertion,
) -> None:
"""Test discovery setup flow during onboarding with a disabled API."""
mock_homewizardenergy.device.side_effect = DisabledError
def mock_initialize():
raise DisabledError
device = get_mock_device()
device.device.side_effect = mock_initialize
with patch(
"homeassistant.components.homewizard.config_flow.HomeWizardEnergy",
return_value=device,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("192.168.43.183"),
ip_addresses=[ip_address("192.168.43.183")],
port=80,
hostname="p1meter-ddeeff.local.",
type="mock_type",
name="mock_name",
properties={
"api_enabled": "0",
"path": "/api/v1",
"product_name": "P1 meter",
"product_type": "HWE-P1",
"serial": "aabbccddeeff",
},
),
)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("127.0.0.1"),
ip_addresses=[ip_address("127.0.0.1")],
port=80,
hostname="p1meter-ddeeff.local.",
type="mock_type",
name="mock_name",
properties={
"api_enabled": "0",
"path": "/api/v1",
"product_name": "P1 meter",
"product_type": "HWE-P1",
"serial": "aabbccddeeff",
},
),
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "discovery_confirm"
assert result["errors"] == {"base": "api_not_enabled"}
# We are onboarded, user enabled API again and picks up from discovery/config flow
device.device.side_effect = None
mock_homewizardenergy.device.side_effect = None
mock_onboarding.return_value = True
with patch(
"homeassistant.components.homewizard.async_setup_entry",
return_value=True,
) as mock_setup_entry, patch(
"homeassistant.components.homewizard.config_flow.HomeWizardEnergy",
return_value=device,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={"ip_address": "192.168.43.183"}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={"ip_address": "127.0.0.1"}
)
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "P1 meter"
assert result["data"][CONF_IP_ADDRESS] == "192.168.43.183"
assert result["result"]
assert result["result"].unique_id == "HWE-P1_aabbccddeeff"
assert result == snapshot
assert len(mock_setup_entry.mock_calls) == 1
assert len(mock_onboarding.mock_calls) == 1
async def test_discovery_disabled_api(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
hass: HomeAssistant,
mock_homewizardenergy: MagicMock,
) -> None:
"""Test discovery detecting disabled api."""
service_info = zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("192.168.43.183"),
ip_addresses=[ip_address("192.168.43.183")],
port=80,
hostname="p1meter-ddeeff.local.",
type="",
name="",
properties={
"api_enabled": "0",
"path": "/api/v1",
"product_name": "P1 meter",
"product_type": "HWE-P1",
"serial": "aabbccddeeff",
},
)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=service_info,
data=zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("127.0.0.1"),
ip_addresses=[ip_address("127.0.0.1")],
port=80,
hostname="p1meter-ddeeff.local.",
type="",
name="",
properties={
"api_enabled": "0",
"path": "/api/v1",
"product_name": "P1 meter",
"product_type": "HWE-P1",
"serial": "aabbccddeeff",
},
),
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "discovery_confirm"
def mock_initialize():
raise DisabledError
mock_homewizardenergy.device.side_effect = DisabledError
device = get_mock_device()
device.device.side_effect = mock_initialize
with patch(
"homeassistant.components.homewizard.async_setup_entry",
return_value=True,
), patch(
"homeassistant.components.homewizard.config_flow.HomeWizardEnergy",
return_value=device,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={"ip_address": "192.168.43.183"}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={"ip_address": "127.0.0.1"}
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {"base": "api_not_enabled"}
async def test_discovery_missing_data_in_service_info(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
async def test_discovery_missing_data_in_service_info(hass: HomeAssistant) -> None:
"""Test discovery detecting missing discovery info."""
service_info = zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("192.168.43.183"),
ip_addresses=[ip_address("192.168.43.183")],
port=80,
hostname="p1meter-ddeeff.local.",
type="",
name="",
properties={
# "api_enabled": "1", --> removed
"path": "/api/v1",
"product_name": "P1 meter",
"product_type": "HWE-P1",
"serial": "aabbccddeeff",
},
)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=service_info,
data=zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("127.0.0.1"),
ip_addresses=[ip_address("127.0.0.1")],
port=80,
hostname="p1meter-ddeeff.local.",
type="",
name="",
properties={
# "api_enabled": "1", --> removed
"path": "/api/v1",
"product_name": "P1 meter",
"product_type": "HWE-P1",
"serial": "aabbccddeeff",
},
),
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "invalid_discovery_parameters"
async def test_discovery_invalid_api(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
async def test_discovery_invalid_api(hass: HomeAssistant) -> None:
"""Test discovery detecting invalid_api."""
service_info = zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("192.168.43.183"),
ip_addresses=[ip_address("192.168.43.183")],
port=80,
hostname="p1meter-ddeeff.local.",
type="",
name="",
properties={
"api_enabled": "1",
"path": "/api/not_v1",
"product_name": "P1 meter",
"product_type": "HWE-P1",
"serial": "aabbccddeeff",
},
)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=service_info,
data=zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("127.0.0.1"),
ip_addresses=[ip_address("127.0.0.1")],
port=80,
hostname="p1meter-ddeeff.local.",
type="",
name="",
properties={
"api_enabled": "1",
"path": "/api/not_v1",
"product_name": "P1 meter",
"product_type": "HWE-P1",
"serial": "aabbccddeeff",
},
),
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "unsupported_api_version"
async def test_check_disabled_api(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
@pytest.mark.usefixtures("mock_setup_entry")
@pytest.mark.parametrize(
("exception", "reason"),
[(DisabledError, "api_not_enabled"), (RequestError, "network_error")],
)
async def test_error_flow(
hass: HomeAssistant,
mock_homewizardenergy: MagicMock,
exception: Exception,
reason: str,
) -> None:
"""Test check detecting disabled api."""
def mock_initialize():
raise DisabledError
device = get_mock_device()
device.device.side_effect = mock_initialize
mock_homewizardenergy.device.side_effect = exception
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
with patch(
"homeassistant.components.homewizard.config_flow.HomeWizardEnergy",
return_value=device,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_IP_ADDRESS: "2.2.2.2"}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_IP_ADDRESS: "127.0.0.1"}
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {"base": "api_not_enabled"}
assert result["errors"] == {"base": reason}
# Recover from error
mock_homewizardenergy.device.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_IP_ADDRESS: "127.0.0.1"}
)
assert result["type"] == FlowResultType.CREATE_ENTRY
async def test_check_error_handling_api(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
@pytest.mark.parametrize(
("exception", "reason"),
[
(Exception, "unknown_error"),
(UnsupportedError, "unsupported_api_version"),
],
)
async def test_abort_flow(
hass: HomeAssistant,
mock_homewizardenergy: MagicMock,
exception: Exception,
reason: str,
) -> None:
"""Test check detecting error with api."""
def mock_initialize():
raise Exception()
device = get_mock_device()
device.device.side_effect = mock_initialize
mock_homewizardenergy.device.side_effect = exception
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
@ -384,146 +323,60 @@ async def test_check_error_handling_api(
assert result["type"] == "form"
assert result["step_id"] == "user"
with patch(
"homeassistant.components.homewizard.config_flow.HomeWizardEnergy",
return_value=device,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_IP_ADDRESS: "2.2.2.2"}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_IP_ADDRESS: "2.2.2.2"}
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "unknown_error"
async def test_check_detects_invalid_api(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test check detecting device endpoint failed fetching data."""
def mock_initialize():
raise UnsupportedError
device = get_mock_device()
device.device.side_effect = mock_initialize
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
with patch(
"homeassistant.components.homewizard.config_flow.HomeWizardEnergy",
return_value=device,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_IP_ADDRESS: "2.2.2.2"}
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "unsupported_api_version"
async def test_check_requesterror(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test check detecting device endpoint failed fetching data due to a requesterror."""
def mock_initialize():
raise RequestError
device = get_mock_device()
device.device.side_effect = mock_initialize
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
with patch(
"homeassistant.components.homewizard.config_flow.HomeWizardEnergy",
return_value=device,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_IP_ADDRESS: "2.2.2.2"}
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {"base": "network_error"}
assert result["reason"] == reason
@pytest.mark.usefixtures("mock_homewizardenergy", "mock_setup_entry")
async def test_reauth_flow(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test reauth flow while API is enabled."""
mock_entry = MockConfigEntry(
domain="homewizard_energy", data={CONF_IP_ADDRESS: "1.2.3.4"}
)
mock_entry.add_to_hass(hass)
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": config_entries.SOURCE_REAUTH,
"entry_id": mock_entry.entry_id,
"entry_id": mock_config_entry.entry_id,
},
)
assert result["type"] == "form"
assert result["step_id"] == "reauth_confirm"
device = get_mock_device()
with patch(
"homeassistant.components.homewizard.async_setup_entry",
return_value=True,
), patch(
"homeassistant.components.homewizard.config_flow.HomeWizardEnergy",
return_value=device,
):
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
async def test_reauth_error(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
hass: HomeAssistant,
mock_homewizardenergy: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test reauth flow while API is still disabled."""
def mock_initialize():
raise DisabledError()
mock_entry = MockConfigEntry(
domain="homewizard_energy", data={CONF_IP_ADDRESS: "1.2.3.4"}
)
mock_entry.add_to_hass(hass)
mock_homewizardenergy.device.side_effect = DisabledError
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": config_entries.SOURCE_REAUTH,
"entry_id": mock_entry.entry_id,
"entry_id": mock_config_entry.entry_id,
},
)
assert result["type"] == "form"
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
device = get_mock_device()
device.device.side_effect = mock_initialize
with patch(
"homeassistant.components.homewizard.config_flow.HomeWizardEnergy",
return_value=device,
):
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {"base": "api_not_enabled"}
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {"base": "api_not_enabled"}

4
tests/components/homewizard/test_diagnostics.py
View File

@ -1,5 +1,6 @@
"""Tests for diagnostics data."""
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.core import HomeAssistant
@ -9,6 +10,9 @@ from tests.components.diagnostics import get_diagnostics_for_config_entry
from tests.typing import ClientSessionGenerator
@pytest.mark.parametrize(
"device_fixture", ["device-HWE-P1.json", "device-HWE-SKT.json"]
)
async def test_diagnostics(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,

215
tests/components/homewizard/test_init.py
View File

@ -1,106 +1,62 @@
"""Tests for the homewizard component."""
from asyncio import TimeoutError
from unittest.mock import patch
from unittest.mock import MagicMock
from homewizard_energy.errors import DisabledError, HomeWizardEnergyException
import pytest
from homeassistant.components.homewizard.const import DOMAIN
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState
from homeassistant.const import CONF_IP_ADDRESS
from homeassistant.core import HomeAssistant
from .generator import get_mock_device
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
async def test_load_unload(
aioclient_mock: AiohttpClientMocker, hass: HomeAssistant
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_homewizardenergy: MagicMock,
) -> None:
"""Test loading and unloading of integration."""
device = get_mock_device()
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_IP_ADDRESS: "1.1.1.1"},
unique_id=DOMAIN,
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=device,
):
await hass.config_entries.async_setup(entry.entry_id)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.LOADED
assert mock_config_entry.state is ConfigEntryState.LOADED
assert len(mock_homewizardenergy.device.mock_calls) == 1
await hass.config_entries.async_unload(entry.entry_id)
await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.NOT_LOADED
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
async def test_load_failed_host_unavailable(
aioclient_mock: AiohttpClientMocker, hass: HomeAssistant
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_homewizardenergy: MagicMock,
) -> None:
"""Test setup handles unreachable host."""
def MockInitialize():
raise TimeoutError()
device = get_mock_device()
device.device.side_effect = MockInitialize
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_IP_ADDRESS: "1.1.1.1"},
unique_id=DOMAIN,
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=device,
):
await hass.config_entries.async_setup(entry.entry_id)
mock_homewizardenergy.device.side_effect = TimeoutError()
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.SETUP_RETRY
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_load_detect_api_disabled(
aioclient_mock: AiohttpClientMocker, hass: HomeAssistant
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_homewizardenergy: MagicMock,
) -> None:
"""Test setup detects disabled API."""
def MockInitialize():
raise DisabledError()
device = get_mock_device()
device.device.side_effect = MockInitialize
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_IP_ADDRESS: "1.1.1.1"},
unique_id=DOMAIN,
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=device,
):
await hass.config_entries.async_setup(entry.entry_id)
mock_homewizardenergy.device.side_effect = DisabledError()
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.SETUP_RETRY
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
@ -111,125 +67,54 @@ async def test_load_detect_api_disabled(
assert "context" in flow
assert flow["context"].get("source") == SOURCE_REAUTH
assert flow["context"].get("entry_id") == entry.entry_id
assert flow["context"].get("entry_id") == mock_config_entry.entry_id
@pytest.mark.usefixtures("mock_homewizardenergy")
async def test_load_removes_reauth_flow(
aioclient_mock: AiohttpClientMocker, hass: HomeAssistant
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test setup removes reauth flow when API is enabled."""
device = get_mock_device()
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_IP_ADDRESS: "1.1.1.1"},
unique_id=DOMAIN,
)
entry.add_to_hass(hass)
mock_config_entry.add_to_hass(hass)
# Add reauth flow from 'previously' failed init
entry.async_start_reauth(hass)
mock_config_entry.async_start_reauth(hass)
await hass.async_block_till_done()
flows = hass.config_entries.flow.async_progress_by_handler(DOMAIN)
assert len(flows) == 1
# Initialize entry
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=device,
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.LOADED
assert mock_config_entry.state is ConfigEntryState.LOADED
# Flow should be removed
flows = hass.config_entries.flow.async_progress_by_handler(DOMAIN)
assert len(flows) == 0
@pytest.mark.parametrize(
"exception",
[
HomeWizardEnergyException,
Exception,
],
)
async def test_load_handles_homewizardenergy_exception(
aioclient_mock: AiohttpClientMocker, hass: HomeAssistant
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_homewizardenergy: MagicMock,
exception: Exception,
) -> None:
"""Test setup handles exception from API."""
def MockInitialize():
raise HomeWizardEnergyException()
device = get_mock_device()
device.device.side_effect = MockInitialize
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_IP_ADDRESS: "1.1.1.1"},
unique_id=DOMAIN,
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=device,
):
await hass.config_entries.async_setup(entry.entry_id)
mock_homewizardenergy.device.side_effect = exception
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.SETUP_RETRY or ConfigEntryState.SETUP_ERROR
async def test_load_handles_generic_exception(
aioclient_mock: AiohttpClientMocker, hass: HomeAssistant
) -> None:
"""Test setup handles global exception."""
def MockInitialize():
raise Exception()
device = get_mock_device()
device.device.side_effect = MockInitialize
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_IP_ADDRESS: "1.1.1.1"},
unique_id=DOMAIN,
assert mock_config_entry.state in (
ConfigEntryState.SETUP_RETRY,
ConfigEntryState.SETUP_ERROR,
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=device,
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.SETUP_RETRY or ConfigEntryState.SETUP_ERROR
async def test_load_handles_initialization_error(
aioclient_mock: AiohttpClientMocker, hass: HomeAssistant
) -> None:
"""Test handles non-exception error."""
device = get_mock_device()
device.device = None
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_IP_ADDRESS: "1.1.1.1"},
unique_id=DOMAIN,
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=device,
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.SETUP_RETRY or ConfigEntryState.SETUP_ERROR

293
tests/components/homewizard/test_number.py
View File

@ -1,275 +1,88 @@
"""Test the update coordinator for HomeWizard."""
from unittest.mock import AsyncMock, patch
"""Test the number entity for HomeWizard."""
from unittest.mock import MagicMock
from homewizard_energy.errors import DisabledError, RequestError
from homewizard_energy.models import State
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components import number
from homeassistant.components.homewizard.const import UPDATE_INTERVAL
from homeassistant.components.number import ATTR_VALUE, SERVICE_SET_VALUE
from homeassistant.const import ATTR_ENTITY_ID, ATTR_FRIENDLY_NAME
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNKNOWN
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers import device_registry as dr, entity_registry as er
import homeassistant.util.dt as dt_util
from .generator import get_mock_device
from tests.common import async_fire_time_changed
async def test_number_entity_not_loaded_when_not_available(
hass: HomeAssistant, mock_config_entry_data, mock_config_entry
@pytest.mark.usefixtures("init_integration")
@pytest.mark.parametrize("device_fixture", ["device-HWE-SKT.json"])
async def test_number_entities(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
mock_homewizardenergy: MagicMock,
snapshot: SnapshotAssertion,
) -> None:
"""Test entity does not load number when brightness is not available."""
"""Test number handles state changes correctly."""
assert (state := hass.states.get("number.device_status_light_brightness"))
assert snapshot == state
api = get_mock_device()
assert (entity_entry := entity_registry.async_get(state.entity_id))
assert snapshot == entity_entry
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=api,
):
entry = mock_config_entry
entry.data = mock_config_entry_data
entry.add_to_hass(hass)
assert entity_entry.device_id
assert (device_entry := device_registry.async_get(entity_entry.device_id))
assert snapshot == device_entry
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert (
hass.states.get("number.product_name_aabbccddeeff_status_light_brightness")
is None
)
async def test_number_loads_entities(
hass: HomeAssistant, mock_config_entry_data, mock_config_entry
) -> None:
"""Test entity does load number when brightness is available."""
api = get_mock_device(product_type="HWE-SKT")
api.state = AsyncMock(return_value=State.from_dict({"brightness": 255}))
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=api,
):
entry = mock_config_entry
entry.data = mock_config_entry_data
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
entity_registry = er.async_get(hass)
state = hass.states.get("number.product_name_aabbccddeeff_status_light_brightness")
assert state
# Test unknown handling
assert state.state == "100"
assert (
state.attributes.get(ATTR_FRIENDLY_NAME)
== "Product Name (aabbccddeeff) Status light brightness"
mock_homewizardenergy.state.return_value.brightness = None
async_fire_time_changed(hass, dt_util.utcnow() + UPDATE_INTERVAL)
await hass.async_block_till_done()
assert (state := hass.states.get(state.entity_id))
assert state.state == STATE_UNKNOWN
# Test service methods
assert len(mock_homewizardenergy.state_set.mock_calls) == 0
await hass.services.async_call(
number.DOMAIN,
SERVICE_SET_VALUE,
{
ATTR_ENTITY_ID: state.entity_id,
ATTR_VALUE: 50,
},
blocking=True,
)
entry = entity_registry.async_get(
"number.product_name_aabbccddeeff_status_light_brightness"
)
assert entry
assert entry.unique_id == "aabbccddeeff_status_light_brightness"
assert not entry.disabled
assert len(mock_homewizardenergy.state_set.mock_calls) == 1
mock_homewizardenergy.state_set.assert_called_with(brightness=127)
async def test_brightness_level_set(
hass: HomeAssistant, mock_config_entry_data, mock_config_entry
) -> None:
"""Test entity turns sets light level."""
api = get_mock_device(product_type="HWE-SKT")
api.state = AsyncMock(return_value=State.from_dict({"brightness": 255}))
def state_set(brightness):
api.state = AsyncMock(return_value=State.from_dict({"brightness": brightness}))
api.state_set = AsyncMock(side_effect=state_set)
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=api,
):
entry = mock_config_entry
entry.data = mock_config_entry_data
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert (
hass.states.get(
"number.product_name_aabbccddeeff_status_light_brightness"
).state
== "100"
)
# Set level halfway
mock_homewizardenergy.state_set.side_effect = RequestError
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
number.DOMAIN,
SERVICE_SET_VALUE,
{
ATTR_ENTITY_ID: (
"number.product_name_aabbccddeeff_status_light_brightness"
),
ATTR_ENTITY_ID: state.entity_id,
ATTR_VALUE: 50,
},
blocking=True,
)
await hass.async_block_till_done()
assert (
hass.states.get(
"number.product_name_aabbccddeeff_status_light_brightness"
).state
== "50"
)
assert len(api.state_set.mock_calls) == 1
# Turn off level
mock_homewizardenergy.state_set.side_effect = DisabledError
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
number.DOMAIN,
SERVICE_SET_VALUE,
{
ATTR_ENTITY_ID: (
"number.product_name_aabbccddeeff_status_light_brightness"
),
ATTR_VALUE: 0,
ATTR_ENTITY_ID: state.entity_id,
ATTR_VALUE: 50,
},
blocking=True,
)
await hass.async_block_till_done()
assert (
hass.states.get(
"number.product_name_aabbccddeeff_status_light_brightness"
).state
== "0"
)
assert len(api.state_set.mock_calls) == 2
async def test_brightness_level_set_catches_requesterror(
hass: HomeAssistant, mock_config_entry_data, mock_config_entry
) -> None:
"""Test entity raises HomeAssistantError when RequestError was raised."""
api = get_mock_device(product_type="HWE-SKT")
api.state = AsyncMock(return_value=State.from_dict({"brightness": 255}))
api.state_set = AsyncMock(side_effect=RequestError())
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=api,
):
entry = mock_config_entry
entry.data = mock_config_entry_data
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
# Set level halfway
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
number.DOMAIN,
SERVICE_SET_VALUE,
{
ATTR_ENTITY_ID: (
"number.product_name_aabbccddeeff_status_light_brightness"
),
ATTR_VALUE: 50,
},
blocking=True,
)
async def test_brightness_level_set_catches_disablederror(
hass: HomeAssistant, mock_config_entry_data, mock_config_entry
) -> None:
"""Test entity raises HomeAssistantError when DisabledError was raised."""
api = get_mock_device(product_type="HWE-SKT")
api.state = AsyncMock(return_value=State.from_dict({"brightness": 255}))
api.state_set = AsyncMock(side_effect=DisabledError())
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=api,
):
entry = mock_config_entry
entry.data = mock_config_entry_data
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
# Set level halfway
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
number.DOMAIN,
SERVICE_SET_VALUE,
{
ATTR_ENTITY_ID: (
"number.product_name_aabbccddeeff_status_light_brightness"
),
ATTR_VALUE: 50,
},
blocking=True,
)
async def test_brightness_level_set_catches_invalid_value(
hass: HomeAssistant, mock_config_entry_data, mock_config_entry
) -> None:
"""Test entity raises ValueError when value was invalid."""
api = get_mock_device(product_type="HWE-SKT")
api.state = AsyncMock(return_value=State.from_dict({"brightness": 255}))
def state_set(brightness):
api.state = AsyncMock(return_value=State.from_dict({"brightness": brightness}))
api.state_set = AsyncMock(side_effect=state_set)
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=api,
):
entry = mock_config_entry
entry.data = mock_config_entry_data
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
with pytest.raises(ValueError):
await hass.services.async_call(
number.DOMAIN,
SERVICE_SET_VALUE,
{
ATTR_ENTITY_ID: (
"number.product_name_aabbccddeeff_status_light_brightness"
),
ATTR_VALUE: -1,
},
blocking=True,
)
with pytest.raises(ValueError):
await hass.services.async_call(
number.DOMAIN,
SERVICE_SET_VALUE,
{
ATTR_ENTITY_ID: (
"number.product_name_aabbccddeeff_status_light_brightness"
),
ATTR_VALUE: 101,
},
blocking=True,
)

1863
tests/components/homewizard/test_sensor.py
View File

File diff suppressed because it is too large Load Diff

583
tests/components/homewizard/test_switch.py
View File

@ -1,545 +1,146 @@
"""Test the update coordinator for HomeWizard."""
from unittest.mock import AsyncMock, patch
"""Test the switch entity for HomeWizard."""
from unittest.mock import MagicMock
from homewizard_energy.errors import DisabledError, RequestError, UnsupportedError
from homewizard_energy.models import State, System
from homewizard_energy import UnsupportedError
from homewizard_energy.errors import DisabledError, RequestError
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components import switch
from homeassistant.components.switch import SwitchDeviceClass
from homeassistant.components.homewizard.const import UPDATE_INTERVAL
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_FRIENDLY_NAME,
ATTR_ICON,
ATTR_ENTITY_ID,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers import device_registry as dr, entity_registry as er
import homeassistant.util.dt as dt_util
from .generator import get_mock_device
from tests.common import async_fire_time_changed
pytestmark = [
pytest.mark.usefixtures("init_integration"),
]
async def test_switch_entity_not_loaded_when_not_available(
hass: HomeAssistant, mock_config_entry_data, mock_config_entry
@pytest.mark.parametrize("device_fixture", ["device-HWE-SKT.json"])
@pytest.mark.parametrize(
("entity_id", "method", "parameter"),
[
("switch.device", "state_set", "power_on"),
("switch.device_switch_lock", "state_set", "switch_lock"),
("switch.device_cloud_connection", "system_set", "cloud_enabled"),
],
)
async def test_switch_entities(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
mock_homewizardenergy: MagicMock,
snapshot: SnapshotAssertion,
entity_id: str,
method: str,
parameter: str,
) -> None:
"""Test entity loads smr version."""
"""Test that switch handles state changes correctly."""
assert (state := hass.states.get(entity_id))
assert snapshot == state
api = get_mock_device()
assert (entity_entry := entity_registry.async_get(entity_id))
assert snapshot == entity_entry
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=api,
):
entry = mock_config_entry
entry.data = mock_config_entry_data
entry.add_to_hass(hass)
assert entity_entry.device_id
assert (device_entry := device_registry.async_get(entity_entry.device_id))
assert snapshot == device_entry
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
mocked_method = getattr(mock_homewizardenergy, method)
state_power_on = hass.states.get("sensor.product_name_aabbccddeeff")
state_switch_lock = hass.states.get("sensor.product_name_aabbccddeeff_switch_lock")
assert state_power_on is None
assert state_switch_lock is None
async def test_switch_loads_entities(
hass: HomeAssistant, mock_config_entry_data, mock_config_entry
) -> None:
"""Test entity loads smr version."""
api = get_mock_device(product_type="HWE-SKT")
api.state = AsyncMock(
return_value=State.from_dict({"power_on": False, "switch_lock": False})
# Turn power_on on
await hass.services.async_call(
switch.DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=api,
):
entry = mock_config_entry
entry.data = mock_config_entry_data
entry.add_to_hass(hass)
assert len(mocked_method.mock_calls) == 1
mocked_method.assert_called_with(**{parameter: True})
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
entity_registry = er.async_get(hass)
state_power_on = hass.states.get("switch.product_name_aabbccddeeff")
entry_power_on = entity_registry.async_get("switch.product_name_aabbccddeeff")
assert state_power_on
assert entry_power_on
assert entry_power_on.unique_id == "aabbccddeeff_power_on"
assert not entry_power_on.disabled
assert state_power_on.state == STATE_OFF
assert (
state_power_on.attributes.get(ATTR_FRIENDLY_NAME)
== "Product Name (aabbccddeeff)"
)
assert state_power_on.attributes.get(ATTR_DEVICE_CLASS) == SwitchDeviceClass.OUTLET
assert ATTR_ICON not in state_power_on.attributes
state_switch_lock = hass.states.get("switch.product_name_aabbccddeeff_switch_lock")
entry_switch_lock = entity_registry.async_get(
"switch.product_name_aabbccddeeff_switch_lock"
# Turn power_on off
await hass.services.async_call(
switch.DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
assert state_switch_lock
assert entry_switch_lock
assert entry_switch_lock.unique_id == "aabbccddeeff_switch_lock"
assert not entry_switch_lock.disabled
assert state_switch_lock.state == STATE_OFF
assert (
state_switch_lock.attributes.get(ATTR_FRIENDLY_NAME)
== "Product Name (aabbccddeeff) Switch lock"
)
assert state_switch_lock.attributes.get(ATTR_ICON) == "mdi:lock-open"
assert ATTR_DEVICE_CLASS not in state_switch_lock.attributes
assert len(mocked_method.mock_calls) == 2
mocked_method.assert_called_with(**{parameter: False})
# Test request error handling
mocked_method.side_effect = RequestError
async def test_switch_power_on_off(
hass: HomeAssistant, mock_config_entry_data, mock_config_entry
) -> None:
"""Test entity turns switch on and off."""
api = get_mock_device(product_type="HWE-SKT")
api.state = AsyncMock(
return_value=State.from_dict({"power_on": False, "switch_lock": False})
)
def state_set(power_on):
api.state = AsyncMock(
return_value=State.from_dict({"power_on": power_on, "switch_lock": False})
)
api.state_set = AsyncMock(side_effect=state_set)
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=api,
):
entry = mock_config_entry
entry.data = mock_config_entry_data
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get("switch.product_name_aabbccddeeff").state == STATE_OFF
# Turn power_on on
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
switch.DOMAIN,
SERVICE_TURN_ON,
{"entity_id": "switch.product_name_aabbccddeeff"},
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
await hass.async_block_till_done()
assert len(api.state_set.mock_calls) == 1
assert hass.states.get("switch.product_name_aabbccddeeff").state == STATE_ON
# Turn power_on off
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
switch.DOMAIN,
SERVICE_TURN_OFF,
{"entity_id": "switch.product_name_aabbccddeeff"},
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
await hass.async_block_till_done()
assert hass.states.get("switch.product_name_aabbccddeeff").state == STATE_OFF
assert len(api.state_set.mock_calls) == 2
# Test disabled error handling
mocked_method.side_effect = DisabledError
async def test_switch_lock_power_on_off(
hass: HomeAssistant, mock_config_entry_data, mock_config_entry
) -> None:
"""Test entity turns switch on and off."""
api = get_mock_device(product_type="HWE-SKT")
api.state = AsyncMock(
return_value=State.from_dict({"power_on": False, "switch_lock": False})
)
def state_set(switch_lock):
api.state = AsyncMock(
return_value=State.from_dict({"power_on": True, "switch_lock": switch_lock})
)
api.state_set = AsyncMock(side_effect=state_set)
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=api,
):
entry = mock_config_entry
entry.data = mock_config_entry_data
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert (
hass.states.get("switch.product_name_aabbccddeeff_switch_lock").state
== STATE_OFF
)
# Turn power_on on
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
switch.DOMAIN,
SERVICE_TURN_ON,
{"entity_id": "switch.product_name_aabbccddeeff_switch_lock"},
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
await hass.async_block_till_done()
assert len(api.state_set.mock_calls) == 1
assert (
hass.states.get("switch.product_name_aabbccddeeff_switch_lock").state
== STATE_ON
)
# Turn power_on off
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
switch.DOMAIN,
SERVICE_TURN_OFF,
{"entity_id": "switch.product_name_aabbccddeeff_switch_lock"},
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
await hass.async_block_till_done()
assert (
hass.states.get("switch.product_name_aabbccddeeff_switch_lock").state
== STATE_OFF
)
assert len(api.state_set.mock_calls) == 2
async def test_switch_lock_sets_power_on_unavailable(
hass: HomeAssistant, mock_config_entry_data, mock_config_entry
@pytest.mark.parametrize("device_fixture", ["device-HWE-SKT.json"])
@pytest.mark.parametrize("exception", [RequestError, DisabledError, UnsupportedError])
@pytest.mark.parametrize(
("entity_id", "method"),
[
("switch.device", "state"),
("switch.device_switch_lock", "state"),
("switch.device_cloud_connection", "system"),
],
)
async def test_switch_unreachable(
hass: HomeAssistant,
mock_homewizardenergy: MagicMock,
exception: Exception,
entity_id: str,
method: str,
) -> None:
"""Test entity turns switch on and off."""
"""Test that unreachable devices are marked as unavailable."""
mocked_method = getattr(mock_homewizardenergy, method)
mocked_method.side_effect = exception
async_fire_time_changed(hass, dt_util.utcnow() + UPDATE_INTERVAL)
await hass.async_block_till_done()
api = get_mock_device(product_type="HWE-SKT")
api.state = AsyncMock(
return_value=State.from_dict({"power_on": True, "switch_lock": False})
)
def state_set(switch_lock):
api.state = AsyncMock(
return_value=State.from_dict({"power_on": True, "switch_lock": switch_lock})
)
api.state_set = AsyncMock(side_effect=state_set)
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=api,
):
entry = mock_config_entry
entry.data = mock_config_entry_data
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get("switch.product_name_aabbccddeeff").state == STATE_ON
assert (
hass.states.get("switch.product_name_aabbccddeeff_switch_lock").state
== STATE_OFF
)
# Turn power_on on
await hass.services.async_call(
switch.DOMAIN,
SERVICE_TURN_ON,
{"entity_id": "switch.product_name_aabbccddeeff_switch_lock"},
blocking=True,
)
await hass.async_block_till_done()
assert len(api.state_set.mock_calls) == 1
assert (
hass.states.get("switch.product_name_aabbccddeeff").state
== STATE_UNAVAILABLE
)
assert (
hass.states.get("switch.product_name_aabbccddeeff_switch_lock").state
== STATE_ON
)
# Turn power_on off
await hass.services.async_call(
switch.DOMAIN,
SERVICE_TURN_OFF,
{"entity_id": "switch.product_name_aabbccddeeff_switch_lock"},
blocking=True,
)
await hass.async_block_till_done()
assert hass.states.get("switch.product_name_aabbccddeeff").state == STATE_ON
assert (
hass.states.get("switch.product_name_aabbccddeeff_switch_lock").state
== STATE_OFF
)
assert len(api.state_set.mock_calls) == 2
async def test_cloud_connection_on_off(
hass: HomeAssistant, mock_config_entry_data, mock_config_entry
) -> None:
"""Test entity turns switch on and off."""
api = get_mock_device(product_type="HWE-SKT", firmware_version="3.02")
api.system = AsyncMock(return_value=System.from_dict({"cloud_enabled": False}))
def system_set(cloud_enabled):
api.system = AsyncMock(
return_value=System.from_dict({"cloud_enabled": cloud_enabled})
)
api.system_set = AsyncMock(side_effect=system_set)
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=api,
):
entry = mock_config_entry
entry.data = mock_config_entry_data
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert (
hass.states.get("switch.product_name_aabbccddeeff_cloud_connection").state
== STATE_OFF
)
# Enable cloud
await hass.services.async_call(
switch.DOMAIN,
SERVICE_TURN_ON,
{"entity_id": "switch.product_name_aabbccddeeff_cloud_connection"},
blocking=True,
)
await hass.async_block_till_done()
assert len(api.system_set.mock_calls) == 1
assert (
hass.states.get("switch.product_name_aabbccddeeff_cloud_connection").state
== STATE_ON
)
# Disable cloud
await hass.services.async_call(
switch.DOMAIN,
SERVICE_TURN_OFF,
{"entity_id": "switch.product_name_aabbccddeeff_cloud_connection"},
blocking=True,
)
await hass.async_block_till_done()
assert (
hass.states.get("switch.product_name_aabbccddeeff_cloud_connection").state
== STATE_OFF
)
assert len(api.system_set.mock_calls) == 2
async def test_switch_handles_requesterror(
hass: HomeAssistant, mock_config_entry_data, mock_config_entry
) -> None:
"""Test entity raises HomeAssistantError when RequestError was raised."""
api = get_mock_device(product_type="HWE-SKT", firmware_version="3.02")
api.state = AsyncMock(
return_value=State.from_dict({"power_on": False, "switch_lock": False})
)
api.system = AsyncMock(return_value=System.from_dict({"cloud_enabled": False}))
api.state_set = AsyncMock(side_effect=RequestError())
api.system_set = AsyncMock(side_effect=RequestError())
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=api,
):
entry = mock_config_entry
entry.data = mock_config_entry_data
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
# Power on toggle
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
switch.DOMAIN,
SERVICE_TURN_ON,
{"entity_id": "switch.product_name_aabbccddeeff"},
blocking=True,
)
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
switch.DOMAIN,
SERVICE_TURN_OFF,
{"entity_id": "switch.product_name_aabbccddeeff_cloud_connection"},
blocking=True,
)
# Switch Lock toggle
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
switch.DOMAIN,
SERVICE_TURN_ON,
{"entity_id": "switch.product_name_aabbccddeeff_switch_lock"},
blocking=True,
)
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
switch.DOMAIN,
SERVICE_TURN_OFF,
{"entity_id": "switch.product_name_aabbccddeeff_switch_lock"},
blocking=True,
)
# Disable Cloud toggle
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
switch.DOMAIN,
SERVICE_TURN_ON,
{"entity_id": "switch.product_name_aabbccddeeff_cloud_connection"},
blocking=True,
)
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
switch.DOMAIN,
SERVICE_TURN_OFF,
{"entity_id": "switch.product_name_aabbccddeeff_cloud_connection"},
blocking=True,
)
async def test_switch_handles_disablederror(
hass: HomeAssistant, mock_config_entry_data, mock_config_entry
) -> None:
"""Test entity raises HomeAssistantError when Disabled was raised."""
api = get_mock_device(product_type="HWE-SKT", firmware_version="3.02")
api.state = AsyncMock(
return_value=State.from_dict({"power_on": False, "switch_lock": False})
)
api.system = AsyncMock(return_value=System.from_dict({"cloud_enabled": False}))
api.state_set = AsyncMock(side_effect=DisabledError())
api.system_set = AsyncMock(side_effect=DisabledError())
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=api,
):
entry = mock_config_entry
entry.data = mock_config_entry_data
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
# Power on toggle
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
switch.DOMAIN,
SERVICE_TURN_ON,
{"entity_id": "switch.product_name_aabbccddeeff"},
blocking=True,
)
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
switch.DOMAIN,
SERVICE_TURN_OFF,
{"entity_id": "switch.product_name_aabbccddeeff_cloud_connection"},
blocking=True,
)
# Switch Lock toggle
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
switch.DOMAIN,
SERVICE_TURN_ON,
{"entity_id": "switch.product_name_aabbccddeeff_switch_lock"},
blocking=True,
)
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
switch.DOMAIN,
SERVICE_TURN_OFF,
{"entity_id": "switch.product_name_aabbccddeeff_switch_lock"},
blocking=True,
)
# Disable Cloud toggle
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
switch.DOMAIN,
SERVICE_TURN_ON,
{"entity_id": "switch.product_name_aabbccddeeff_cloud_connection"},
blocking=True,
)
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
switch.DOMAIN,
SERVICE_TURN_OFF,
{"entity_id": "switch.product_name_aabbccddeeff_cloud_connection"},
blocking=True,
)
async def test_switch_handles_unsupportedrrror(
hass: HomeAssistant, mock_config_entry_data, mock_config_entry
) -> None:
"""Test entity raises HomeAssistantError when Disabled was raised."""
api = get_mock_device(product_type="HWE-SKT", firmware_version="3.02")
api.state = AsyncMock(side_effect=UnsupportedError())
api.system = AsyncMock(side_effect=UnsupportedError())
with patch(
"homeassistant.components.homewizard.coordinator.HomeWizardEnergy",
return_value=api,
):
entry = mock_config_entry
entry.data = mock_config_entry_data
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert (
hass.states.get("switch.product_name_aabbccddeeff_cloud_connection").state
== STATE_UNAVAILABLE
)
assert (
hass.states.get("switch.product_name_aabbccddeeff_switch_lock").state
== STATE_UNAVAILABLE
)
assert (
hass.states.get("switch.product_name_aabbccddeeff").state
== STATE_UNAVAILABLE
)
assert (state := hass.states.get(entity_id))
assert state.state == STATE_UNAVAILABLE