Skip to content
Snippets Groups Projects
Unverified Commit b15e08ca authored by Nerdix's avatar Nerdix Committed by GitHub
Browse files

Add sleep switch for all Foscam cameras if more than 1 camera are configured (#126064)

parent 3fb98090
No related branches found
No related tags found
No related merge requests found
......@@ -11,7 +11,7 @@ from homeassistant.const import (
Platform,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_registry import async_migrate_entries
from homeassistant.helpers.entity_registry import RegistryEntry, async_migrate_entries
from .config_flow import DEFAULT_RTSP_PORT
from .const import CONF_RTSP_PORT, DOMAIN, LOGGER, SERVICE_PTZ, SERVICE_PTZ_PRESET
......@@ -36,6 +36,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
# Migrate to correct unique IDs for switches
await async_migrate_entities(hass, entry)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
......@@ -92,3 +95,24 @@ async def async_migrate_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
LOGGER.debug("Migration to version %s successful", entry.version)
return True
async def async_migrate_entities(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Migrate old entry."""
@callback
def _update_unique_id(
entity_entry: RegistryEntry,
) -> dict[str, str] | None:
"""Update unique ID of entity entry."""
if (
entity_entry.domain == Platform.SWITCH
and entity_entry.unique_id == "sleep_switch"
):
entity_new_unique_id = f"{entity_entry.config_entry_id}_sleep_switch"
return {"new_unique_id": entity_new_unique_id}
return None
# Migrate entities
await async_migrate_entries(hass, entry.entry_id, _update_unique_id)
......@@ -41,7 +41,7 @@ class FoscamSleepSwitch(FoscamEntity, SwitchEntity):
"""Initialize a Foscam Sleep Switch."""
super().__init__(coordinator, config_entry.entry_id)
self._attr_unique_id = "sleep_switch"
self._attr_unique_id = f"{config_entry.entry_id}_sleep_switch"
self._attr_translation_key = "sleep_switch"
self._attr_has_entity_name = True
......
"""Common stuff for Foscam tests."""
from libpyfoscam.foscam import (
ERROR_FOSCAM_AUTH,
ERROR_FOSCAM_CMD,
ERROR_FOSCAM_UNAVAILABLE,
ERROR_FOSCAM_UNKNOWN,
)
from homeassistant.components.foscam import config_flow
from .const import (
CAMERA_MAC,
CAMERA_NAME,
INVALID_RESPONSE_CONFIG,
OPERATOR_CONFIG,
VALID_CONFIG,
)
def setup_mock_foscam_camera(mock_foscam_camera):
"""Mock FoscamCamera simulating behaviour using a base valid config."""
def configure_mock_on_init(host, port, user, passwd, verbose=False):
product_all_info_rc = 0
dev_info_rc = 0
dev_info_data = {}
if (
host != VALID_CONFIG[config_flow.CONF_HOST]
or port != VALID_CONFIG[config_flow.CONF_PORT]
):
product_all_info_rc = dev_info_rc = ERROR_FOSCAM_UNAVAILABLE
elif (
user
not in [
VALID_CONFIG[config_flow.CONF_USERNAME],
OPERATOR_CONFIG[config_flow.CONF_USERNAME],
INVALID_RESPONSE_CONFIG[config_flow.CONF_USERNAME],
]
or passwd != VALID_CONFIG[config_flow.CONF_PASSWORD]
):
product_all_info_rc = dev_info_rc = ERROR_FOSCAM_AUTH
elif user == INVALID_RESPONSE_CONFIG[config_flow.CONF_USERNAME]:
product_all_info_rc = dev_info_rc = ERROR_FOSCAM_UNKNOWN
elif user == OPERATOR_CONFIG[config_flow.CONF_USERNAME]:
dev_info_rc = ERROR_FOSCAM_CMD
else:
dev_info_data["devName"] = CAMERA_NAME
dev_info_data["mac"] = CAMERA_MAC
dev_info_data["productName"] = "Foscam Product"
dev_info_data["firmwareVer"] = "1.2.3"
dev_info_data["hardwareVer"] = "4.5.6"
mock_foscam_camera.get_product_all_info.return_value = (product_all_info_rc, {})
mock_foscam_camera.get_dev_info.return_value = (dev_info_rc, dev_info_data)
mock_foscam_camera.get_port_info.return_value = (dev_info_rc, {})
mock_foscam_camera.is_asleep.return_value = (0, True)
return mock_foscam_camera
mock_foscam_camera.side_effect = configure_mock_on_init
"""Constants for Foscam tests."""
from homeassistant.components.foscam import config_flow
VALID_CONFIG = {
config_flow.CONF_HOST: "10.0.0.2",
config_flow.CONF_PORT: 88,
config_flow.CONF_USERNAME: "admin",
config_flow.CONF_PASSWORD: "1234",
config_flow.CONF_STREAM: "Main",
config_flow.CONF_RTSP_PORT: 554,
}
OPERATOR_CONFIG = {
config_flow.CONF_USERNAME: "operator",
}
INVALID_RESPONSE_CONFIG = {
config_flow.CONF_USERNAME: "interr",
}
CAMERA_NAME = "Mocked Foscam Camera"
CAMERA_MAC = "C0:C1:D0:F4:B4:D4"
ENTRY_ID = "123ABC"
......@@ -2,79 +2,15 @@
from unittest.mock import patch
from libpyfoscam.foscam import (
ERROR_FOSCAM_AUTH,
ERROR_FOSCAM_CMD,
ERROR_FOSCAM_UNAVAILABLE,
ERROR_FOSCAM_UNKNOWN,
)
from homeassistant import config_entries
from homeassistant.components.foscam import config_flow
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
from .conftest import setup_mock_foscam_camera
from .const import CAMERA_NAME, INVALID_RESPONSE_CONFIG, VALID_CONFIG
VALID_CONFIG = {
config_flow.CONF_HOST: "10.0.0.2",
config_flow.CONF_PORT: 88,
config_flow.CONF_USERNAME: "admin",
config_flow.CONF_PASSWORD: "1234",
config_flow.CONF_STREAM: "Main",
config_flow.CONF_RTSP_PORT: 554,
}
OPERATOR_CONFIG = {
config_flow.CONF_USERNAME: "operator",
}
INVALID_RESPONSE_CONFIG = {
config_flow.CONF_USERNAME: "interr",
}
CAMERA_NAME = "Mocked Foscam Camera"
CAMERA_MAC = "C0:C1:D0:F4:B4:D4"
def setup_mock_foscam_camera(mock_foscam_camera):
"""Mock FoscamCamera simulating behaviour using a base valid config."""
def configure_mock_on_init(host, port, user, passwd, verbose=False):
product_all_info_rc = 0
dev_info_rc = 0
dev_info_data = {}
if (
host != VALID_CONFIG[config_flow.CONF_HOST]
or port != VALID_CONFIG[config_flow.CONF_PORT]
):
product_all_info_rc = dev_info_rc = ERROR_FOSCAM_UNAVAILABLE
elif (
user
not in [
VALID_CONFIG[config_flow.CONF_USERNAME],
OPERATOR_CONFIG[config_flow.CONF_USERNAME],
INVALID_RESPONSE_CONFIG[config_flow.CONF_USERNAME],
]
or passwd != VALID_CONFIG[config_flow.CONF_PASSWORD]
):
product_all_info_rc = dev_info_rc = ERROR_FOSCAM_AUTH
elif user == INVALID_RESPONSE_CONFIG[config_flow.CONF_USERNAME]:
product_all_info_rc = dev_info_rc = ERROR_FOSCAM_UNKNOWN
elif user == OPERATOR_CONFIG[config_flow.CONF_USERNAME]:
dev_info_rc = ERROR_FOSCAM_CMD
else:
dev_info_data["devName"] = CAMERA_NAME
dev_info_data["mac"] = CAMERA_MAC
mock_foscam_camera.get_product_all_info.return_value = (product_all_info_rc, {})
mock_foscam_camera.get_dev_info.return_value = (dev_info_rc, dev_info_data)
return mock_foscam_camera
mock_foscam_camera.side_effect = configure_mock_on_init
from tests.common import MockConfigEntry
async def test_user_valid(hass: HomeAssistant) -> None:
......
"""Test the Foscam component."""
from unittest.mock import patch
from homeassistant.components.foscam import DOMAIN, config_flow
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from .conftest import setup_mock_foscam_camera
from .const import ENTRY_ID, VALID_CONFIG
from tests.common import MockConfigEntry
async def test_unique_id_new_entry(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
) -> None:
"""Test unique ID for a newly added device is correct."""
entry = MockConfigEntry(
domain=config_flow.DOMAIN, data=VALID_CONFIG, entry_id=ENTRY_ID
)
entry.add_to_hass(hass)
with (
# Mock a valid camera instance"
patch("homeassistant.components.foscam.FoscamCamera") as mock_foscam_camera,
):
setup_mock_foscam_camera(mock_foscam_camera)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
# Test that unique_id remains the same.
entity_id = entity_registry.async_get_entity_id(
SWITCH_DOMAIN, DOMAIN, f"{ENTRY_ID}_sleep_switch"
)
entity_new = entity_registry.async_get(entity_id)
assert entity_new.unique_id == f"{ENTRY_ID}_sleep_switch"
async def test_switch_unique_id_migration_ok(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that the unique ID for a sleep switch is migrated to the new format."""
entry = MockConfigEntry(
domain=config_flow.DOMAIN, data=VALID_CONFIG, entry_id=ENTRY_ID, version=1
)
entry.add_to_hass(hass)
entity_before = entity_registry.async_get_or_create(
SWITCH_DOMAIN, DOMAIN, "sleep_switch", config_entry=entry
)
assert entity_before.unique_id == "sleep_switch"
# Update config entry with version 2
entry = MockConfigEntry(
domain=config_flow.DOMAIN, data=VALID_CONFIG, entry_id=ENTRY_ID, version=2
)
entry.add_to_hass(hass)
with (
# Mock a valid camera instance"
patch("homeassistant.components.foscam.FoscamCamera") as mock_foscam_camera,
):
setup_mock_foscam_camera(mock_foscam_camera)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
entity_id_new = entity_registry.async_get_entity_id(
SWITCH_DOMAIN, DOMAIN, f"{ENTRY_ID}_sleep_switch"
)
assert hass.states.get(entity_id_new)
entity_after = entity_registry.async_get(entity_id_new)
assert entity_after.previous_unique_id == "sleep_switch"
assert entity_after.unique_id == f"{ENTRY_ID}_sleep_switch"
async def test_unique_id_migration_not_needed(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that the unique ID for a sleep switch is not executed if already in right format."""
entry = MockConfigEntry(
domain=config_flow.DOMAIN, data=VALID_CONFIG, entry_id=ENTRY_ID
)
entry.add_to_hass(hass)
entity_registry.async_get_or_create(
SWITCH_DOMAIN, DOMAIN, f"{ENTRY_ID}_sleep_switch", config_entry=entry
)
entity_id = entity_registry.async_get_entity_id(
SWITCH_DOMAIN, DOMAIN, f"{ENTRY_ID}_sleep_switch"
)
entity_before = entity_registry.async_get(entity_id)
assert entity_before.unique_id == f"{ENTRY_ID}_sleep_switch"
with (
# Mock a valid camera instance"
patch("homeassistant.components.foscam.FoscamCamera") as mock_foscam_camera,
patch(
"homeassistant.components.foscam.async_migrate_entry",
return_value=True,
),
):
setup_mock_foscam_camera(mock_foscam_camera)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
# Test that unique_id remains the same.
assert hass.states.get(entity_id)
entity_after = entity_registry.async_get(entity_id)
assert entity_after.unique_id == entity_before.unique_id
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment