"""Mocks for the august component.""" import json import os import time from unittest.mock import MagicMock, PropertyMock from asynctest import mock from august.activity import DoorOperationActivity, LockOperationActivity from august.authenticator import AuthenticationState from august.doorbell import Doorbell, DoorbellDetail from august.lock import Lock, LockDetail from homeassistant.components.august import ( CONF_LOGIN_METHOD, CONF_PASSWORD, CONF_USERNAME, DOMAIN, ) from homeassistant.setup import async_setup_component from tests.common import load_fixture def _mock_get_config(): """Return a default august config.""" return { DOMAIN: { CONF_LOGIN_METHOD: "email", CONF_USERNAME: "mocked_username", CONF_PASSWORD: "mocked_password", } } @mock.patch("homeassistant.components.august.gateway.Api") @mock.patch("homeassistant.components.august.gateway.Authenticator.authenticate") async def _mock_setup_august(hass, api_instance, authenticate_mock, api_mock): """Set up august integration.""" authenticate_mock.side_effect = MagicMock( return_value=_mock_august_authentication("original_token", 1234) ) api_mock.return_value = api_instance assert await async_setup_component(hass, DOMAIN, _mock_get_config()) await hass.async_block_till_done() return True async def _create_august_with_devices(hass, devices, api_call_side_effects=None): if api_call_side_effects is None: api_call_side_effects = {} device_data = { "doorbells": [], "locks": [], } for device in devices: if isinstance(device, LockDetail): device_data["locks"].append( {"base": _mock_august_lock(device.device_id), "detail": device} ) elif isinstance(device, DoorbellDetail): device_data["doorbells"].append( {"base": _mock_august_doorbell(device.device_id), "detail": device} ) else: raise ValueError def _get_device_detail(device_type, device_id): for device in device_data[device_type]: if device["detail"].device_id == device_id: return device["detail"] raise ValueError def _get_base_devices(device_type): base_devices = [] for device in device_data[device_type]: base_devices.append(device["base"]) return base_devices def get_lock_detail_side_effect(access_token, device_id): return _get_device_detail("locks", device_id) def get_doorbell_detail_side_effect(access_token, device_id): return _get_device_detail("doorbells", device_id) def get_operable_locks_side_effect(access_token): return _get_base_devices("locks") def get_doorbells_side_effect(access_token): return _get_base_devices("doorbells") def get_house_activities_side_effect(access_token, house_id, limit=10): return [] def lock_return_activities_side_effect(access_token, device_id): lock = _get_device_detail("locks", device_id) return [ _mock_lock_operation_activity(lock, "lock"), _mock_door_operation_activity(lock, "doorclosed"), ] def unlock_return_activities_side_effect(access_token, device_id): lock = _get_device_detail("locks", device_id) return [ _mock_lock_operation_activity(lock, "unlock"), _mock_door_operation_activity(lock, "dooropen"), ] if "get_lock_detail" not in api_call_side_effects: api_call_side_effects["get_lock_detail"] = get_lock_detail_side_effect if "get_doorbell_detail" not in api_call_side_effects: api_call_side_effects["get_doorbell_detail"] = get_doorbell_detail_side_effect if "get_operable_locks" not in api_call_side_effects: api_call_side_effects["get_operable_locks"] = get_operable_locks_side_effect if "get_doorbells" not in api_call_side_effects: api_call_side_effects["get_doorbells"] = get_doorbells_side_effect if "get_house_activities" not in api_call_side_effects: api_call_side_effects["get_house_activities"] = get_house_activities_side_effect if "lock_return_activities" not in api_call_side_effects: api_call_side_effects[ "lock_return_activities" ] = lock_return_activities_side_effect if "unlock_return_activities" not in api_call_side_effects: api_call_side_effects[ "unlock_return_activities" ] = unlock_return_activities_side_effect return await _mock_setup_august_with_api_side_effects(hass, api_call_side_effects) async def _mock_setup_august_with_api_side_effects(hass, api_call_side_effects): api_instance = MagicMock(name="Api") if api_call_side_effects["get_lock_detail"]: api_instance.get_lock_detail.side_effect = api_call_side_effects[ "get_lock_detail" ] if api_call_side_effects["get_operable_locks"]: api_instance.get_operable_locks.side_effect = api_call_side_effects[ "get_operable_locks" ] if api_call_side_effects["get_doorbells"]: api_instance.get_doorbells.side_effect = api_call_side_effects["get_doorbells"] if api_call_side_effects["get_doorbell_detail"]: api_instance.get_doorbell_detail.side_effect = api_call_side_effects[ "get_doorbell_detail" ] if api_call_side_effects["get_house_activities"]: api_instance.get_house_activities.side_effect = api_call_side_effects[ "get_house_activities" ] if api_call_side_effects["lock_return_activities"]: api_instance.lock_return_activities.side_effect = api_call_side_effects[ "lock_return_activities" ] if api_call_side_effects["unlock_return_activities"]: api_instance.unlock_return_activities.side_effect = api_call_side_effects[ "unlock_return_activities" ] return await _mock_setup_august(hass, api_instance) def _mock_august_authentication(token_text, token_timestamp): authentication = MagicMock(name="august.authentication") type(authentication).state = PropertyMock( return_value=AuthenticationState.AUTHENTICATED ) type(authentication).access_token = PropertyMock(return_value=token_text) type(authentication).access_token_expires = PropertyMock( return_value=token_timestamp ) return authentication def _mock_august_lock(lockid="mocklockid1", houseid="mockhouseid1"): return Lock(lockid, _mock_august_lock_data(lockid=lockid, houseid=houseid)) def _mock_august_doorbell(deviceid="mockdeviceid1", houseid="mockhouseid1"): return Doorbell( deviceid, _mock_august_doorbell_data(deviceid=deviceid, houseid=houseid) ) def _mock_august_doorbell_data(deviceid="mockdeviceid1", houseid="mockhouseid1"): return { "_id": deviceid, "DeviceID": deviceid, "name": deviceid + " Name", "HouseID": houseid, "UserType": "owner", "serialNumber": "mockserial", "battery": 90, "status": "standby", "currentFirmwareVersion": "mockfirmware", "Bridge": { "_id": "bridgeid1", "firmwareVersion": "mockfirm", "operative": True, }, "LockStatus": {"doorState": "open"}, } def _mock_august_lock_data(lockid="mocklockid1", houseid="mockhouseid1"): return { "_id": lockid, "LockID": lockid, "LockName": lockid + " Name", "HouseID": houseid, "UserType": "owner", "SerialNumber": "mockserial", "battery": 90, "currentFirmwareVersion": "mockfirmware", "Bridge": { "_id": "bridgeid1", "firmwareVersion": "mockfirm", "operative": True, }, "LockStatus": {"doorState": "open"}, } async def _mock_operative_august_lock_detail(hass): return await _mock_lock_from_fixture(hass, "get_lock.online.json") async def _mock_inoperative_august_lock_detail(hass): return await _mock_lock_from_fixture(hass, "get_lock.offline.json") async def _mock_lock_from_fixture(hass, path): json_dict = await _load_json_fixture(hass, path) return LockDetail(json_dict) async def _mock_doorbell_from_fixture(hass, path): json_dict = await _load_json_fixture(hass, path) return DoorbellDetail(json_dict) async def _load_json_fixture(hass, path): fixture = await hass.async_add_executor_job( load_fixture, os.path.join("august", path) ) return json.loads(fixture) async def _mock_doorsense_enabled_august_lock_detail(hass): return await _mock_lock_from_fixture(hass, "get_lock.online_with_doorsense.json") async def _mock_doorsense_missing_august_lock_detail(hass): return await _mock_lock_from_fixture(hass, "get_lock.online_missing_doorsense.json") def _mock_lock_operation_activity(lock, action): return LockOperationActivity( { "dateTime": time.time() * 1000, "deviceID": lock.device_id, "deviceType": "lock", "action": action, } ) def _mock_door_operation_activity(lock, action): return DoorOperationActivity( { "dateTime": time.time() * 1000, "deviceID": lock.device_id, "deviceType": "lock", "action": action, } )