"""The lock tests for the august platform.""" import datetime from unittest.mock import Mock from aiohttp import ClientResponseError from freezegun.api import FrozenDateTimeFactory import pytest from syrupy import SnapshotAssertion from yalexs.manager.activity import INITIAL_LOCK_RESYNC_TIME from yalexs.pubnub_async import AugustPubNub from homeassistant.components.lock import ( DOMAIN as LOCK_DOMAIN, STATE_JAMMED, STATE_LOCKING, STATE_UNLOCKING, ) from homeassistant.const import ( ATTR_ENTITY_ID, SERVICE_LOCK, SERVICE_OPEN, SERVICE_UNLOCK, STATE_LOCKED, STATE_UNAVAILABLE, STATE_UNKNOWN, STATE_UNLOCKED, ) from homeassistant.core import HomeAssistant from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers import device_registry as dr, entity_registry as er import homeassistant.util.dt as dt_util from .mocks import ( _create_august_with_devices, _mock_activities_from_fixture, _mock_doorsense_enabled_august_lock_detail, _mock_lock_from_fixture, _mock_lock_with_unlatch, _mock_operative_august_lock_detail, ) from tests.common import async_fire_time_changed async def test_lock_device_registry( hass: HomeAssistant, device_registry: dr.DeviceRegistry, snapshot: SnapshotAssertion ) -> None: """Test creation of a lock with doorsense and bridge ands up in the registry.""" lock_one = await _mock_doorsense_enabled_august_lock_detail(hass) await _create_august_with_devices(hass, [lock_one]) reg_device = device_registry.async_get_device( identifiers={("august", "online_with_doorsense")} ) assert reg_device == snapshot async def test_lock_changed_by(hass: HomeAssistant) -> None: """Test creation of a lock with doorsense and bridge.""" lock_one = await _mock_doorsense_enabled_august_lock_detail(hass) activities = await _mock_activities_from_fixture(hass, "get_activity.lock.json") await _create_august_with_devices(hass, [lock_one], activities=activities) lock_state = hass.states.get("lock.online_with_doorsense_name") assert lock_state.state == STATE_LOCKED assert lock_state.attributes["changed_by"] == "Your favorite elven princess" async def test_state_locking(hass: HomeAssistant) -> None: """Test creation of a lock with doorsense and bridge that is locking.""" lock_one = await _mock_doorsense_enabled_august_lock_detail(hass) activities = await _mock_activities_from_fixture(hass, "get_activity.locking.json") await _create_august_with_devices(hass, [lock_one], activities=activities) assert hass.states.get("lock.online_with_doorsense_name").state == STATE_LOCKING async def test_state_unlocking(hass: HomeAssistant) -> None: """Test creation of a lock with doorsense and bridge that is unlocking.""" lock_one = await _mock_doorsense_enabled_august_lock_detail(hass) activities = await _mock_activities_from_fixture( hass, "get_activity.unlocking.json" ) await _create_august_with_devices(hass, [lock_one], activities=activities) assert hass.states.get("lock.online_with_doorsense_name").state == STATE_UNLOCKING async def test_state_jammed(hass: HomeAssistant) -> None: """Test creation of a lock with doorsense and bridge that is jammed.""" lock_one = await _mock_doorsense_enabled_august_lock_detail(hass) activities = await _mock_activities_from_fixture(hass, "get_activity.jammed.json") await _create_august_with_devices(hass, [lock_one], activities=activities) assert hass.states.get("lock.online_with_doorsense_name").state == STATE_JAMMED async def test_one_lock_operation( hass: HomeAssistant, entity_registry: er.EntityRegistry ) -> None: """Test creation of a lock with doorsense and bridge.""" lock_one = await _mock_doorsense_enabled_august_lock_detail(hass) await _create_august_with_devices(hass, [lock_one]) states = hass.states lock_state = states.get("lock.online_with_doorsense_name") assert lock_state.state == STATE_LOCKED assert lock_state.attributes["battery_level"] == 92 assert lock_state.attributes["friendly_name"] == "online_with_doorsense Name" data = {ATTR_ENTITY_ID: "lock.online_with_doorsense_name"} await hass.services.async_call(LOCK_DOMAIN, SERVICE_UNLOCK, data, blocking=True) lock_state = states.get("lock.online_with_doorsense_name") assert lock_state.state == STATE_UNLOCKED assert lock_state.attributes["battery_level"] == 92 assert lock_state.attributes["friendly_name"] == "online_with_doorsense Name" await hass.services.async_call(LOCK_DOMAIN, SERVICE_LOCK, data, blocking=True) assert states.get("lock.online_with_doorsense_name").state == STATE_LOCKED # No activity means it will be unavailable until the activity feed has data lock_operator_sensor = entity_registry.async_get( "sensor.online_with_doorsense_name_operator" ) assert lock_operator_sensor assert ( states.get("sensor.online_with_doorsense_name_operator").state == STATE_UNKNOWN ) async def test_open_lock_operation(hass: HomeAssistant) -> None: """Test open lock operation using the open service.""" lock_with_unlatch = await _mock_lock_with_unlatch(hass) await _create_august_with_devices(hass, [lock_with_unlatch]) lock_online_with_unlatch_name = hass.states.get("lock.online_with_unlatch_name") assert lock_online_with_unlatch_name.state == STATE_LOCKED data = {ATTR_ENTITY_ID: "lock.online_with_unlatch_name"} await hass.services.async_call(LOCK_DOMAIN, SERVICE_OPEN, data, blocking=True) lock_online_with_unlatch_name = hass.states.get("lock.online_with_unlatch_name") assert lock_online_with_unlatch_name.state == STATE_UNLOCKED async def test_open_lock_operation_pubnub_connected( hass: HomeAssistant, entity_registry: er.EntityRegistry, freezer: FrozenDateTimeFactory, ) -> None: """Test open lock operation using the open service when pubnub is connected.""" lock_with_unlatch = await _mock_lock_with_unlatch(hass) assert lock_with_unlatch.pubsub_channel == "pubsub" pubnub = AugustPubNub() await _create_august_with_devices(hass, [lock_with_unlatch], pubnub=pubnub) pubnub.connected = True assert hass.states.get("lock.online_with_unlatch_name").state == STATE_LOCKED data = {ATTR_ENTITY_ID: "lock.online_with_unlatch_name"} await hass.services.async_call(LOCK_DOMAIN, SERVICE_OPEN, data, blocking=True) pubnub.message( pubnub, Mock( channel=lock_with_unlatch.pubsub_channel, timetoken=(dt_util.utcnow().timestamp() + 2) * 10000000, message={ "status": "kAugLockState_Unlocked", }, ), ) await hass.async_block_till_done() await hass.async_block_till_done() assert hass.states.get("lock.online_with_unlatch_name").state == STATE_UNLOCKED await hass.async_block_till_done() async def test_one_lock_operation_pubnub_connected( hass: HomeAssistant, entity_registry: er.EntityRegistry, freezer: FrozenDateTimeFactory, ) -> None: """Test lock and unlock operations are async when pubnub is connected.""" lock_one = await _mock_doorsense_enabled_august_lock_detail(hass) assert lock_one.pubsub_channel == "pubsub" pubnub = AugustPubNub() await _create_august_with_devices(hass, [lock_one], pubnub=pubnub) pubnub.connected = True lock_state = hass.states.get("lock.online_with_doorsense_name") assert lock_state.state == STATE_LOCKED assert lock_state.attributes["battery_level"] == 92 assert lock_state.attributes["friendly_name"] == "online_with_doorsense Name" data = {ATTR_ENTITY_ID: "lock.online_with_doorsense_name"} await hass.services.async_call(LOCK_DOMAIN, SERVICE_UNLOCK, data, blocking=True) pubnub.message( pubnub, Mock( channel=lock_one.pubsub_channel, timetoken=(dt_util.utcnow().timestamp() + 1) * 10000000, message={ "status": "kAugLockState_Unlocked", }, ), ) await hass.async_block_till_done() await hass.async_block_till_done() lock_state = hass.states.get("lock.online_with_doorsense_name") assert lock_state.state == STATE_UNLOCKED assert lock_state.attributes["battery_level"] == 92 assert lock_state.attributes["friendly_name"] == "online_with_doorsense Name" await hass.services.async_call(LOCK_DOMAIN, SERVICE_LOCK, data, blocking=True) pubnub.message( pubnub, Mock( channel=lock_one.pubsub_channel, timetoken=(dt_util.utcnow().timestamp() + 2) * 10000000, message={ "status": "kAugLockState_Locked", }, ), ) await hass.async_block_till_done() await hass.async_block_till_done() lock_state = hass.states.get("lock.online_with_doorsense_name") assert lock_state.state == STATE_LOCKED # No activity means it will be unavailable until the activity feed has data lock_operator_sensor = entity_registry.async_get( "sensor.online_with_doorsense_name_operator" ) assert lock_operator_sensor assert ( hass.states.get("sensor.online_with_doorsense_name_operator").state == STATE_UNKNOWN ) freezer.tick(INITIAL_LOCK_RESYNC_TIME) pubnub.message( pubnub, Mock( channel=lock_one.pubsub_channel, timetoken=(dt_util.utcnow().timestamp() + 2) * 10000000, message={ "status": "kAugLockState_Unlocked", }, ), ) await hass.async_block_till_done() lock_state = hass.states.get("lock.online_with_doorsense_name") assert lock_state.state == STATE_UNLOCKED async def test_lock_jammed(hass: HomeAssistant) -> None: """Test lock gets jammed on unlock.""" def _unlock_return_activities_side_effect(access_token, device_id): raise ClientResponseError(None, None, status=531) lock_one = await _mock_doorsense_enabled_august_lock_detail(hass) await _create_august_with_devices( hass, [lock_one], api_call_side_effects={ "unlock_return_activities": _unlock_return_activities_side_effect }, ) lock_state = hass.states.get("lock.online_with_doorsense_name") assert lock_state.state == STATE_LOCKED assert lock_state.attributes["battery_level"] == 92 assert lock_state.attributes["friendly_name"] == "online_with_doorsense Name" data = {ATTR_ENTITY_ID: "lock.online_with_doorsense_name"} await hass.services.async_call(LOCK_DOMAIN, SERVICE_UNLOCK, data, blocking=True) lock_state = hass.states.get("lock.online_with_doorsense_name") assert lock_state.state == STATE_JAMMED async def test_lock_throws_exception_on_unknown_status_code( hass: HomeAssistant, ) -> None: """Test lock throws exception.""" def _unlock_return_activities_side_effect(access_token, device_id): raise ClientResponseError(None, None, status=500) lock_one = await _mock_doorsense_enabled_august_lock_detail(hass) await _create_august_with_devices( hass, [lock_one], api_call_side_effects={ "unlock_return_activities": _unlock_return_activities_side_effect }, ) lock_state = hass.states.get("lock.online_with_doorsense_name") assert lock_state.state == STATE_LOCKED assert lock_state.attributes["battery_level"] == 92 assert lock_state.attributes["friendly_name"] == "online_with_doorsense Name" data = {ATTR_ENTITY_ID: "lock.online_with_doorsense_name"} with pytest.raises(ClientResponseError): await hass.services.async_call(LOCK_DOMAIN, SERVICE_UNLOCK, data, blocking=True) async def test_one_lock_unknown_state(hass: HomeAssistant) -> None: """Test creation of a lock with doorsense and bridge.""" lock_one = await _mock_lock_from_fixture( hass, "get_lock.online.unknown_state.json", ) await _create_august_with_devices(hass, [lock_one]) assert hass.states.get("lock.brokenid_name").state == STATE_UNKNOWN async def test_lock_bridge_offline(hass: HomeAssistant) -> None: """Test creation of a lock with doorsense and bridge that goes offline.""" lock_one = await _mock_doorsense_enabled_august_lock_detail(hass) activities = await _mock_activities_from_fixture( hass, "get_activity.bridge_offline.json" ) await _create_august_with_devices(hass, [lock_one], activities=activities) assert hass.states.get("lock.online_with_doorsense_name").state == STATE_UNAVAILABLE async def test_lock_bridge_online(hass: HomeAssistant) -> None: """Test creation of a lock with doorsense and bridge that goes offline.""" lock_one = await _mock_doorsense_enabled_august_lock_detail(hass) activities = await _mock_activities_from_fixture( hass, "get_activity.bridge_online.json" ) await _create_august_with_devices(hass, [lock_one], activities=activities) assert hass.states.get("lock.online_with_doorsense_name").state == STATE_LOCKED async def test_lock_update_via_pubnub(hass: HomeAssistant) -> None: """Test creation of a lock with doorsense and bridge.""" lock_one = await _mock_doorsense_enabled_august_lock_detail(hass) states = hass.states assert lock_one.pubsub_channel == "pubsub" pubnub = AugustPubNub() activities = await _mock_activities_from_fixture(hass, "get_activity.lock.json") config_entry = await _create_august_with_devices( hass, [lock_one], activities=activities, pubnub=pubnub ) pubnub.connected = True assert states.get("lock.online_with_doorsense_name").state == STATE_LOCKED pubnub.message( pubnub, Mock( channel=lock_one.pubsub_channel, timetoken=dt_util.utcnow().timestamp() * 10000000, message={ "status": "kAugLockState_Unlocking", }, ), ) await hass.async_block_till_done() await hass.async_block_till_done() assert states.get("lock.online_with_doorsense_name").state == STATE_UNLOCKING pubnub.message( pubnub, Mock( channel=lock_one.pubsub_channel, timetoken=(dt_util.utcnow().timestamp() + 1) * 10000000, message={ "status": "kAugLockState_Locking", }, ), ) await hass.async_block_till_done() await hass.async_block_till_done() assert states.get("lock.online_with_doorsense_name").state == STATE_LOCKING async_fire_time_changed(hass, dt_util.utcnow() + datetime.timedelta(seconds=30)) await hass.async_block_till_done() assert hass.states.get("lock.online_with_doorsense_name").state == STATE_LOCKING pubnub.connected = True async_fire_time_changed(hass, dt_util.utcnow() + datetime.timedelta(seconds=30)) await hass.async_block_till_done() assert states.get("lock.online_with_doorsense_name").state == STATE_LOCKING # Ensure pubnub status is always preserved async_fire_time_changed(hass, dt_util.utcnow() + datetime.timedelta(hours=2)) await hass.async_block_till_done() assert states.get("lock.online_with_doorsense_name").state == STATE_LOCKING pubnub.message( pubnub, Mock( channel=lock_one.pubsub_channel, timetoken=(dt_util.utcnow().timestamp() + 2) * 10000000, message={ "status": "kAugLockState_Unlocking", }, ), ) await hass.async_block_till_done() await hass.async_block_till_done() assert states.get("lock.online_with_doorsense_name").state == STATE_UNLOCKING async_fire_time_changed(hass, dt_util.utcnow() + datetime.timedelta(hours=4)) await hass.async_block_till_done() assert states.get("lock.online_with_doorsense_name").state == STATE_UNLOCKING await hass.config_entries.async_unload(config_entry.entry_id) await hass.async_block_till_done() async def test_open_throws_hass_service_not_supported_error( hass: HomeAssistant, ) -> None: """Test open throws correct error on entity does not support this service error.""" mocked_lock_detail = await _mock_operative_august_lock_detail(hass) await _create_august_with_devices(hass, [mocked_lock_detail]) data = {ATTR_ENTITY_ID: "lock.a6697750d607098bae8d6baa11ef8063_name"} with pytest.raises(HomeAssistantError, match="does not support this service"): await hass.services.async_call(LOCK_DOMAIN, SERVICE_OPEN, data, blocking=True)