diff --git a/.github/PULL_REQUEST_TEMPLATE/pull_request_template.md b/.github/PULL_REQUEST_TEMPLATE/pull_request_template.md new file mode 100644 index 00000000..e69de29b diff --git a/.github/workflows/pythonpackage.yml b/.github/workflows/pythonpackage.yml index 09fd22b2..5b23a13e 100644 --- a/.github/workflows/pythonpackage.yml +++ b/.github/workflows/pythonpackage.yml @@ -29,6 +29,12 @@ jobs: flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --ignore=E303,W503 --statistics - name: Test with pytest run: | - pip install pytest pytest-mock + pip install pytest pytest-mock pytest-cov cd tests - pytest + pytest --cov=devolo_home_control_api + - name: Coveralls + run: | + pip install coveralls==1.10.0 + export COVERALLS_REPO_TOKEN=${{ secrets.COVERALLS_TOKEN }} + cd tests + coveralls diff --git a/README.md b/README.md index fe5eae80..15215780 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,10 @@ # devolo_home_control_api -![PyPI - Downloads](https://img.shields.io/pypi/dd/devolo-home-control-api) ![Libraries.io SourceRank](https://img.shields.io/librariesio/sourcerank/pypi/devolo-home-control-api) +[![GitHub Workflow Status](https://img.shields.io/github/workflow/status/2Fake/devolo_home_control_api/Python%20package)](https://github.com/2Fake/devolo_home_control_api/actions?query=workflow%3A%22Python+package%22) +[![PyPI - Downloads](https://img.shields.io/pypi/dd/devolo-home-control-api)](https://pypi.org/project/devolo-home-control-api/) +[![Libraries.io SourceRank](https://img.shields.io/librariesio/sourcerank/pypi/devolo-home-control-api)](https://libraries.io/pypi/devolo-home-control-api) +[![Code Climate maintainability](https://img.shields.io/codeclimate/maintainability/2Fake/devolo_home_control_api)](https://codeclimate.com/github/2Fake/devolo_home_control_api) +[![Coverage Status](https://coveralls.io/repos/github/2Fake/devolo_home_control_api/badge.svg?branch=master)](https://coveralls.io/github/2Fake/devolo_home_control_api?branch=master) This project implements parts of the devolo Home Control API in Python. It is based on reverse engineering and therefore may fail with any new devolo update. If you discover a breakage, please feel free to [report an issue](https://github.com/2Fake/devolo_home_control_api/issues). @@ -77,31 +81,17 @@ for gateway_id in mydevolo.gateway_ids: ### Collecting Home Control data -There are three ways of getting data: +There are two ways of getting data: -1. Poll the gateway 1. Let the websocket push data into your object, but still poll the object 1. Subscribe to the publisher and let it push (preferred) -#### Poll the gateway - -When polling the gateway, each property will be checked at the time of accessing it. - -```python -mprm = MprmRest(gateway_id=gateway_id) -for binary_switch in mprm.binary_switch_devices: - for state in binary_switch.binary_switch_property: - print (f"State of {binary_switch.name} ({binary_switch.binary_switch_property[state].element_uid}): {binary_switch.binary_switch_property[state].state}") -``` - -To execute this example, you need a configured instance of Mydevolo. - #### Using websockets -Your way of accessing the data is more or less the same. Websocket events will keep the object up to date. This method uses less resources on the devolo Home Control Central Unit. +When using websocket events, messages will keep the object up to date. Nevertheless, no further action is triggered. So you have to ask yourself. The following example will list the current state of all binary switches. If the state changes, you will not notice unless you ask again. ```python -mprm = MprmWebsocket(gateway_id=gateway_id) +homecontrol = HomeControl(gateway_id=gateway_id) for binary_switch in mprm.binary_switch_devices: for state in binary_switch.binary_switch_property: print (f"State of {binary_switch.name} ({binary_switch.binary_switch_property[state].element_uid}): {binary_switch.binary_switch_property[state].state}") diff --git a/devolo_home_control_api/backend/__init__.py b/devolo_home_control_api/backend/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/devolo_home_control_api/backend/mprm_rest.py b/devolo_home_control_api/backend/mprm_rest.py new file mode 100644 index 00000000..df911521 --- /dev/null +++ b/devolo_home_control_api/backend/mprm_rest.py @@ -0,0 +1,185 @@ +import json +import logging +import socket +import time +import requests +import threading +from zeroconf import ServiceBrowser, ServiceStateChange, Zeroconf + +from ..devices.gateway import Gateway +from ..mydevolo import Mydevolo + + +class MprmRest: + """ + The MprmRest object handles calls to the so called mPRM as singleton. It does not cover all API calls, just those + requested up to now. All calls are done in a gateway context, so you need to provide the ID of that gateway. + + :param gateway_id: Gateway ID + :param url: URL of the mPRM + .. todo:: Make __instance a dict to handle multiple gateways at the same time + """ + + __instance = None + + @classmethod + def get_instance(cls): + if cls.__instance is None: + raise SyntaxError(f"Please init {cls.__name__}() once to establish a connection to the gateway's backend.") + return cls.__instance + + @classmethod + def del_instance(cls): + cls.__instance = None + + + def __init__(self, gateway_id: str, url: str): + if self.__class__.__instance is not None: + raise SyntaxError(f"Please use {self.__class__.__name__}.get_instance() to reuse the connection to the backend.") + self._logger = logging.getLogger(self.__class__.__name__) + self._gateway = Gateway(gateway_id) + self._mydevolo = Mydevolo.get_instance() + self._session = requests.Session() + self._data_id = 0 + self._mprm_url = url + self._local_ip = None + + self.__class__.__instance = self + + + def detect_gateway_in_lan(self): + """ Detects a gateway in local network and check if it is the desired one. """ + def on_service_state_change(zeroconf, service_type, name, state_change): + if state_change is ServiceStateChange.Added: + zeroconf.get_service_info(service_type, name) + + zeroconf = Zeroconf() + ServiceBrowser(zeroconf, "_http._tcp.local.", handlers=[on_service_state_change]) + self._logger.info("Searching for gateway in LAN") + start_time = time.time() + while not time.time() > start_time + 3 and self._local_ip is None: + for mdns_name in zeroconf.cache.entries(): + try: + ip = socket.inet_ntoa(mdns_name.address) + if mdns_name.key.startswith("devolo-homecontrol") and \ + requests.get("http://" + ip + "/dhlp/port/full", + auth=(self._gateway.local_user, self._gateway.local_passkey), + timeout=0.5).status_code == requests.codes.ok: + self._logger.debug(f"Got successful answer from ip {ip}. Setting this as local gateway") + self._local_ip = ip + break + except OSError: + # Got IPv6 address which isn't supported by socket.inet_ntoa and the gateway as well. + self._logger.debug(f"Found an IPv6 address. This cannot be a gateway.") + except AttributeError: + # The MDNS entry does not provide address information + pass + else: + time.sleep(0.05) + threading.Thread(target=zeroconf.close).start() + return self._local_ip + + def create_connection(self): + """ Create session, either locally or via cloud. """ + if self._local_ip: + self._gateway.local_connection = True + self.get_local_session() + elif self._gateway.external_access and not self._mydevolo.maintenance: + self.get_remote_session() + else: + self._logger.error("Cannot connect to gateway. No gateway found in LAN and external access is not possible.") + raise ConnectionError("Cannot connect to gateway.") + + def extract_data_from_element_uid(self, uid: str) -> dict: + """ + Returns data from an element UID using an RPC call. + + :param uid: Element UID, something like devolo.MultiLevelSensor:hdm:ZWave:CBC56091/24#2 + :return: Data connected to the element UID, payload so to say + """ + data = {"method": "FIM/getFunctionalItems", + "params": [[uid], 0]} + response = self.post(data) + return response.get("result").get("items")[0] + + def get_all_devices(self) -> dict: + """ + Get all devices. + + :return: Dict with all devices and their properties. + """ + self._logger.info("Inspecting devices") + data = {"method": "FIM/getFunctionalItems", + "params": [['devolo.DevicesPage'], 0]} + response = self.post(data) + return response.get("result").get("items")[0].get("properties").get("deviceUIDs") + + def get_local_session(self): + """ Connect to the gateway locally. """ + self._logger.info("Connecting to gateway locally") + self._mprm_url = "http://" + self._local_ip + try: + self._token_url = self._session.get(self._mprm_url + "/dhlp/portal/full", + auth=(self._gateway.local_user, self._gateway.local_passkey), timeout=5).json() + except json.JSONDecodeError: + self._logger.error("Could not connect to the gateway locally.") + raise MprmDeviceCommunicationError("Could not connect to the gateway locally.") from None + except requests.ConnectTimeout: + self._logger.error("Timeout during connecting to the gateway.") + raise + self._session.get(self._token_url.get('link')) + + def get_name_and_element_uids(self, uid: str): + """ + Returns the name, all element UIDs and the device model of the given device UID. + + :param uid: Element UID, something like devolo.MultiLevelSensor:hdm:ZWave:CBC56091/24#2 + """ + data = {"method": "FIM/getFunctionalItems", + "params": [[uid], 0]} + response = self.post(data) + properties = response.get("result").get("items")[0].get("properties") + return properties + + def get_remote_session(self): + """ Connect to the gateway remotely. """ + self._logger.info("Connecting to gateway via cloud") + try: + self._session.get(self._gateway.full_url, timeout=15) + except json.JSONDecodeError: + raise MprmDeviceCommunicationError("Gateway is offline.") from None + + def post(self, data: dict) -> dict: + """ + Communicate with the RPC interface. + + :param data: Data to be send + :return: Response to the data + """ + if not(self._gateway.online or self._gateway.sync) and not self._gateway.local_connection: + raise MprmDeviceCommunicationError("Gateway is offline.") + + self._data_id += 1 + data['jsonrpc'] = "2.0" + data['id'] = self._data_id + try: + response = self._session.post(self._mprm_url + "/remote/json-rpc", + data=json.dumps(data), + headers={"content-type": "application/json"}, + timeout=15).json() + except requests.ReadTimeout: + self._logger.error("Gateway is offline.") + self._gateway.update_state(False) + raise MprmDeviceCommunicationError("Gateway is offline.") from None + if response['id'] != data['id']: + self._logger.error("Got an unexpected response after posting data.") + raise ValueError("Got an unexpected response after posting data.") + return response + + +class MprmDeviceCommunicationError(Exception): + """ Communicating to a device via mPRM failed """ + + +class MprmDeviceNotFoundError(Exception): + """ A device like this was not found """ diff --git a/devolo_home_control_api/backend/mprm_websocket.py b/devolo_home_control_api/backend/mprm_websocket.py new file mode 100644 index 00000000..7f5a3720 --- /dev/null +++ b/devolo_home_control_api/backend/mprm_websocket.py @@ -0,0 +1,95 @@ +import json +import threading +import time + +import websocket +from requests import ConnectionError, ReadTimeout +from urllib3.connection import ConnectTimeoutError + +from devolo_home_control_api.backend.mprm_rest import MprmRest, MprmDeviceCommunicationError + + +class MprmWebsocket(MprmRest): + """ + The MprmWebsocket object handles calls to the mPRM via websockets. It does not cover all API calls, just those + requested up to now. All calls are done in a gateway context, so you need to provide the ID of that gateway. As + it inherites from MprmRest, it is a singleton as well. + + :param gateway_id: Gateway ID (aka serial number), typically found on the label of the device + :param url: URL of the mPRM + """ + + def __init__(self, gateway_id: str, url: str): + super().__init__(gateway_id, url) + self._ws = None + self._event_sequence = 0 + + self.publisher = None + self.on_update = None + + + def websocket_connection(self): + """ Set up the websocket connection """ + ws_url = self._mprm_url.replace("https://", "wss://").replace("http://", "ws://") + cookie = "; ".join([str(name) + "=" + str(value) for name, value in self._session.cookies.items()]) + ws_url = f"{ws_url}/remote/events/?topics=com/prosyst/mbs/services/fim/FunctionalItemEvent/PROPERTY_CHANGED," \ + f"com/prosyst/mbs/services/fim/FunctionalItemEvent/UNREGISTERED" \ + f"&filter=(|(GW_ID={self._gateway.id})(!(GW_ID=*)))" + self._logger.debug(f"Connecting to {ws_url}") + self._ws = websocket.WebSocketApp(ws_url, + cookie=cookie, + on_open=self._on_open, + on_message=self._on_message, + on_error=self._on_error, + on_close=self._on_close) + self._ws.run_forever(ping_interval=30, ping_timeout=5) + + + def _on_close(self): + """ Callback function to react on closing the websocket. """ + self._logger.info("Closed web socket connection.") + + def _on_error(self, error: str): + """ Callback function to react on errors. We will try reconnecting with prolonging intervals. """ + self._logger.error(error) + i = 16 + connected = False + self._ws.close() + + self._event_sequence = 0 + + while not connected: + try: + self._logger.info("Trying to reconnect to the gateway.") + # TODO: Check if local_ip is still correct after lost connection + self.get_local_session() if self._local_ip else self.get_remote_session() + connected = True + except (json.JSONDecodeError, ConnectTimeoutError, ReadTimeout, ConnectionError, MprmDeviceCommunicationError): + self._logger.info(f"Sleeping for {i} seconds.") + time.sleep(i) + i = i * 2 if i < 2048 else 3600 + self.websocket_connection() + + def _on_message(self, message: str): + """ Callback function to react on a message. """ + message = json.loads(message) + + event_sequence = message.get("properties").get("com.prosyst.mbs.services.remote.event.sequence.number") + if event_sequence == self._event_sequence: + self._event_sequence += 1 + else: + self._logger.warning("We missed a websocket message.") + self._event_sequence = event_sequence + + try: + self.on_update(message) + except TypeError: + self._logger.error("on_update is not set.") + + def _on_open(self): + """ Callback function to keep the websocket open. """ + def run(*args): + self._logger.info("Starting web socket connection") + while self._ws.sock is not None and self._ws.sock.connected: + time.sleep(1) + threading.Thread(target=run).start() diff --git a/devolo_home_control_api/devices/gateway.py b/devolo_home_control_api/devices/gateway.py index e7d6f759..eb2950d7 100644 --- a/devolo_home_control_api/devices/gateway.py +++ b/devolo_home_control_api/devices/gateway.py @@ -32,7 +32,7 @@ def __init__(self, gateway_id: str): @property def full_url(self): - """ The full URL is used to get a valid remote session """ + """ The full URL is used to get a valid remote session. """ if self._full_url is None: self._full_url = self._mydevolo.get_full_url(self.id) self._logger.debug(f"Setting full URL to {self._full_url}") diff --git a/devolo_home_control_api/devices/zwave.py b/devolo_home_control_api/devices/zwave.py index bbbe14b5..ff4e4f36 100644 --- a/devolo_home_control_api/devices/zwave.py +++ b/devolo_home_control_api/devices/zwave.py @@ -1,31 +1,34 @@ import logging +from ..mydevolo import Mydevolo + class Zwave: """ - Representing object for Z-Wave devices connected to the devolo Home Control Central Unit + Representing object for Z-Wave devices. :param device_uid: Device UID, something like hdm:ZWave:CBC56091/24 """ - def __init__(self, name: str, device_uid: str, zone: str, battery_level: int, icon: str, online_state: int): + def __init__(self, **kwargs): self._logger = logging.getLogger(self.__class__.__name__) - self.name = name - self.zone = zone - if battery_level != -1: - self.battery_level = battery_level - self.icon = icon - self.device_uid = device_uid - self.subscriber = None - - # Online state is returned as numbers. 1 --> Offline, 2 --> Online, 7 (?) --> Not initialized - if online_state == 1: - self.online = "offline" - elif online_state == 2: - self.online = "online" - else: - self.online = "unknown state" - self._logger.warning(f"Unknown state {online_state} for device {self.name}") + + for key, value in kwargs.items(): + setattr(self, key, value) + + self.mydevolo = Mydevolo.get_instance() + device_info = self.mydevolo.get_zwave_products(manufacturer=self.manID, + product_type=self.prodTypeID, + product=self.prodID) + for key, value in device_info.items(): + setattr(self, key, value) + + self.uid = get_device_uid_from_element_uid(self.elementUIDs[0]) + + if self.batteryLevel == -1: + delattr(self, "batteryLevel") + delattr(self, "batteryLow") + def get_property(self, name: str) -> list: """ @@ -35,4 +38,34 @@ def get_property(self, name: str) -> list: :return: List of UIDs in this property :raises: AttributeError: The property does not exist in this device type """ - return [*getattr(self, f"{name}_property").keys()] + return [*getattr(self, f"{name}_property").values()] + + +def get_device_type_from_element_uid(element_uid: str) -> str: + """ + Return the device type of the given element UID. + + :param element_uid: Element UID, something like devolo.MultiLevelSensor:hdm:ZWave:CBC56091/24#2 + :return: Device type, something like devolo.MultiLevelSensor + """ + return element_uid.split(":")[0] + + +def get_device_uid_from_setting_uid(setting_uid: str) -> str: + """ + Return the device uid of the given setting UID. + + :param setting_uid: Setting UID, something like lis.hdm:ZWave:EB5A9F6C/2 + :return: Device UID, something like hdm:ZWave:EB5A9F6C/2 + """ + return setting_uid.split(".", 1)[-1] + + +def get_device_uid_from_element_uid(element_uid: str) -> str: + """ + Return device UID from the given element UID. + + :param element_uid: Element UID, something like devolo.MultiLevelSensor:hdm:ZWave:CBC56091/24#2 + :return: Device UID, something like hdm:ZWave:CBC56091/24 + """ + return element_uid.split(":", 1)[1].split("#")[0] diff --git a/devolo_home_control_api/homecontrol.py b/devolo_home_control_api/homecontrol.py new file mode 100644 index 00000000..db4acb23 --- /dev/null +++ b/devolo_home_control_api/homecontrol.py @@ -0,0 +1,173 @@ +import logging +import threading + +import requests + +from .backend.mprm_websocket import MprmWebsocket +from .devices.gateway import Gateway +from .devices.zwave import Zwave, get_device_type_from_element_uid +from .properties.binary_switch_property import BinarySwitchProperty +from .properties.consumption_property import ConsumptionProperty +from .properties.settings_property import SettingsProperty +from .properties.voltage_property import VoltageProperty +from .publisher.publisher import Publisher +from .publisher.updater import Updater + + +class HomeControl: + """ + Representing object for your Home Control setup. This is more or less the glue between your devolo Home Control Central + Unit, your devices and their properties. + + :param gateway_id: Gateway ID (aka serial number), typically found on the label of the device + :param url: URL of the mPRM (typically leave it at default) + """ + + def __init__(self, gateway_id: str, url: str = "https://homecontrol.mydevolo.com"): + self._logger = logging.getLogger(self.__class__.__name__) + self._gateway = Gateway(gateway_id) + self._session = requests.Session() + + self.mprm = MprmWebsocket(gateway_id=gateway_id, url=url) + self.mprm.on_update = self.update + self.mprm.detect_gateway_in_lan() + self.mprm.create_connection() + + # Create the initial device dict + self.devices = {} + self._inspect_devices() + self.device_names = dict(zip([self.devices.get(device).itemName for device in self.devices], + [self.devices.get(device).uid for device in self.devices])) + + self.create_pub() + self.updater = Updater(devices=self.devices, gateway=self._gateway, publisher=self.mprm.publisher) + + threading.Thread(target=self.mprm.websocket_connection).start() + + + @property + def binary_switch_devices(self) -> list: + """ Get all binary switch devices. """ + return [self.devices.get(uid) for uid in self.devices if hasattr(self.devices.get(uid), "binary_switch_property")] + + @property + def publisher(self) -> Publisher: + """ Get all publisher. """ + return self.mprm.publisher + + + def create_pub(self): + """ Create a publisher for every device. """ + publisher_list = [device for device in self.devices] + self.mprm.publisher = Publisher(publisher_list) + + def is_online(self, uid: str) -> bool: + """ + Get the online state of a device. + + :param uid: Device UID, something like hdm:ZWave:CBC56091/24 + :return: True, if device is online + """ + return False if self.devices.get(uid).status == 1 else True + + def update(self, message: str): + """ Initialize steps needed to update properties on a new message. """ + self.updater.update(message) + + + def _inspect_devices(self): + """ Create the initial internal device dict. """ + for device in self.mprm.get_all_devices(): + properties = dict([(key, value) for key, value in self.mprm.get_name_and_element_uids(uid=device).items()]) + self.devices[device] = Zwave(**properties) + self._process_element_uids(device=device, element_uids=properties.get("elementUIDs")) + self._process_settings_uids(device=device, setting_uids=properties.get("settingUIDs")) + + def _process_element_uids(self, device: str, element_uids: list): + """ Generate properties depending on the element uid. """ + def binary_switch(element_uid: str): + if not hasattr(self.devices[device], "binary_switch_property"): + self.devices[device].binary_switch_property = {} + self._logger.debug(f"Adding binary switch property to {device}.") + self.devices[device].binary_switch_property[element_uid] = BinarySwitchProperty(element_uid) + self.devices[device].binary_switch_property[element_uid].is_online = self.is_online + self.devices[device].binary_switch_property[element_uid].fetch_binary_switch_state() + + def meter(element_uid: str): + if not hasattr(self.devices[device], "consumption_property"): + self.devices[device].consumption_property = {} + self._logger.debug(f"Adding consumption property to {device}.") + self.devices[device].consumption_property[element_uid] = ConsumptionProperty(element_uid) + self.devices[device].consumption_property[element_uid].fetch_consumption('current') + self.devices[device].consumption_property[element_uid].fetch_consumption('total') + + def voltage_multi_level_sensor(element_uid: str): + if not hasattr(self.devices[device], "voltage_property"): + self.devices[device].voltage_property = {} + self._logger.debug(f"Adding voltage property to {device}.") + self.devices[device].voltage_property[element_uid] = VoltageProperty(element_uid) + self.devices[device].voltage_property[element_uid].fetch_voltage() + + device_type = {"devolo.BinarySwitch": binary_switch, + "devolo.Meter": meter, + "devolo.VoltageMultiLevelSensor": voltage_multi_level_sensor} + + for element_uid in element_uids: + try: + device_type[get_device_type_from_element_uid(element_uid)](element_uid) + except KeyError: + self._logger.debug(f"Found an unexpected element uid: {element_uid}") + + def _process_settings_uids(self, device: str, setting_uids: list): + """ Generate properties depending on the setting uid. """ + def led(setting_uid: str): + self._logger.debug(f"Adding led settings to {device}.") + self.devices[device].settings_property["led"] = SettingsProperty(element_uid=setting_uid, led_setting=None) + self.devices[device].settings_property["led"].fetch_led_setting() + + def general_device(setting_uid: str): + self._logger.debug(f"Adding general device settings to {device}.") + self.devices[device].settings_property["general_device_settings"] = SettingsProperty(element_uid=setting_uid, + events_enabled=None, + name=None, + zone_id=None, + icon=None) + self.devices[device].settings_property["general_device_settings"].fetch_general_device_settings() + + def parameter(setting_uid: str): + self._logger.debug(f"Adding parameter settings to {device}.") + self.devices[device].settings_property["param_changed"] = SettingsProperty(element_uid=setting_uid, + param_changed=None) + self.devices[device].settings_property["param_changed"].fetch_param_changed_setting() + + def protection(setting_uid: str): + self._logger.debug(f"Adding protection settings to {device}.") + self.devices[device].settings_property["protection"] = SettingsProperty(element_uid=setting_uid, + local_switching=None, + remote_switching=None) + self.devices[device].settings_property["protection"].fetch_protection_setting(protection_setting="local") + self.devices[device].settings_property["protection"].fetch_protection_setting(protection_setting="remote") + + if not hasattr(self.devices[device], "settings_property"): + self.devices[device].settings_property = {} + + setting = {"lis.hdm": led, + "gds.hdm": general_device, + "cps.hdm": parameter, + "ps.hdm": protection} + + for setting_uid in setting_uids: + try: + setting[get_device_type_from_element_uid(setting_uid)](setting_uid) + except KeyError: + self._logger.debug(f"Found an unexpected element uid: {setting_uid}") + + +def get_sub_device_uid_from_element_uid(element_uid: str) -> int: + """ + Return the sub device uid of the given element UID. + + :param element_uid: Element UID, something like devolo.MultiLevelSensor:hdm:ZWave:CBC56091/24#2 + :return: Sub device UID, something like 2 + """ + return None if "#" not in element_uid else int(element_uid.split("#")[-1]) diff --git a/devolo_home_control_api/mprm_rest.py b/devolo_home_control_api/mprm_rest.py deleted file mode 100644 index 17746c2f..00000000 --- a/devolo_home_control_api/mprm_rest.py +++ /dev/null @@ -1,440 +0,0 @@ -import json -import logging -import socket -import time - -import requests -from zeroconf import ServiceBrowser, ServiceStateChange, Zeroconf - -from .devices.gateway import Gateway -from .devices.zwave import Zwave -from .mydevolo import Mydevolo -from .properties.binary_switch_property import BinarySwitchProperty -from .properties.consumption_property import ConsumptionProperty -from .properties.settings_property import SettingsProperty -from .properties.voltage_property import VoltageProperty - - -class MprmRest: - """ - The MprmRest object handles calls to the so called mPRM. It does not cover all API calls, just those requested - up to now. All calls are done in a gateway context, so you need to provide the ID of that gateway. - - :param gateway_id: Gateway ID (aka serial number), typically found on the label of the device - :param url: URL of the mPRM (typically leave it at default) - :raises: JSONDecodeError: Connecting to the gateway was not possible - """ - - def __init__(self, gateway_id: str, url: str = "https://homecontrol.mydevolo.com"): - self._logger = logging.getLogger(self.__class__.__name__) - self._gateway = Gateway(gateway_id) - self._session = requests.Session() - self._data_id = 0 - self._mprm_url = url - - mydevolo = Mydevolo.get_instance() - self.local_ip = self._detect_gateway_in_lan() - - if self.local_ip: - self._gateway.local_connection = True - self._get_local_session() - elif self._gateway.external_access and not mydevolo.maintenance: - self._get_remote_session() - else: - self._logger.error("Cannot connect to gateway. No gateway found in LAN and external access is not possible.") - raise ConnectionError("Cannot connect to gateway.") - - # create the initial device dict - self.devices = {} - self._inspect_devices() - - @property - def binary_switch_devices(self): - """Returns all binary switch devices.""" - return [self.devices.get(uid) for uid in self.devices if hasattr(self.devices.get(uid), - "binary_switch_property")] - - - def get_binary_switch_state(self, element_uid: str) -> bool: - """ - Update and return the binary switch state for the given uid. - - :param element_uid: element UID of the consumption. Usually starts with devolo.BinarySwitch - :return: Binary switch state - """ - if not element_uid.startswith("devolo.BinarySwitch"): - raise ValueError("Not a valid uid to get binary switch data.") - response = self._extract_data_from_element_uid(element_uid) - self.devices.get(get_device_uid_from_element_uid(element_uid)).binary_switch_property.get(element_uid).state = \ - True if response.get("properties").get("state") == 1 else False - return self.devices.get(get_device_uid_from_element_uid(element_uid)).binary_switch_property.get(element_uid).state - - def get_consumption(self, element_uid: str, consumption_type: str = "current") -> float: - """ - Update and return the consumption, specified in consumption_type for the given uid. - - :param element_uid: Element UID of the consumption. Usually starts with devolo.Meter. - :param consumption_type: Current or total consumption - :return: Consumption - """ - if not element_uid.startswith("devolo.Meter"): - raise ValueError("Not a valid uid to get consumption data.") - if consumption_type not in ["current", "total"]: - raise ValueError('Unknown consumption type. "current" and "total" are valid consumption types.') - response = self._extract_data_from_element_uid(element_uid) - if consumption_type == "current": - self.devices.get(get_device_uid_from_element_uid(element_uid)).consumption_property.get(element_uid).current = \ - response.get("properties").get("currentValue") - return self.devices.get(get_device_uid_from_element_uid(element_uid)).consumption_property.get(element_uid).current - else: - self.devices.get(get_device_uid_from_element_uid(element_uid)).consumption_property.get(element_uid).total = \ - response.get("properties").get("totalValue") - return self.devices.get(get_device_uid_from_element_uid(element_uid)).consumption_property.get(element_uid).total - - def get_device_uid_from_name(self, name: str, zone: str = "") -> str: - """ - Get device from name. Sometimes, the name is ambiguous. Then hopefully the zone makes it unique. - - :param name: Name of the device - :param zone: Zone the device is in. Only needed, if device name is ambiguous. - :return: Device UID - """ - device_list = [] - for device in self.devices.values(): - if device.name == name: - device_list.append(device) - if len(device_list) == 0: - raise MprmDeviceNotFoundError(f'There is no device "{name}"') - elif len(device_list) > 1 and zone == "": - raise MprmDeviceNotFoundError(f'The name "{name}" is ambiguous ({len(device_list)} times). Please provide a zone.') - elif len(device_list) > 1: - for device in device_list: - if device.zone == zone: - return device.device_uid - else: - raise MprmDeviceNotFoundError(f'There is no device "{name}" in zone "{zone}".') - else: - return device_list[0].device_uid - - def get_led_setting(self, setting_uid: str) -> bool: - """ - Update and return the led setting. - - :param setting_uid: Setting UID of the LED setting. Usually starts with lis.hdm. - :return: LED setting - """ - if not setting_uid.startswith("lis.hdm"): - raise ValueError("Not a valid uid to get the led setting") - response = self._extract_data_from_element_uid(setting_uid) - self.devices.get(get_device_uid_from_setting_uid(setting_uid)).settings_property.get("led").led_setting = \ - response.get("properties").get("led") - return self.devices.get(get_device_uid_from_setting_uid(setting_uid)).settings_property.get("led").led_setting - - def get_general_device_settings(self, setting_uid: str) -> bool: - """ - Update and return the events enabled setting. If a device shall report to the diary, this is true. - - :param setting_uid: Settings UID to look at. Usually starts with gds.hdm. - :return: Events enabled or not - """ - if not setting_uid.startswith("gds.hdm"): - raise ValueError("Not a valid uid to get the events enabled setting.") - response = self._extract_data_from_element_uid(setting_uid) - gds = self.devices.get(get_device_uid_from_setting_uid(setting_uid)).settings_property.get("general_device_settings") - gds.name = response.get("properties").get("name") - gds.icon = response.get("properties").get("icon") - gds.zone_id = response.get("properties").get("zoneID") - gds.events_enabled = response.get("properties").get("eventsEnabled") - return gds.name, gds.icon, gds.zone_id, gds.events_enabled - - def get_param_changed_setting(self, setting_uid: str) -> bool: - """ - Update and return the param changed setting. If a device has modified Z-Wave parameters, this is true. - - :param setting_uid: Settings UID to look at. Usually starts with cps.hdm. - :return: Parameter changed or not - """ - if not setting_uid.startswith("cps.hdm"): - raise ValueError("Not a valid uid to get the param changed setting") - response = self._extract_data_from_element_uid(setting_uid) - device_uid = get_device_uid_from_setting_uid(setting_uid) - self.devices.get(device_uid).settings_property.get("param_changed").param_changed = \ - response.get("properties").get("paramChanged") - return self.devices.get(device_uid).settings_property.get("param_changed").param_changed - - def get_protection_setting(self, setting_uid, protection_setting): - """ - Update and return the protection setting. There are only two protection settings. Local and remote switching. - - :param setting_uid: Settings UID to look at. Usually starts with ps.hdm. - :param protection_setting: Look at local or remote switching. - :return: Switching is protected or not - """ - if not setting_uid.startswith("ps.hdm"): - raise ValueError("Not a valid uid to get the protection setting") - if protection_setting not in ["local", "remote"]: - raise ValueError("Only local and remote are possible protection settings") - response = self._extract_data_from_element_uid(setting_uid) - setting_property = self.devices.get(get_device_uid_from_setting_uid(setting_uid)).settings_property.get("led") - if protection_setting == "local": - setting_property.local_switching = response.get("properties").get("localSwitch") - return setting_property.local_switching - else: - setting_property.remote_switching = response.get("properties").get("remoteSwitch") - return setting_property.remote_switching - - def get_voltage(self, element_uid: str) -> float: - """ - Update and return the voltage - - :param element_uid: Element UID of the voltage. Usually starts with devolo.VoltageMultiLevelSensor - :return: Voltage value - """ - if not element_uid.startswith("devolo.VoltageMultiLevelSensor"): - raise ValueError("Not a valid uid to get consumption data.") - response = self._extract_data_from_element_uid(element_uid) - self.devices.get(get_device_uid_from_element_uid(element_uid)).voltage_property.get(element_uid).current = \ - response.get("properties").get("value") - return self.devices.get(get_device_uid_from_element_uid(element_uid)).voltage_property.get(element_uid).current - - def set_binary_switch(self, element_uid: str, state: bool): - """ - Set the binary switch of the given element_uid to the given state. - - :param element_uid: element_uid - :param state: True if switching on, False if switching off - """ - if not element_uid.startswith("devolo.BinarySwitch"): - raise ValueError("Not a valid uid to set binary switch data.") - if type(state) != bool: - raise ValueError("Not a valid binary switch state.") - data = {"method": "FIM/invokeOperation", - "params": [element_uid, "turnOn" if state else "turnOff", []]} - device_uid = get_device_uid_from_element_uid(element_uid) - response = self._post(data) - if response.get("result").get("status") == 1: - self.devices.get(device_uid).binary_switch_property.get(element_uid).state = state - elif response.get("result").get("status") == 2 \ - and not self._device_usable(get_device_uid_from_element_uid(element_uid)): - raise MprmDeviceCommunicationError("The device is offline.") - else: - self._logger.info(f"Could not set state of device {device_uid}. Maybe it is already at this state.") - self._logger.info(f"Target state is {state}.") - self._logger.info(f"Actual state is {self.devices.get(device_uid).binary_switch_property.get(element_uid).state}.") - - - def _detect_gateway_in_lan(self): - """ Detects a gateway in local network and check if it is the desired one. """ - def on_service_state_change(zeroconf, service_type, name, state_change): - if state_change is ServiceStateChange.Added: - zeroconf.get_service_info(service_type, name) - - local_ip = None - zeroconf = Zeroconf() - ServiceBrowser(zeroconf, "_http._tcp.local.", handlers=[on_service_state_change]) - self._logger.info("Searching for gateway in LAN") - # TODO: Optimize the sleep - time.sleep(2) - for mdns_name in zeroconf.cache.entries(): - if hasattr(mdns_name, "address"): - try: - ip = socket.inet_ntoa(mdns_name.address) - if requests.get("http://" + ip + "/dhlp/port/full", - auth=(self._gateway.local_user, self._gateway.local_passkey), - timeout=0.5).status_code == requests.codes.ok: - self._logger.debug(f"Got successful answer from ip {ip}. Setting this as local gateway") - local_ip = ip - break - except OSError: - # Got IPv6 address which isn't supported by socket.inet_ntoa and the gateway as well. - self._logger.debug(f"Found an IPv6 address. This cannot be a gateway.") - zeroconf.close() - return local_ip - - def _device_usable(self, uid): - """ - Return the 'online' state of the given device as bool. - We consider everything as 'online' if the device can receive new values. - """ - return True if self.devices.get(uid).online in ["online"] else False - - def _extract_data_from_element_uid(self, element_uid): - """ Returns data from an element_uid using a RPC call """ - data = {"method": "FIM/getFunctionalItems", - "params": [[element_uid], 0]} - response = self._post(data) - # TODO: Catch error! - return response.get("result").get("items")[0] - - def _get_local_session(self): - """ Connect to the gateway locally. """ - self._logger.info("Connecting to gateway locally") - self._mprm_url = "http://" + self.local_ip - try: - self._token_url = self._session.get(self._mprm_url + "/dhlp/portal/full", - auth=(self._gateway.local_user, self._gateway.local_passkey), timeout=5).json() - except json.JSONDecodeError: - self._logger.error("Could not connect to the gateway locally.") - raise MprmDeviceCommunicationError("Could not connect to the gateway locally.") from None - except requests.ConnectTimeout: - self._logger.error("Timeout during connecting to the gateway.") - raise - self._session.get(self._token_url.get('link')) - - def _get_name_and_element_uids(self, uid): - """ Returns the name, all element UIDs and the device model of the given device UID. """ - data = {"method": "FIM/getFunctionalItems", - "params": [[uid], 0]} - response = self._post(data) - properties = response.get("result").get("items")[0].get("properties") - return properties.get("itemName"),\ - properties.get("zone"),\ - properties.get("batteryLevel"),\ - properties.get("icon"),\ - properties.get("elementUIDs"),\ - properties.get("settingUIDs"),\ - properties.get("deviceModelUID"),\ - properties.get("status") - - def _get_remote_session(self): - """ Connect to the gateway remotely. """ - self._logger.info("Connecting to gateway via cloud") - try: - self._session.get(self._gateway.full_url) - except json.JSONDecodeError: - raise MprmDeviceCommunicationError("Gateway is offline.") from None - - def _inspect_devices(self): - """ Create the initial internal device dict. """ - self._logger.info("Inspecting devices") - data = {"method": "FIM/getFunctionalItems", - "params": [['devolo.DevicesPage'], 0]} - response = self._post(data) - all_devices_list = response.get("result").get("items")[0].get("properties").get("deviceUIDs") - for device in all_devices_list: - name, zone, battery_level, icon, element_uids, setting_uids, deviceModelUID, online_state = \ - self._get_name_and_element_uids(uid=device) - # Process device uids - self.devices[device] = Zwave(name=name, - device_uid=device, - zone=zone, - battery_level=battery_level, - icon=icon, - online_state=online_state) - self._process_element_uids(device=device, name=name, element_uids=element_uids) - self._process_settings_uids(device=device, name=name, setting_uids=setting_uids) - - def _post(self, data: dict) -> dict: - """ Communicate with the RPC interface. """ - if not(self._gateway.online or self._gateway.sync) and not self._gateway.local_connection: - raise MprmDeviceCommunicationError("Gateway is offline.") - - self._data_id += 1 - data['jsonrpc'] = "2.0" - data['id'] = self._data_id - try: - response = self._session.post(self._mprm_url + "/remote/json-rpc", - data=json.dumps(data), - headers={"content-type": "application/json"}, - timeout=15).json() - except requests.ReadTimeout: - self._logger.error("Gateway is offline.") - self._gateway.update_state(False) - raise MprmDeviceCommunicationError("Gateway is offline.") from None - if response['id'] != data['id']: - self._logger.error("Got an unexpected response after posting data.") - raise ValueError("Got an unexpected response after posting data.") - return response - - def _process_element_uids(self, device, name, element_uids): - """ Generate properties depending on the element uid """ - for element_uid in element_uids: - if get_device_type_from_element_uid(element_uid) == "devolo.BinarySwitch": - if not hasattr(self.devices[device], "binary_switch_property"): - self.devices[device].binary_switch_property = {} - self._logger.debug(f"Adding {name} ({device}) to device list as binary switch property.") - self.devices[device].binary_switch_property[element_uid] = BinarySwitchProperty(element_uid) - self.get_binary_switch_state(element_uid) - elif get_device_type_from_element_uid(element_uid) == "devolo.Meter": - if not hasattr(self.devices[device], "consumption_property"): - self.devices[device].consumption_property = {} - self._logger.debug(f"Adding {name} ({device}) to device list as consumption property.") - self.devices[device].consumption_property[element_uid] = ConsumptionProperty(element_uid) - for consumption in ['current', 'total']: - self.get_consumption(element_uid, consumption) - elif get_device_type_from_element_uid(element_uid) == "devolo.VoltageMultiLevelSensor": - if not hasattr(self.devices[device], "voltage_property"): - self.devices[device].voltage_property = {} - self._logger.debug(f"Adding {name} ({device}) to device list as voltage property.") - self.devices[device].voltage_property[element_uid] = VoltageProperty(element_uid) - self.get_voltage(element_uid) - else: - self._logger.debug(f"Found an unexpected element uid: {element_uid}") - - def _process_settings_uids(self, device, name, setting_uids): - """Generate properties depending on the setting uid""" - for setting_uid in setting_uids: - if not hasattr(self.devices[device], "settings_property"): - self.devices[device].settings_property = {} - if get_device_type_from_element_uid(setting_uid) == "lis.hdm": - self._logger.debug(f"Adding {name} ({device}) to device list as settings property") - self.devices[device].settings_property["led"] = SettingsProperty(element_uid=setting_uid, - led_setting=None) - self.get_led_setting(setting_uid) - elif get_device_type_from_element_uid(setting_uid) == "gds.hdm": - self.devices[device].settings_property["general_device_settings"] = SettingsProperty(element_uid=setting_uid, - events_enabled=None, - name=None, - zone_id=None, - icon=None) - self.get_general_device_settings(setting_uid) - elif get_device_type_from_element_uid(setting_uid) == "cps.hdm": - self.devices[device].settings_property["param_changed"] = SettingsProperty(element_uid=setting_uid, - param_changed=None) - self.get_param_changed_setting(setting_uid) - elif get_device_type_from_element_uid(setting_uid) == "ps.hdm": - self.devices[device].settings_property["protection"] = SettingsProperty(element_uid=setting_uid, - local_switching=None, - remote_switching=None) - for protection in ["local", "remote"]: - # TODO: find a better way for this loop. - self.get_protection_setting(setting_uid=setting_uid, protection_setting=protection) - else: - self._logger.debug(f"Found an unexpected element uid: {setting_uid}") - - -def get_device_uid_from_element_uid(element_uid: str) -> str: - """ - Return device UID from the given element UID - - :param element_uid: Element UID, something like devolo.MultiLevelSensor:hdm:ZWave:CBC56091/24#2 - :return: Device UID, something like hdm:ZWave:CBC56091/24 - """ - return element_uid.split(":", 1)[1].split("#")[0] - - -def get_device_type_from_element_uid(element_uid): - """ - Return the device type of the given element uid - - :param element_uid: Element UID, something like devolo.MultiLevelSensor:hdm:ZWave:CBC56091/24#2 - :return: Device type, something like devolo.MultiLevelSensor - """ - return element_uid.split(":")[0] - - -def get_device_uid_from_setting_uid(setting_uid): - """ - Return the device uid of the given setting uid - :param setting_uid: Setting UID, something like lis.hdm:ZWave:EB5A9F6C/2 - :return: Device UID, something like hdm:ZWave:EB5A9F6C/2 - """ - return setting_uid.split(".", 1)[-1] - - -class MprmDeviceCommunicationError(Exception): - """ Communicating to a device via mPRM failed """ - - -class MprmDeviceNotFoundError(Exception): - """ A device like this was not found """ diff --git a/devolo_home_control_api/mprm_websocket.py b/devolo_home_control_api/mprm_websocket.py deleted file mode 100644 index b4c27cd7..00000000 --- a/devolo_home_control_api/mprm_websocket.py +++ /dev/null @@ -1,253 +0,0 @@ -import json -import threading -import time - -import websocket -from requests import ConnectionError, ReadTimeout -from urllib3.connection import ConnectTimeoutError - -from .mprm_rest import MprmRest, get_device_uid_from_element_uid - - -class MprmWebsocket(MprmRest): - """ - The MprmWebsocket object handles calls to the mPRM via websockets. It does not cover all API calls, just those - requested up to now. All calls are done in a user context, so you need to provide credentials of that user. - - :param gateway_id: Gateway ID - :param url: URL of the mPRM stage (typically use default value) - """ - - def __init__(self, gateway_id: str, url: str = "https://homecontrol.mydevolo.com"): - super().__init__(gateway_id, url) - self._ws = None - self._event_sequence = 0 - - self.publisher = None - - self._create_pub() - threading.Thread(target=self._websocket_connection).start() - - - def get_consumption(self, element_uid: str, consumption_type: str = "current") -> float: - """ - Return the internal saved consumption, specified in consumption_type for the given uid. - - :param element_uid: element UID of the consumption. Usually starts with devolo.Meter - :param consumption_type: current or total consumption - :return: Consumption - """ - if not element_uid.startswith("devolo.Meter"): - raise ValueError("Not a valid uid to get consumption data.") - if consumption_type not in ["current", "total"]: - raise ValueError('Unknown consumption type. "current" and "total" are valid consumption types.') - if consumption_type == "current": - return self.devices.get(get_device_uid_from_element_uid(element_uid)).consumption_property.get(element_uid).current - else: - return self.devices.get(get_device_uid_from_element_uid(element_uid)).consumption_property.get(element_uid).total - - def update_binary_switch_state(self, element_uid: str, value: bool = None): - """ - Function to update the internal binary switch state of a device. - If value is None, it uses a RPC-Call to retrieve the value. If a value is given, e.g. from a websocket, - the value is written into the internal dict. - - :param element_uid: Element UID, something like, devolo.BinarySwitch:hdm:ZWave:CBC56091/24#2 - :param value: Value so be set - """ - if not element_uid.startswith("devolo.BinarySwitch"): - raise ValueError("Not a valid uid to set binary switch state.") - if value is None: - self.get_binary_switch_state(element_uid=element_uid) - else: - for binary_switch_name, binary_switch_property_value in \ - self.devices.get(get_device_uid_from_element_uid(element_uid)).binary_switch_property.items(): - if binary_switch_name == element_uid: - self._logger.debug(f"Updating state of {element_uid}") - binary_switch_property_value.state = value - message = (element_uid, value) - self.publisher.dispatch(get_device_uid_from_element_uid(element_uid), message) - - def update_consumption(self, element_uid: str, consumption: str, value: float = None): - """ - Function to update the internal consumption of a device. - If value is None, it uses a RPC-Call to retrieve the value. If a value is given, e.g. from a websocket, - the value is written into the internal dict. - - :param element_uid: Element UID, something like , something like devolo.MultiLevelSensor:hdm:ZWave:CBC56091/24#2 - :param consumption: current or total consumption - :param value: Value so be set - """ - if not element_uid.startswith("devolo.Meter"): - raise ValueError("Not a valid uid to set consumption data.") - if consumption not in ["current", "total"]: - raise ValueError(f'Consumption value "{consumption}" is not valid. Only "current" and "total" are allowed.') - if value is None: - super().get_consumption(element_uid=element_uid, consumption_type=consumption) - else: - for consumption_property_name, consumption_property_value in \ - self.devices.get(get_device_uid_from_element_uid(element_uid)).consumption_property.items(): - if element_uid == consumption_property_name: - self._logger.debug(f"Updating {consumption} consumption of {element_uid} to {value}") - # TODO : make one liner - if consumption == "current": - consumption_property_value.current = value - else: - consumption_property_value.total = value - message = (element_uid, value) - self.publisher.dispatch(get_device_uid_from_element_uid(element_uid), message) - - def update_gateway_state(self, accessible: bool, online_sync: bool): - """ - Function to update the gateway status. A gateway might go on- or offline while we listen to the websocket. - - :param accessible: Online state of the gateway - :param online_sync: Sync state of the gateway - """ - self._logger.debug(f"Updating status and state of gateway to status: {accessible} and state: {online_sync}") - self._gateway.online = accessible - self._gateway.sync = online_sync - - def update_voltage(self, element_uid: str, value: float = None): - """ - Function to update the internal voltage of a device. - If value is None, it uses a RPC-Call to retrieve the value. If a value is given, e.g. from a websocket, - the value is written into the internal dict. - - :param element_uid: Element UID, something like devolo.VoltageMultiLevelSensor:hdm:ZWave:CBC56091/24 - :param value: Value so be set - """ - if not element_uid.startswith("devolo.VoltageMultiLevelSensor"): - raise ValueError("Not a valid uid to set voltage data.") - if value is None: - self.get_voltage(element_uid=element_uid) - else: - for voltage_property_name, voltage_property_value in \ - self.devices.get(get_device_uid_from_element_uid(element_uid)).voltage_property.items(): - if element_uid == voltage_property_name: - self._logger.debug(f"Updating voltage of {element_uid}") - voltage_property_value.current = value - message = (element_uid, value) - self.publisher.dispatch(get_device_uid_from_element_uid(element_uid), message) - - - def _create_pub(self): - """ - Create a publisher for every element we support at the moment. - Actually, there are publisher for current consumption and binary state. Current consumption publisher is create as - "current_consumption_ELEMENT_UID" and binary state publisher is created as "binary_state_ELEMENT_UID". - """ - publisher_list = [device for device in self.devices] - self.publisher = Publisher(publisher_list) - - def _on_open(self): - """ Callback function to keep the websocket open. """ - def run(*args): - self._logger.info("Starting web socket connection") - while self._ws.sock.connected: - time.sleep(1) - threading.Thread(target=run).start() - - def _on_message(self, message): - """ Callback function to react on a message. """ - message = json.loads(message) - event_sequence = message.get("properties").get("com.prosyst.mbs.services.remote.event.sequence.number") - if event_sequence == self._event_sequence: - self._event_sequence += 1 - else: - self._logger.warning("We missed a websocket message.") - self._event_sequence = event_sequence - if message.get("properties").get("uid").startswith("devolo.Meter"): - if message.get("properties").get("property.name") == "currentValue": - self.update_consumption(element_uid=message.get("properties").get("uid"), - consumption="current", - value=message.get("properties").get("property.value.new")) - elif message.get("properties").get("property.name") == "totalValue": - self.update_consumption(element_uid=message.get("properties").get("uid"), - consumption="total", value=message.get("properties").get("property.value.new")) - else: - self._logger.info(f'Unknown meter message received for {message.get("properties").get("uid")}.') - self._logger.info(message.get("properties")) - elif message.get("properties").get("uid").startswith("devolo.BinarySwitch") \ - and message.get("properties").get("property.name") == "state": - self.update_binary_switch_state(element_uid=message.get("properties").get("uid"), - value=True if message.get("properties").get("property.value.new") == 1 - else False) - elif message.get("properties").get("uid").startswith("devolo.VoltageMultiLevelSensor"): - self.update_voltage(element_uid=message.get("properties").get("uid"), - value=message.get("properties").get("property.value.new")) - elif message.get("properties").get("uid") == "devolo.mprm.gw.GatewayAccessibilityFI" \ - and message.get("properties").get("property.name") == "gatewayAccessible": - self.update_gateway_state(accessible=message.get("properties").get("property.value.new").get("accessible"), - online_sync=message.get("properties").get("property.value.new").get("onlineSync")) - - else: - # Unknown messages shall be ignored - self._logger.debug(json.dumps(message, indent=4)) - - def _on_error(self, error): - """ Callback function to react on errors. We will try reconnecting with prolonging intervals. """ - self._logger.error(error) - - - def _on_close(self): - """ Callback function to react on closing the websocket. """ - self._logger.info("Closed web socket connection") - i = 16 - while not self._ws.sock.connected: - try: - self._logger.info("Trying to reconnect to the gateway.") - if self.local_ip: - self._get_local_session() - else: - self._get_remote_session() - self._websocket_connection() - except (json.JSONDecodeError, ConnectTimeoutError, ReadTimeout, ConnectionError, websocket.WebSocketException): - self._logger.info(f"Sleeping for {i} seconds.") - time.sleep(i) - if i < 3600: - i *= 2 - else: - i = 3600 - - def _websocket_connection(self): - """ Set up the websocket connection """ - ws_url = self._mprm_url.replace("https://", "wss://").replace("http://", "ws://") - cookie = "; ".join([str(name) + "=" + str(value) for name, value in self._session.cookies.items()]) - ws_url = f"{ws_url}/remote/events/?topics=com/prosyst/mbs/services/fim/FunctionalItemEvent/PROPERTY_CHANGED," \ - f"com/prosyst/mbs/services/fim/FunctionalItemEvent/UNREGISTERED" \ - f"&filter=(|(GW_ID={self._gateway.id})(!(GW_ID=*)))" - self._logger.debug(f"Connecting to {ws_url}") - self._ws = websocket.WebSocketApp(ws_url, - cookie=cookie, - on_open=self._on_open, - on_message=self._on_message, - on_error=self._on_error, - on_close=self._on_close) - self._ws.run_forever(ping_interval=30, ping_timeout=5) - - -class Publisher: - def __init__(self, events): - # maps event names to subscribers - # str -> dict - self.events = {event: dict() - for event in events} - - def dispatch(self, event, message): - for callback in self.get_subscribers(event).values(): - callback(message) - - def get_events(self): - return self.events - - def get_subscribers(self, event): - return self.events[event] - - def register(self, event, who, callback=None): - if callback is None: - callback = getattr(who, 'update') - self.get_subscribers(event)[who] = callback - - def unregister(self, event, who): - del self.get_subscribers(event)[who] diff --git a/devolo_home_control_api/mydevolo.py b/devolo_home_control_api/mydevolo.py index ed6af3ac..65b057bf 100644 --- a/devolo_home_control_api/mydevolo.py +++ b/devolo_home_control_api/mydevolo.py @@ -16,30 +16,30 @@ class Mydevolo: __instance = None - @staticmethod - def get_instance(): - if Mydevolo.__instance is None: - Mydevolo() - return Mydevolo.__instance + @classmethod + def get_instance(cls): + if cls.__instance is None: + raise SyntaxError(f"Please init {cls.__name__}() once to establish a connection to my devolo.") + return cls.__instance - @staticmethod - def del_instance(): - Mydevolo.__instance = None + @classmethod + def del_instance(cls): + cls.__instance = None def __init__(self): - if Mydevolo.__instance is not None: - raise SyntaxError("Please use Mydevolo.get_instance() to connect to my devolo.") - else: - self._logger = logging.getLogger(self.__class__.__name__) - self._user = None - self._password = None - self._uuid = None - self._gateway_ids = [] + if self.__class__.__instance is not None: + raise SyntaxError(f"Please use {self.__class__.__name__}.get_instance() and reuse the connection to my devolo.") + + self._logger = logging.getLogger(self.__class__.__name__) + self._user = None + self._password = None + self._uuid = None + self._gateway_ids = [] - self.url = "https://www.mydevolo.com" + self.url = "https://www.mydevolo.com" - Mydevolo.__instance = self + self.__class__.__instance = self @property @@ -71,13 +71,13 @@ def uuid(self) -> str: """ The uuid is a central attribute in my devolo. Most URLs in the user context contain it. """ if self._uuid is None: self._logger.debug("Getting UUID") - self._uuid = self._call(self.url + "/v1/users/uuid").get("uuid") + self._uuid = self._call(f"{self.url}/v1/users/uuid").get("uuid") return self._uuid @property def maintenance(self) -> bool: """ If devolo Home Control is in maintenance, there is not much we can do via cloud. """ - state = self._call(self.url + "/v1/hc/maintenance").get("state") + state = self._call(f"{self.url}/v1/hc/maintenance").get("state") if state == "on": return False else: @@ -89,7 +89,7 @@ def gateway_ids(self) -> list: """ Get gateway IDs. """ if not self._gateway_ids: self._logger.debug(f"Getting list of gateways") - items = self._call(self.url + "/v1/users/" + self.uuid + "/hc/gateways/status").get("items") + items = self._call(f"{self.url}/v1/users/{self.uuid}/hc/gateways/status").get("items") for gateway in items: self._gateway_ids.append(gateway.get("gatewayId")) self._logger.debug(f'Adding {gateway.get("gatewayId")} to list of gateways.') @@ -107,7 +107,13 @@ def get_gateway(self, gateway_id: str) -> dict: :return: Gateway object """ self._logger.debug(f"Getting details for gateway {gateway_id}") - return self._call(self.url + "/v1/users/" + self.uuid + "/hc/gateways/" + gateway_id) + details = {} + try: + details = self._call(f"{self.url}/v1/users/{self.uuid}/hc/gateways/{gateway_id}") + except WrongUrlError: + self._logger.error("Could not get full URL. Wrong gateway ID used?") + raise + return details def get_full_url(self, gateway_id: str) -> str: """ @@ -117,14 +123,29 @@ def get_full_url(self, gateway_id: str) -> str: :return: URL """ self._logger.debug("Getting full URL of gateway.") - return self._call(self.url + "/v1/users/" - + self.uuid + "/hc/gateways/" + gateway_id + "/fullURL").get("url") - + return self._call(f"{self.url}/v1/users/{self.uuid}/hc/gateways/{gateway_id}/fullURL").get("url") - def _call(self, url: str) -> dict: + def get_zwave_products(self, manufacturer: str, product_type: str, product: str) -> dict: """ - Make a call to any entry point with the user's context. + Get information about a Z-Wave device. + + :param manufacturer: The manufacturer ID in hex. + :param product_type: The product type ID in hex. + :param product: The product ID in hex. + :return: All known product information. """ + self._logger.debug(f"Getting information for {manufacturer}/{product_type}/{product}") + device_info = {} + try: + device_info = self._call(f"{self.url}/v1/zwave/products/{manufacturer}/{product_type}/{product}") + except WrongUrlError: + # At some devices no device information are returned + self._logger.debug("No device info found") + return device_info + + + def _call(self, url: str) -> dict: + """ Make a call to any entry point with the user's context. """ responds = requests.get(url, auth=(self._user, self._password), headers={'content-type': 'application/json'}, @@ -132,8 +153,7 @@ def _call(self, url: str) -> dict: if responds.status_code == requests.codes.forbidden: self._logger.error("Could not get full URL. Wrong username or password?") raise WrongCredentialsError("Wrong username or password.") - elif responds.status_code == requests.codes.not_found: - self._logger.error("Could not get full URL. Wrong gateway ID used?") + if responds.status_code == requests.codes.not_found: raise WrongUrlError(f"Wrong URL: {url}") return responds.json() diff --git a/devolo_home_control_api/properties/binary_switch_property.py b/devolo_home_control_api/properties/binary_switch_property.py index 8b412688..cce1e97c 100644 --- a/devolo_home_control_api/properties/binary_switch_property.py +++ b/devolo_home_control_api/properties/binary_switch_property.py @@ -1,4 +1,5 @@ from .property import Property, WrongElementError +from ..backend.mprm_rest import MprmDeviceCommunicationError class BinarySwitchProperty(Property): @@ -14,3 +15,31 @@ def __init__(self, element_uid): super().__init__(element_uid=element_uid) self.state = None + + + def fetch_binary_switch_state(self) -> bool: + """ + Update and return the binary switch state for the given uid. + + :return: Binary switch state + """ + response = self.mprm.extract_data_from_element_uid(self.element_uid) + self.state = True if response.get("properties").get("state") == 1 else False + return self.state + + def set_binary_switch(self, state: bool): + """ + Set the binary switch of the given element_uid to the given state. + + :param state: True if switching on, False if switching off + """ + data = {"method": "FIM/invokeOperation", + "params": [self.element_uid, "turnOn" if state else "turnOff", []]} + response = self.mprm.post(data) + if response.get("result").get("status") == 2 and not self.is_online(self.device_uid): + raise MprmDeviceCommunicationError("The device is offline.") + if response.get("result").get("status") == 1: + self.state = state + else: + self._logger.info(f"Could not set state of device {self.device_uid}. Maybe it is already at this state.") + self._logger.info(f"Target state is {state}. Actual state is {self.state}.") diff --git a/devolo_home_control_api/properties/consumption_property.py b/devolo_home_control_api/properties/consumption_property.py index fe4cb148..2a5cdcda 100644 --- a/devolo_home_control_api/properties/consumption_property.py +++ b/devolo_home_control_api/properties/consumption_property.py @@ -18,3 +18,21 @@ def __init__(self, element_uid): self.total = None self.total_since = None self.total_unit = "kWh" + + + def fetch_consumption(self, consumption_type: str = "current") -> float: + """ + Update and return the consumption, specified in consumption_type for the given uid. + + :param consumption_type: Current or total consumption + :return: Consumption + """ + if consumption_type not in ["current", "total"]: + raise ValueError('Unknown consumption type. "current" and "total" are valid consumption types.') + response = self.mprm.extract_data_from_element_uid(self.element_uid) + if consumption_type == "current": + self.current = response.get("properties").get("currentValue") + return self.current + else: + self.total = response.get("properties").get("totalValue") + return self.total diff --git a/devolo_home_control_api/properties/property.py b/devolo_home_control_api/properties/property.py index 28fe9fee..069efde6 100644 --- a/devolo_home_control_api/properties/property.py +++ b/devolo_home_control_api/properties/property.py @@ -1,4 +1,5 @@ import logging +from ..backend.mprm_websocket import MprmWebsocket class Property: @@ -11,7 +12,10 @@ class Property: def __init__(self, element_uid): self._logger = logging.getLogger(self.__class__.__name__) self.element_uid = element_uid + self.device_uid = element_uid.split(":", 1)[1].split("#")[0] + self.mprm = MprmWebsocket.get_instance() + self.is_online = None class WrongElementError(Exception): - """ This element was not meant for this property """ + """ This element was not meant for this property. """ diff --git a/devolo_home_control_api/properties/settings_property.py b/devolo_home_control_api/properties/settings_property.py index a2d2afb3..5dfdf364 100644 --- a/devolo_home_control_api/properties/settings_property.py +++ b/devolo_home_control_api/properties/settings_property.py @@ -8,22 +8,63 @@ class SettingsProperty(Property): :param element_uid: Element UID, something like devolo.BinarySwitch:hdm:ZWave:CBC56091/24#2 """ - def __init__(self, element_uid, **kwargs): - """ - - :param kwargs: - """ - super().__init__(element_uid=element_uid) + def __init__(self, element_uid: str, **kwargs): if element_uid.split(".")[0] not in ["lis", "gds", "cps", "ps"]: raise WrongElementError() + + super().__init__(element_uid=element_uid) + self.setting_uid = element_uid for key, value in kwargs.items(): - if key == "led_setting": - self.led_setting = value - elif key == "events_enabled": - self.events_enabled = value - elif key == "param_changed": - self.param_changed = value - elif key == "local_switching": - self.local_switching = value - elif key == "remote_switching": - self.remote_switching = value + setattr(self, key, value) + + + def fetch_general_device_settings(self) -> bool: + """ + Update and return the events enabled setting. If a device shall report to the diary, this is true. + + :return: Events enabled or not + """ + response = self.mprm.extract_data_from_element_uid(self.setting_uid) + self.name = response.get("properties").get("name") + self.icon = response.get("properties").get("icon") + self.zone_id = response.get("properties").get("zoneID") + self.events_enabled = response.get("properties").get("eventsEnabled") + return self.name, self.icon, self.zone_id, self.events_enabled + + def fetch_led_setting(self) -> bool: + """ + Update and return the led setting. + + :return: LED setting + """ + response = self.mprm.extract_data_from_element_uid(self.setting_uid) + self.led_setting = response.get("properties").get("led") + return self.led_setting + + def fetch_param_changed_setting(self) -> bool: + """ + Update and return the param changed setting. + + :param setting_uid: Settings UID to look at. Usually starts with cps.hdm. + :return: True, if parameter was changed + """ + response = self.mprm.extract_data_from_element_uid(self.setting_uid) + self.param_changed = response.get("properties").get("paramChanged") + return self.param_changed + + def fetch_protection_setting(self, protection_setting: str) -> bool: + """ + Update and return the protection setting. There are only two protection settings: local and remote switching. + + :param protection_setting: Look at local or remote switching. + :return: Switching is protected or not + """ + if protection_setting not in ["local", "remote"]: + raise ValueError("Only local and remote are possible protection settings") + response = self.mprm.extract_data_from_element_uid(self.setting_uid) + if protection_setting == "local": + self.local_switching = response.get("properties").get("localSwitch") + return self.local_switching + else: + self.remote_switching = response.get("properties").get("remoteSwitch") + return self.remote_switching diff --git a/devolo_home_control_api/properties/voltage_property.py b/devolo_home_control_api/properties/voltage_property.py index 52dd33b6..ca12062a 100644 --- a/devolo_home_control_api/properties/voltage_property.py +++ b/devolo_home_control_api/properties/voltage_property.py @@ -15,3 +15,14 @@ def __init__(self, element_uid): super().__init__(element_uid=element_uid) self.current = None self.current_unit = "V" + + + def fetch_voltage(self) -> float: + """ + Update and return the current voltage. + + :return: Voltage value + """ + response = self.mprm.extract_data_from_element_uid(self.element_uid) + self.current = response.get("properties").get("value") + return self.current diff --git a/devolo_home_control_api/publisher/__init__.py b/devolo_home_control_api/publisher/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/devolo_home_control_api/publisher/publisher.py b/devolo_home_control_api/publisher/publisher.py new file mode 100644 index 00000000..4b59f2dc --- /dev/null +++ b/devolo_home_control_api/publisher/publisher.py @@ -0,0 +1,33 @@ +class Publisher: + """ + The Publisher send messages to attached subscribers. + """ + + def __init__(self, events: list): + self._events = {event: dict() for event in events} + + + def dispatch(self, event: str, message: tuple): + """ Dispatch the message to the subscribers. """ + for callback in self._get_subscribers_for_specific_event(event).values(): + callback(message) + + def register(self, event: str, who: object, callback: callable = None): + """ + As a new subscriber for an event, add a callback function to call on new message. + If no callback is given, it registers update(). + + :raises AttributeError: The supposed callback is not callable. + """ + if callback is None: + callback = getattr(who, 'update') + self._get_subscribers_for_specific_event(event)[who] = callback + + def unregister(self, event: str, who: object): + """ Remove a subscriber for a specific event. """ + del self._get_subscribers_for_specific_event(event)[who] + + + def _get_subscribers_for_specific_event(self, event: str): + """ All subscribers listening to an event. """ + return self._events[event] diff --git a/devolo_home_control_api/publisher/updater.py b/devolo_home_control_api/publisher/updater.py new file mode 100644 index 00000000..c6a8e07d --- /dev/null +++ b/devolo_home_control_api/publisher/updater.py @@ -0,0 +1,133 @@ +import json +import logging + +from ..devices.zwave import get_device_type_from_element_uid, get_device_uid_from_element_uid +from .publisher import Publisher +from ..devices.gateway import Gateway + + +class Updater: + """ + The Updater takes care of new states and values of devices and sends them to the Publisher object. + + :param devices: List of devices to await updates for + :param gateway: Instance of a Gateway object + :param publisher: Instance of a Publisher object + """ + def __init__(self, devices: dict, gateway: Gateway, publisher: Publisher): + self._logger = logging.getLogger(self.__class__.__name__) + self._devices = devices + self._gateway = gateway + self._publisher = publisher + + + def update(self, message: dict): + """ + Update states and values depending on the message type. + + :param message: Message to process + """ + message_type = {"devolo.BinarySwitch": self._binary_switch, + "devolo.mprm.gw.GatewayAccessibilityFI": self._gateway_accessible, + "devolo.Meter": self._meter, + "devolo.VoltageMultiLevelSensor": self._voltage_multi_level_sensor, + "hdm": self._device_online_state} + try: + message_type[get_device_type_from_element_uid(message.get("properties").get("uid"))](message) + except KeyError: + self._logger.debug(json.dumps(message, indent=4)) + + def update_device_online_state(self, uid: str, value: int): + """ + + :param uid: + :param value: + """ + self._logger.debug(f"Updating device online state of {uid} to {value}") + self._devices.get(uid).status = value + self._publisher.dispatch(uid, (uid, value)) + + def update_binary_switch_state(self, element_uid: str, value: bool): + """ + Update the binary switch state of a device externally. The value is written into the internal dict. + + :param element_uid: Element UID, something like, devolo.BinarySwitch:hdm:ZWave:CBC56091/24#2 + :param value: Value so be set + """ + device_uid = get_device_uid_from_element_uid(element_uid) + self._devices.get(device_uid).binary_switch_property.get(element_uid).state = value + self._logger.debug(f"Updating state of {element_uid} to {value}") + self._publisher.dispatch(device_uid, (element_uid, value)) + + def update_consumption(self, element_uid: str, consumption: str, value: float): + """ + Update the consumption of a device externally. The value is written into the internal dict. + + :param element_uid: Element UID, something like , something like devolo.MultiLevelSensor:hdm:ZWave:CBC56091/24#2 + :param consumption: current or total consumption + :param value: Value so be set + """ + device_uid = get_device_uid_from_element_uid(element_uid) + if consumption == "current": + self._devices.get(device_uid).consumption_property.get(element_uid).current = value + else: + self._devices.get(device_uid).consumption_property.get(element_uid).total = value + self._logger.debug(f"Updating {consumption} consumption of {element_uid} to {value}") + self._publisher.dispatch(device_uid, (element_uid, value)) + + def update_voltage(self, element_uid: str, value: float): + """ + Update the voltage of a device externally. The value is written into the internal dict. + + :param element_uid: Element UID, something like devolo.VoltageMultiLevelSensor:hdm:ZWave:CBC56091/24 + :param value: Value so be set + """ + device_uid = get_device_uid_from_element_uid(element_uid) + self._devices.get(device_uid).voltage_property.get(element_uid).current = value + self._logger.debug(f"Updating voltage of {element_uid} to {value}") + self._publisher.dispatch(device_uid, (element_uid, value)) + + def update_gateway_state(self, accessible: bool, online_sync: bool): + """ + Update the gateway status externally. A gateway might go on- or offline while we listen to the websocket. + + :param accessible: Online state of the gateway + :param online_sync: Sync state of the gateway + """ + self._logger.debug(f"Updating status and state of gateway to status: {accessible} and state: {online_sync}") + self._gateway.online = accessible + self._gateway.sync = online_sync + + + def _binary_switch(self, message): + """ Update a binary switch's state. """ + if message.get("properties").get("property.name") == "state": + self.update_binary_switch_state(element_uid=message.get("properties").get("uid"), + value=True if message.get("properties").get("property.value.new") == 1 + else False) + + def _device_online_state(self, message): + """ Update the device online state. """ + self.update_device_online_state(uid=message.get("properties").get("uid"), + value=message.get("properties").get("property.value.new")) + + def _gateway_accessible(self, message): + """ Update the gateway's state. """ + if message.get("properties").get("property.name") == "gatewayAccessible": + self.update_gateway_state(accessible=message.get("properties").get("property.value.new").get("accessible"), + online_sync=message.get("properties").get("property.value.new").get("onlineSync")) + + def _meter(self, message): + """ Update a meter value. """ + if message.get("properties").get("property.name") == "currentValue": + self.update_consumption(element_uid=message.get("properties").get("uid"), + consumption="current", + value=message.get("properties").get("property.value.new")) + elif message.get("properties").get("property.name") == "totalValue": + self.update_consumption(element_uid=message.get("properties").get("uid"), + consumption="total", value=message.get("properties").get("property.value.new")) + + def _voltage_multi_level_sensor(self, message): + """ Update a voltage value. """ + self.update_voltage(element_uid=message.get("properties").get("uid"), + value=message.get("properties").get("property.value.new")) diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 95c98a8a..5fdf5737 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -4,7 +4,19 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [v0.2.0] - 05.02.2020 +## [v0.3.0] - 2020/02/21 + +### Added + +- Support for Fibaro Double Relay Switch +- Lookup for Z-Wave device information +- React on device online state + +### Changed + +- **BREAKING**: The relation between the objects changed completely. + +## [v0.2.0] - 2020/02/05 ### Added @@ -18,7 +30,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), - Publisher now returns a list of element_uid and value - Rename MprmDeviceError to MprmDeviceCommunicationError -## [v0.1.0] - 31.01.2020 +## [v0.1.0] - 2020/01/31 ### Added diff --git a/docs/CONTRIBUTING.md b/docs/CONTRIBUTING.md index a0d40fca..97cccf5d 100644 --- a/docs/CONTRIBUTING.md +++ b/docs/CONTRIBUTING.md @@ -34,4 +34,4 @@ If you find code where we violated our own rules, feel free to [tell us](https:/ ## Testing -We cover our code with unit tests written in pytest, but we do not push them to hard. We want public methods covered, but we skip internal and trivial methods. If you want to contribute, please make sure to keep the unit tests green and to deliver new once, if you extend the functionality. +We cover our code with unit tests written in pytest, but we do not push them to hard. We want public methods covered, but we skip internal, nested and trivial methods. Often we also skip constructors. If you want to contribute, please make sure to keep the unit tests green and to deliver new ones, if you extend the functionality. diff --git a/example.py b/example.py index ed1d6f1b..16452c9c 100644 --- a/example.py +++ b/example.py @@ -1,9 +1,9 @@ import logging -from devolo_home_control_api.mprm_websocket import MprmWebsocket +from devolo_home_control_api.homecontrol import HomeControl from devolo_home_control_api.mydevolo import Mydevolo -logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s") +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s: %(message)s") user = "username" password = "password" @@ -17,13 +17,13 @@ def update(self, message): print(f'{self.name} got message "{message}"') -mydevolo = Mydevolo.get_instance() +mydevolo = Mydevolo() mydevolo.user = user mydevolo.password = password gateway_id = mydevolo.gateway_ids[0] -mprm_websocket = MprmWebsocket(gateway_id=gateway_id) +homecontrol = HomeControl(gateway_id=gateway_id) -for device in mprm_websocket.devices: - mprm_websocket.devices[device].subscriber = Subscriber(device) - mprm_websocket.publisher.register(device, mprm_websocket.devices[device].subscriber) +for device in homecontrol.devices: + homecontrol.devices[device].subscriber = Subscriber(device) + homecontrol.mprm.publisher.register(device, homecontrol.devices[device].subscriber) diff --git a/setup.py b/setup.py index 538c16f9..def71353 100644 --- a/setup.py +++ b/setup.py @@ -5,7 +5,7 @@ setuptools.setup( name="devolo_home_control_api", - version="0.2.0", + version="0.3.0", author="Markus Bong, Guido Schmitz", author_email="m.bong@famabo.de, guido.schmitz@fedaix.de", description="devolo Home Control API in Python", diff --git a/tests/conftest.py b/tests/conftest.py index fadc5c63..a9bb5fd7 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -3,121 +3,22 @@ import pytest -from devolo_home_control_api.mprm_rest import MprmRest -from devolo_home_control_api.mprm_websocket import MprmWebsocket -from devolo_home_control_api.mydevolo import Mydevolo, WrongUrlError -from tests.mock_dummy_device import dummy_device -from tests.mock_gateway import Gateway -from tests.mock_metering_plug import metering_plug try: - with open('test_data.json') as file: + with open("test_data.json") as file: test_data = json.load(file) except FileNotFoundError: print("Please run tests from within the tests directory.") sys.exit(127) -@pytest.fixture() -def mock_gateway(mocker): - mocker.patch("devolo_home_control_api.devices.gateway.Gateway.__init__", Gateway.__init__) - - -@pytest.fixture() -def mock_inspect_devices_metering_plug(mocker): - def mock__inspect_devices(self): - for device_type, device in test_data.get("devices").items(): - device_uid = device.get("uid") - if device_type == "mains": - self.devices[device_uid] = metering_plug(device_uid=device_uid) - else: - self.devices[device_uid] = dummy_device(key=device_type) - - mocker.patch("devolo_home_control_api.mprm_rest.MprmRest._inspect_devices", mock__inspect_devices) - - -@pytest.fixture() -def mock_mprmrest__detect_gateway_in_lan(mocker): - def _detect_gateway_in_lan(self): - return None - - mocker.patch("devolo_home_control_api.mprm_rest.MprmRest._detect_gateway_in_lan", _detect_gateway_in_lan) - - -@pytest.fixture() -def mock_mydevolo__call(mocker, request): - def _call_mock(url): - uuid = test_data.get("user").get("uuid") - gateway_id = test_data.get("gateway").get("id") - full_url = test_data.get("gateway").get("full_url") - - if url == f"https://www.mydevolo.com/v1/users/{uuid}/hc/gateways/{gateway_id}/fullURL": - return {"url": full_url} - elif url == "https://www.mydevolo.com/v1/users/uuid": - return {"uuid": uuid} - elif url == f"https://www.mydevolo.com/v1/users/{uuid}/hc/gateways/status": - if request.node.name == "test_gateway_ids_empty": - return {"items": []} - else: - return {"items": [{"gatewayId": gateway_id}]} - elif url == f"https://www.mydevolo.com/v1/users/{uuid}/hc/gateways/{gateway_id}": - return {"gatewayId": gateway_id, - "status": "devolo.hc_gateway.status.online", - "state": "devolo.hc_gateway.state.idle"} - elif url == "https://www.mydevolo.com/v1/hc/maintenance": - if request.node.name == "test_maintenance_off": - return {"state": "off"} - else: - return {"state": "on"} - else: - raise WrongUrlError(f"This URL was not mocked: {url}") - - mocker.patch("devolo_home_control_api.mydevolo.Mydevolo._call", side_effect=_call_mock) - - -@pytest.fixture() -def mock_mprmrest__extract_data_from_element_uid(mocker, request): - def _extract_data_from_element_uid(element_uid): - properties = {} - properties['test_get_binary_switch_state_valid_on'] = { - "properties": {"state": 1}} - properties['test_get_binary_switch_state_valid_off'] = { - "properties": {"state": 0}} - properties['test_get_consumption_valid'] = { - "properties": {"currentValue": test_data.get("devices").get("mains").get("current_consumption"), - "totalValue": test_data.get("devices").get("mains").get("total_consumption")}} - properties['test_get_led_setting_valid'] = { - "properties": {"led": test_data.get("devices").get("mains").get("led_setting")}} - properties['test_get_param_changed_valid'] = { - "properties": {"paramChanged": test_data.get("devices").get("mains").get("param_changed")}} - properties['test_get_general_device_settings_valid'] = { - "properties": {"eventsEnabled": test_data.get("devices").get("mains").get("events_enabled"), - "name": test_data.get("devices").get("mains").get("name"), - "icon": test_data.get("devices").get("mains").get("icon"), - "zoneID": test_data.get("devices").get("mains").get("zone_id")}} - properties['test_get_protection_setting_valid'] = { - "properties": {"localSwitch": test_data.get("devices").get("mains").get("local_switch"), - "remoteSwitch": test_data.get("devices").get("mains").get("remote_switch")}} - properties['test_get_voltage_valid'] = { - "properties": {"value": test_data.get("devices").get("mains").get("voltage")}} - properties['test_update_consumption_valid'] = { - "properties": {"currentValue": test_data.get("devices").get("mains").get("current_consumption"), - "totalValue": test_data.get("devices").get("mains").get("total_consumption")}} - return properties.get(request.node.name) - - mocker.patch("devolo_home_control_api.mprm_rest.MprmRest._extract_data_from_element_uid", - side_effect=_extract_data_from_element_uid) - - -@pytest.fixture() -def mock_mprmrest__post_set(mocker, request): - def _post_mock(data): - if request.node.name == "test_set_binary_switch_valid": - return {"result": {"status": 1}} - elif request.node.name == "test_set_binary_switch_error": - return {"result": {"status": 2}} - - mocker.patch("devolo_home_control_api.mprm_rest.MprmRest._post", side_effect=_post_mock) +pytest_plugins = ['tests.fixtures.gateway', + 'tests.fixtures.homecontrol', + 'tests.fixtures.mprm', + 'tests.fixtures.mydevolo', + 'tests.fixtures.properties', + 'tests.fixtures.publisher', + 'tests.fixtures.requests'] @pytest.fixture(autouse=True) @@ -127,33 +28,16 @@ def test_data_fixture(request): request.cls.devices = test_data.get("devices") -@pytest.fixture() -def mprm_instance(request, mocker, mock_gateway, mock_inspect_devices_metering_plug, mock_mprmrest__detect_gateway_in_lan): - if "TestMprmRest" in request.node.nodeid: - request.cls.mprm = MprmRest(test_data.get("gateway").get("id")) - else: - def _websocket_connection_mock(): - pass - - mocker.patch("devolo_home_control_api.mprm_websocket.MprmWebsocket._websocket_connection", - side_effect=_websocket_connection_mock) - request.cls.mprm = MprmWebsocket(test_data.get("gateway").get("id")) - - @pytest.fixture() def fill_device_data(request): - consumption_property = request.cls.mprm.devices.get(test_data.get('devices').get("mains").get("uid")).consumption_property + consumption_property = request.cls.homecontrol.devices.get(test_data.get('devices').get("mains").get("uid")) \ + .consumption_property consumption_property.get(f"devolo.Meter:{test_data.get('devices').get('mains').get('uid')}").current = 0.58 consumption_property.get(f"devolo.Meter:{test_data.get('devices').get('mains').get('uid')}").total = 125.68 binary_switch_property = \ - request.cls.mprm.devices.get(test_data.get('devices').get("mains").get("uid")).binary_switch_property + request.cls.homecontrol.devices.get(test_data.get('devices').get("mains").get("uid")).binary_switch_property binary_switch_property.get(f"devolo.BinarySwitch:{test_data.get('devices').get('mains').get('uid')}").state = False - voltage_property = request.cls.mprm.devices.get(test_data.get('devices').get("mains").get("uid")).voltage_property + voltage_property = request.cls.homecontrol.devices.get(test_data.get('devices').get("mains").get("uid")).voltage_property voltage_property.get(f"devolo.VoltageMultiLevelSensor:{test_data.get('devices').get('mains').get('uid')}").current = 236 - - -@pytest.fixture(autouse=True) -def clear_mydevolo(): - Mydevolo.del_instance() diff --git a/tests/fixtures/gateway.py b/tests/fixtures/gateway.py new file mode 100644 index 00000000..009ac1da --- /dev/null +++ b/tests/fixtures/gateway.py @@ -0,0 +1,8 @@ +import pytest + +from ..mocks.mock_gateway import MockGateway + + +@pytest.fixture() +def mock_gateway(mocker): + mocker.patch("devolo_home_control_api.devices.gateway.Gateway.__init__", MockGateway.__init__) diff --git a/tests/fixtures/homecontrol.py b/tests/fixtures/homecontrol.py new file mode 100644 index 00000000..d4d19833 --- /dev/null +++ b/tests/fixtures/homecontrol.py @@ -0,0 +1,26 @@ +import pytest + +from devolo_home_control_api.backend.mprm_websocket import MprmWebsocket +from devolo_home_control_api.homecontrol import HomeControl + +from ..mocks.mock_homecontrol import mock__inspect_devices + + +@pytest.fixture() +def home_control_instance(request, mydevolo, mock_gateway, mock_mprmwebsocket_websocket_connection, + mock_inspect_devices_metering_plug, mock_mprmrest__detect_gateway_in_lan): + request.cls.homecontrol = HomeControl(request.cls.gateway.get("id")) + request.cls.homecontrol.devices['hdm:ZWave:F6BF9812/4'] \ + .binary_switch_property['devolo.BinarySwitch:hdm:ZWave:F6BF9812/4'].is_online = request.cls.homecontrol.is_online + yield + MprmWebsocket.del_instance() + + +@pytest.fixture() +def mock_inspect_devices_metering_plug(mocker, mock_mydevolo__call): + mocker.patch("devolo_home_control_api.homecontrol.HomeControl._inspect_devices", mock__inspect_devices) + + +@pytest.fixture() +def mock_homecontrol_is_online(mocker): + mocker.patch("devolo_home_control_api.homecontrol.HomeControl.is_online", return_value=False) diff --git a/tests/fixtures/mprm.py b/tests/fixtures/mprm.py new file mode 100644 index 00000000..820a1b26 --- /dev/null +++ b/tests/fixtures/mprm.py @@ -0,0 +1,110 @@ +import json + +import pytest + +from devolo_home_control_api.backend.mprm_rest import MprmRest +from devolo_home_control_api.backend.mprm_websocket import MprmWebsocket + +from ..mocks.mock_websocketapp import MockWebsocketapp + + +@pytest.fixture() +def mprm_instance(request, mocker, mydevolo, mock_gateway, mock_inspect_devices_metering_plug, mock_mprmrest__detect_gateway_in_lan): + if "TestMprmRest" in request.node.nodeid: + request.cls.mprm = MprmRest(gateway_id=request.cls.gateway.get("id"), url="https://homecontrol.mydevolo.com") + elif "TestMprmWebsocket" in request.node.nodeid: + request.cls.mprm = MprmWebsocket(gateway_id=request.cls.gateway.get("id"), url="https://homecontrol.mydevolo.com") + else: + mocker.patch("devolo_home_control_api.backend.mprm_websocket.MprmWebsocket.websocket_connection", return_value=None) + request.cls.mprm = MprmWebsocket(gateway_id=request.cls.gateway.get("id"), url="https://homecontrol.mydevolo.com") + yield + request.cls.mprm.del_instance() + + +@pytest.fixture() +def mock_mprmrest_get_local_session_json_decode_error(mocker): + mocker.patch("devolo_home_control_api.backend.mprm_rest.MprmRest.get_local_session", side_effect=json.JSONDecodeError) + + +@pytest.fixture() +def mock_mprmrest_get_local_session(mocker): + mocker.patch("devolo_home_control_api.backend.mprm_rest.MprmRest.get_local_session", return_value=True) + + +@pytest.fixture() +def mock_mprmrest_get_remote_session(mocker): + mocker.patch("devolo_home_control_api.backend.mprm_rest.MprmRest.get_remote_session", return_value=True) + + +@pytest.fixture() +def mock_mprmrest__detect_gateway_in_lan(mocker, request): + if request.node.name not in ["test_detect_gateway_in_lan_valid"]: + mocker.patch("devolo_home_control_api.backend.mprm_rest.MprmRest.detect_gateway_in_lan", return_value=None) + + +@pytest.fixture() +def mock_mprmrest__extract_data_from_element_uid(mocker, request): + properties = {} + properties['test_fetch_binary_switch_state_valid_on'] = {"properties": {"state": 1}} + properties['test_fetch_binary_switch_state_valid_off'] = {"properties": {"state": 0}} + properties['test_fetch_consumption_valid'] = { + "properties": {"currentValue": request.cls.devices.get("mains").get("current_consumption"), + "totalValue": request.cls.devices.get("mains").get("total_consumption")}} + properties['test_fetch_led_setting_valid'] = { + "properties": {"led": request.cls.devices.get("mains").get("led_setting")}} + properties['test_fetch_param_changed_valid'] = { + "properties": {"paramChanged": request.cls.devices.get("mains").get("param_changed")}} + properties['test_fetch_general_device_settings_valid'] = { + "properties": {"eventsEnabled": request.cls.devices.get("mains").get("events_enabled"), + "name": request.cls.devices.get("mains").get("name"), + "icon": request.cls.devices.get("mains").get("icon"), + "zoneID": request.cls.devices.get("mains").get("zone_id")}} + properties['test_fetch_protection_setting_valid'] = { + "properties": {"localSwitch": request.cls.devices.get("mains").get("local_switch"), + "remoteSwitch": request.cls.devices.get("mains").get("remote_switch")}} + properties['test_fetch_voltage_valid'] = { + "properties": {"value": request.cls.devices.get("mains").get("voltage")}} + properties['test_update_consumption_valid'] = { + "properties": {"currentValue": request.cls.devices.get("mains").get("current_consumption"), + "totalValue": request.cls.devices.get("mains").get("total_consumption")}} + + mocker.patch("devolo_home_control_api.backend.mprm_rest.MprmRest.extract_data_from_element_uid", + return_value=properties.get(request.node.name)) + + +@pytest.fixture() +def mock_mprmrest__post(mocker, request): + properties = {} + properties["test_get_name_and_element_uids"] = {"result": {"items": [{"properties": + {"itemName": "test_name", + "zone": "test_zone", + "batteryLevel": "test_battery", + "icon": "test_icon", + "elementUIDs": "test_element_uids", + "settingUIDs": "test_setting_uids", + "deviceModelUID": "test_device_model_uid", + "status": "test_status"}}]}} + properties["test_extract_data_from_element_uid"] = {"result": {"items": [{"properties": {"itemName": "test_name"}}]}} + properties["test_get_all_devices"] = {"result": {"items": [{"properties": {"deviceUIDs": "deviceUIDs"}}]}} + + mocker.patch("devolo_home_control_api.backend.mprm_rest.MprmRest.post", return_value=properties.get(request.node.name)) + + +@pytest.fixture() +def mock_mprmrest__post_set(mocker, request): + status = {} + status['test_set_binary_switch_valid'] = {"result": {"status": 1}} + status['test_set_binary_switch_error'] = {"result": {"status": 2}} + + mocker.patch("devolo_home_control_api.backend.mprm_rest.MprmRest.post", return_value=status.get(request.node.name)) + + +@pytest.fixture() +def mock_mprmwebsocket_websocketapp(mocker): + mocker.patch("websocket.WebSocketApp.__init__", MockWebsocketapp.__init__) + mocker.patch("websocket.WebSocketApp.run_forever", MockWebsocketapp.run_forever) + + +@pytest.fixture() +def mock_mprmwebsocket_websocket_connection(mocker, request): + mocker.patch("devolo_home_control_api.backend.mprm_websocket.MprmWebsocket.websocket_connection", return_value=None) diff --git a/tests/fixtures/mydevolo.py b/tests/fixtures/mydevolo.py new file mode 100644 index 00000000..6eba25d9 --- /dev/null +++ b/tests/fixtures/mydevolo.py @@ -0,0 +1,30 @@ +import pytest + +from devolo_home_control_api.mydevolo import Mydevolo, WrongUrlError + +from ..mocks.mock_mydevolo import MockMydevolo + + +@pytest.fixture() +def mydevolo(request): + mydevolo = Mydevolo() + mydevolo._uuid = request.cls.user.get("uuid") + yield mydevolo + Mydevolo.del_instance() + + +@pytest.fixture() +def mock_mydevolo_full_url(mocker): + mocker.patch("devolo_home_control_api.mydevolo.Mydevolo.get_full_url", side_effect=MockMydevolo.get_full_url) + + +@pytest.fixture() +def mock_mydevolo__call(mocker, request): + mock_mydevolo = MockMydevolo(request) + mocker.patch("devolo_home_control_api.mydevolo.Mydevolo._call", side_effect=mock_mydevolo._call) + del mock_mydevolo + + +@pytest.fixture() +def mock_mydevolo__call_raise_WrongUrlError(mocker): + mocker.patch("devolo_home_control_api.mydevolo.Mydevolo._call", side_effect=WrongUrlError) diff --git a/tests/fixtures/properties.py b/tests/fixtures/properties.py new file mode 100644 index 00000000..fb84c4fd --- /dev/null +++ b/tests/fixtures/properties.py @@ -0,0 +1,19 @@ +import pytest + + +@pytest.fixture() +def mock_properties(mocker): + mocker.patch("devolo_home_control_api.properties.consumption_property.ConsumptionProperty.fetch_consumption", + return_value=None) + mocker.patch("devolo_home_control_api.properties.binary_switch_property.BinarySwitchProperty.fetch_binary_switch_state", + return_value=None) + mocker.patch("devolo_home_control_api.properties.voltage_property.VoltageProperty.fetch_voltage", + return_value=None) + mocker.patch("devolo_home_control_api.properties.settings_property.SettingsProperty.fetch_general_device_settings", + return_value=None) + mocker.patch("devolo_home_control_api.properties.settings_property.SettingsProperty.fetch_param_changed_setting", + return_value=None) + mocker.patch("devolo_home_control_api.properties.settings_property.SettingsProperty.fetch_protection_setting", + return_value=None) + mocker.patch("devolo_home_control_api.properties.settings_property.SettingsProperty.fetch_led_setting", + return_value=None) diff --git a/tests/fixtures/publisher.py b/tests/fixtures/publisher.py new file mode 100644 index 00000000..01f1285e --- /dev/null +++ b/tests/fixtures/publisher.py @@ -0,0 +1,6 @@ +import pytest + + +@pytest.fixture() +def mock_publisher_dispatch(mocker): + mocker.patch("devolo_home_control_api.publisher.publisher.Publisher.dispatch", return_value=None) diff --git a/tests/fixtures/requests.py b/tests/fixtures/requests.py new file mode 100644 index 00000000..e0b3ea05 --- /dev/null +++ b/tests/fixtures/requests.py @@ -0,0 +1,57 @@ +import pytest + +from ..mocks.mock_response import (MockResponseConnectTimeout, MockResponseGet, + MockResponseJsonError, MockResponsePost, + MockResponseReadTimeout) + + +@pytest.fixture() +def mock_response_json(mocker): + mocker.patch("requests.Session", return_value=MockResponseGet({"link": "test_link"}, status_code=200)) + + +@pytest.fixture() +def mock_response_requests_ConnectTimeout(mocker): + mocker.patch("requests.Session", return_value=MockResponseConnectTimeout({"link": "test_link"}, status_code=200)) + + +@pytest.fixture() +def mock_response_json_JSONDecodeError(mocker): + mocker.patch("requests.Session", return_value=MockResponseJsonError({"link": "test_link"}, status_code=200)) + + +@pytest.fixture() +def mock_response_requests_invalid_id(mocker): + mocker.patch("requests.Session", return_value=MockResponsePost({"link": "test_link"}, status_code=200)) + + +@pytest.fixture() +def mock_response_requests_valid(mocker): + mocker.patch("requests.Session", return_value=MockResponsePost({"link": "test_link"}, status_code=200)) + + +@pytest.fixture() +def mock_response_valid(mocker): + mocker.patch("requests.get", return_value=MockResponseGet({"response": "response"}, status_code=200)) + + +@pytest.fixture() +def mock_response_wrong_credentials_error(mocker): + mocker.patch("requests.get", return_value=MockResponseGet({"link": "test_link"}, status_code=403)) + + +@pytest.fixture() +def mock_response_wrong_url_error(mocker): + mocker.patch("requests.get", return_value=MockResponseGet({"link": "test_link"}, status_code=404)) + + +@pytest.fixture() +def mock_response_requests_ReadTimeout(mocker): + mocker.patch("requests.Session", return_value=MockResponseReadTimeout({"link": "test_link"}, status_code=200)) + + +@pytest.fixture() +def mock_session_post(mocker, request): + properties = {} + properties["test_get_local_session_valid"] = {"link": "test_link"} + mocker.patch("requests.Session.get", return_value=properties.get(request.node.name)) diff --git a/tests/mock_dummy_device.py b/tests/mocks/mock_dummy_device.py similarity index 63% rename from tests/mock_dummy_device.py rename to tests/mocks/mock_dummy_device.py index 1f960adb..f5900c87 100644 --- a/tests/mock_dummy_device.py +++ b/tests/mocks/mock_dummy_device.py @@ -14,12 +14,7 @@ def dummy_device(key: str) -> Zwave: with open('test_data.json') as file: test_data = json.load(file) - device = Zwave(name=test_data.get("devices").get(key).get("name"), - device_uid=test_data.get("devices").get(key).get("uid"), - zone=test_data.get("devices").get(key).get("zone_name"), - battery_level=-1, - icon=test_data.get("devices").get(key).get("icon"), - online_state=test_data.get("devices").get(key).get("online")) + device = Zwave(**test_data.get("devices").get(key)) device.binary_switch_property = {} device.binary_switch_property[f'devolo.BinarySwitch:{test_data.get("devices").get(key).get("uid")}'] = \ diff --git a/tests/mock_gateway.py b/tests/mocks/mock_gateway.py similarity index 93% rename from tests/mock_gateway.py rename to tests/mocks/mock_gateway.py index e53e5403..18fcb372 100644 --- a/tests/mock_gateway.py +++ b/tests/mocks/mock_gateway.py @@ -1,9 +1,7 @@ import json -class Gateway: - """ Represent a gateway in tests """ - +class MockGateway: def __init__(self, gateway_id: str): with open('test_data.json') as file: test_data = json.load(file) @@ -18,3 +16,4 @@ def __init__(self, gateway_id: str): self.status = test_data.get("gateway").get("status") self.state = test_data.get("gateway").get("state") self.firmware_version = test_data.get("gateway").get("firmware_version") + self.online = True diff --git a/tests/mocks/mock_homecontrol.py b/tests/mocks/mock_homecontrol.py new file mode 100644 index 00000000..8dff1c5d --- /dev/null +++ b/tests/mocks/mock_homecontrol.py @@ -0,0 +1,16 @@ +import json + +from .mock_dummy_device import dummy_device +from .mock_metering_plug import metering_plug + + +def mock__inspect_devices(self): + with open('test_data.json') as file: + test_data = json.load(file) + + for device_type, device in test_data.get("devices").items(): + device_uid = device.get("uid") + if device_type == "mains": + self.devices[device_uid] = metering_plug(device_uid=device_uid) + else: + self.devices[device_uid] = dummy_device(key=device_type) diff --git a/tests/mock_metering_plug.py b/tests/mocks/mock_metering_plug.py similarity index 77% rename from tests/mock_metering_plug.py rename to tests/mocks/mock_metering_plug.py index fa98d2a0..f002573a 100644 --- a/tests/mock_metering_plug.py +++ b/tests/mocks/mock_metering_plug.py @@ -17,12 +17,7 @@ def metering_plug(device_uid: str) -> Zwave: with open('test_data.json') as file: test_data = json.load(file) - device = Zwave(name=test_data.get("devices").get("mains").get("name"), - device_uid=test_data.get("devices").get("mains").get("uid"), - zone=test_data.get("devices").get("mains").get("zone_name"), - battery_level=test_data.get("devices").get("mains").get("battery_level"), - icon=test_data.get("devices").get("mains").get("icon"), - online_state=test_data.get("devices").get("mains").get("online")) + device = Zwave(**test_data.get("devices").get("mains")) device.binary_switch_property = {} device.consumption_property = {} diff --git a/tests/mocks/mock_mydevolo.py b/tests/mocks/mock_mydevolo.py new file mode 100644 index 00000000..f9cef641 --- /dev/null +++ b/tests/mocks/mock_mydevolo.py @@ -0,0 +1,46 @@ +class MockMydevolo: + @staticmethod + def get_full_url(gateway_id): + return gateway_id + + + def __init__(self, request): + self._request = request + + + def _call(self, url): + uuid = self._request.cls.user.get("uuid") + gateway_id = self._request.cls.gateway.get("id") + full_url = self._request.cls.gateway.get("full_url") + + response = {} + response[f'https://www.mydevolo.com/v1/users/{uuid}/hc/gateways/{gateway_id}/fullURL'] = {"url": full_url} + response['https://www.mydevolo.com/v1/users/uuid'] = {"uuid": uuid} + response[f'https://www.mydevolo.com/v1/users/{uuid}/hc/gateways/status'] = {"items": []} \ + if self._request.node.name == "test_gateway_ids_empty" else {"items": [{"gatewayId": gateway_id}]} + response[f'https://www.mydevolo.com/v1/users/{uuid}/hc/gateways/{gateway_id}'] = \ + {"gatewayId": gateway_id, "status": "devolo.hc_gateway.status.online", "state": "devolo.hc_gateway.state.idle"} + response['https://www.mydevolo.com/v1/hc/maintenance'] = {"state": "off"} \ + if self._request.node.name == "test_maintenance_off" else {"state": "on"} + response['https://www.mydevolo.com/v1/zwave/products/0x0060/0x0001/0x000'] = {"brand": "Everspring", + "deviceType": "Door Lock Keypad", + "genericDeviceClass": "Entry Control", + "identifier": "SP814-US", + "isZWavePlus": True, + "manufacturerId": "0x0060", + "name": "Everspring PIR Sensor SP814", + "productId": "0x0002", + "productTypeId": "0x0001", + "zwaveVersion": "6.51.07"} + response['https://www.mydevolo.com/v1/zwave/products/0x0175/0x0001/0x0011'] = {"manufacturerId": "0x0175", + "productTypeId": "0x0001", + "productId": "0x0011", + "name": "Metering Plug", + "brand": "devolo", + "identifier": "MT02646", + "isZWavePlus": True, + "deviceType": "On/Off Power Switch", + "zwaveVersion": "6.51.00", + "specificDeviceClass": None, + "genericDeviceClass": None} + return response[url] diff --git a/tests/mocks/mock_response.py b/tests/mocks/mock_response.py new file mode 100644 index 00000000..b2c0a9e3 --- /dev/null +++ b/tests/mocks/mock_response.py @@ -0,0 +1,40 @@ +from json import JSONDecodeError + +from requests import ConnectTimeout, ReadTimeout + + +class MockResponse: + def __init__(self, json_data, status_code): + self.json_data = json_data + self.status_code = status_code + + +class MockResponseConnectTimeout(MockResponse): + def get(self, url, auth=None, timeout=None): + raise ConnectTimeout + + +class MockResponseGet(MockResponse): + def get(self, url, auth=None, timeout=None): + return MockResponseGet({"link": "test_link"}, status_code=200) + + def json(self): + return self.json_data + + +class MockResponseJsonError(MockResponse): + def get(self, url, auth=None, timeout=None): + raise JSONDecodeError(msg="message", doc="doc", pos=1) + + +class MockResponsePost(MockResponse): + def post(self, url, auth=None, timeout=None, headers=None, data=None): + return MockResponsePost({"link": "test_link"}, status_code=200) + + def json(self): + return {"id": 2} + + +class MockResponseReadTimeout(MockResponse): + def post(self, url, auth=None, timeout=None, headers=None, data=None): + raise ReadTimeout diff --git a/tests/mocks/mock_websocket.py b/tests/mocks/mock_websocket.py new file mode 100644 index 00000000..a4fa1e51 --- /dev/null +++ b/tests/mocks/mock_websocket.py @@ -0,0 +1,3 @@ +class MockWebsocket: + def close(self): + pass diff --git a/tests/mocks/mock_websocketapp.py b/tests/mocks/mock_websocketapp.py new file mode 100644 index 00000000..b4821cec --- /dev/null +++ b/tests/mocks/mock_websocketapp.py @@ -0,0 +1,8 @@ +class MockWebsocketapp: + def __init__(self, ws_url, **kwargs): + print("init") + pass + + def run_forever(self, **kwargs): + print("running") + raise AssertionError \ No newline at end of file diff --git a/tests/requirements.txt b/tests/requirements.txt new file mode 100644 index 00000000..73224b00 --- /dev/null +++ b/tests/requirements.txt @@ -0,0 +1,2 @@ +pytest==4.5.0 +pytest-mock==1.11.2 diff --git a/tests/test_binary_switch_property.py b/tests/test_binary_switch_property.py new file mode 100644 index 00000000..27f61fe0 --- /dev/null +++ b/tests/test_binary_switch_property.py @@ -0,0 +1,25 @@ +import pytest + +from devolo_home_control_api.backend.mprm_rest import MprmDeviceCommunicationError + + +@pytest.mark.usefixtures("home_control_instance") +class TestBinarySwitchProperty: + def test_fetch_binary_switch_state_valid_on(self, mock_mprmrest__extract_data_from_element_uid): + assert self.homecontrol.devices.get(self.devices.get("mains").get("uid"))\ + .binary_switch_property.get(self.devices.get("mains").get("elementUIDs")[1]).fetch_binary_switch_state() + + def test_fetch_binary_switch_state_valid_off(self, mock_mprmrest__extract_data_from_element_uid): + assert not self.homecontrol.devices.get(self.devices.get("mains").get("uid"))\ + .binary_switch_property.get(self.devices.get("mains").get("elementUIDs")[1]).fetch_binary_switch_state() + + def test_set_binary_switch_valid(self, mock_mprmrest__post_set): + self.homecontrol.devices.get(self.devices.get("mains").get("uid"))\ + .binary_switch_property.get(self.devices.get("mains").get("elementUIDs")[1]).set_binary_switch(True) + assert self.homecontrol.devices.get(self.devices.get("mains").get("uid"))\ + .binary_switch_property.get(self.devices.get("mains").get("elementUIDs")[1]).state + + def test_set_binary_switch_error(self, mock_mprmrest__post_set, mock_homecontrol_is_online): + with pytest.raises(MprmDeviceCommunicationError): + self.homecontrol.devices[self.devices.get("ambiguous_2").get("uid")]\ + .binary_switch_property[self.devices.get("ambiguous_2").get("elementUIDs")[1]].set_binary_switch(True) diff --git a/tests/test_consumption_property.py b/tests/test_consumption_property.py new file mode 100644 index 00000000..5113e9c1 --- /dev/null +++ b/tests/test_consumption_property.py @@ -0,0 +1,19 @@ +import pytest + + +@pytest.mark.usefixtures("home_control_instance") +class TestConsumption: + def test_fetch_consumption_invalid(self): + with pytest.raises(ValueError): + self.homecontrol.devices.get(self.devices.get("mains").get("uid"))\ + .consumption_property.get(self.devices.get("mains").get("elementUIDs")[0]).fetch_consumption("invalid") + + def test_fetch_consumption_valid(self, mock_mprmrest__extract_data_from_element_uid): + current = self.homecontrol.devices.get(self.devices.get("mains").get("uid"))\ + .consumption_property.get(self.devices.get("mains").get("elementUIDs")[0])\ + .fetch_consumption(consumption_type="current") + total = self.homecontrol.devices.get(self.devices.get("mains").get("uid"))\ + .consumption_property.get(self.devices.get("mains").get("elementUIDs")[0])\ + .fetch_consumption(consumption_type="total") + assert current == 0.58 + assert total == 125.68 diff --git a/tests/test_data.json b/tests/test_data.json index 903459eb..b52069c5 100644 --- a/tests/test_data.json +++ b/tests/test_data.json @@ -13,24 +13,28 @@ "external_access": true, "online": true, "sync": true, - "firmware": "8.0.45_2016-11-17" + "firmware": "8.0.45_2016-11-17", + "local_ip": "123.456.789.123" }, "devices": { "mains":{ "uid": "hdm:ZWave:F6BF9812/2", - "name": "Light", + "itemName": "Light", + "batteryLevel": -1, + "batteryLow": null, "icon": "light-bulb", "zone_id": "hz_2", "zone_name": "Office", "battery_level": -1, - "settings_uids": [ + "settingsUIDs": [ "gds.hdm:ZWave:F6BF9812/2", "cps.hdm:ZWave:F6BF9812/2", "lis.hdm:ZWave:F6BF9812/2", "ps.hdm:ZWave:F6BF9812/2"], - "element_uids": [ + "elementUIDs": [ "devolo.Meter:hdm:ZWave:F6BF9812/2", - "devolo.BinarySwitch:hdm:ZWave:F6BF9812/2"], + "devolo.BinarySwitch:hdm:ZWave:F6BF9812/2", + "devolo.VoltageMultiLevelSensor:hdm:ZWave:F6BF9812/2"], "state": 1, "current_consumption": 0.58, "total_consumption": 125.68, @@ -40,23 +44,38 @@ "local_switch": true, "remote_switch": false, "voltage": 237, - "online": 2 + "status": 2, + "manID": "0x0175", + "prodTypeID": "0x0001", + "prodID": "0x0011" }, "ambiguous_1":{ "uid": "hdm:ZWave:F6BF9812/3", - "name": "Heating", + "itemName": "Heating", + "batteryLevel": 55, + "batteryLow": true, + "elementUIDs": ["devolo.Meter:hdm:ZWave:F6BF9812/3"], "icon": "icon_1", "zone_name": "Kitchen", - "online": 2 + "status": 25, + "manID": "0x0175", + "prodTypeID": "0x0001", + "prodID": "0x0011" }, "ambiguous_2":{ "uid": "hdm:ZWave:F6BF9812/4", - "name": "Heating", + "itemName": "Heating", + "batteryLevel": -1, + "batteryLow": null, + "elementUIDs": [ + "devolo.Meter:hdm:ZWave:F6BF9812/4", + "devolo.BinarySwitch:hdm:ZWave:F6BF9812/4"], "icon": "icon_1", "zone_name": "Living Room", - "element_uids": [ - "devolo.BinarySwitch:hdm:ZWave:F6BF9812/4"], - "online": 1 + "status": 1, + "manID": "0x0175", + "prodTypeID": "0x0001", + "prodID": "0x0011" } } } diff --git a/tests/test_gateway.py b/tests/test_gateway.py index 747c9356..1f7d89e4 100644 --- a/tests/test_gateway.py +++ b/tests/test_gateway.py @@ -3,6 +3,7 @@ from devolo_home_control_api.devices.gateway import Gateway +@pytest.mark.usefixtures("mydevolo") @pytest.mark.usefixtures("mock_mydevolo__call") class TestGateway: def test_update_state_known(self): @@ -12,7 +13,14 @@ def test_update_state_known(self): assert not gateway.online assert not gateway.sync - def test_update_state_unknow(self): + def test_update_state_offline(self): + gateway = Gateway(self.gateway.get("id")) + gateway._update_state(status="devolo.hc_gateway.status.offline", state="devolo.hc_gateway.state.update") + + assert not gateway.online + assert not gateway.sync + + def test_update_state_unknown(self): gateway = Gateway(self.gateway.get("id")) gateway.online = False gateway.sync = False @@ -20,3 +28,8 @@ def test_update_state_unknow(self): assert gateway.online assert gateway.sync + + @pytest.mark.usefixtures("mock_mydevolo_full_url") + def test_full_url(self): + gateway = Gateway(self.gateway.get("id")) + assert gateway.full_url == self.gateway.get("id") diff --git a/tests/test_homecontrol.py b/tests/test_homecontrol.py new file mode 100644 index 00000000..d3ecb060 --- /dev/null +++ b/tests/test_homecontrol.py @@ -0,0 +1,60 @@ +import pytest + +from devolo_home_control_api.homecontrol import get_sub_device_uid_from_element_uid + + +@pytest.mark.usefixtures("mock_inspect_devices_metering_plug") +@pytest.mark.usefixtures("home_control_instance") +@pytest.mark.usefixtures("mock_mydevolo__call") +class TestHomeControl: + + def test_binary_switch_devices(self): + assert hasattr(self.homecontrol.binary_switch_devices[0], "binary_switch_property") + + def test_get_publisher(self): + assert len(self.homecontrol.publisher._events) == 3 + + def test_is_online(self): + assert self.homecontrol.is_online(self.devices.get("mains").get("uid")) + assert not self.homecontrol.is_online(self.devices.get("ambiguous_2").get("uid")) + + def test_get_sub_device_uid_from_element_uid(self): + assert get_sub_device_uid_from_element_uid("devolo.Meter:hdm:ZWave:F6BF9812/2#2") == 2 + assert get_sub_device_uid_from_element_uid("devolo.Meter:hdm:ZWave:F6BF9812/2") is None + + def test_process_element_uids(self, mock_properties): + device = self.devices.get("mains").get("uid") + element_uids = self.devices.get("mains").get("elementUIDs") + del self.homecontrol.devices['hdm:ZWave:F6BF9812/2'].binary_switch_property + del self.homecontrol.devices['hdm:ZWave:F6BF9812/2'].consumption_property + del self.homecontrol.devices['hdm:ZWave:F6BF9812/2'].voltage_property + assert not hasattr(self.homecontrol.devices['hdm:ZWave:F6BF9812/2'], "binary_switch_property") + assert not hasattr(self.homecontrol.devices['hdm:ZWave:F6BF9812/2'], "consumption_property") + assert not hasattr(self.homecontrol.devices['hdm:ZWave:F6BF9812/2'], "voltage_property") + self.homecontrol._process_element_uids(device, element_uids) + assert hasattr(self.homecontrol.devices['hdm:ZWave:F6BF9812/2'], "binary_switch_property") + assert hasattr(self.homecontrol.devices['hdm:ZWave:F6BF9812/2'], "consumption_property") + assert hasattr(self.homecontrol.devices['hdm:ZWave:F6BF9812/2'], "voltage_property") + + def test_process_element_uids_invalid(self): + device = self.devices.get("mains").get("uid") + element_uids = ['fibaro:hdm:ZWave:F6BF9812/2'] + self.homecontrol._process_element_uids(device, element_uids) + + def test_process_settings_uids(self, mock_properties): + device = self.devices.get("mains").get("uid") + setting_uids = self.devices.get("mains").get("settingsUIDs") + self.homecontrol._process_settings_uids(device, setting_uids) + + def test_process_settings_uids_invalid(self): + device = self.devices.get("mains").get("uid") + settings_uids = ['fibaro:hdm:ZWave:F6BF9812/2'] + self.homecontrol._process_settings_uids(device, settings_uids) + + def test_process_settings_property_empty(self, mock_properties): + del self.homecontrol.devices['hdm:ZWave:F6BF9812/2'].settings_property + assert not hasattr(self.homecontrol.devices['hdm:ZWave:F6BF9812/2'], "settings_property") + device = self.devices.get("mains").get("uid") + setting_uids = self.devices.get("mains").get("settingsUIDs") + self.homecontrol._process_settings_uids(device, setting_uids) + assert len(self.homecontrol.devices['hdm:ZWave:F6BF9812/2'].settings_property) > 0 diff --git a/tests/test_mprm_rest.py b/tests/test_mprm_rest.py index 8d1141c8..ee25a9aa 100644 --- a/tests/test_mprm_rest.py +++ b/tests/test_mprm_rest.py @@ -1,125 +1,100 @@ import pytest +from requests import ConnectTimeout -from devolo_home_control_api.mprm_rest import MprmDeviceCommunicationError, MprmDeviceNotFoundError +from devolo_home_control_api.backend.mprm_rest import MprmDeviceCommunicationError, MprmRest +from devolo_home_control_api.mydevolo import Mydevolo @pytest.mark.usefixtures("mprm_instance") -@pytest.mark.usefixtures("mock_mprmrest__extract_data_from_element_uid") -@pytest.mark.usefixtures("mock_mydevolo__call") class TestMprmRest: - def test_binary_switch_devices(self): - assert hasattr(self.mprm.binary_switch_devices[0], "binary_switch_property") - def test_get_binary_switch_state_invalid(self): - with pytest.raises(ValueError): - self.mprm.get_binary_switch_state("invalid") - - def test_get_binary_switch_state_valid_on(self): - assert self.mprm.get_binary_switch_state(element_uid=f"devolo.BinarySwitch:{self.devices.get('mains').get('uid')}") - - def test_get_binary_switch_state_valid_off(self): - assert not self.mprm.get_binary_switch_state(element_uid=f"devolo.BinarySwitch:{self.devices.get('mains').get('uid')}") - - def test_get_device_uid_from_name_unique(self): - uid = self.mprm.get_device_uid_from_name(name=self.devices.get("mains").get("name")) - assert uid == self.devices.get("mains").get("uid") - - def test_get_device_uid_from_name_unknow(self): - with pytest.raises(MprmDeviceNotFoundError): - self.mprm.get_device_uid_from_name(name="unknown") - - def test_get_device_uid_from_name_ambiguous(self): - with pytest.raises(MprmDeviceNotFoundError): - self.mprm.get_device_uid_from_name(name=self.devices.get("ambiguous_1").get("name")) - - def test_get_device_uid_from_name_ambiguous_with_zone(self): - uid = self.mprm.get_device_uid_from_name(name=self.devices.get("ambiguous_1").get("name"), - zone=self.devices.get("ambiguous_1").get("zone_name")) - assert uid == self.devices.get("ambiguous_1").get("uid") - - def test_get_device_uid_from_name_ambiguous_with_zone_invalid(self): - with pytest.raises(MprmDeviceNotFoundError): - self.mprm.get_device_uid_from_name(name=self.devices.get("ambiguous_1").get("name"), zone="invalid") - - def test_get_consumption_invalid(self): - with pytest.raises(ValueError): - self.mprm.get_consumption(element_uid="invalid", consumption_type="invalid") - with pytest.raises(ValueError): - self.mprm.get_consumption(element_uid="devolo.Meter:", consumption_type="invalid") - - def test_get_consumption_valid(self): - current = self.mprm.get_consumption(element_uid=f"devolo.Meter:{self.devices.get('mains').get('uid')}", - consumption_type="current") - total = self.mprm.get_consumption(element_uid=f"devolo.Meter:{self.devices.get('mains').get('uid')}", - consumption_type="total") - assert current == 0.58 - assert total == 125.68 - - def test_get_general_device_settings_invalid(self): - with pytest.raises(ValueError): - self.mprm.get_general_device_settings(setting_uid="invalid") - - def test_get_general_device_settings_valid(self): - name, icon, zone_id, events_enabled = \ - self.mprm.get_general_device_settings(setting_uid=f"gds.{self.devices.get('mains').get('uid')}") - assert name == self.devices.get('mains').get('name') - assert icon == self.devices.get('mains').get('icon') - assert zone_id == self.devices.get('mains').get('zone_id') - assert events_enabled - - def test_get_led_setting_invalid(self): - with pytest.raises(ValueError): - self.mprm.get_led_setting("invalid") - - def test_get_led_setting_valid(self): - assert self.mprm.get_led_setting(setting_uid=f"lis.{self.devices.get('mains').get('uid')}") - - def test_get_param_changed_invalid(self): - with pytest.raises(ValueError): - self.mprm.get_param_changed_setting(setting_uid="invalid") - - def test_get_param_changed_valid(self): - assert not self.mprm.get_param_changed_setting(setting_uid=f"cps.{self.devices.get('mains').get('uid')}") + def test_get_name_and_element_uids(self, mock_mprmrest__extract_data_from_element_uid, mock_mprmrest__post): + properties = self.mprm.get_name_and_element_uids("test") + assert properties == {"itemName": "test_name", + "zone": "test_zone", + "batteryLevel": "test_battery", + "icon": "test_icon", + "elementUIDs": "test_element_uids", + "settingUIDs": "test_setting_uids", + "deviceModelUID": "test_device_model_uid", + "status": "test_status"} + + def test_singleton(self): + MprmRest.del_instance() + + with pytest.raises(SyntaxError): + MprmRest.get_instance() + + first = MprmRest(gateway_id=self.gateway.get("id"), url="https://homecontrol.mydevolo.com") + with pytest.raises(SyntaxError): + MprmRest(gateway_id=self.gateway.get("id"), url="https://homecontrol.mydevolo.com") + + second = MprmRest.get_instance() + assert first is second + + def test_create_connection_local(self, mock_mprmrest_get_local_session): + self.mprm._local_ip = "123.456.789.123" + self.mprm.create_connection() + + def test_create_connection_remote(self, mock_mprmrest_get_remote_session, mydevolo): + self.mprm._gateway.external_access = True + self._mydevolo = Mydevolo.get_instance() + self.mprm.create_connection() + + def test_create_connection_invalid(self): + with pytest.raises(ConnectionError): + self.mprm._gateway.external_access = False + self.mprm.create_connection() + + def test_extract_data_from_element_uid(self, mock_mprmrest__post): + properties = self.mprm.extract_data_from_element_uid(uid="test") + assert properties.get("properties").get("itemName") == "test_name" + + def test_get_all_devices(self, mock_mprmrest__post): + devices = self.mprm.get_all_devices() + assert devices == "deviceUIDs" + + @pytest.mark.usefixtures("mock_session_post") + @pytest.mark.usefixtures("mock_response_json") + def test_get_local_session_valid(self): + self.mprm._local_ip = self.gateway.get("local_ip") + self.mprm.get_local_session() + + @pytest.mark.usefixtures("mock_response_requests_ConnectTimeout") + def test_get_local_session_ConnectTimeout(self): + self.mprm._local_ip = self.gateway.get("local_ip") + with pytest.raises(ConnectTimeout): + self.mprm.get_local_session() + + @pytest.mark.usefixtures("mock_response_json_JSONDecodeError") + def test_get_local_session_JSONDecodeError(self): + self.mprm._local_ip = self.gateway.get("local_ip") + with pytest.raises(MprmDeviceCommunicationError): + self.mprm.get_local_session() - def test_get_protection_setting_invalid(self): - with pytest.raises(ValueError): - self.mprm.get_protection_setting(setting_uid="invalid", protection_setting="local") + @pytest.mark.usefixtures("mock_response_json_JSONDecodeError") + def test_get_remote_session_JSONDecodeError(self): + with pytest.raises(MprmDeviceCommunicationError): + self.mprm.get_remote_session() - with pytest.raises(ValueError): - self.mprm.get_protection_setting(setting_uid=f"ps.{self.devices.get('mains').get('uid')}", - protection_setting="invalid") - - def test_get_protection_setting_valid(self): - local_switch = self.mprm.get_protection_setting(setting_uid=f"ps.{self.devices.get('mains').get('uid')}", - protection_setting="local") - remote_switch = self.mprm.get_protection_setting(setting_uid=f"ps.{self.devices.get('mains').get('uid')}", - protection_setting="remote") - assert local_switch - assert not remote_switch - - def test_get_voltage_invalid(self): - with pytest.raises(ValueError): - self.mprm.get_voltage(element_uid="invalid") + @pytest.mark.usefixtures("mock_response_requests_ReadTimeout") + def test_post_ReadTimeOut(self): + with pytest.raises(MprmDeviceCommunicationError): + self.mprm.post({"data": "test"}) - def test_get_voltage_valid(self): - voltage = self.mprm.get_voltage(element_uid=f"devolo.VoltageMultiLevelSensor:{self.devices.get('mains').get('uid')}") - assert voltage == 237 + def test_post_gateway_offline(self): + self.mprm._gateway.online = False + self.mprm._gateway.sync = False + self.mprm._gateway.local_connection = False + with pytest.raises(MprmDeviceCommunicationError): + self.mprm.post({"data": "test"}) - def test_set_binary_switch_invalid(self): + @pytest.mark.usefixtures("mock_response_requests_invalid_id") + def test_post_invalid_id(self): + self.mprm._data_id = 0 with pytest.raises(ValueError): - self.mprm.set_binary_switch(element_uid="invalid", state=True) - with pytest.raises(ValueError): - self.mprm.set_binary_switch(element_uid=f"devolo.BinarySwitch:{self.devices.get('mains').get('uid')}", - state="invalid") - - @pytest.mark.usefixtures("mock_mprmrest__post_set") - def test_set_binary_switch_valid(self): - element_uid = f"devolo.BinarySwitch:{self.devices.get('mains').get('uid')}" - self.mprm.set_binary_switch(element_uid=element_uid, state=True) - assert self.mprm.devices.get(self.devices.get('mains').get('uid')).binary_switch_property.get(element_uid).state + self.mprm.post({"data": "test"}) - @pytest.mark.usefixtures("mock_mprmrest__post_set") - def test_set_binary_switch_error(self): - with pytest.raises(MprmDeviceCommunicationError): - element_uid = f"devolo.BinarySwitch:{self.devices.get('ambiguous_2').get('uid')}" - self.mprm.set_binary_switch(element_uid=element_uid, state=True) + def test_post_valid(self, mock_response_requests_valid): + self.mprm._data_id = 1 + assert self.mprm.post({"data": "test"}).get("id") == 2 diff --git a/tests/test_mprm_websocket.py b/tests/test_mprm_websocket.py index 4f5a933a..8dc8c13b 100644 --- a/tests/test_mprm_websocket.py +++ b/tests/test_mprm_websocket.py @@ -1,70 +1,54 @@ +import threading +import time + import pytest +from .mocks.mock_websocket import MockWebsocket + @pytest.mark.usefixtures("mprm_instance") -@pytest.mark.usefixtures("mock_mprmrest__extract_data_from_element_uid") -@pytest.mark.usefixtures("mock_mydevolo__call") class TestMprmWebsocket: - def test_get_consumption_invalid(self): - with pytest.raises(ValueError): - self.mprm.get_consumption("invalid", "invalid") - with pytest.raises(ValueError): - self.mprm.get_consumption("devolo.Meter:", "invalid") - - def test_get_consumption_valid(self, fill_device_data): - current = self.mprm.get_consumption(f"devolo.Meter:{self.devices.get('mains').get('uid')}", "current") - total = self.mprm.get_consumption(f"devolo.Meter:{self.devices.get('mains').get('uid')}", "total") - assert current == self.devices.get('mains').get('current_consumption') - assert total == self.devices.get('mains').get('total_consumption') - - def test_get_binary_switch_state_invalid(self): - with pytest.raises(ValueError): - self.mprm.get_binary_switch_state("invalid") - - def test_update_binary_switch_state_invalid(self): - with pytest.raises(ValueError): - self.mprm.update_binary_switch_state("invalid") - - def test_update_binary_switch_state_valid(self, fill_device_data): - binary_switch_property = self.mprm.devices.get(self.devices.get("mains").get("uid")).binary_switch_property - state = binary_switch_property.get(f"devolo.BinarySwitch:{self.devices.get('mains').get('uid')}").state - self.mprm.update_binary_switch_state(f"devolo.BinarySwitch:{self.devices.get('mains').get('uid')}", value=True) - assert state != binary_switch_property.get(f"devolo.BinarySwitch:{self.devices.get('mains').get('uid')}").state - - def test_update_consumption_invalid(self): - with pytest.raises(ValueError): - self.mprm.update_consumption("invalid", "current") - with pytest.raises(ValueError): - self.mprm.update_consumption("devolo.Meter", "invalid") - - def test_update_consumption_valid(self, fill_device_data): - consumption_property = self.mprm.devices.get(self.devices.get("mains").get("uid")).consumption_property - current_before = consumption_property.get(f"devolo.Meter:{self.devices.get('mains').get('uid')}").current - total_before = consumption_property.get(f"devolo.Meter:{self.devices.get('mains').get('uid')}").total - self.mprm.update_consumption(element_uid=f"devolo.Meter:{self.devices.get('mains').get('uid')}", - consumption="current", value=1.58) - self.mprm.update_consumption(element_uid=f"devolo.Meter:{self.devices.get('mains').get('uid')}", - consumption="total", value=254) - assert current_before != consumption_property.get(f"devolo.Meter:{self.devices.get('mains').get('uid')}").current - assert total_before != consumption_property.get(f"devolo.Meter:{self.devices.get('mains').get('uid')}").total - assert consumption_property.get(f"devolo.Meter:{self.devices.get('mains').get('uid')}").current == 1.58 - assert consumption_property.get(f"devolo.Meter:{self.devices.get('mains').get('uid')}").total == 254 - - def test_update_gateway_state(self): - self.mprm.update_gateway_state(accessible=True, online_sync=False) - assert self.mprm._gateway.online - assert not self.mprm._gateway.sync - - def test_update_voltage_valid(self, fill_device_data): - voltage_property = self.mprm.devices.get(self.devices.get("mains").get("uid")).voltage_property - current_voltage = \ - voltage_property.get(f"devolo.VoltageMultiLevelSensor:{self.devices.get('mains').get('uid')}").current - self.mprm.update_voltage(element_uid=f"devolo.VoltageMultiLevelSensor:{self.devices.get('mains').get('uid')}", - value=257) - assert current_voltage != \ - voltage_property.get(f"devolo.VoltageMultiLevelSensor:{self.devices.get('mains').get('uid')}").current - assert voltage_property.get(f"devolo.VoltageMultiLevelSensor:{self.devices.get('mains').get('uid')}").current == 257 - def test_update_voltage_invalid(self): - with pytest.raises(ValueError): - self.mprm.update_voltage(element_uid="invalid", value=123) + def test_websocket_connection(self, mock_mprmwebsocket_websocketapp): + with pytest.raises(AssertionError): + self.mprm.websocket_connection() + + def test__on_message(self): + message = '{"properties": {"com.prosyst.mbs.services.remote.event.sequence.number": 0}}' + self.mprm.on_update = lambda: AssertionError() + try: + self.mprm._on_message(message) + assert False + except AssertionError: + assert True + + def test__on_message_event_sequence(self): + event_sequence = self.mprm._event_sequence + message = '{"properties": {"com.prosyst.mbs.services.remote.event.sequence.number": 5}}' + self.mprm._on_message(message) + assert event_sequence != self.mprm._event_sequence + assert self.mprm._event_sequence == 5 + + def test__on_update_not_set(self): + # TypeError should be caught by _on_message + self.mprm.on_update = None + message = '{"properties": {"com.prosyst.mbs.services.remote.event.sequence.number": 0}}' + self.mprm._on_message(message) + + def test__on_error(self, mock_mprmrest_get_remote_session, mock_mprmwebsocket_websocket_connection): + self.mprm._ws = MockWebsocket() + self.mprm._on_error("error") + + def test__on_error_errors(self, mock_mprmrest_get_remote_session, mock_mprmwebsocket_websocket_connection, + mock_mprmrest_get_local_session_json_decode_error): + self.mprm._ws = MockWebsocket() + + self.mprm._local_ip = "123.456.789.123" + threading.Thread(target=self.mprm._on_error).start() + # local ip is set --> self.get_local_session() will throw an error because of the fixture + # mock_get_local_session_json_decode_error. + # After first run we remove the local ip and self.get_remote_session() will pass. + time.sleep(2) + self.mprm._local_ip = None + + self.mprm.websocket_connection = lambda: None diff --git a/tests/test_mydevolo.py b/tests/test_mydevolo.py index b7aaa11d..b4d7ebe2 100644 --- a/tests/test_mydevolo.py +++ b/tests/test_mydevolo.py @@ -1,51 +1,43 @@ import pytest -from devolo_home_control_api.mydevolo import Mydevolo +from devolo_home_control_api.mydevolo import Mydevolo, WrongUrlError, WrongCredentialsError class TestMydevolo: - - def test_gateway_ids(self, mock_mydevolo__call): - mydevolo = Mydevolo.get_instance() - mydevolo._uuid = self.user.get("uuid") - + def test_gateway_ids(self, mydevolo, mock_mydevolo__call): assert mydevolo.gateway_ids == [self.gateway.get("id")] - def test_gateway_ids_empty(self, mock_mydevolo__call): - mydevolo = Mydevolo.get_instance() - mydevolo._uuid = self.user.get("uuid") - + def test_gateway_ids_empty(self, mydevolo, mock_mydevolo__call): with pytest.raises(IndexError): mydevolo.gateway_ids - def test_get_full_url(self, mock_mydevolo__call): - - mydevolo = Mydevolo.get_instance() - mydevolo._uuid = self.user.get("uuid") - + def test_get_full_url(self, mydevolo, mock_mydevolo__call): full_url = mydevolo.get_full_url(self.gateway.get("id")) - assert full_url == self.gateway.get("full_url") - def test_get_gateway(self, mock_mydevolo__call): - mydevolo = Mydevolo.get_instance() - mydevolo._uuid = self.user.get("uuid") - + def test_get_gateway(self, mydevolo, mock_mydevolo__call): details = mydevolo.get_gateway(self.gateway.get("id")) - assert details.get("gatewayId") == self.gateway.get("id") - def test_maintenance_on(self, mock_mydevolo__call): - mydevolo = Mydevolo.get_instance() + def test_get_gateway_invalid(self, mydevolo, mock_mydevolo__call_raise_WrongUrlError): + with pytest.raises(WrongUrlError): + mydevolo.get_gateway(self.gateway.get("id")) + + def test_get_zwave_products(self, mydevolo, mock_mydevolo__call): + product = mydevolo.get_zwave_products(manufacturer="0x0060", product_type="0x0001", product="0x000") + assert product.get("name") == "Everspring PIR Sensor SP814" + + def test_get_zwave_products_invalid(self, mydevolo, mock_mydevolo__call_raise_WrongUrlError): + device_infos = mydevolo.get_zwave_products(manufacturer="0x0070", product_type="0x0001", product="0x000") + assert len(device_infos) == 0 + + def test_maintenance_on(self, mydevolo, mock_mydevolo__call): assert not mydevolo.maintenance - def test_maintenance_off(self, mock_mydevolo__call): - mydevolo = Mydevolo.get_instance() + def test_maintenance_off(self, mydevolo, mock_mydevolo__call): assert mydevolo.maintenance - def test_set_password(self): - mydevolo = Mydevolo.get_instance() - mydevolo._uuid = self.user.get("uuid") + def test_set_password(self, mydevolo): mydevolo._gateway_ids = [self.gateway.get("id")] mydevolo.password = self.user.get("password") @@ -53,9 +45,7 @@ def test_set_password(self): assert mydevolo._uuid is None assert mydevolo._gateway_ids == [] - def test_set_user(self): - mydevolo = Mydevolo.get_instance() - mydevolo._uuid = self.user.get("uuid") + def test_set_user(self, mydevolo): mydevolo._gateway_ids = [self.gateway.get("id")] mydevolo.user = self.user.get("username") @@ -63,12 +53,40 @@ def test_set_user(self): assert mydevolo._uuid is None assert mydevolo._gateway_ids == [] + def test_get_user(self, mydevolo): + mydevolo.user = self.user.get("username") + assert mydevolo.user == self.user.get("username") + + def test_get_password(self, mydevolo): + mydevolo.password = self.user.get("password") + assert mydevolo.password == self.user.get("password") + def test_singleton_mydevolo(self): - Mydevolo.get_instance() + with pytest.raises(SyntaxError): + Mydevolo.get_instance() + first = Mydevolo() with pytest.raises(SyntaxError): Mydevolo() - def test_uuid(self, mock_mydevolo__call): - mydevolo = Mydevolo.get_instance() + second = Mydevolo.get_instance() + assert first is second + Mydevolo.del_instance() + + def test_uuid(self, mydevolo, mock_mydevolo__call): assert mydevolo.uuid == self.user.get("uuid") + + def test_call_WrongCredentialsError(self, mock_response_wrong_credentials_error): + mydevolo = Mydevolo() + with pytest.raises(WrongCredentialsError): + mydevolo._call("test") + Mydevolo.del_instance() + + def test_call_WrongUrlError(self, mock_response_wrong_url_error): + mydevolo = Mydevolo() + with pytest.raises(WrongUrlError): + mydevolo._call("test") + Mydevolo.del_instance() + + def test_call_valid(self, mydevolo, mock_response_valid): + assert mydevolo._call("test").get("response") == "response" diff --git a/tests/test_properties.py b/tests/test_properties.py index c0aa113d..a3b412f9 100644 --- a/tests/test_properties.py +++ b/tests/test_properties.py @@ -7,6 +7,7 @@ from devolo_home_control_api.properties.voltage_property import VoltageProperty +@pytest.mark.usefixtures("home_control_instance") class TestProperties: def test_settings_property_valid(self): setting_property = SettingsProperty(f"lis.{self.devices.get('mains').get('uid')}", @@ -30,6 +31,7 @@ def test_binary_switch_property_invalid(self): with pytest.raises(WrongElementError): BinarySwitchProperty("invalid") + def test_consumption_property_invalid(self): with pytest.raises(WrongElementError): ConsumptionProperty("invalid") diff --git a/tests/test_publisher.py b/tests/test_publisher.py new file mode 100644 index 00000000..b4becfe2 --- /dev/null +++ b/tests/test_publisher.py @@ -0,0 +1,32 @@ +import pytest + + +@pytest.mark.usefixtures("mock_inspect_devices_metering_plug") +@pytest.mark.usefixtures("home_control_instance") +class TestPublisher: + + def test_register_unregister(self): + for device in self.homecontrol.devices: + self.homecontrol.devices[device].subscriber = Subscriber(device) + self.homecontrol.mprm.publisher.register(device, self.homecontrol.devices[device].subscriber) + assert len(self.homecontrol.publisher._get_subscribers_for_specific_event(device)) == 1 + for device in self.homecontrol.devices: + self.homecontrol.mprm.publisher.unregister(device, self.homecontrol.devices[device].subscriber) + assert len(self.homecontrol.publisher._get_subscribers_for_specific_event(device)) == 0 + + def test_dispatch(self): + for device in self.homecontrol.devices: + self.homecontrol.devices[device].subscriber = Subscriber(device) + self.homecontrol.mprm.publisher.register(device, self.homecontrol.devices[device].subscriber) + with pytest.raises(FileExistsError): + self.homecontrol.publisher.dispatch(event="hdm:ZWave:F6BF9812/4", message=()) + + + +class Subscriber: + def __init__(self, name): + self.name = name + + def update(self, message): + # We raise an error here so we can check for it in the test case. + raise FileExistsError diff --git a/tests/test_settings_property.py b/tests/test_settings_property.py new file mode 100644 index 00000000..5a1954ad --- /dev/null +++ b/tests/test_settings_property.py @@ -0,0 +1,34 @@ +import pytest + + +@pytest.mark.usefixtures("home_control_instance") +class TestSettingsProperty: + def test_fetch_general_device_settings_valid(self, mock_mprmrest__extract_data_from_element_uid): + name, icon, zone_id, events_enabled = \ + self.homecontrol.devices.get(self.devices.get("mains").get("uid"))\ + .settings_property.get("general_device_settings").fetch_general_device_settings() + assert name == self.devices.get('mains').get('name') + assert icon == self.devices.get('mains').get('icon') + assert zone_id == self.devices.get('mains').get('zone_id') + assert events_enabled + + def test_fetch_led_setting_valid(self, mock_mprmrest__extract_data_from_element_uid): + assert self.homecontrol.devices.get(self.devices.get("mains").get("uid"))\ + .settings_property.get("led").fetch_led_setting() + + def test_fetch_param_changed_valid(self, mock_mprmrest__extract_data_from_element_uid): + assert not self.homecontrol.devices.get(self.devices.get("mains").get("uid"))\ + .settings_property.get("param_changed").fetch_param_changed_setting() + + def test_fetch_protection_setting_invalid(self): + with pytest.raises(ValueError): + self.homecontrol.devices.get(self.devices.get("mains").get("uid"))\ + .settings_property.get("protection_setting").fetch_protection_setting(protection_setting="invalid") + + def test_fetch_protection_setting_valid(self, mock_mprmrest__extract_data_from_element_uid): + local_switch = self.homecontrol.devices.get(self.devices.get("mains").get("uid"))\ + .settings_property.get("protection_setting").fetch_protection_setting(protection_setting="local") + remote_switch = self.homecontrol.devices.get(self.devices.get("mains").get("uid"))\ + .settings_property.get("protection_setting").fetch_protection_setting(protection_setting="remote") + assert local_switch + assert not remote_switch diff --git a/tests/test_updater.py b/tests/test_updater.py new file mode 100644 index 00000000..ef1dd744 --- /dev/null +++ b/tests/test_updater.py @@ -0,0 +1,132 @@ +import pytest + + +@pytest.mark.usefixtures("home_control_instance") +@pytest.mark.usefixtures("mock_publisher_dispatch") +class TestUpdater: + def test_update_binary_switch_state_valid(self, fill_device_data): + binary_switch_property = self.homecontrol.devices.get(self.devices.get("mains").get("uid")).binary_switch_property + state = binary_switch_property.get(f"devolo.BinarySwitch:{self.devices.get('mains').get('uid')}").state + self.homecontrol.updater.update_binary_switch_state(element_uid=f"devolo.BinarySwitch:" + f"{self.devices.get('mains').get('uid')}", + value=True) + assert state != binary_switch_property.get(f"devolo.BinarySwitch:{self.devices.get('mains').get('uid')}").state + + def test_update_consumption_valid(self, fill_device_data): + consumption_property = self.homecontrol.devices.get(self.devices.get("mains").get("uid")).consumption_property + current_before = consumption_property.get(f"devolo.Meter:{self.devices.get('mains').get('uid')}").current + total_before = consumption_property.get(f"devolo.Meter:{self.devices.get('mains').get('uid')}").total + self.homecontrol.updater.update_consumption(element_uid=f"devolo.Meter:{self.devices.get('mains').get('uid')}", + consumption="current", value=1.58) + self.homecontrol.updater.update_consumption(element_uid=f"devolo.Meter:{self.devices.get('mains').get('uid')}", + consumption="total", value=254) + assert current_before != consumption_property.get(f"devolo.Meter:{self.devices.get('mains').get('uid')}").current + assert total_before != consumption_property.get(f"devolo.Meter:{self.devices.get('mains').get('uid')}").total + assert consumption_property.get(f"devolo.Meter:{self.devices.get('mains').get('uid')}").current == 1.58 + assert consumption_property.get(f"devolo.Meter:{self.devices.get('mains').get('uid')}").total == 254 + + def test_device_online_state(self): + online_state = self.homecontrol.devices.get(self.devices.get("mains").get("uid")).status + self.homecontrol.updater.update_device_online_state(uid=self.devices.get('mains').get('uid'), + value=1) + assert self.homecontrol.devices.get(self.devices.get("mains").get("uid")).status == 1 + assert self.homecontrol.devices.get(self.devices.get("mains").get("uid")).status != online_state + + def test_update_gateway_state(self): + self.homecontrol.updater.update_gateway_state(accessible=True, online_sync=False) + assert self.homecontrol._gateway.online + assert not self.homecontrol._gateway.sync + + def test_update_voltage_valid(self, fill_device_data): + voltage_property = self.homecontrol.devices.get(self.devices.get("mains").get("uid")).voltage_property + current_voltage = \ + voltage_property.get(f"devolo.VoltageMultiLevelSensor:{self.devices.get('mains').get('uid')}").current + self.homecontrol.updater.update_voltage(element_uid=f"devolo.VoltageMultiLevelSensor:" + f"{self.devices.get('mains').get('uid')}", + value=257) + assert current_voltage != \ + voltage_property.get(f"devolo.VoltageMultiLevelSensor:{self.devices.get('mains').get('uid')}").current + assert voltage_property.get(f"devolo.VoltageMultiLevelSensor:{self.devices.get('mains').get('uid')}").current == 257 + + def test_update(self): + self.homecontrol.updater.update(message={"properties": + {"uid": f"devolo.BinarySwitch:{self.devices.get('mains').get('uid')}"}}) + + def test_update_invalid(self): + self.homecontrol.updater.update(message={"properties": + {"uid": "fibaro"}}) + + def test__binary_switch(self): + self.homecontrol.devices.get(self.devices.get('mains').get("uid")).binary_switch_property \ + .get(f"devolo.BinarySwitch:{self.devices.get('mains').get('uid')}").state = True + state = self.homecontrol.devices.get(self.devices.get('mains').get("uid")).binary_switch_property \ + .get(f"devolo.BinarySwitch:{self.devices.get('mains').get('uid')}").state + self.homecontrol.updater._binary_switch(message={"properties": {"property.name": "state", + "uid": f"devolo.BinarySwitch:{self.devices.get('mains').get('uid')}", + "property.value.new": 0}}) + state_new = self.homecontrol.devices.get(self.devices.get('mains').get("uid")).binary_switch_property \ + .get(f"devolo.BinarySwitch:{self.devices.get('mains').get('uid')}").state + assert state != state_new + + def test__device_online_state(self): + online_state = self.homecontrol.devices.get(self.devices.get("mains").get("uid")).status + self.homecontrol.updater._device_online_state(message={"properties": {"uid": self.devices.get('mains').get('uid'), + "property.value.new": 1}}) + assert self.homecontrol.devices.get(self.devices.get("mains").get("uid")).status == 1 + assert online_state != self.homecontrol.devices.get(self.devices.get("mains").get("uid")).status + + def test__gateway_accessible(self): + self.homecontrol._gateway.online = True + self.homecontrol._gateway.sync = True + accessible = self.homecontrol._gateway.online + online_sync = self.homecontrol._gateway.sync + self.homecontrol.updater._gateway_accessible(message={"properties": {"property.name": "gatewayAccessible", + "property.value.new": {"accessible": False, + "onlineSync": False}}}) + accessible_new = self.homecontrol._gateway.online + online_sync_new = self.homecontrol._gateway.sync + assert accessible != accessible_new + assert online_sync != online_sync_new + + def test__meter(self): + self.homecontrol.devices.get(self.devices.get('mains').get("uid")).consumption_property \ + .get(f"devolo.Meter:{self.devices.get('mains').get('uid')}").current = 5 + self.homecontrol.devices.get(self.devices.get('mains').get("uid")).consumption_property \ + .get(f"devolo.Meter:{self.devices.get('mains').get('uid')}").total = 230 + total = self.homecontrol.devices.get(self.devices.get('mains').get("uid")).consumption_property \ + .get(f"devolo.Meter:{self.devices.get('mains').get('uid')}").total + # Changing current value + self.homecontrol.updater._meter(message={"properties": {"property.name": "currentValue", + "uid": f"devolo.Meter:{self.devices.get('mains').get('uid')}", + "property.value.new": 7}}) + current_new = self.homecontrol.devices.get(self.devices.get('mains').get("uid")).consumption_property \ + .get(f"devolo.Meter:{self.devices.get('mains').get('uid')}").current + # Check if current value has changed + assert current_new == 7 + # Check if total has not changed + assert total == self.homecontrol.devices.get(self.devices.get('mains').get("uid")).consumption_property \ + .get(f"devolo.Meter:{self.devices.get('mains').get('uid')}").total + # Changing total value + self.homecontrol.updater._meter(message={"properties": {"property.name": "totalValue", + "uid": f"devolo.Meter:{self.devices.get('mains').get('uid')}", + "property.value.new": 235}}) + total_new = self.homecontrol.devices.get(self.devices.get('mains').get("uid")).consumption_property \ + .get(f"devolo.Meter:{self.devices.get('mains').get('uid')}").total + # Check if total value has changed + assert total_new == 235 + # Check if current value has not changed + assert self.homecontrol.devices.get(self.devices.get('mains').get("uid")).consumption_property \ + .get(f"devolo.Meter:{self.devices.get('mains').get('uid')}").current == current_new + + def test__voltage_multi_level_sensor(self): + self.homecontrol.devices.get(self.devices.get('mains').get("uid")).voltage_property \ + .get(f"devolo.VoltageMultiLevelSensor:{self.devices.get('mains').get('uid')}").current = 231 + current = self.homecontrol.devices.get(self.devices.get('mains').get("uid")).voltage_property \ + .get(f"devolo.VoltageMultiLevelSensor:{self.devices.get('mains').get('uid')}").current + self.homecontrol.updater._voltage_multi_level_sensor(message={"properties": {"uid": f"devolo.VoltageMultiLevelSensor:{self.devices.get('mains').get('uid')}", + "property.value.new": 234}}) + current_new = self.homecontrol.devices.get(self.devices.get('mains').get("uid")).voltage_property \ + .get(f"devolo.VoltageMultiLevelSensor:{self.devices.get('mains').get('uid')}").current + + assert current_new == 234 + assert current != current_new diff --git a/tests/test_voltage_property.py b/tests/test_voltage_property.py new file mode 100644 index 00000000..a95e1b49 --- /dev/null +++ b/tests/test_voltage_property.py @@ -0,0 +1,10 @@ +import pytest + + +@pytest.mark.usefixtures("home_control_instance") +@pytest.mark.usefixtures("mock_mprmrest__extract_data_from_element_uid") +class TestVoltageProperty: + def test_fetch_voltage_valid(self): + voltage = self.homecontrol.devices.get(self.devices.get("mains").get("uid")).\ + voltage_property.get(f"devolo.VoltageMultiLevelSensor:{self.devices.get('mains').get('uid')}").fetch_voltage() + assert voltage == 237 diff --git a/tests/test_zwave.py b/tests/test_zwave.py index 07ffbafc..b506078f 100644 --- a/tests/test_zwave.py +++ b/tests/test_zwave.py @@ -1,67 +1,47 @@ import pytest -from devolo_home_control_api.devices.zwave import Zwave +from devolo_home_control_api.devices.zwave import Zwave,\ + get_device_type_from_element_uid, get_device_uid_from_element_uid, get_device_uid_from_setting_uid from devolo_home_control_api.properties.binary_switch_property import BinarySwitchProperty class TestZwave: - def test_get_property(self): - device = Zwave(name=self.devices.get("mains").get("name"), - device_uid=self.devices.get("mains").get("uid"), - zone=self.devices.get("mains").get("zone"), - battery_level=-1, - icon=self.devices.get("mains").get("icon"), - online_state=2) + def test_get_property(self, home_control_instance, mock_mprmrest__extract_data_from_element_uid): + device = Zwave(**self.devices.get("mains")) device.binary_switch_property = {} element_uid = f'devolo.BinarySwitch:{self.devices.get("mains").get("uid")}' device.binary_switch_property[element_uid] = BinarySwitchProperty(element_uid=element_uid) - assert device.get_property("binary_switch")[0] == f'devolo.BinarySwitch:{self.devices.get("mains").get("uid")}' + assert isinstance(device.get_property("binary_switch")[0], BinarySwitchProperty) - def test_get_property_invalid(self): - device = Zwave(name=self.devices.get("mains").get("name"), - device_uid=self.devices.get("mains").get("uid"), - zone=self.devices.get("mains").get("zone"), - battery_level=-1, - icon=self.devices.get("mains").get("icon"), - online_state=2) + def test_get_property_invalid(self, mydevolo, mock_mydevolo__call): + device = Zwave(**self.devices.get("mains")) with pytest.raises(AttributeError): device.get_property("binary_switch") - def test_battery_level(self): + def test_battery_level(self, mydevolo, mock_mydevolo__call): # TODO: Use battery driven device - device = Zwave(name=self.devices.get("mains").get("name"), - device_uid=self.devices.get("mains").get("uid"), - zone=self.devices.get("mains").get("zone"), - battery_level=55, - icon=self.devices.get("mains").get("icon"), - online_state=2) + device = Zwave(**self.devices.get("ambiguous_1")) - assert device.battery_level == 55 + assert device.batteryLevel == 55 - def test_device_online_state_state(self): - device = Zwave(name=self.devices.get("mains").get("name"), - device_uid=self.devices.get("mains").get("uid"), - zone=self.devices.get("mains").get("zone"), - battery_level=-1, - icon=self.devices.get("mains").get("icon"), - online_state=1) - assert device.online == "offline" + def test_device_online_state_state(self, mydevolo, mock_mydevolo__call): + device = Zwave(**self.devices.get("ambiguous_2")) + assert device.status == 1 - device = Zwave(name=self.devices.get("mains").get("name"), - device_uid=self.devices.get("mains").get("uid"), - zone=self.devices.get("mains").get("zone"), - battery_level=-1, - icon=self.devices.get("mains").get("icon"), - online_state=2) - assert device.online == "online" + device = Zwave(**self.devices.get("mains")) + assert device.status == 2 - device = Zwave(name=self.devices.get("mains").get("name"), - device_uid=self.devices.get("mains").get("uid"), - zone=self.devices.get("mains").get("zone"), - battery_level=-1, - icon=self.devices.get("mains").get("icon"), - online_state=27) - device.online = "unknown state" + device = Zwave(**self.devices.get("ambiguous_1")) + assert device.status not in [1, 2] + + def test_get_device_type_from_element_uid(self): + assert get_device_type_from_element_uid("devolo.Meter:hdm:ZWave:F6BF9812/2#2") == "devolo.Meter" + + def test_get_device_uid_from_setting_uid(self): + assert get_device_uid_from_setting_uid("lis.hdm:ZWave:EB5A9F6C/2") == "hdm:ZWave:EB5A9F6C/2" + + def test_get_device_uid_from_element_uid(self): + assert get_device_uid_from_element_uid("devolo.Meter:hdm:ZWave:F6BF9812/2#2") == "hdm:ZWave:F6BF9812/2"