mirror of https://github.com/home-assistant/core
193 lines
7.3 KiB
Python
193 lines
7.3 KiB
Python
"""Tests for the HomeKit IID manager."""
|
|
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
from homeassistant.components.homekit.const import DOMAIN
|
|
from homeassistant.components.homekit.iidmanager import (
|
|
AccessoryIIDStorage,
|
|
get_iid_storage_filename_for_entry_id,
|
|
)
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.util.json import json_loads
|
|
from homeassistant.util.uuid import random_uuid_hex
|
|
|
|
from tests.common import MockConfigEntry, load_fixture
|
|
|
|
|
|
async def test_iid_generation_and_restore(
|
|
hass: HomeAssistant, iid_storage, hass_storage: dict[str, Any]
|
|
) -> None:
|
|
"""Test generating iids and restoring them from storage."""
|
|
entry = MockConfigEntry(domain=DOMAIN)
|
|
|
|
iid_storage = AccessoryIIDStorage(hass, entry.entry_id)
|
|
await iid_storage.async_initialize()
|
|
|
|
random_service_uuid = UUID(random_uuid_hex())
|
|
random_characteristic_uuid = UUID(random_uuid_hex())
|
|
|
|
iid1 = iid_storage.get_or_allocate_iid(
|
|
1, random_service_uuid, None, random_characteristic_uuid, None
|
|
)
|
|
iid2 = iid_storage.get_or_allocate_iid(
|
|
1, random_service_uuid, None, random_characteristic_uuid, None
|
|
)
|
|
assert iid1 == iid2
|
|
|
|
service_only_iid1 = iid_storage.get_or_allocate_iid(
|
|
1, random_service_uuid, None, None, None
|
|
)
|
|
service_only_iid2 = iid_storage.get_or_allocate_iid(
|
|
1, random_service_uuid, None, None, None
|
|
)
|
|
assert service_only_iid1 == service_only_iid2
|
|
assert service_only_iid1 != iid1
|
|
|
|
service_only_iid_with_unique_id1 = iid_storage.get_or_allocate_iid(
|
|
1, random_service_uuid, "any", None, None
|
|
)
|
|
service_only_iid_with_unique_id2 = iid_storage.get_or_allocate_iid(
|
|
1, random_service_uuid, "any", None, None
|
|
)
|
|
assert service_only_iid_with_unique_id1 == service_only_iid_with_unique_id2
|
|
assert service_only_iid_with_unique_id1 != service_only_iid1
|
|
|
|
unique_char_iid1 = iid_storage.get_or_allocate_iid(
|
|
1, random_service_uuid, None, random_characteristic_uuid, "any"
|
|
)
|
|
unique_char_iid2 = iid_storage.get_or_allocate_iid(
|
|
1, random_service_uuid, None, random_characteristic_uuid, "any"
|
|
)
|
|
assert unique_char_iid1 == unique_char_iid2
|
|
assert unique_char_iid1 != iid1
|
|
|
|
unique_service_unique_char_iid1 = iid_storage.get_or_allocate_iid(
|
|
1, random_service_uuid, "any", random_characteristic_uuid, "any"
|
|
)
|
|
unique_service_unique_char_iid2 = iid_storage.get_or_allocate_iid(
|
|
1, random_service_uuid, "any", random_characteristic_uuid, "any"
|
|
)
|
|
assert unique_service_unique_char_iid1 == unique_service_unique_char_iid2
|
|
assert unique_service_unique_char_iid1 != iid1
|
|
|
|
unique_service_unique_char_new_aid_iid1 = iid_storage.get_or_allocate_iid(
|
|
2, random_service_uuid, "any", random_characteristic_uuid, "any"
|
|
)
|
|
unique_service_unique_char_new_aid_iid2 = iid_storage.get_or_allocate_iid(
|
|
2, random_service_uuid, "any", random_characteristic_uuid, "any"
|
|
)
|
|
assert (
|
|
unique_service_unique_char_new_aid_iid1
|
|
== unique_service_unique_char_new_aid_iid2
|
|
)
|
|
await iid_storage.async_save()
|
|
|
|
iid_storage2 = AccessoryIIDStorage(hass, entry.entry_id)
|
|
await iid_storage2.async_initialize()
|
|
iid3 = iid_storage2.get_or_allocate_iid(
|
|
1, random_service_uuid, None, random_characteristic_uuid, None
|
|
)
|
|
assert iid3 == iid1
|
|
|
|
|
|
async def test_iid_storage_filename(
|
|
hass: HomeAssistant, iid_storage, hass_storage: dict[str, Any]
|
|
) -> None:
|
|
"""Test iid storage uses the expected filename."""
|
|
entry = MockConfigEntry(domain=DOMAIN)
|
|
|
|
iid_storage = AccessoryIIDStorage(hass, entry.entry_id)
|
|
await iid_storage.async_initialize()
|
|
assert iid_storage.store.path.endswith(
|
|
get_iid_storage_filename_for_entry_id(entry.entry_id)
|
|
)
|
|
|
|
|
|
async def test_iid_migration_to_v2(
|
|
hass: HomeAssistant, iid_storage, hass_storage: dict[str, Any]
|
|
) -> None:
|
|
"""Test iid storage migration."""
|
|
v1_iids = json_loads(load_fixture("iids_v1", DOMAIN))
|
|
v2_iids = json_loads(load_fixture("iids_v2", DOMAIN))
|
|
hass_storage["homekit.v1.iids"] = v1_iids
|
|
hass_storage["homekit.v2.iids"] = v2_iids
|
|
|
|
iid_storage_v2 = AccessoryIIDStorage(hass, "v1")
|
|
await iid_storage_v2.async_initialize()
|
|
|
|
iid_storage_v1 = AccessoryIIDStorage(hass, "v2")
|
|
await iid_storage_v1.async_initialize()
|
|
|
|
assert iid_storage_v1.allocations == iid_storage_v2.allocations
|
|
assert iid_storage_v1.allocated_iids == iid_storage_v2.allocated_iids
|
|
|
|
assert len(iid_storage_v2.allocations) == 12
|
|
|
|
for allocations in iid_storage_v2.allocations.values():
|
|
assert allocations["3E___"] == 1
|
|
|
|
|
|
async def test_iid_migration_to_v2_with_underscore(
|
|
hass: HomeAssistant, iid_storage, hass_storage: dict[str, Any]
|
|
) -> None:
|
|
"""Test iid storage migration with underscore."""
|
|
v1_iids = json_loads(load_fixture("iids_v1_with_underscore", DOMAIN))
|
|
v2_iids = json_loads(load_fixture("iids_v2_with_underscore", DOMAIN))
|
|
hass_storage["homekit.v1_with_underscore.iids"] = v1_iids
|
|
hass_storage["homekit.v2_with_underscore.iids"] = v2_iids
|
|
|
|
iid_storage_v2 = AccessoryIIDStorage(hass, "v1_with_underscore")
|
|
await iid_storage_v2.async_initialize()
|
|
|
|
iid_storage_v1 = AccessoryIIDStorage(hass, "v2_with_underscore")
|
|
await iid_storage_v1.async_initialize()
|
|
|
|
assert iid_storage_v1.allocations == iid_storage_v2.allocations
|
|
assert iid_storage_v1.allocated_iids == iid_storage_v2.allocated_iids
|
|
|
|
assert len(iid_storage_v2.allocations) == 2
|
|
|
|
for allocations in iid_storage_v2.allocations.values():
|
|
assert allocations["3E___"] == 1
|
|
|
|
|
|
async def test_iid_generation_and_restore_v2(
|
|
hass: HomeAssistant, iid_storage, hass_storage: dict[str, Any]
|
|
) -> None:
|
|
"""Test generating iids and restoring them from storage."""
|
|
entry = MockConfigEntry(domain=DOMAIN)
|
|
|
|
iid_storage = AccessoryIIDStorage(hass, entry.entry_id)
|
|
await iid_storage.async_initialize()
|
|
not_accessory_info_service_iid = iid_storage.get_or_allocate_iid(
|
|
1, "000000AA-0000-1000-8000-0026BB765291", None, None, None
|
|
)
|
|
assert not_accessory_info_service_iid == 2
|
|
assert iid_storage.allocated_iids == {"1": [1, 2]}
|
|
not_accessory_info_service_iid_2 = iid_storage.get_or_allocate_iid(
|
|
1, "000000BB-0000-1000-8000-0026BB765291", None, None, None
|
|
)
|
|
assert not_accessory_info_service_iid_2 == 3
|
|
assert iid_storage.allocated_iids == {"1": [1, 2, 3]}
|
|
not_accessory_info_service_iid_2 = iid_storage.get_or_allocate_iid(
|
|
1, "000000BB-0000-1000-8000-0026BB765291", None, None, None
|
|
)
|
|
assert not_accessory_info_service_iid_2 == 3
|
|
assert iid_storage.allocated_iids == {"1": [1, 2, 3]}
|
|
accessory_info_service_iid = iid_storage.get_or_allocate_iid(
|
|
1, "0000003E-0000-1000-8000-0026BB765291", None, None, None
|
|
)
|
|
assert accessory_info_service_iid == 1
|
|
assert iid_storage.allocated_iids == {"1": [1, 2, 3]}
|
|
accessory_info_service_iid = iid_storage.get_or_allocate_iid(
|
|
1, "0000003E-0000-1000-8000-0026BB765291", None, None, None
|
|
)
|
|
assert accessory_info_service_iid == 1
|
|
assert iid_storage.allocated_iids == {"1": [1, 2, 3]}
|
|
accessory_info_service_iid = iid_storage.get_or_allocate_iid(
|
|
2, "0000003E-0000-1000-8000-0026BB765291", None, None, None
|
|
)
|
|
assert accessory_info_service_iid == 1
|
|
assert iid_storage.allocated_iids == {"1": [1, 2, 3], "2": [1]}
|