diff --git a/.travis.yml b/.travis.yml deleted file mode 100644 index 6a82e93..0000000 --- a/.travis.yml +++ /dev/null @@ -1,9 +0,0 @@ -language: python -python: - - "3.4" - - "3.5" - - "3.6" -install: - - pip install -e .[dev] - -script: nosetests \ No newline at end of file diff --git a/README.md b/README.md deleted file mode 100644 index 99f99cf..0000000 --- a/README.md +++ /dev/null @@ -1,152 +0,0 @@ -Linux: [![Build Status](https://travis-ci.org/wpietri/sucks.svg?branch=master)](https://travis-ci.org/wpietri/sucks) -Windows: [![Build Status](https://ci.appveyor.com/api/projects/status/github/wpietri/sucks)](https://ci.appveyor.com/project/wpietri/sucks) - - -sucks -===== - -A simple command-line python script to drive a robot vacuum. Currently -known to work with the Ecovacs Deebot N79, M80 Pro, M81, M88 -Pro, and R95 MKII from both North America and Europe. - -Does it work for your model as well? Join the discussion on the -[sucks-users mailing -list](https://groups.google.com/forum/#!forum/sucks-users). - -If you're curious about the protocol, I have [a rough doc](http://github.com/wpietri/sucks/blob/master/protocol.md) -started. I'll happily accept pull requests for it. - -Why the project name? Well, a) it's ridiculous that I needed to MITM -my own vacuum. This is not the future I signed up for. And b), -it's a vacuum. - -## Installation - -If you have a recent version of Python 3, you should be able to -do `pip install sucks` to get the most recently released version of -this. - -## Usage - -To get started, you'll need to have already set up an EcoVacs account -using your smartphone. - -With that ready, step one is to log in: -``` - % sucks login - Ecovacs app email: [your email] - Ecovacs app password: [your password] - your two-letter country code: us - your two-letter continent code: na - Config saved. -``` - -That creates a config file in a platform-appropriate place. The password -is hashed before saving, so it's reasonably safe. (If it doesn't appear -to work for your continent, try "ww", their world-wide catchall.) - -With that set up, you could have it clean in auto mode for 10 minutes -and return to its charger: - -``` - % sucks clean 10 -``` - -You could have it clean for 15 minutes and then do an extra 10 minutes -of edging: - -``` - % sucks clean 15 edge 10 -``` - -If you wanted it to clean for 5 minutes and then stop without charging: - -``` - % sucks clean 5 stop -``` - -If it's running amok and you'd just like it to stop where it is: - -``` - % sucks stop -``` - -To tell it to go plug in: - -``` - % sucks charge -``` - -I run mine from my crontab, but I didn't want it to clean every day, -so it also has a mode where it randomly decides to run or not based on -a frequency you give it. My crontab entry looks like this: - -``` -0 10 * * * /home/william/projects/sucks/sucks.sh clean -f 4/7 15 edge -f 1/14 10 -``` - -This means that every day at 10 am, it might do something. 4 days out -of 7, it will do 15 minutes of automatic cleaning. 1 day out of 14, -it will do 10 minutes of edging. And afterward it will always go back to -charge. - -## Library use - -You are welcome to try using this as a python library for other efforts. The -API is still experimental, so expect changes. Please join the [mailing -list](https://groups.google.com/forum/#!forum/sucks-users) to participate in -shaping the API. - -A simple usage might go something like this: - -```python -from sucks import * - -config = ... - -api = EcoVacsAPI(config['device_id'], config['email'], config['password_hash'], - config['country'], config['continent']) -my_vac = api.devices()[0] -vacbot = VacBot(api.uid, api.REALM, api.resource, api.user_access_token, my_vac, config['continent']) -vacbot.connect_and_wait_until_ready() - -vacbot.run(Clean()) # start cleaning -time.sleep(900) # clean for 15 minutes -vacbot.run(Charge()) # return to the charger -``` - -## Developing - -If you'd like to join in on developing, I recommend checking out the code, -setting up a virtual environment, and installing this package in editable -mode. You can confirm your environment works by running the tests. And please -do join the [mailing list](https://groups.google.com/forum/#!forum/sucks-users) -to discuss your plans. - -For more information see [the development documentation](developing.md). - - - -## See also - -There are now similar libraries in [Javascript](https://github.com/joostth/sucks.js) -and [Go](https://github.com/skburgart/go-vacbot). - -## Thanks - -My heartfelt thanks to: - -* [xmpppeek](https://www.beneaththewaves.net/Software/XMPPPeek.html), -a great library for examining XMPP traffic flows (yes, your vacuum -speaks Jabbber!), -* [mitmproxy](https://mitmproxy.org/), a fantastic tool for analyzing HTTPS, -* [click](http://click.pocoo.org/), a wonderfully complete and thoughtful -library for making Python command-line interfaces, -* [requests](http://docs.python-requests.org/en/master/), a polished Python -library for HTTP requests, -* [Decompilers online](http://www.javadecompilers.com/apk), which was -very helpful in figuring out what the Android app was up to, -* Albert Louw, who was kind enough to post code from [his own -experiments](https://community.smartthings.com/t/ecovacs-deebot-n79/93410/33) -with his device, and -* All the users who have given useful feedback and contributed code! diff --git a/appveyor.yml b/appveyor.yml deleted file mode 100644 index 47aad33..0000000 --- a/appveyor.yml +++ /dev/null @@ -1,24 +0,0 @@ -environment: - - matrix: - - # For Python versions available on Appveyor, see - # http://www.appveyor.com/docs/installed-software#python - - - PYTHON: "C:\\Python35" - -install: - # We need wheel installed to build wheels - - "SET PATH=%PYTHON%;%PYTHON%\\Scripts;%PATH%" - - python --version - - pip install -e .[dev] - -build: off - -test_script: - # Put your test command here. - # Note that you must use the environment variable %PYTHON% to refer to - # the interpreter you're using - Appveyor does not do anything special - # to put the Python version you want to use on PATH. - - "nosetests" - diff --git a/ecovacs/__init__.py b/custom_components/ecovacs/__init__.py similarity index 86% rename from ecovacs/__init__.py rename to custom_components/ecovacs/__init__.py index f0aa2d9..b9c6100 100644 --- a/ecovacs/__init__.py +++ b/custom_components/ecovacs/__init__.py @@ -24,6 +24,10 @@ DOMAIN = "ecovacs" CONF_COUNTRY = "country" CONF_CONTINENT = "continent" +CONF_BUMPER = "bumper" +CONF_BUMPER_SERVER = "bumper_server" +server_address = None + CONFIG_SCHEMA = vol.Schema( { @@ -33,7 +37,9 @@ CONFIG_SCHEMA = vol.Schema( vol.Required(CONF_PASSWORD): cv.string, vol.Required(CONF_COUNTRY): vol.All(vol.Lower, cv.string), vol.Required(CONF_CONTINENT): vol.All(vol.Lower, cv.string), - vol.Optional(CONF_VERIFY_SSL, default=DEFAULT_VERIFY_SSL): cv.boolean, + vol.Optional(CONF_BUMPER, default=False): cv.boolean, + vol.Optional(CONF_BUMPER_SERVER): cv.string, + vol.Optional(CONF_VERIFY_SSL, default=True): cv.boolean, } ) }, @@ -53,6 +59,10 @@ def setup(hass: HomeAssistant, config: ConfigType) -> bool: _LOGGER.debug("Creating new Ecovacs component") hass.data[ECOVACS_DEVICES] = [] + if CONF_BUMPER == True: + server_address = (config[DOMAIN].get(CONF_BUMPER_SERVER), 5223) + else: + server_address = None ecovacs_api = EcoVacsAPI( ECOVACS_API_DEVICEID, @@ -79,8 +89,9 @@ def setup(hass: HomeAssistant, config: ConfigType) -> bool: ecovacs_api.user_access_token, device, config[DOMAIN].get(CONF_CONTINENT).lower(), - monitor=True, + server_address, config[DOMAIN].get(CONF_VERIFY_SSL), + monitor=True, ) hass.data[ECOVACS_DEVICES].append(vacbot) diff --git a/ecovacs/manifest.json b/custom_components/ecovacs/manifest.json similarity index 58% rename from ecovacs/manifest.json rename to custom_components/ecovacs/manifest.json index 6879ed1..20075dd 100644 --- a/ecovacs/manifest.json +++ b/custom_components/ecovacs/manifest.json @@ -1,8 +1,9 @@ { "domain": "ecovacs", "name": "Ecovacs Bumper", + "version": "1.0.2", "documentation": "https://www.home-assistant.io/integrations/ecovacs", - "requirements": [""], + "requirements": ["sleekxmppfs>=1.3.4", "click>=6", "requests>=2.18", "pycryptodome>=3.4", "pycountry-convert>=0.5", "paho-mqtt>=1.4", "stringcase>=1.2"], "codeowners": ["@OverloadUT", "@mib1185"], "iot_class": "cloud_push", "loggers": ["sleekxmppfs", "sucksbumper"] diff --git a/custom_components/ecovacs/services.yaml b/custom_components/ecovacs/services.yaml new file mode 100644 index 0000000..26c8d74 --- /dev/null +++ b/custom_components/ecovacs/services.yaml @@ -0,0 +1,102 @@ +# Describes the format for available vacuum services + +turn_on: + name: Turn on + description: Start a new cleaning task. + target: + entity: + domain: vacuum + +turn_off: + name: Turn off + description: Stop the current cleaning task and return to home. + target: + entity: + domain: vacuum + +stop: + name: Stop + description: Stop the current cleaning task. + target: + entity: + domain: vacuum + +locate: + name: Locate + description: Locate the vacuum cleaner robot. + target: + entity: + domain: vacuum + +start_pause: + name: Start/Pause + description: Start, pause, or resume the cleaning task. + target: + entity: + domain: vacuum + +start: + name: Start + description: Start or resume the cleaning task. + target: + entity: + domain: vacuum + +pause: + name: Pause + description: Pause the cleaning task. + target: + entity: + domain: vacuum + +return_to_base: + name: Return to base + description: Tell the vacuum cleaner to return to its dock. + target: + entity: + domain: vacuum + +clean_spot: + name: Clean spot + description: Tell the vacuum cleaner to do a spot clean-up. + target: + entity: + domain: vacuum + +send_command: + name: Send command + description: Send a raw command to the vacuum cleaner. + target: + entity: + domain: vacuum + fields: + command: + name: Command + description: Command to execute. + required: true + example: "set_dnd_timer" + selector: + text: + params: + name: Parameters + description: Parameters for the command. + example: '{ "key": "value" }' + selector: + object: + +set_fan_speed: + name: Set fan speed + description: Set the fan speed of the vacuum cleaner. + target: + entity: + domain: vacuum + fields: + fan_speed: + name: Fan speed + description: + Platform dependent vacuum cleaner fan speed, with speed steps, like + 'medium' or by percentage, between 0 and 100. + required: true + example: "low" + selector: + text: diff --git a/ecovacs/sucksbumper.py b/custom_components/ecovacs/sucksbumper.py similarity index 96% rename from ecovacs/sucksbumper.py rename to custom_components/ecovacs/sucksbumper.py index e129da3..2386f7d 100644 --- a/ecovacs/sucksbumper.py +++ b/custom_components/ecovacs/sucksbumper.py @@ -376,10 +376,11 @@ class EventListener(object): self._emitter.unsubscribe(self) class VacBot(): - def __init__(self, user, domain, resource, secret, vacuum, continent, server_address=None, monitor=False, verify_ssl=True): + def __init__(self, user, domain, resource, secret, vacuum, continent, server_address=None, verify_ssl=True, monitor=False): self.vacuum = vacuum + self.server_address = server_address # If True, the VacBot object will handle keeping track of all statuses, # including the initial request for statuses, and new requests after the # VacBot returns from being offline. It will also cause it to regularly @@ -410,10 +411,14 @@ class VacBot(): self.iotmq = None if not vacuum['iotmq']: - self.xmpp = EcoVacsXMPP(user, domain, resource, secret, continent, vacuum, server_address) - #Uncomment line to allow unencrypted plain auth - #self.xmpp['feature_mechanisms'].unencrypted_plain = True - self.xmpp.subscribe_to_ctls(self._handle_ctl) + if self.server_address is not None: + vacuum = {"did": "none", "class": "none"} + super().__init__("sucks", "ecouser.net", "", "", vacuum, "") + else: + self.xmpp = EcoVacsXMPP(user, domain, resource, secret, continent, vacuum, server_address) + #Uncomment line to allow unencrypted plain auth + #self.xmpp['feature_mechanisms'].unencrypted_plain = True + self.xmpp.subscribe_to_ctls(self._handle_ctl) else: self.iotmq = EcoVacsIOTMQ(user, domain, resource, secret, continent, vacuum, server_address, verify_ssl=verify_ssl) @@ -426,21 +431,27 @@ class VacBot(): #self.xmpp.subscribe_to_ctls(self._handle_ctl) def connect_and_wait_until_ready(self): - if not self.vacuum['iotmq']: - self.xmpp.connect_and_wait_until_ready() - self.xmpp.schedule('Ping', 30, lambda: self.send_ping(), repeat=True) + if self.server_address: + logging.info("connecting") + self.xmpp.connect(self.server_address) + self.xmpp.process() + self.xmpp.wait_until_ready() else: - self.iotmq.connect_and_wait_until_ready() - self.iotmq.schedule(30, self.send_ping) - #self.xmpp.connect_and_wait_until_ready() #Leaving in case xmpp is given to iotmq in the future - - if self._monitor: - # Do a first ping, which will also fetch initial statuses if the ping succeeds - self.send_ping() - if not self.vacuum['iotmq']: - self.xmpp.schedule('Components', 3600, lambda: self.refresh_components(), repeat=True) + if not self.vacuum['iotmq']: + self.xmpp.connect_and_wait_until_ready() + self.xmpp.schedule('Ping', 30, lambda: self.send_ping(), repeat=True) else: - self.iotmq.schedule(3600,self.refresh_components) + self.iotmq.connect_and_wait_until_ready() + self.iotmq.schedule(30, self.send_ping) + #self.xmpp.connect_and_wait_until_ready() #Leaving in case xmpp is given to iotmq in the future + + if self._monitor: + # Do a first ping, which will also fetch initial statuses if the ping succeeds + self.send_ping() + if not self.vacuum['iotmq']: + self.xmpp.schedule('Components', 3600, lambda: self.refresh_components(), repeat=True) + else: + self.iotmq.schedule(3600,self.refresh_components) def _handle_ctl(self, ctl): method = '_handle_' + ctl['event'] diff --git a/ecovacs/vacuum.py b/custom_components/ecovacs/vacuum.py similarity index 99% rename from ecovacs/vacuum.py rename to custom_components/ecovacs/vacuum.py index 61030e8..f3a15bd 100644 --- a/ecovacs/vacuum.py +++ b/custom_components/ecovacs/vacuum.py @@ -4,7 +4,7 @@ from __future__ import annotations import logging from typing import Any -import .sucksbumper +from . import sucksbumper from homeassistant.components.vacuum import VacuumEntity, VacuumEntityFeature from homeassistant.core import HomeAssistant diff --git a/log_clean.py b/log_clean.py deleted file mode 100644 index 25dea57..0000000 --- a/log_clean.py +++ /dev/null @@ -1,86 +0,0 @@ -import base64 -import re -import sys - -# a script to take an xmpppeek log of a Ecovacs app session with a Deebot N79 and strip out some of the nonsense, -# including any private identifiers - -source_ip = None -userid = None -resourceid = None -robotid = None -auth_glob = None - -for line in sys.stdin: - # remove the garbage - line = line.rstrip() - line = re.sub("\[\\d{4}-\\d{2}-\\d{2} ", '', line) - line = re.sub("\.\\d{6}-\\d{2}:\\d{2}\] \[", ' ', line) - line = re.sub("]$", ' ', line) - line = re.sub("\(([SC])2[SC]\) [.0-9]+:\\d+ -> [.0-9]+:\d+\]", '\\1', line) - line = re.sub("\}\}\}", '', line) - line = re.sub("\{\{\{", '', line) - - # find the private bits and remove them - if not source_ip: - match = re.search('Client connect from ([.0-9]+)', line) - if match: - source_ip = match.group(1) - if not userid: - match = re.search('(20\d{6}[0-9a-f]{13})@ecouser.net/([0-9a-f]{8})', line) - if match: - userid = match.group(1) - resourceid = match.group(2) - if not robotid: - match = re.search('(E\d{8,})@126.ecorobot.net/atom', line) - if match: - robotid = match.group(1) - if not auth_glob: - match = re.search('([-A-Za-z0-9+/=]+)', - line) - if match: - auth_glob = match.group(1) - if source_ip: - line = re.sub(source_ip, 'SOURCEIP', line) - if userid: - line = re.sub(userid, 'USERID', line) - if resourceid: - line = re.sub(resourceid, 'RESOURCEID', line) - if robotid: - line = re.sub(robotid, 'ROBOTID', line) - if auth_glob: - line = re.sub(auth_glob, 'AUTHGLOB', line) - - # translate client commmands - - line = re.sub( - '()', - 'id=\\1 command=\\2', line) - - # translate server responses - - line = re.sub( - '', - 'id=\\1 result =empty', line) - line = re.sub( - '', - 'id=\\1 id=\\2 result=\\3', line) - line = re.sub( - '(', - 'id=\\1 response=\\2', line) - - print(line) - -# per SASL plain auth: https://tools.ietf.org/html/rfc4616 -(authentication_id, authorization_id, password) = base64.b64decode(auth_glob).decode().split(sep='\0') - -# no idea what the leading field is, and the resource appears to be the same -(mystery, resource, secret) = password.split('/') - -print("------------------") -print("sample config:") -print("user=" + userid) -print("domain=ecouser.net") -print("resource=" + resourceid) -print("secret=" + secret) -print("vacuum=" + robotid + "@126.ecorobot.net") diff --git a/setup.py b/setup.py deleted file mode 100644 index 22bb6d1..0000000 --- a/setup.py +++ /dev/null @@ -1,101 +0,0 @@ -from codecs import open -from os import path - -from setuptools import setup, find_packages - -here = path.abspath(path.dirname(__file__)) - -# Get the long description from the README file -long_description = '' -try: - with open(path.join(here, 'README.rst'), encoding='utf-8') as f: - long_description = f.read() -except FileNotFoundError: - print("can't find python README; skipping") - -setup( - name='sucks', - version='0.9.3', - - description='a library for controlling certain robot vacuums', - long_description=long_description, - - url='https://github.com/wpietri/sucks', - - # Author details - author='William Pietri', - author_email='sucks-users@googlegroups.com', - - # Choose your license - license='GPL-3.0', - - # See https://pypi.python.org/pypi?%3Aaction=list_classifiers - classifiers=[ - # How mature is this project? Common values are - # 3 - Alpha - # 4 - Beta - # 5 - Production/Stable - 'Development Status :: 4 - Beta', - - # Indicate who your project is intended for - 'Intended Audience :: Developers', - 'Topic :: Software Development :: Libraries', - 'Topic :: Home Automation', - - # Pick your license as you wish (should match "license" above) - 'License :: OSI Approved :: GNU General Public License v3 (GPLv3)', - - # Specify the Python versions you support here. In particular, ensure - # that you indicate whether you support Python 2, Python 3 or both. - 'Programming Language :: Python :: 3.5', - ], - - # What does your project relate to? - keywords='home automation vacuum robot', - - # You can just specify the packages manually here if your project is - # simple. Or you can use find_packages(). - packages=find_packages(exclude=['contrib', 'docs', 'tests']), - - # List run-time dependencies here. These will be installed by pip when - # your project is installed. For an analysis of "install_requires" vs pip's - # requirements files see: - # https://packaging.python.org/en/latest/requirements.html - - install_requires=[ - 'sleekxmppfs>=1.3.4', - 'click>=6', - 'requests>=2.18', - 'pycryptodome>=3.4', - 'pycountry-convert>=0.5', - 'paho-mqtt>=1.4', - 'stringcase>=1.2' - ], - - # List additional groups of dependencies here (e.g. development - # dependencies). You can install these using the following syntax, - # for example: - # $ pip install -e .[dev,test] - extras_require={ - 'dev': [ - 'nose', - 'requests-mock>=1.3' - ], - }, - - # If there are data files included in your packages that need to be - # installed, specify them here. If using Python 2.6 or less, then these - # have to be included in MANIFEST.in as well. - # package_data={ - # 'sample': ['package_data.dat'], - # }, - - # To provide executable scripts, use entry points in preference to the - # "scripts" keyword. Entry points provide cross-platform support and allow - # pip to create the appropriate form of executable for the target platform. - entry_points={ - 'console_scripts': [ - 'sucks=sucks.cli:cli', - ], - }, -) diff --git a/sucks.sh b/sucks.sh deleted file mode 100644 index 3c5dfe4..0000000 --- a/sucks.sh +++ /dev/null @@ -1,11 +0,0 @@ -#!/bin/bash -# -# A command-line script for running the cli if -# you haven't installed it using pip. Mainly -# useful for developers, I think. - - -DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )" - -cd ${DIR} -pipenv run -- python -m sucks.cli "$@" diff --git a/sucks/__init__.py b/sucks/__init__.py deleted file mode 100644 index e129da3..0000000 --- a/sucks/__init__.py +++ /dev/null @@ -1,1096 +0,0 @@ -import hashlib -import logging -import time -from base64 import b64decode, b64encode -from collections import OrderedDict -from threading import Event -import threading -import sched -import random -import ssl -import requests -import stringcase -import os -from sleekxmppfs import ClientXMPP, Callback, MatchXPath -from sleekxmppfs.xmlstream import ET -from sleekxmppfs.exceptions import XMPPError - -from paho.mqtt.client import Client as ClientMQTT -from paho.mqtt import publish as MQTTPublish -from paho.mqtt import subscribe as MQTTSubscribe - -_LOGGER = logging.getLogger(__name__) - -# These consts define all of the vocabulary used by this library when presenting various states and components. -# Applications implementing this library should import these rather than hard-code the strings, for future-proofing. - -CLEAN_MODE_AUTO = 'auto' -CLEAN_MODE_EDGE = 'edge' -CLEAN_MODE_SPOT = 'spot' -CLEAN_MODE_SPOT_AREA = 'spot_area' -CLEAN_MODE_SINGLE_ROOM = 'single_room' -CLEAN_MODE_STOP = 'stop' - -CLEAN_ACTION_START = 'start' -CLEAN_ACTION_PAUSE = 'pause' -CLEAN_ACTION_RESUME = 'resume' -CLEAN_ACTION_STOP = 'stop' - -FAN_SPEED_NORMAL = 'normal' -FAN_SPEED_HIGH = 'high' - -CHARGE_MODE_RETURN = 'return' -CHARGE_MODE_RETURNING = 'returning' -CHARGE_MODE_CHARGING = 'charging' -CHARGE_MODE_IDLE = 'idle' - -COMPONENT_SIDE_BRUSH = 'side_brush' -COMPONENT_MAIN_BRUSH = 'main_brush' -COMPONENT_FILTER = 'filter' - -VACUUM_STATUS_OFFLINE = 'offline' - -CLEANING_STATES = {CLEAN_MODE_AUTO, CLEAN_MODE_EDGE, CLEAN_MODE_SPOT, CLEAN_MODE_SPOT_AREA, CLEAN_MODE_SINGLE_ROOM} -CHARGING_STATES = {CHARGE_MODE_CHARGING} - -# These dictionaries convert to and from Sucks's consts (which closely match what the UI and manuals use) -# to and from what the Ecovacs API uses (which are sometimes very oddly named and have random capitalization.) -CLEAN_MODE_TO_ECOVACS = { - CLEAN_MODE_AUTO: 'auto', - CLEAN_MODE_EDGE: 'border', - CLEAN_MODE_SPOT: 'spot', - CLEAN_MODE_SPOT_AREA: 'SpotArea', - CLEAN_MODE_SINGLE_ROOM: 'singleroom', - CLEAN_MODE_STOP: 'stop' -} - -CLEAN_ACTION_TO_ECOVACS = { - CLEAN_ACTION_START: 's', - CLEAN_ACTION_PAUSE: 'p', - CLEAN_ACTION_RESUME: 'r', - CLEAN_ACTION_STOP: 'h', -} - -CLEAN_ACTION_FROM_ECOVACS = { - 's': CLEAN_ACTION_START, - 'p': CLEAN_ACTION_PAUSE, - 'r': CLEAN_ACTION_RESUME, - 'h': CLEAN_ACTION_STOP, -} - -CLEAN_MODE_FROM_ECOVACS = { - 'auto': CLEAN_MODE_AUTO, - 'border': CLEAN_MODE_EDGE, - 'spot': CLEAN_MODE_SPOT, - 'spot_area': CLEAN_MODE_SPOT_AREA, - 'singleroom': CLEAN_MODE_SINGLE_ROOM, - 'stop': CLEAN_MODE_STOP, - 'going': CHARGE_MODE_RETURNING -} - -FAN_SPEED_TO_ECOVACS = { - FAN_SPEED_NORMAL: 'standard', - FAN_SPEED_HIGH: 'strong' -} - -FAN_SPEED_FROM_ECOVACS = { - 'standard': FAN_SPEED_NORMAL, - 'strong': FAN_SPEED_HIGH -} - -CHARGE_MODE_TO_ECOVACS = { - CHARGE_MODE_RETURN: 'go', - CHARGE_MODE_RETURNING: 'Going', - CHARGE_MODE_CHARGING: 'SlotCharging', - CHARGE_MODE_IDLE: 'Idle' -} - -CHARGE_MODE_FROM_ECOVACS = { - 'going': CHARGE_MODE_RETURNING, - 'slot_charging': CHARGE_MODE_CHARGING, - 'idle': CHARGE_MODE_IDLE -} - -COMPONENT_TO_ECOVACS = { - COMPONENT_MAIN_BRUSH: 'Brush', - COMPONENT_SIDE_BRUSH: 'SideBrush', - COMPONENT_FILTER: 'DustCaseHeap' -} - -COMPONENT_FROM_ECOVACS = { - 'brush': COMPONENT_MAIN_BRUSH, - 'side_brush': COMPONENT_SIDE_BRUSH, - 'dust_case_heap': COMPONENT_FILTER -} - -def str_to_bool_or_cert(s): - if s == 'True' or s == True: - return True - elif s == 'False' or s == False: - return False - else: - if not s == None: - if os.path.exists(s): # User could provide a path to a CA Cert as well, which is useful for Bumper - if os.path.isfile(s): - return s - else: - raise ValueError("Certificate path provided is not a file - {}".format(s)) - - raise ValueError("Cannot covert {} to a bool or certificate path".format(s)) - - -class EcoVacsAPI: - CLIENT_KEY = "eJUWrzRv34qFSaYk" - SECRET = "Cyu5jcR4zyK6QEPn1hdIGXB5QIDAQABMA0GC" - PUBLIC_KEY = 'MIIB/TCCAWYCCQDJ7TMYJFzqYDANBgkqhkiG9w0BAQUFADBCMQswCQYDVQQGEwJjbjEVMBMGA1UEBwwMRGVmYXVsdCBDaXR5MRwwGgYDVQQKDBNEZWZhdWx0IENvbXBhbnkgTHRkMCAXDTE3MDUwOTA1MTkxMFoYDzIxMTcwNDE1MDUxOTEwWjBCMQswCQYDVQQGEwJjbjEVMBMGA1UEBwwMRGVmYXVsdCBDaXR5MRwwGgYDVQQKDBNEZWZhdWx0IENvbXBhbnkgTHRkMIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDb8V0OYUGP3Fs63E1gJzJh+7iqeymjFUKJUqSD60nhWReZ+Fg3tZvKKqgNcgl7EGXp1yNifJKUNC/SedFG1IJRh5hBeDMGq0m0RQYDpf9l0umqYURpJ5fmfvH/gjfHe3Eg/NTLm7QEa0a0Il2t3Cyu5jcR4zyK6QEPn1hdIGXB5QIDAQABMA0GCSqGSIb3DQEBBQUAA4GBANhIMT0+IyJa9SU8AEyaWZZmT2KEYrjakuadOvlkn3vFdhpvNpnnXiL+cyWy2oU1Q9MAdCTiOPfXmAQt8zIvP2JC8j6yRTcxJCvBwORDyv/uBtXFxBPEC6MDfzU2gKAaHeeJUWrzRv34qFSaYkYta8canK+PSInylQTjJK9VqmjQ' - MAIN_URL_FORMAT = 'https://eco-{country}-api.ecovacs.com/v1/private/{country}/{lang}/{deviceId}/{appCode}/{appVersion}/{channel}/{deviceType}' - USER_URL_FORMAT = 'https://users-{continent}.ecouser.net:8000/user.do' - PORTAL_URL_FORMAT = 'https://portal-{continent}.ecouser.net/api' - - USERSAPI = 'users/user.do' - IOTDEVMANAGERAPI = 'iot/devmanager.do' # IOT Device Manager - This provides control of "IOT" products via RestAPI, some bots use this instead of XMPP - PRODUCTAPI = 'pim/product' # Leaving this open, the only endpoint known currently is "Product IOT Map" - pim/product/getProductIotMap - This provides a list of "IOT" products. Not sure what this provides the app. - - - REALM = 'ecouser.net' - - def __init__(self, device_id, account_id, password_hash, country, continent, verify_ssl=True): - self.meta = { - 'country': country, - 'lang': 'en', - 'deviceId': device_id, - 'appCode': 'i_eco_e', - #'appCode': 'i_eco_a' - iphone - 'appVersion': '1.3.5', - #'appVersion': '1.4.6' - iphone - 'channel': 'c_googleplay', - #'channel': 'c_iphone', - iphone - 'deviceType': '1' - #'deviceType': '2' - iphone - } - - self.verify_ssl = str_to_bool_or_cert(verify_ssl) - _LOGGER.debug("Setting up EcoVacsAPI") - self.resource = device_id[0:8] - self.country = country - self.continent = continent - login_info = self.__call_main_api('user/login', - ('account', self.encrypt(account_id)), - ('password', self.encrypt(password_hash))) - self.uid = login_info['uid'] - self.login_access_token = login_info['accessToken'] - self.auth_code = self.__call_main_api('user/getAuthCode', - ('uid', self.uid), - ('accessToken', self.login_access_token))['authCode'] - login_response = self.__call_login_by_it_token() - self.user_access_token = login_response['token'] - if login_response['userId'] != self.uid: - logging.debug("Switching to shorter UID " + login_response['userId']) - self.uid = login_response['userId'] - logging.debug("EcoVacsAPI connection complete") - - def __sign(self, params): - result = params.copy() - result['authTimespan'] = int(time.time() * 1000) - result['authTimeZone'] = 'GMT-8' - - sign_on = self.meta.copy() - sign_on.update(result) - sign_on_text = EcoVacsAPI.CLIENT_KEY + ''.join( - [k + '=' + str(sign_on[k]) for k in sorted(sign_on.keys())]) + EcoVacsAPI.SECRET - - result['authAppkey'] = EcoVacsAPI.CLIENT_KEY - result['authSign'] = self.md5(sign_on_text) - return result - - def __call_main_api(self, function, *args): - _LOGGER.debug("calling main api {} with {}".format(function, args)) - params = OrderedDict(args) - params['requestId'] = self.md5(time.time()) - url = (EcoVacsAPI.MAIN_URL_FORMAT + "/" + function).format(**self.meta) - api_response = requests.get(url, self.__sign(params), verify=self.verify_ssl) - json = api_response.json() - _LOGGER.debug("got {}".format(json)) - if json['code'] == '0000': - return json['data'] - elif json['code'] == '1005': - _LOGGER.warning("incorrect email or password") - raise ValueError("incorrect email or password") - else: - _LOGGER.error("call to {} failed with {}".format(function, json)) - raise RuntimeError("failure code {} ({}) for call {} and parameters {}".format( - json['code'], json['msg'], function, args)) - - def __call_user_api(self, function, args): - _LOGGER.debug("calling user api {} with {}".format(function, args)) - params = {'todo': function} - params.update(args) - response = requests.post(EcoVacsAPI.USER_URL_FORMAT.format(continent=self.continent), json=params, verify=self.verify_ssl) - json = response.json() - _LOGGER.debug("got {}".format(json)) - if json['result'] == 'ok': - return json - else: - _LOGGER.error("call to {} failed with {}".format(function, json)) - raise RuntimeError( - "failure {} ({}) for call {} and parameters {}".format(json['error'], json['errno'], function, params)) - - def __call_portal_api(self, api, function, args, verify_ssl=True, **kwargs): - - if api == self.USERSAPI: - params = {'todo': function} - params.update(args) - else: - params = {} - params.update(args) - - _LOGGER.debug("calling portal api {} function {} with {}".format(api, function, params)) - - continent = self.continent - if 'continent' in kwargs: - continent = kwargs.get('continent') - - url = (EcoVacsAPI.PORTAL_URL_FORMAT + "/" + api).format(continent=continent, **self.meta) - - response = requests.post(url, json=params, verify=verify_ssl) - - json = response.json() - _LOGGER.debug("got {}".format(json)) - if api == self.USERSAPI: - if json['result'] == 'ok': - return json - elif json['result'] == 'fail': - if json['error'] == 'set token error.': # If it is a set token error try again - if not 'set_token' in kwargs: - _LOGGER.debug("loginByItToken set token error, trying again (2/3)") - return self.__call_portal_api(self.USERSAPI, function, args, verify_ssl=verify_ssl, set_token=1) - elif kwargs.get('set_token') == 1: - _LOGGER.debug("loginByItToken set token error, trying again with ww (3/3)") - return self.__call_portal_api(self.USERSAPI, function, args, verify_ssl=verify_ssl, set_token=2, continent="ww") - else: - _LOGGER.debug("loginByItToken set token error, failed after 3 attempts") - - if api.startswith(self.PRODUCTAPI): - if json['code'] == 0: - return json - - else: - _LOGGER.error("call to {} failed with {}".format(function, json)) - raise RuntimeError( - "failure {} ({}) for call {} and parameters {}".format(json['error'], json['errno'], function, params)) - - def __call_login_by_it_token(self): - return self.__call_portal_api(self.USERSAPI,'loginByItToken', - {'country': self.meta['country'].upper(), - 'resource': self.resource, - 'realm': EcoVacsAPI.REALM, - 'userId': self.uid, - 'token': self.auth_code} - , verify_ssl=self.verify_ssl) - - def getdevices(self): - return self.__call_portal_api(self.USERSAPI,'GetDeviceList', { - 'userid': self.uid, - 'auth': { - 'with': 'users', - 'userid': self.uid, - 'realm': EcoVacsAPI.REALM, - 'token': self.user_access_token, - 'resource': self.resource - } - }, verify_ssl=self.verify_ssl)['devices'] - - def getiotProducts(self): - return self.__call_portal_api(self.PRODUCTAPI + '/getProductIotMap','', { - 'channel': '', - 'auth': { - 'with': 'users', - 'userid': self.uid, - 'realm': EcoVacsAPI.REALM, - 'token': self.user_access_token, - 'resource': self.resource - } - }, verify_ssl=self.verify_ssl)['data'] - - def SetIOTDevices(self, devices, iotproducts): - #Originally added for D900, and not actively used in code now - Not sure what the app checks the items in this list for - for device in devices: #Check if the device is part of iotProducts - device['iot_product'] = False - for iotProduct in iotproducts: - if device['class'] in iotProduct['classid']: - device['iot_product'] = True - - return devices - - def SetIOTMQDevices(self, devices): - #Added for devices that utilize MQTT instead of XMPP for communication - for device in devices: - device['iotmq'] = False - if device['company'] == 'eco-ng': #Check if the device is part of the list - device['iotmq'] = True - - return devices - - def devices(self): - return self.SetIOTMQDevices(self.getdevices()) - - @staticmethod - def md5(text): - return hashlib.md5(bytes(str(text), 'utf8')).hexdigest() - - @staticmethod - def encrypt(text): - from Crypto.PublicKey import RSA - from Crypto.Cipher import PKCS1_v1_5 - key = RSA.import_key(b64decode(EcoVacsAPI.PUBLIC_KEY)) - cipher = PKCS1_v1_5.new(key) - result = cipher.encrypt(bytes(text, 'utf8')) - return str(b64encode(result), 'utf8') - - -class EventEmitter(object): - """A very simple event emitting system.""" - def __init__(self): - self._subscribers = [] - - def subscribe(self, callback): - listener = EventListener(self, callback) - self._subscribers.append(listener) - return listener - - def unsubscribe(self, listener): - self._subscribers.remove(listener) - - def notify(self, event): - for subscriber in self._subscribers: - subscriber.callback(event) - - -class EventListener(object): - """Object that allows event consumers to easily unsubscribe from events.""" - def __init__(self, emitter, callback): - self._emitter = emitter - self.callback = callback - - def unsubscribe(self): - self._emitter.unsubscribe(self) - -class VacBot(): - def __init__(self, user, domain, resource, secret, vacuum, continent, server_address=None, monitor=False, verify_ssl=True): - - self.vacuum = vacuum - - # If True, the VacBot object will handle keeping track of all statuses, - # including the initial request for statuses, and new requests after the - # VacBot returns from being offline. It will also cause it to regularly - # request component lifespans - self._monitor = monitor - - self._failed_pings = 0 - - # These three are representations of the vacuum state as reported by the API - self.clean_status = None - self.charge_status = None - self.battery_status = None - - # This is an aggregate state managed by the sucks library, combining the clean and charge events to a single state - self.vacuum_status = None - self.fan_speed = None - - # Populated by component Lifespan reports - self.components = {} - - self.statusEvents = EventEmitter() - self.batteryEvents = EventEmitter() - self.lifespanEvents = EventEmitter() - self.errorEvents = EventEmitter() - - #Set none for clients to start - self.xmpp = None - self.iotmq = None - - if not vacuum['iotmq']: - self.xmpp = EcoVacsXMPP(user, domain, resource, secret, continent, vacuum, server_address) - #Uncomment line to allow unencrypted plain auth - #self.xmpp['feature_mechanisms'].unencrypted_plain = True - self.xmpp.subscribe_to_ctls(self._handle_ctl) - - else: - self.iotmq = EcoVacsIOTMQ(user, domain, resource, secret, continent, vacuum, server_address, verify_ssl=verify_ssl) - self.iotmq.subscribe_to_ctls(self._handle_ctl) - #The app still connects to XMPP as well, but only issues ping commands. - #Everything works without XMPP, so leaving the below commented out. - #self.xmpp = EcoVacsXMPP(user, domain, resource, secret, continent, vacuum, server_address) - #Uncomment line to allow unencrypted plain auth - #self.xmpp['feature_mechanisms'].unencrypted_plain = True - #self.xmpp.subscribe_to_ctls(self._handle_ctl) - - def connect_and_wait_until_ready(self): - if not self.vacuum['iotmq']: - self.xmpp.connect_and_wait_until_ready() - self.xmpp.schedule('Ping', 30, lambda: self.send_ping(), repeat=True) - else: - self.iotmq.connect_and_wait_until_ready() - self.iotmq.schedule(30, self.send_ping) - #self.xmpp.connect_and_wait_until_ready() #Leaving in case xmpp is given to iotmq in the future - - if self._monitor: - # Do a first ping, which will also fetch initial statuses if the ping succeeds - self.send_ping() - if not self.vacuum['iotmq']: - self.xmpp.schedule('Components', 3600, lambda: self.refresh_components(), repeat=True) - else: - self.iotmq.schedule(3600,self.refresh_components) - - def _handle_ctl(self, ctl): - method = '_handle_' + ctl['event'] - if hasattr(self, method): - getattr(self, method)(ctl) - - def _handle_error(self, event): - if 'error' in event: - error = event['error'] - elif 'errs' in event: - error = event['errs'] - - if not error == '': - self.errorEvents.notify(error) - _LOGGER.debug("*** error = " + error) - - def _handle_life_span(self, event): - type = event['type'] - try: - type = COMPONENT_FROM_ECOVACS[type] - except KeyError: - _LOGGER.warning("Unknown component type: '" + type + "'") - - if 'val' in event: - lifespan = int(event['val']) / 100 - else: - lifespan = int(event['left']) / 60 #This works for a D901 - self.components[type] = lifespan - - lifespan_event = {'type': type, 'lifespan': lifespan} - self.lifespanEvents.notify(lifespan_event) - _LOGGER.debug("*** life_span " + type + " = " + str(lifespan)) - - def _handle_clean_report(self, event): - type = event['type'] - try: - type = CLEAN_MODE_FROM_ECOVACS[type] - if self.vacuum['iotmq']: #Was able to parse additional status from the IOTMQ, may apply to XMPP too - statustype = event['st'] - statustype = CLEAN_ACTION_FROM_ECOVACS[statustype] - if statustype == CLEAN_ACTION_STOP or statustype == CLEAN_ACTION_PAUSE: - type = statustype - except KeyError: - _LOGGER.warning("Unknown cleaning status '" + type + "'") - self.clean_status = type - self.vacuum_status = type - - fan = event.get('speed', None) - if fan is not None: - try: - fan = FAN_SPEED_FROM_ECOVACS[fan] - except KeyError: - _LOGGER.warning("Unknown fan speed: '" + fan + "'") - self.fan_speed = fan - self.statusEvents.notify(self.vacuum_status) - if self.fan_speed: - _LOGGER.debug("*** clean_status = " + self.clean_status + " fan_speed = " + self.fan_speed) - else: - _LOGGER.debug("*** clean_status = " + self.clean_status + " fan_speed = None") - - def _handle_battery_info(self, iq): - try: - self.battery_status = float(iq['power']) / 100 - except ValueError: - _LOGGER.warning("couldn't parse battery status " + ET.tostring(iq)) - else: - self.batteryEvents.notify(self.battery_status) - _LOGGER.debug("*** battery_status = {:.0%}".format(self.battery_status)) - - def _handle_charge_state(self, event): - if 'type' in event: - status = event['type'] - elif 'errno' in event: #Handle error - if event['ret'] == 'fail' and event['errno'] == '8': #Already charging - status = 'slot_charging' - elif event['ret'] == 'fail' and event['errno'] == '5': #Busy with another command - status = 'idle' - elif event['ret'] == 'fail' and event['errno'] == '3': #Bot in stuck state, example dust bin out - status = 'idle' - else: - status = 'idle' #Fall back to Idle status - _LOGGER.error("Unknown charging status '" + event['errno'] + "'") #Log this so we can identify more errors - - try: - status = CHARGE_MODE_FROM_ECOVACS[status] - except KeyError: - _LOGGER.warning("Unknown charging status '" + status + "'") - - self.charge_status = status - if status != 'idle' or self.vacuum_status == 'charging': - # We have to ignore the idle messages, because all it means is that it's not - # currently charging, in which case the clean_status is a better indicator - # of what the vacuum is currently up to. - self.vacuum_status = status - self.statusEvents.notify(self.vacuum_status) - _LOGGER.debug("*** charge_status = " + self.charge_status) - - def _vacuum_address(self): - if not self.vacuum['iotmq']: - return self.vacuum['did'] + '@' + self.vacuum['class'] + '.ecorobot.net/atom' - else: - return self.vacuum['did'] #IOTMQ only uses the did - - @property - def is_charging(self) -> bool: - return self.vacuum_status in CHARGING_STATES - - @property - def is_cleaning(self) -> bool: - return self.vacuum_status in CLEANING_STATES - - def send_ping(self): - try: - if not self.vacuum['iotmq']: - self.xmpp.send_ping(self._vacuum_address()) - elif self.vacuum['iotmq']: - if not self.iotmq.send_ping(): - raise RuntimeError() - - except XMPPError as err: - _LOGGER.warning("Ping did not reach VacBot. Will retry.") - _LOGGER.debug("*** Error type: " + err.etype) - _LOGGER.debug("*** Error condition: " + err.condition) - self._failed_pings += 1 - if self._failed_pings >= 4: - self.vacuum_status = 'offline' - self.statusEvents.notify(self.vacuum_status) - - except RuntimeError as err: - _LOGGER.warning("Ping did not reach VacBot. Will retry.") - self._failed_pings += 1 - if self._failed_pings >= 4: - self.vacuum_status = 'offline' - self.statusEvents.notify(self.vacuum_status) - - else: - self._failed_pings = 0 - if self._monitor: - # If we don't yet have a vacuum status, request initial statuses again now that the ping succeeded - if self.vacuum_status == 'offline' or self.vacuum_status is None: - self.request_all_statuses() - else: - # If we're not auto-monitoring the status, then just reset the status to None, which indicates unknown - if self.vacuum_status == 'offline': - self.vacuum_status = None - self.statusEvents.notify(self.vacuum_status) - - def refresh_components(self): - try: - self.run(GetLifeSpan('main_brush')) - self.run(GetLifeSpan('side_brush')) - self.run(GetLifeSpan('filter')) - except XMPPError as err: - _LOGGER.warning("Component refresh requests failed to reach VacBot. Will try again later.") - _LOGGER.debug("*** Error type: " + err.etype) - _LOGGER.debug("*** Error condition: " + err.condition) - - def refresh_statuses(self): - try: - self.run(GetCleanState()) - self.run(GetChargeState()) - self.run(GetBatteryState()) - except XMPPError as err: - _LOGGER.warning("Initial status requests failed to reach VacBot. Will try again on next ping.") - _LOGGER.debug("*** Error type: " + err.etype) - _LOGGER.debug("*** Error condition: " + err.condition) - - def request_all_statuses(self): - self.refresh_statuses() - self.refresh_components() - - def send_command(self, action): - if not self.vacuum['iotmq']: - self.xmpp.send_command(action.to_xml(), self._vacuum_address()) - else: - #IOTMQ issues commands via RestAPI, and listens on MQTT for status updates - self.iotmq.send_command(action, self._vacuum_address()) #IOTMQ devices need the full action for additional parsing - - def run(self, action): - self.send_command(action) - - def disconnect(self, wait=False): - if not self.vacuum['iotmq']: - self.xmpp.disconnect(wait=wait) - else: - self.iotmq._disconnect() - #self.xmpp.disconnect(wait=wait) #Leaving in case xmpp is added to iotmq in the future - -#This is used by EcoVacsIOTMQ and EcoVacsXMPP for _ctl_to_dict -def RepresentsInt(stringvar): - try: - int(stringvar) - return True - except ValueError: - return False - -class EcoVacsIOTMQ(ClientMQTT): - def __init__(self, user, domain, resource, secret, continent, vacuum, server_address=None, verify_ssl=True): - ClientMQTT.__init__(self) - self.ctl_subscribers = [] - self.user = user - self.domain = str(domain).split(".")[0] #MQTT is using domain without tld extension - self.resource = resource - self.secret = secret - self.continent = continent - self.vacuum = vacuum - self.scheduler = sched.scheduler(time.time, time.sleep) - self.scheduler_thread = threading.Thread(target=self.scheduler.run, daemon=True, name="mqtt_schedule_thread") - self.verify_ssl = str_to_bool_or_cert(verify_ssl) - - if server_address is None: - self.hostname = ('mq-{}.ecouser.net'.format(self.continent)) - self.port = 8883 - else: - saddress = server_address.split(":") - if len(saddress) > 1: - self.hostname = saddress[0] - if RepresentsInt(saddress[1]): - self.port = int(saddress[1]) - else: - self.port = 8883 - - self._client_id = self.user + '@' + self.domain.split(".")[0] + '/' + self.resource - self.username_pw_set(self.user + '@' + self.domain, secret) - - self.ready_flag = Event() - - def connect_and_wait_until_ready(self): - #self._on_log = self.on_log #This provides more logging than needed, even for debug - self._on_message = self._handle_ctl_mqtt - self._on_connect = self.on_connect - - #TODO: This is pretty insecure and accepts any cert, maybe actually check? - ssl_ctx = ssl.create_default_context() - ssl_ctx.check_hostname = False - ssl_ctx.verify_mode = ssl.CERT_NONE - self.tls_set_context(ssl_ctx) - self.tls_insecure_set(True) - - self.connect(self.hostname, self.port) - self.loop_start() - self.wait_until_ready() - - def subscribe_to_ctls(self, function): - self.ctl_subscribers.append(function) - - def _disconnect(self): - self.disconnect() #disconnect mqtt connection - self.scheduler.empty() #Clear schedule queue - - def _run_scheduled_func(self, timer_seconds, timer_function): - timer_function() - self.schedule(timer_seconds, timer_function) - - def schedule(self, timer_seconds, timer_function): - self.scheduler.enter(timer_seconds, 1, self._run_scheduled_func,(timer_seconds, timer_function)) - if not self.scheduler_thread.isAlive(): - self.scheduler_thread.start() - - def wait_until_ready(self): - self.ready_flag.wait() - - def on_connect(self, client, userdata, flags, rc): - if rc != 0: - _LOGGER.error("EcoVacsMQTT - error connecting with MQTT Return {}".format(rc)) - raise RuntimeError("EcoVacsMQTT - error connecting with MQTT Return {}".format(rc)) - - else: - _LOGGER.debug("EcoVacsMQTT - Connected with result code "+str(rc)) - _LOGGER.debug("EcoVacsMQTT - Subscribing to all") - - self.subscribe('iot/atr/+/' + self.vacuum['did'] + '/' + self.vacuum['class'] + '/' + self.vacuum['resource'] + '/+', qos=0) - self.ready_flag.set() - - #def on_log(self, client, userdata, level, buf): #This is very noisy and verbose - # _LOGGER.debug("EcoVacsMQTT Log: {} ".format(buf)) - - def send_ping(self): - _LOGGER.debug("*** MQTT sending ping ***") - rc = self._send_simple_command(MQTTPublish.paho.PINGREQ) - if rc == MQTTPublish.paho.MQTT_ERR_SUCCESS: - return True - else: - return False - - def send_command(self, action, recipient): - if action.name == "Clean": #For handling Clean when action not specified (i.e. CLI) - action.args['clean']['act'] = CLEAN_ACTION_TO_ECOVACS['start'] #Inject a start action - c = self._wrap_command(action, recipient) - _LOGGER.debug('Sending command {0}'.format(c)) - self._handle_ctl_api(action, - self.__call_iotdevmanager_api(c ,verify_ssl=self.verify_ssl ) - ) - - def _wrap_command(self, cmd, recipient): - #Remove the td from ctl xml for RestAPI - payloadxml = cmd.to_xml() - payloadxml.attrib.pop("td") - - return { - 'auth': { - 'realm': EcoVacsAPI.REALM, - 'resource': self.resource, - 'token': self.secret, - 'userid': self.user, - 'with': 'users', - }, - "cmdName": cmd.name, - "payload": ET.tostring(payloadxml).decode(), - - "payloadType": "x", - "td": "q", - "toId": recipient, - "toRes": self.vacuum['resource'], - "toType": self.vacuum['class'] - } - - def __call_iotdevmanager_api(self, args, verify_ssl=True): - _LOGGER.debug("calling iotdevmanager api with {}".format(args)) - params = {} - params.update(args) - - url = (EcoVacsAPI.PORTAL_URL_FORMAT + "/iot/devmanager.do").format(continent=self.continent) - response = None - try: #The RestAPI sometimes doesnt provide a response depending on command, reduce timeout to 3 to accomodate and make requests faster - response = requests.post(url, json=params, timeout=3, verify=verify_ssl) #May think about having timeout as an arg that could be provided in the future - except requests.exceptions.ReadTimeout: - _LOGGER.debug("call to iotdevmanager failed with ReadTimeout") - return {} - - json = response.json() - if json['ret'] == 'ok': - return json - elif json['ret'] == 'fail': - if 'debug' in json: - if json['debug'] == 'wait for response timed out': - #TODO - Maybe handle timeout for IOT better in the future - _LOGGER.error("call to iotdevmanager failed with {}".format(json)) - return {} - else: - #TODO - Not sure if we want to raise an error yet, just return empty for now - _LOGGER.error("call to iotdevmanager failed with {}".format(json)) - return {} - #raise RuntimeError( - #"failure {} ({}) for call {} and parameters {}".format(json['error'], json['errno'], function, params)) - - def _handle_ctl_api(self, action, message): - if not message == {}: - resp = self._ctl_to_dict_api(action, message['resp']) - if resp is not None: - for s in self.ctl_subscribers: - s(resp) - - def _ctl_to_dict_api(self, action, xmlstring): - xml = ET.fromstring(xmlstring) - - xmlchild = xml.getchildren() - if len(xmlchild) > 0: - result = xmlchild[0].attrib.copy() - #Fix for difference in XMPP vs API response - #Depending on the report will use the tag and add "report" to fit the mold of sucks library - if xmlchild[0].tag == "clean": - result['event'] = "CleanReport" - elif xmlchild[0].tag == "charge": - result['event'] = "ChargeState" - elif xmlchild[0].tag == "battery": - result['event'] = "BatteryInfo" - else: #Default back to replacing Get from the api cmdName - result['event'] = action.name.replace("Get","",1) - - else: - result = xml.attrib.copy() - result['event'] = action.name.replace("Get","",1) - if 'ret' in result: #Handle errors as needed - if result['ret'] == 'fail': - if action.name == "Charge": #So far only seen this with Charge, when already docked - result['event'] = "ChargeState" - - for key in result: - if not RepresentsInt(result[key]): #Fix to handle negative int values - result[key] = stringcase.snakecase(result[key]) - - return result - - def _handle_ctl_mqtt(self, client, userdata, message): - #_LOGGER.debug("EcoVacs MQTT Received Message on Topic: {} - Message: {}".format(message.topic, str(message.payload.decode("utf-8")))) - as_dict = self._ctl_to_dict_mqtt(message.topic, str(message.payload.decode("utf-8"))) - if as_dict is not None: - for s in self.ctl_subscribers: - s(as_dict) - - def _ctl_to_dict_mqtt(self, topic, xmlstring): - #I haven't seen the need to fall back to data within the topic (like we do with IOT rest call actions), but it is here in case of future need - xml = ET.fromstring(xmlstring) #Convert from string to xml (like IOT rest calls), other than this it is similar to XMPP - - #Including changes from jasonarends @ 28da7c2 below - result = xml.attrib.copy() - if 'td' not in result: - # This happens for commands with no response data, such as PlaySound - # Handle response data with no 'td' - - if 'type' in result: # single element with type and val - result['event'] = "LifeSpan" # seems to always be LifeSpan type - - else: - if len(xml) > 0: # case where there is child element - if 'clean' in xml[0].tag: - result['event'] = "CleanReport" - elif 'charge' in xml[0].tag: - result['event'] = "ChargeState" - elif 'battery' in xml[0].tag: - result['event'] = "BatteryInfo" - else: - return - result.update(xml[0].attrib) - else: # for non-'type' result with no child element, e.g., result of PlaySound - return - else: # response includes 'td' - result['event'] = result.pop('td') - if xml: - result.update(xml[0].attrib) - - for key in result: - #Check for RepresentInt to handle negative int values, and ',' for ignoring position updates - if not RepresentsInt(result[key]) and ',' not in result[key]: - result[key] = stringcase.snakecase(result[key]) - - return result - - -class EcoVacsXMPP(ClientXMPP): - def __init__(self, user, domain, resource, secret, continent, vacuum, server_address=None ): - ClientXMPP.__init__(self, "{}@{}/{}".format(user, domain,resource), '0/' + resource + '/' + secret) #Init with resource to bind it - self.user = user - self.domain = domain - self.resource = resource - self.continent = continent - self.vacuum = vacuum - self.credentials['authzid'] = user - if server_address is None: - self.server_address = ('msg-{}.ecouser.net'.format(self.continent), '5223') - else: - self.server_address = server_address - self.add_event_handler("session_start", self.session_start) - self.ctl_subscribers = [] - self.ready_flag = Event() - - - def wait_until_ready(self): - self.ready_flag.wait() - - def session_start(self, event): - _LOGGER.debug("----------------- starting session ----------------") - _LOGGER.debug("event = {}".format(event)) - self.register_handler(Callback("general", - MatchXPath('{jabber:client}iq/{com:ctl}query/{com:ctl}'), - self._handle_ctl)) - self.ready_flag.set() - - def subscribe_to_ctls(self, function): - self.ctl_subscribers.append(function) - - def _handle_ctl(self, message): - the_good_part = message.get_payload()[0][0] - as_dict = self._ctl_to_dict(the_good_part) - if as_dict is not None: - for s in self.ctl_subscribers: - s(as_dict) - - def _ctl_to_dict(self, xml): - result = xml.attrib.copy() - if 'td' not in result: - # This happens for commands with no response data, such as PlaySound - return - - result['event'] = result.pop('td') - if xml: - result.update(xml[0].attrib) - - for key in result: - if not RepresentsInt(result[key]): #Fix to handle negative int values - result[key] = stringcase.snakecase(result[key]) - - return result - - def register_callback(self, userdata, message): - - self.register_handler(Callback(kind, - MatchXPath('{jabber:client}iq/{com:ctl}query/{com:ctl}ctl[@td="' + kind + '"]'), - function)) - - def send_command(self, xml, recipient): - c = self._wrap_command(xml, recipient) - _LOGGER.debug('Sending command {0}'.format(c)) - c.send() - - def _wrap_command(self, ctl, recipient): - q = self.make_iq_query(xmlns=u'com:ctl', ito=recipient, ifrom=self._my_address()) - q['type'] = 'set' - if not "id" in ctl.attrib: - ctl.attrib["id"] = self.getReqID() #If no ctl id provided, add an id to the ctl. This was required for the ozmo930 and shouldn't hurt others - for child in q.xml: - if child.tag.endswith('query'): - child.append(ctl) - return q - - def getReqID(self, customid="0"): #Generate a somewhat random string for request id, with minium 8 chars. Works similar to ecovacs app. - if customid != "0": - return "{}".format(customid) #return provided id as string - else: - rtnval = str(random.randint(1,50)) - while len(str(rtnval)) <= 8: - rtnval = "{}{}".format(rtnval,random.randint(0,50)) - - return "{}".format(rtnval) #return as string - - def _my_address(self): - if not self.vacuum['iotmq']: - return self.user + '@' + self.domain + '/' + self.boundjid.resource - else: - return self.user + '@' + self.domain + '/' + self.resource - - - def send_ping(self, to): - q = self.make_iq_get(ito=to, ifrom=self._my_address()) - q.xml.append(ET.Element('ping', {'xmlns': 'urn:xmpp:ping'})) - _LOGGER.debug("*** sending ping ***") - q.send() - - def connect_and_wait_until_ready(self): - self.connect(self.server_address) - self.process() - self.wait_until_ready() - -class VacBotCommand: - ACTION = { - 'forward': 'forward', - 'backward': 'backward', - 'left': 'SpinLeft', - 'right': 'SpinRight', - 'turn_around': 'TurnAround', - 'stop': 'stop' - } - - def __init__(self, name, args=None, **kwargs): - if args is None: - args = {} - self.name = name - self.args = args - - def to_xml(self): - ctl = ET.Element('ctl', {'td': self.name}) - for key, value in self.args.items(): - if type(value) is dict: - inner = ET.Element(key, value) - ctl.append(inner) - elif type(value) is list: - for item in value: - ixml = self.listobject_to_xml(key, item) - ctl.append(ixml) - else: - ctl.set(key, value) - - return ctl - - def __str__(self, *args, **kwargs): - return self.command_name() + " command" - - def command_name(self): - return self.__class__.__name__.lower() - - def listobject_to_xml(self, tag, conv_object): - rtnobject = ET.Element(tag) - if type(conv_object) is dict: - for key, value in conv_object.items(): - rtnobject.set(key, value) - else: - rtnobject.set(tag, conv_object) - return rtnobject - -class Clean(VacBotCommand): - def __init__(self, mode='auto', speed='normal', iotmq=False, action='start',terminal=False, **kwargs): - if kwargs == {}: - #Looks like action is needed for some bots, shouldn't affect older models - super().__init__('Clean', {'clean': {'type': CLEAN_MODE_TO_ECOVACS[mode], 'speed': FAN_SPEED_TO_ECOVACS[speed],'act': CLEAN_ACTION_TO_ECOVACS[action]}}) - else: - initcmd = {'type': CLEAN_MODE_TO_ECOVACS[mode], 'speed': FAN_SPEED_TO_ECOVACS[speed]} - for kkey, kvalue in kwargs.items(): - initcmd[kkey] = kvalue - super().__init__('Clean', {'clean': initcmd}) - -class Edge(Clean): - def __init__(self): - super().__init__('edge', 'high') - - -class Spot(Clean): - def __init__(self): - super().__init__('spot', 'high') - - -class Stop(Clean): - def __init__(self): - super().__init__('stop', 'normal') - -class SpotArea(Clean): - def __init__(self, action='start', area='', map_position='', cleanings='1'): - if area != '': #For cleaning specified area - super().__init__('spot_area', 'normal', act=CLEAN_ACTION_TO_ECOVACS[action], mid=area) - elif map_position != '': #For cleaning custom map area, and specify deep amount 1x/2x - super().__init__('spot_area' ,'normal',act=CLEAN_ACTION_TO_ECOVACS[action], p=map_position, deep=cleanings) - else: - #no valid entries - raise ValueError("must provide area or map_position for spotarea clean") - -class Charge(VacBotCommand): - def __init__(self): - super().__init__('Charge', {'charge': {'type': CHARGE_MODE_TO_ECOVACS['return']}}) - - -class Move(VacBotCommand): - def __init__(self, action): - super().__init__('Move', {'move': {'action': self.ACTION[action]}}) - - -class PlaySound(VacBotCommand): - def __init__(self, sid="0"): - super().__init__('PlaySound', {'sid': sid}) - - -class GetCleanState(VacBotCommand): - def __init__(self): - super().__init__('GetCleanState') - - -class GetChargeState(VacBotCommand): - def __init__(self): - super().__init__('GetChargeState') - - -class GetBatteryState(VacBotCommand): - def __init__(self): - super().__init__('GetBatteryInfo') - - -class GetLifeSpan(VacBotCommand): - def __init__(self, component): - super().__init__('GetLifeSpan', {'type': COMPONENT_TO_ECOVACS[component]}) - - -class SetTime(VacBotCommand): - def __init__(self, timestamp, timezone): - super().__init__('SetTime', {'time': {'t': timestamp, 'tz': timezone}}) diff --git a/sucks/cli.py b/sucks/cli.py deleted file mode 100644 index 843cc5a..0000000 --- a/sucks/cli.py +++ /dev/null @@ -1,244 +0,0 @@ -import configparser -import itertools -import os -import platform -import random -import re - -import click -from pycountry_convert import country_alpha2_to_continent_code - -from sucks import * - -_LOGGER = logging.getLogger(__name__) - - -class FrequencyParamType(click.ParamType): - name = 'frequency' - RATIONAL_PATTERN = re.compile(r'([.0-9]+)/([.0-9]+)') - - def convert(self, value, param, ctx): - result = None - try: - search = self.RATIONAL_PATTERN.search(value) - if search: - result = float(search.group(1)) / float(search.group(2)) - else: - try: - result = float(value) - except ValueError: - pass - except (ValueError, ArithmeticError): - pass - - if result is None: - self.fail('%s is not a valid frequency' % value, param, ctx) - if 0 <= result <= 1: - return result - - self.fail('%s is not between 0 and 1' % value, param, ctx) - - -FREQUENCY = FrequencyParamType() - - -class BotWait(): - pass - - def wait(self, bot): - raise NotImplementedError() - - -class TimeWait(BotWait): - def __init__(self, seconds): - super().__init__() - self.seconds = seconds - - def wait(self, bot): - click.echo("waiting for " + str(self.seconds) + "s") - time.sleep(self.seconds) - - -class StatusWait(BotWait): - def __init__(self, wait_on, wait_for): - super().__init__() - self.wait_on = wait_on - self.wait_for = wait_for - - def wait(self, bot): - if not hasattr(bot, self.wait_on): - raise ValueError("object " + bot + " does not have method " + self.wait_on) - _LOGGER.debug("waiting on " + self.wait_on + " for value " + self.wait_for) - - while getattr(bot, self.wait_on) != self.wait_for: - time.sleep(0.5) - _LOGGER.debug("wait complete; " + self.wait_on + " is now " + self.wait_for) - - -class CliAction: - def __init__(self, vac_command, terminal=False, wait=None): - self.vac_command = vac_command - self.terminal = terminal - self.wait = wait - - -def config_file(): - if platform.system() == 'Windows': - return os.path.join(os.getenv('APPDATA'), 'sucks.conf') - else: - return os.path.expanduser('~/.config/sucks.conf') - - -def config_file_exists(): - return os.path.isfile(config_file()) - - -def read_config(): - parser = configparser.ConfigParser() - with open(config_file()) as fp: - parser.read_file(itertools.chain(['[global]'], fp), source=config_file()) - return parser['global'] - - -def write_config(config): - os.makedirs(os.path.dirname(config_file()), exist_ok=True) - with open(config_file(), 'w') as fp: - for key in config: - fp.write(key + '=' + str(config[key]) + "\n") - - -def current_country(): - # noinspection PyBroadException - try: - return requests.get('http://ipinfo.io/json').json()['country'].lower() - except: - return 'us' - - -def continent_for_country(country_code): - return country_alpha2_to_continent_code(country_code.upper()).lower() - - -def should_run(frequency): - if frequency is None: - return True - n = random.random() - result = n <= frequency - _LOGGER.debug("tossing coin: {:0.3f} <= {:0.3f}: {}".format(n, frequency, result)) - return result - - -@click.group(chain=True) -@click.option('--debug/--no-debug', default=False) -def cli(debug): - logging.basicConfig(format='%(name)-10s %(levelname)-8s %(message)s') - _LOGGER.parent.setLevel(logging.DEBUG if debug else logging.ERROR) - - -@cli.command(help='logs in with specified email; run this first') -@click.option('--email', prompt='Ecovacs app email') -@click.option('--password', prompt='Ecovacs app password', hide_input=True) -@click.option('--country-code', prompt='your two-letter country code', default=lambda: current_country()) -@click.option('--continent-code', prompt='your two-letter continent code', - default=lambda: continent_for_country(click.get_current_context().params['country_code'])) -@click.option('--verify-ssl', prompt='Verify SSL for API requests', default=True) -def login(email, password, country_code, continent_code, verify_ssl): - if config_file_exists() and not click.confirm('overwrite existing config?'): - click.echo("Skipping login.") - exit(0) - config = OrderedDict() - password_hash = EcoVacsAPI.md5(password) - device_id = EcoVacsAPI.md5(str(time.time())) - try: - EcoVacsAPI(device_id, email, password_hash, country_code, continent_code, verify_ssl) - except ValueError as e: - click.echo(e.args[0]) - exit(1) - config['email'] = email - config['password_hash'] = password_hash - config['device_id'] = device_id - config['country'] = country_code.lower() - config['continent'] = continent_code.lower() - config['verify_ssl'] = verify_ssl - write_config(config) - click.echo("Config saved.") - exit(0) - - -@cli.command(help='auto-cleans for the specified number of minutes, if minutes is 0 auto clean until bot returns to charger by itself') -@click.option('--frequency', '-f', type=FREQUENCY, help='frequency with which to run; e.g. 0.5 or 3/7') -@click.argument('minutes', type=click.FLOAT) -def clean(frequency, minutes): - waiter = StatusWait('charge_status', 'charging') - if minutes > 0: - waiter = TimeWait(minutes * 60) - - if should_run(frequency): - return CliAction(Clean(), wait=waiter) - - -@cli.command(help='cleans room edges for the specified number of minutes') -@click.option('--frequency', '-f', type=FREQUENCY, help='frequency with which to run; e.g. 0.5 or 3/7') -@click.argument('minutes', type=click.FLOAT) -def edge(frequency, minutes): - if should_run(frequency): - return CliAction(Edge(), wait=TimeWait(minutes * 60)) - - -@cli.command(help='cleans provided area(s), ex: "0,1"',context_settings={"ignore_unknown_options": True}) #ignore_unknown for map coordinates with negatives -@click.option("--map-position","-p", is_flag=True, help='clean provided map position instead of area, ex: "-602,1812,800,723"') -@click.argument('area', type=click.STRING, required=True) -def area(area, map_position): - if map_position: - return CliAction(SpotArea('start', map_position=area), wait=StatusWait('charge_status', 'returning')) - else: - return CliAction(SpotArea('start', area=area), wait=StatusWait('charge_status', 'returning')) - - -@cli.command(help='returns to charger') -def charge(): - return charge_action() - - -def charge_action(): - return CliAction(Charge(), terminal=True, wait=StatusWait('charge_status', 'charging')) - - -@cli.command(help='stops the robot in its current position') -def stop(): - return CliAction(Stop(), terminal=True, wait=StatusWait('clean_status', 'stop')) - - -@cli.resultcallback() -def run(actions, debug): - actions = list(filter(None.__ne__, actions)) - if actions and charge and not actions[-1].terminal: - actions.append(charge_action()) - - if not config_file_exists(): - click.echo("Not logged in. Do 'click login' first.") - exit(1) - - if debug: - _LOGGER.debug("will run {}".format(actions)) - - if actions: - config = read_config() - api = EcoVacsAPI(config['device_id'], config['email'], config['password_hash'], - config['country'], config['continent'], verify_ssl=config['verify_ssl']) - vacuum = api.devices()[0] - vacbot = VacBot(api.uid, api.REALM, api.resource, api.user_access_token, vacuum, config['continent'], verify_ssl=config['verify_ssl']) - vacbot.connect_and_wait_until_ready() - - for action in actions: - click.echo("performing " + str(action.vac_command)) - vacbot.run(action.vac_command) - action.wait.wait(vacbot) - - vacbot.disconnect(wait=True) - - click.echo("done") - - -if __name__ == '__main__': - cli() diff --git a/tests/test_cli.py b/tests/test_cli.py deleted file mode 100644 index 0ccd861..0000000 --- a/tests/test_cli.py +++ /dev/null @@ -1,62 +0,0 @@ -import tempfile -from unittest.mock import Mock, patch - -from nose.tools import * - -from sucks.cli import * - - -def test_config_file_name(): - if platform.system() == 'Windows': - print(config_file()) - assert_true(re.match(r'[A-Z]:\\.+\\\w+\\AppData(\\Roaming)?\\sucks.conf', config_file())) - else: - assert_true(re.match(r'/.+/\w+/.config/sucks.conf', config_file())) - - -def test_write_and_read_config(): - with patch('sucks.cli.config_file', - Mock(return_value=os.path.join(tempfile.mkdtemp(), 'some_other_dir', 'sucks.conf'))): - write_config({'a': "ayyy", 'b': 2}) - config2 = read_config() - assert_equals(config2['a'], 'ayyy') - assert_equals(config2['b'], '2') - - -def test_frequency_param_type(): - t = FREQUENCY - assert_equals(t.convert('0', None, None), 0) - assert_equals(t.convert('1', None, None), 1.0) - assert_equals(t.convert('1/2', None, None), 0.5) - assert_equals(t.convert('1/7', None, None), 1.0 / 7.0) - assert_equals(t.convert('1/14', None, None), 1.0 / 14.0) - assert_equals(t.convert('1/1000', None, None), 1.0 / 1000.0) - assert_equals(t.convert('1.5/2', None, None), 1.5 / 2.0) - assert_equals(t.convert('0/2', None, None), 0) - with assert_raises(click.exceptions.BadParameter): - t.convert('bob', None, None) - with assert_raises(click.exceptions.BadParameter): - t.convert('2', None, None) - with assert_raises(click.exceptions.BadParameter): - t.convert('7/5', None, None) - with assert_raises(click.exceptions.BadParameter): - t.convert('1/0', None, None) - - -def test_should_run(): - count = 0 - for _ in range(10000): - if should_run(0.1): - count += 1 - assert_almost_equal(count, 1000, delta=200) - - count = 0 - for _ in range(10000): - if should_run(0.9): - count += 1 - assert_almost_equal(count, 9000, delta=200) - - -def test_continent_for_country(): - assert_equal(continent_for_country('us'), 'na') - assert_equal(continent_for_country('fr'), 'eu') diff --git a/tests/test_commands.py b/tests/test_commands.py deleted file mode 100644 index c40baf4..0000000 --- a/tests/test_commands.py +++ /dev/null @@ -1,184 +0,0 @@ -from xml.etree import ElementTree - -from nose.tools import * - -from sucks import * - - -def test_custom_command(): - # Ensure a custom-built command generates the expected XML payload - c = VacBotCommand('CustomCommand', {'type': 'customtype'}) - assert_equals(ElementTree.tostring(c.to_xml()), - b'') - - -def test_custom_command_inner_tag(): - # Ensure a custom-built command generates the expected XML payload - c = VacBotCommand('CustomCommand', {'customtag': {'customvar': 'customvalue'}}) - assert_equals(ElementTree.tostring(c.to_xml()), - b'') - - -def test_custom_command_multiple_inner_tag(): - # Ensure a custom-built command with multiple inner tags generates the expected XML payload - c = VacBotCommand('CustomCommand', {"customtag":[{"customvar":"customvalue1"},{"customvar":"customvalue2"}]}) - logging.info(ElementTree.tostring(c.to_xml())) - assert_equals(ElementTree.tostring(c.to_xml()), - b'') - -def test_custom_command_args_multiple_inner_tag(): - # Ensure a custom-built command with args and multiple inner tags generates the expected XML payload - c = VacBotCommand('CustomCommand', {"arg1":"value1","customtag":[{"customvar":"customvalue1"},{"customvar":"customvalue2"}]}) - assert_equals(ElementTree.tostring(c.to_xml()), - b'') - - -def test_custom_command_noargs(): - # Ensure a custom-built command with no args generates XML without an args element - c = VacBotCommand('CustomCommand') - assert_equals(ElementTree.tostring(c.to_xml()), - b'') - - -def test_clean_command(): - c = Clean() - assert_equals(ElementTree.tostring(c.to_xml()), - b'') # protocol has attribs in other order - - c = Clean('edge', 'high') - assert_equals(ElementTree.tostring(c.to_xml()), - b'') # protocol has attribs in other order - - c = Clean(iotmq=True) - assert_equals(ElementTree.tostring(c.to_xml()), - b'') # test for iot act is added - - -def test_spotarea_command(): - assert_raises(ValueError, SpotArea, 'start') #Value error if SpotArea doesn't include a mid or p - - c = SpotArea('start', '0') - assert_equals(ElementTree.tostring(c.to_xml()), - b'') #Test namedarea clean - - c = SpotArea('start', area='0') - assert_equals(ElementTree.tostring(c.to_xml()), - b'') #Test namedarea keyword clean - - c = SpotArea('start', '', '-602,1812,800,723') - assert_equals(ElementTree.tostring(c.to_xml()), - b'') #Test customarea clean - - c = SpotArea('start', '', '-602,1812,800,723', '2') - assert_equals(ElementTree.tostring(c.to_xml()), - b'') #Test customarea clean with deep 2 - - c = SpotArea('start', '', map_position='-602,1812,800,723') - assert_equals(ElementTree.tostring(c.to_xml()), - b'') #Test customarea keyword clean with deep default - - c = SpotArea('start', map_position='-602,1812,800,723', cleanings='2') - assert_equals(ElementTree.tostring(c.to_xml()), - b'') #Test customarea keyword and cleanings keyword clean with deep default - - c = SpotArea('start', area='0', map_position='-602,1812,800,723', cleanings='2') - assert_equals(ElementTree.tostring(c.to_xml()), - b'') #Test all keywords specified, should default to only mid - - c = SpotArea('start', '0', '-602,1812,800,723','2') - assert_equals(ElementTree.tostring(c.to_xml()), - b'') #Test all keywords specified, should default to only mid - - -def test_edge_command(): - c = Edge() - assert_equals(ElementTree.tostring(c.to_xml()), - b'') # protocol has attribs in other order - - -def test_spot_command(): - c = Spot() - assert_equals(ElementTree.tostring(c.to_xml()), - b'') # protocol has attribs in other order - - -def test_charge_command(): - c = Charge() - assert_equals(ElementTree.tostring(c.to_xml()), - b'') - - -def test_stop_command(): - c = Stop() - assert_equals(ElementTree.tostring(c.to_xml()), - b'') - - -def test_play_sound_command(): - c = PlaySound() - assert_equals(ElementTree.tostring(c.to_xml()), - b'') - - -def test_play_sound_command_with_sid(): - c = PlaySound(sid="1") - assert_equals(ElementTree.tostring(c.to_xml()), - b'') - - -def test_get_clean_state_command(): - c = GetCleanState() - assert_equals(ElementTree.tostring(c.to_xml()), - b'') - - -def test_get_charge_state_command(): - c = GetChargeState() - assert_equals(ElementTree.tostring(c.to_xml()), - b'') - - -def test_get_battery_state_command(): - c = GetBatteryState() - assert_equals(ElementTree.tostring(c.to_xml()), - b'') - - -def test_move_command(): - c = Move(action='left') - assert_equals(ElementTree.tostring(c.to_xml()), - b'') - c = Move(action='right') - assert_equals(ElementTree.tostring(c.to_xml()), - b'') - c = Move(action='turn_around') - assert_equals(ElementTree.tostring(c.to_xml()), - b'') - c = Move(action='forward') - assert_equals(ElementTree.tostring(c.to_xml()), - b'') - c = Move(action='backward') - assert_equals(ElementTree.tostring(c.to_xml()), - b'') - c = Move(action='stop') - assert_equals(ElementTree.tostring(c.to_xml()), - b'') - - -def test_get_lifepsan_command(): - c = GetLifeSpan('main_brush') - assert_equals(ElementTree.tostring(c.to_xml()), - b'') - - c = GetLifeSpan('side_brush') - assert_equals(ElementTree.tostring(c.to_xml()), - b'') - - c = GetLifeSpan('filter') - assert_equals(ElementTree.tostring(c.to_xml()), - b'') - -def test_set_time_command(): - c = SetTime('1234', 'GMT-5') - assert_equals(ElementTree.tostring(c.to_xml()), - b'') diff --git a/tests/test_ecovacs_api.py b/tests/test_ecovacs_api.py deleted file mode 100644 index 65d4f08..0000000 --- a/tests/test_ecovacs_api.py +++ /dev/null @@ -1,190 +0,0 @@ -from re import compile - -import requests_mock -from nose.tools import * - -from sucks import * - - -def test_md5(): - assert_equal(EcoVacsAPI.md5("fnord"), "b15e400c8dbd6697f26385216d32a40f") - - -def test_encrypt(): - assert_equal(len(EcoVacsAPI.encrypt("fnord")), 172) - - -def test_main_api_setup(): - with requests_mock.mock() as m: - r1 = m.get(compile('user/login'), - text='{"time": 1511200804243, "data": {"accessToken": "7a375650b0b1efd780029284479c4e41", "uid": "2017102559f0ee63c588d", "username": null, "email": "william-ecovacs@pota.to", "country": "us"}, "code": "0000", "msg": "X"}') - r2 = m.get(compile('user/getAuthCode'), - text='{"time": 1511200804607, "data": {"authCode": "5c28dac1ff580210e11292df57e87bef"}, "code": "0000", "msg": "X"}') - r3 = m.post(compile('user.do'), - text='{"todo": "result", "token": "jt5O7oDR3gPHdVKCeb8Czx8xw8mDXM6s", "result": "ok", "userId": "2017102559f0ee63c588d", "resource": "f8d99c4d"}') - - api = EcoVacsAPI("long_device_id", "account_id", "password_hash", 'us', 'na') - - # verify setup - assert_equals(api.resource, "long_dev") - - # verify calls - assert_equals(r1.call_count, 1) - assert_equals(r2.call_count, 1) - assert_equals(r3.call_count, 1) - - # verify state - assert_equals(api.uid, "2017102559f0ee63c588d") - assert_equals(api.login_access_token, "7a375650b0b1efd780029284479c4e41") - assert_equals(api.auth_code, "5c28dac1ff580210e11292df57e87bef") - assert_equals(api.user_access_token, "jt5O7oDR3gPHdVKCeb8Czx8xw8mDXM6s") - - #Test old user api endpoint - postdata = {'country': 'US', - 'resource': "f8d99c4d", - 'realm': EcoVacsAPI.REALM, - 'userId': "2017102559f0ee63c588d", - 'token': "jt5O7oDR3gPHdVKCeb8Czx8xw8mDXM6s"} - - r = api._EcoVacsAPI__call_user_api("loginByItToken", postdata) - assert_equals(r3.call_count, 2) - # verify state - assert_equals(api.uid, "2017102559f0ee63c588d") - assert_equals(api.login_access_token, "7a375650b0b1efd780029284479c4e41") - assert_equals(api.auth_code, "5c28dac1ff580210e11292df57e87bef") - assert_equals(api.user_access_token, "jt5O7oDR3gPHdVKCeb8Czx8xw8mDXM6s") - - - - -def test_main_api_setup_with_alternate_uid(): - # Under mysterious circumstances, for certain people the last call sometimes returns a different userId - # along with the user access token. If that's the case, we should use that as the UID for future calls - - with requests_mock.mock() as m: - r1 = m.get(compile('user/login'), - text='{"time": 1511200804243, "data": {"accessToken": "7a375650b0b1efd780029284479c4e41", "uid": "2017102559f0ee63c588d", "username": null, "email": "william-ecovacs@pota.to", "country": "us"}, "code": "0000", "msg": "X"}') - r2 = m.get(compile('user/getAuthCode'), - text='{"time": 1511200804607, "data": {"authCode": "5c28dac1ff580210e11292df57e87bef"}, "code": "0000", "msg": "X"}') - r3 = m.post(compile('user.do'), - text='{"todo": "result", "token": "jt5O7oDR3gPHdVKCeb8Czx8xw8mDXM6s", "result": "ok", "userId": "abcdef", "resource": "f8d99c4d"}') - - api = EcoVacsAPI("long_device_id", "account_id", "password_hash", 'us', 'na') - - assert_equals(r1.call_count, 1) - assert_equals(r2.call_count, 1) - assert_equals(r3.call_count, 1) - - # verify state - assert_equals(api.uid, "abcdef") - assert_equals(api.login_access_token, "7a375650b0b1efd780029284479c4e41") - assert_equals(api.auth_code, "5c28dac1ff580210e11292df57e87bef") - assert_equals(api.user_access_token, "jt5O7oDR3gPHdVKCeb8Czx8xw8mDXM6s") - -def test_main_api_errorcode(): - with requests_mock.mock() as m: - r1 = m.get(compile('user/login'), #test with 0004 (invalid token) - text='{"time": 1511200804243, "code": "0004", "msg": "X", "data": null}') - - assert_raises(RuntimeError, EcoVacsAPI, "long_device_id", "account_id", "password_hash", 'us', 'na') #Runtime error from code 0004 - - -def test_main_api_badpassword(): - with requests_mock.mock() as m: - r1 = m.get(compile('user/login'), #test with 1005 (incorrect email or password) - text='{"time": 1511200804243, "code": "1005", "msg": "X", "data": null}') - - assert_raises(ValueError, EcoVacsAPI, "long_device_id", "account_id", "password_hash", 'us', 'na') #ValueError error from code 1005 - - -def test_device_lookup(): - api = make_api() - with requests_mock.mock() as m: - - #Not IOTMQ - device_id = 'E0000001234567890123' - device_class = '126' - device_company = 'eco-legacy' - r = m.post(compile('user.do'), - text='{"todo": "result", "devices": [{"did": "%s", "company": "%s", "class": "%s", "nick": "bob"}], "result": "ok"}' %(device_id, device_company, device_class)) - - d = api.devices() - assert_equals(r.call_count, 1) - assert_equals(len(d), 1) - vacuum = d[0] - assert_equals(vacuum['did'], device_id) - assert_equals(vacuum['class'], '126') - assert_equals(vacuum['iotmq'], False) - - #Is IOTMQ - device_class = 'ls1ok3' #D900 - device_company = 'eco-ng' - r = m.post(compile('user.do'), - text='{"todo": "result", "devices": [{"did": "%s", "company": "%s", "class": "%s", "nick": "bob"}], "result": "ok"}' %(device_id, device_company, device_class)) - - d = api.devices() - assert_equals(r.call_count, 1) - assert_equals(len(d), 1) - vacuum = d[0] - assert_equals(vacuum['did'], device_id) - assert_equals(vacuum['class'], device_class) - assert_equals(vacuum['iotmq'], True) - - -def test_device_lookup_IOTProduct(): - api = make_api() - with requests_mock.mock() as m: - - #Is IOTProduct - device_id = 'E0000001234567890123' - device_class = 'ls1ok3' #D900 - device_company = 'eco-ng' - - r = m.post(compile('user.do'), - text='{"todo": "result", "devices": [{"did": "%s", "company": "%s", "class": "%s", "nick": "bob"}], "result": "ok"}' %(device_id, device_company, device_class)) - r = m.post(compile('pim/product/getProductIotMap'), - text='{"code":0,"data":[{"classid":"dl8fht","product":{"_id":"5acb0fa87c295c0001876ecf","name":"DEEBOT 600 Series","icon":"5acc32067c295c0001876eea","UILogicId":"dl8fht","ota":false,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5acc32067c295c0001876eea"}},{"classid":"02uwxm","product":{"_id":"5ae1481e7ccd1a0001e1f69e","name":"DEEBOT OZMO Slim10 Series","icon":"5b1dddc48bc45700014035a1","UILogicId":"02uwxm","ota":false,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b1dddc48bc45700014035a1"}},{"classid":"y79a7u","product":{"_id":"5b04c0227ccd1a0001e1f6a8","name":"DEEBOT OZMO 900","icon":"5b04c0217ccd1a0001e1f6a7","UILogicId":"y79a7u","ota":true,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b04c0217ccd1a0001e1f6a7"}},{"classid":"jr3pqa","product":{"_id":"5b43077b8bc457000140363e","name":"DEEBOT 711","icon":"5b5ac4cc8d5a56000111e769","UILogicId":"jr3pqa","ota":true,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b5ac4cc8d5a56000111e769"}},{"classid":"uv242z","product":{"_id":"5b5149b4ac0b87000148c128","name":"DEEBOT 710","icon":"5b5ac4e45f21100001882bb9","UILogicId":"uv242z","ota":true,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b5ac4e45f21100001882bb9"}},{"classid":"ls1ok3","product":{"_id":"5b6561060506b100015c8868","name":"DEEBOT 900 Series","icon":"5ba4a2cb6c2f120001c32839","UILogicId":"ls1ok3","ota":true,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5ba4a2cb6c2f120001c32839"}}]}') - - d = api.devices() - d = api.SetIOTDevices(d, api.getiotProducts()) - - assert_equals(r.call_count, 1) - assert_equals(len(d), 1) - vacuum = d[0] - assert_equals(vacuum['did'], device_id) - assert_equals(vacuum['class'], device_class) - assert_equals(vacuum['iot_product'], True) - assert_equals(vacuum['iotmq'], True) - - #Not IOTProduct - device_id = 'E0000001234567890123' - device_class = '126' - device_company = 'eco-legacy' - - r = m.post(compile('user.do'), - text='{"todo": "result", "devices": [{"did": "%s", "company": "%s", "class": "%s", "nick": "bob"}], "result": "ok"}' %(device_id, device_company, device_class)) - r = m.post(compile('pim/product/getProductIotMap'), - text='{"code":0,"data":[{"classid":"dl8fht","product":{"_id":"5acb0fa87c295c0001876ecf","name":"DEEBOT 600 Series","icon":"5acc32067c295c0001876eea","UILogicId":"dl8fht","ota":false,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5acc32067c295c0001876eea"}},{"classid":"02uwxm","product":{"_id":"5ae1481e7ccd1a0001e1f69e","name":"DEEBOT OZMO Slim10 Series","icon":"5b1dddc48bc45700014035a1","UILogicId":"02uwxm","ota":false,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b1dddc48bc45700014035a1"}},{"classid":"y79a7u","product":{"_id":"5b04c0227ccd1a0001e1f6a8","name":"DEEBOT OZMO 900","icon":"5b04c0217ccd1a0001e1f6a7","UILogicId":"y79a7u","ota":true,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b04c0217ccd1a0001e1f6a7"}},{"classid":"jr3pqa","product":{"_id":"5b43077b8bc457000140363e","name":"DEEBOT 711","icon":"5b5ac4cc8d5a56000111e769","UILogicId":"jr3pqa","ota":true,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b5ac4cc8d5a56000111e769"}},{"classid":"uv242z","product":{"_id":"5b5149b4ac0b87000148c128","name":"DEEBOT 710","icon":"5b5ac4e45f21100001882bb9","UILogicId":"uv242z","ota":true,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b5ac4e45f21100001882bb9"}},{"classid":"ls1ok3","product":{"_id":"5b6561060506b100015c8868","name":"DEEBOT 900 Series","icon":"5ba4a2cb6c2f120001c32839","UILogicId":"ls1ok3","ota":true,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5ba4a2cb6c2f120001c32839"}}]}') - - d = api.devices() - d = api.SetIOTDevices(d, api.getiotProducts()) - - assert_equals(r.call_count, 1) - assert_equals(len(d), 1) - vacuum = d[0] - assert_equals(vacuum['did'], device_id) - assert_equals(vacuum['class'], device_class) - assert_equals(vacuum['iot_product'], False) - assert_equals(vacuum['iotmq'], False) - -def make_api(): - with requests_mock.mock() as m: - m.get(compile('user/login'), - text='{"time": 1511200804243, "data": {"accessToken": "0123456789abcdef0123456789abcdef", "uid": "20170101abcdefabcdefa", "username": null, "email": "username@example.com", "country": "us"}, "code": "0000", "msg": "X"}') - m.get(compile('user/getAuthCode'), - text='{"time": 1511200804607, "data": {"authCode": "abcdef01234567890abcdef012345678"}, "code": "0000", "msg": "X"}') - m.post(compile('user.do'), - text='{"todo": "result", "token": "base64base64base64base64base64ba", "result": "ok", "userId": "20170101abcdefabcdefa", "resource": "abcdef12"}') - m.post(compile('pim/product/getProductIotMap'), - text='{"code":0,"data":[{"classid":"dl8fht","product":{"_id":"5acb0fa87c295c0001876ecf","name":"DEEBOT 600 Series","icon":"5acc32067c295c0001876eea","UILogicId":"dl8fht","ota":false,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5acc32067c295c0001876eea"}},{"classid":"02uwxm","product":{"_id":"5ae1481e7ccd1a0001e1f69e","name":"DEEBOT OZMO Slim10 Series","icon":"5b1dddc48bc45700014035a1","UILogicId":"02uwxm","ota":false,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b1dddc48bc45700014035a1"}},{"classid":"y79a7u","product":{"_id":"5b04c0227ccd1a0001e1f6a8","name":"DEEBOT OZMO 900","icon":"5b04c0217ccd1a0001e1f6a7","UILogicId":"y79a7u","ota":true,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b04c0217ccd1a0001e1f6a7"}},{"classid":"jr3pqa","product":{"_id":"5b43077b8bc457000140363e","name":"DEEBOT 711","icon":"5b5ac4cc8d5a56000111e769","UILogicId":"jr3pqa","ota":true,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b5ac4cc8d5a56000111e769"}},{"classid":"uv242z","product":{"_id":"5b5149b4ac0b87000148c128","name":"DEEBOT 710","icon":"5b5ac4e45f21100001882bb9","UILogicId":"uv242z","ota":true,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b5ac4e45f21100001882bb9"}},{"classid":"ls1ok3","product":{"_id":"5b6561060506b100015c8868","name":"DEEBOT 900 Series","icon":"5ba4a2cb6c2f120001c32839","UILogicId":"ls1ok3","ota":true,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5ba4a2cb6c2f120001c32839"}}]}') - return EcoVacsAPI("long_device_id", "account_id", "password_hash", 'us', 'na') diff --git a/tests/test_ecovacs_iotmq.py b/tests/test_ecovacs_iotmq.py deleted file mode 100644 index ceeff32..0000000 --- a/tests/test_ecovacs_iotmq.py +++ /dev/null @@ -1,236 +0,0 @@ -from re import search - -from nose.tools import * - -import requests_mock -import requests - -from sucks import * -import paho.mqtt - -# There are few tests for the MQTT stuff here because it's relatively complicated to test given -# the library's design and its multithreaded nature and lack of explicit testing support. - -def test_subscribe_to_ctls(): - response = None - - def save_response(value): - nonlocal response - response = value - - x = make_ecovacs_iotmq() - x.subscribe_to_ctls(save_response) - - #Test MQTT ctl - mqtt_message = paho.mqtt.client.MQTTMessage - mqtt_message.topic = 'iot/atr/CleanReport/%s/%s/%s/x'.format(x.vacuum['did'], x.vacuum['class'], x.vacuum['resource']) - mqtt_message.payload = b"" - x._handle_ctl_mqtt('','',mqtt_message) - assert_dict_equal(response, {'event': 'clean_report', 'ts':'1547824270099','type': 'auto','speed':'standard', 'st':'h','rsn':'a', 'a':'', 'l':'', 'sts':''}) - - #Test API ctl - api_message = {} - api_message['resp'] = ' ' - x.subscribe_to_ctls(save_response) - x._handle_ctl_api("Clean", api_message) - assert_dict_equal(response, {'event': 'clean_report', 'type': 'auto'}) - - -def test_is_iotmq(): - x = make_ecovacs_iotmq() - assert_equal(x.vacuum['iotmq'], True) - -def test_wrap_command(): - x = make_ecovacs_iotmq() - - c = x._wrap_command(Charge(), 'E0000000001234567890') - assert_equal(c['cmdName'], Charge().name) - assert_equal(c['toId'], 'E0000000001234567890') - assert_equal(c['payload'], '') - - -def test_iotapi_response(): - x = make_ecovacs_iotmq() - - with requests_mock.mock() as m: - url = (EcoVacsAPI.PORTAL_URL_FORMAT + "/iot/devmanager.do").format(continent=x.continent) - #Test GetCleanState - resp = {"ret":"ok","resp":"","id":"Qgxa"} - r1 = m.post(url, json=resp) - #r1 = m.post(compile('devmanager.do'), json=resp) - cmd = VacBotCommand("GetCleanState") - c = x._wrap_command(cmd, x.vacuum['did']) - rtnval = x._EcoVacsIOTMQ__call_iotdevmanager_api(c) - assert_equal(rtnval, {'ret':'ok','resp':"",'id':'Qgxa'}) - - #Test Exception ReadTimeout - r2 = m.post(url, exc=requests.exceptions.ReadTimeout) - #r2 = m.post(compile('iot/devmanager.do'),exc=requests.exceptions.ReadTimeout) - cmd = VacBotCommand("GetCleanState") - c = x._wrap_command(cmd, x.vacuum['did']) - rtnval = x._EcoVacsIOTMQ__call_iotdevmanager_api(c) - assert_equal(rtnval, {}) #Right now it sends back a blank object - - #Test Response Fail - Timeout - resp = {"ret":"fail","resp": None, "debug":"wait for response timed out" ,"id":"Qgxa"} - r2 = m.post(url, json=resp) - #r1 = m.post(compile('devmanager.do'), json=resp) - cmd = VacBotCommand("TestCommand") - c = x._wrap_command(cmd, x.vacuum['did']) - rtnval = x._EcoVacsIOTMQ__call_iotdevmanager_api(c) - assert_equal(rtnval, {}) - - #Test Response Fail - No debug - resp = {"ret":"fail","resp": None ,"id":"Qgxa"} - r2 = m.post(url, json=resp) - #r1 = m.post(compile('devmanager.do'), json=resp) - cmd = VacBotCommand("TestCommand") - c = x._wrap_command(cmd, x.vacuum['did']) - rtnval = x._EcoVacsIOTMQ__call_iotdevmanager_api(c) - assert_equal(rtnval, {}) - -def test_send_command(): - from unittest.mock import MagicMock - x = make_ecovacs_iotmq() - x._handle_ctl_api = MagicMock() - EcoVacsIOTMQ._EcoVacsIOTMQ__call_iotdevmanager_api = MagicMock() - x.send_command(Clean(iotmq=True), '123') - -def test_send_ping(): - from unittest.mock import MagicMock - x = make_ecovacs_iotmq() - EcoVacsIOTMQ._send_simple_command = MagicMock(return_value=MQTTPublish.paho.MQTT_ERR_SUCCESS) - assert_true(x.send_ping()) #Test ping response success - - EcoVacsIOTMQ._send_simple_command = MagicMock(return_value=MQTTPublish.paho.MQTT_ERR_NOT_FOUND) - assert_false(x.send_ping()) #Test ping response fail - -def test_on_connect_rc_nonzero(): - x = make_ecovacs_iotmq() - assert_raises(RuntimeError, x.on_connect, "client", "userdata", "flags", 1) - -def test_xml_to_dict_mqtt(): - x = make_ecovacs_iotmq() - - test_topic = 'iot/atr/CleanReport/%s/%s/%s/x'.format(x.vacuum['did'], x.vacuum['class'], x.vacuum['resource']) - assert_dict_equal( - x._ctl_to_dict_mqtt(test_topic, ""), - {'event': 'clean_report', 'ts':'1547824270099','type': 'auto','speed':'standard', 'st':'h','rsn':'a', 'a':'', 'l':'', 'sts':''}) - - assert_dict_equal( - x._ctl_to_dict_mqtt(test_topic, ""), - {'event': 'clean_report', 'ts':'1547824270099','type': 'auto','speed':'strong', 'st':'h','rsn':'a', 'a':'', 'l':'', 'sts':''}) - - assert_dict_equal( - x._ctl_to_dict_mqtt(test_topic, ""), - {'event': 'clean_report', 'ts':'1547824270099','type': 'auto','speed':'strong', 'st':'h','rsn':'a', 'a':'', 'l':'', 'sts':''}) #Test without td - - test_topic = 'iot/atr/BatteryInfo/%s/%s/%s/x'.format(x.vacuum['did'], x.vacuum['class'], x.vacuum['resource']) - assert_dict_equal( - x._ctl_to_dict_mqtt(test_topic, ""), - {'event': 'battery_info', 'ts':'1547823289924', 'power': '64'}) - - assert_dict_equal( - x._ctl_to_dict_mqtt(test_topic, ""), - {'event': 'battery_info', 'ts':'1547823289924', 'power': '64'}) #Test without td - - test_topic = 'iot/atr/SleepStatus/%s/%s/%s/x'.format(x.vacuum['did'], x.vacuum['class'], x.vacuum['resource']) - assert_dict_equal( - x._ctl_to_dict_mqtt(test_topic, ""), - {'event': 'sleep_status', 'ts':'1547823129670', 'st': '1'}) - - test_topic = 'iot/atr/errors/%s/%s/%s/x'.format(x.vacuum['did'], x.vacuum['class'], x.vacuum['resource']) - assert_dict_equal( - x._ctl_to_dict_mqtt(test_topic, ""), - {'event': 'errors', 'ts':'1547822982581','old':'','new':'102'}) - assert_dict_equal( - x._ctl_to_dict_mqtt(test_topic, ""), - {'event': 'errors', 'ts':'1547822982581','old':'102','new':''}) - - test_topic = 'iot/atr/Pos/%s/%s/%s/x'.format(x.vacuum['did'], x.vacuum['class'], x.vacuum['resource']) - assert_dict_equal( - x._ctl_to_dict_mqtt(test_topic, ""), - {'event': 'pos', 't':'p', 'p':'7,-10', 'a':'-42','valid':'0'}) - - test_topic = 'iot/atr/DustCaseST/%s/%s/%s/x'.format(x.vacuum['did'], x.vacuum['class'], x.vacuum['resource']) - assert_dict_equal( - x._ctl_to_dict_mqtt(test_topic, ""), - {'event': 'dust_case_s_t', 'ts':'1547822871328','st':'1'}) - - test_topic = 'iot/atr/MapSt/%s/%s/%s/x'.format(x.vacuum['did'], x.vacuum['class'], x.vacuum['resource']) - assert_dict_equal( - x._ctl_to_dict_mqtt(test_topic, ""), - {'event': 'map_st', 'ts':'1547823592934', 'st':'reloc_go_chg_start', 'method':'', 'info':''}) - - test_topic = 'iot/atr/LifeSpan/%s/%s/%s/x'.format(x.vacuum['did'], x.vacuum['class'], x.vacuum['resource']) - assert_dict_equal( - x._ctl_to_dict_mqtt(test_topic, ""), - {'event': 'life_span','ret':'ok', 'type': 'brush', 'left': '9876', 'total': '18000'}) - - test_topic = 'iot/atr/CustomCommand/%s/%s/%s/x'.format(x.vacuum['did'], x.vacuum['class'], x.vacuum['resource']) - assert_dict_equal( - x._ctl_to_dict_mqtt(test_topic, ''), - {'event': 'custom_command', 'customvar': 'customvalue1'}) - - -def test_xml_to_dict_api(): - x = make_ecovacs_iotmq() - message = {} - - cmd = VacBotCommand("Clean") - message['resp'] = "" - assert_dict_equal( - x._ctl_to_dict_api(cmd,message['resp']), - {'event': 'clean_report', 'type': 'auto', 'speed': 'standard', 'st':'h','t':'1159','a':'15','s':'0','tr':''}) - - cmd = VacBotCommand("Clean") - message['resp'] = "" - assert_dict_equal( - x._ctl_to_dict_api(cmd,message['resp']), - {'event': 'clean_report', 'type': 'auto', 'speed': 'strong', 'st':'h','t':'1159','a':'15','s':'0','tr':''}) - - cmd = VacBotCommand("GetBatteryInfo") - message['resp'] = "" - assert_dict_equal( - x._ctl_to_dict_api(cmd,message['resp']), - {'event': 'battery_info', 'power': '82'}) - - cmd = VacBotCommand("GetLifeSpan") - message['resp'] = "" - assert_dict_equal( - x._ctl_to_dict_api(cmd,message['resp']), - {'event': 'life_span','ret':'ok', 'type': 'brush', 'left': '9876', 'total': '18000'}) - - cmd = VacBotCommand("Charge") - message['resp'] = "" - assert_dict_equal( - x._ctl_to_dict_api(cmd,message['resp']), - {'type': 'going', 'h': '', 'r': 'a', 's': '', 'g': '0', 'event': 'charge_state'}) - - cmd = VacBotCommand("GetTestCommand") - message['resp'] = "" - assert_dict_equal( - x._ctl_to_dict_api(cmd,message['resp']), - {'type': 'command', 'event': 'test_command'}) #Test action.name.replace Get - - cmd = VacBotCommand("Charge") - message['resp'] = "" - assert_dict_equal( - x._ctl_to_dict_api(cmd,message['resp']), - {'event': 'charge_state','ret':'fail', 'errno': '8'}) #Test fail from charge command - - -def test_bad_port(): - bot = {"did": "E0000000001234567890", "class": "126","resource":"test_resource", "nick": "bob", "iotmq": True} - mqtt = EcoVacsIOTMQ('20170101abcdefabcdefa', 'ecouser.net', 'abcdef12', 'A1b2C3d4efghijklmNOPQrstuvwxyz12', 'na', bot, server_address='test.com:f123') - assert_equal(8883, mqtt.port) - -def test_good_port(): - bot = {"did": "E0000000001234567890", "class": "126","resource":"test_resource", "nick": "bob", "iotmq": True} - mqtt = EcoVacsIOTMQ('20170101abcdefabcdefa', 'ecouser.net', 'abcdef12', 'A1b2C3d4efghijklmNOPQrstuvwxyz12', 'na', bot, server_address='test.com:8000') - assert_equal(8000, mqtt.port) - -def make_ecovacs_iotmq(bot=None): - if bot is None: - bot = {"did": "E0000000001234567890", "class": "126","resource":"test_resource", "nick": "bob", "iotmq": True} - return EcoVacsIOTMQ('20170101abcdefabcdefa', 'ecouser.net', 'abcdef12', 'A1b2C3d4efghijklmNOPQrstuvwxyz12', 'na', bot) diff --git a/tests/test_ecovacs_xmpp.py b/tests/test_ecovacs_xmpp.py deleted file mode 100644 index 184aeaa..0000000 --- a/tests/test_ecovacs_xmpp.py +++ /dev/null @@ -1,91 +0,0 @@ -from re import search - -from nose.tools import * - -from sucks import * - - -# There are few tests for the XMPP stuff here because it's relatively complicated to test given -# the library's design and its multithreaded nature and lack of explicit testing support. - -def test_wrap_command(): - x = make_ecovacs_xmpp() - c = str(x._wrap_command(Clean().to_xml(), 'E0000000001234567890@126.ecorobot.net/atom')) - assert_true(search(r'from="20170101abcdefabcdefa@ecouser.net/abcdef12"', c)) - assert_true(search(r'to="E0000000001234567890@126.ecorobot.net/atom"', c)) - #Convert to XML to make it easy to see if id was added to ctl - xml_test = ET.fromstring(c) - ctl = xml_test.getchildren()[0][0] - assert_true(ctl.get("id")) #Check that an id was added to ctl - - #Test if customid is added to ctl - cwithid = Clean().to_xml() - cwithid.attrib["id"] = "12345678" #customid 12345678 - c = str(x._wrap_command(cwithid, 'E0000000001234567890@126.ecorobot.net/atom')) - #Convert to XML to make it easy to see if id was added to ctl - xml_test = ET.fromstring(c) - ctl = xml_test.getchildren()[0][0] - assert_equals(ctl.get("id"), "12345678") #Check that an id was added to ctl - - -def test_getReqID(): - x = make_ecovacs_xmpp() - rid = x.getReqID("12345678") - assert_equals(rid, "12345678") #Check returned ID is the same as provided - - rid2 = x.getReqID() - assert_true(len(rid2) >= 8) #Check returned random ID is at least 8 chars - -def test_subscribe_to_ctls(): - response = None - - def save_response(value): - nonlocal response - response = value - - x = make_ecovacs_xmpp() - - query = x.make_iq_query() - query.set_payload( - ET.fromstring(' ')) - - x.subscribe_to_ctls(save_response) - x._handle_ctl(query) - assert_dict_equal(response, {'event': 'clean_report', 'type': 'auto'}) - -def test_xml_to_dict(): - x = make_ecovacs_xmpp() - - assert_dict_equal( - x._ctl_to_dict(make_ctl(' ')), - {'event': 'clean_report', 'type': 'auto'}) - assert_dict_equal( - x._ctl_to_dict(make_ctl(' ')), - {'event': 'clean_report', 'type': 'auto', 'speed': 'strong'}) - - assert_dict_equal( - x._ctl_to_dict(make_ctl('')), - {'event': 'battery_info', 'power': '095'}) - - assert_dict_equal( - x._ctl_to_dict(make_ctl('# ')), - {'event': 'life_span', 'type': 'brush', 'val': '099', 'total': '365'}) - - assert_dict_equal( - x._ctl_to_dict(make_ctl('')), - {'event': 'life_span', 'type': 'dust_case_heap', 'val': '-050', 'total': '365'}) - - assert_equals(x._ctl_to_dict(make_ctl('')), None) - - -def make_ecovacs_xmpp(bot=None, server_address=None): - if bot is None: - bot = {"did": "E0000000001234567890", "class": "126", "nick": "bob", "iotmq": False} - return EcoVacsXMPP('20170101abcdefabcdefa', 'ecouser.net', 'abcdef12', 'A1b2C3d4efghijklmNOPQrstuvwxyz12', 'na', bot, server_address=server_address) - -def test_xmpp_customaddress(): - x = make_ecovacs_xmpp(server_address="test.xmppserver.com") - assert_equals(x.server_address, "test.xmppserver.com") - -def make_ctl(string): - return ET.fromstring('' + string + '')[0] diff --git a/tests/test_vacbot.py b/tests/test_vacbot.py deleted file mode 100644 index c4e436a..0000000 --- a/tests/test_vacbot.py +++ /dev/null @@ -1,398 +0,0 @@ -from nose.tools import * - -from sucks import * -from unittest.mock import Mock, patch -from sleekxmppfs.exceptions import XMPPError -from paho.mqtt.client import MQTT_ERR_UNKNOWN as MQTTError - - -def test_handle_clean_report(): - v = a_vacbot() - assert_equals(None, v.clean_status) - - v._handle_ctl({'event': 'clean_report', 'type': 'auto', 'speed': 'strong'}) - assert_equals('auto', v.clean_status) - assert_equals('high', v.fan_speed) - - v._handle_ctl({'event': 'clean_report', 'type': 'border', 'speed': 'standard'}) - assert_equals('edge', v.clean_status) - assert_equals('normal', v.fan_speed) - - # Missing fan_speed - v = a_vacbot() - v._handle_ctl({'event': 'clean_report', 'type': 'border'}) - assert_equals('edge', v.clean_status) - assert_is_none(v.fan_speed) - - # For states not handled by sucks constants, fall back to just using whatever the vacuum said - v._handle_ctl({'event': 'clean_report', 'type': 'a_type_not_supported_by_sucks', 'speed': 'a_weird_speed'}) - assert_equals('a_type_not_supported_by_sucks', v.clean_status) - assert_equals('a_weird_speed', v.fan_speed) - - - -def test_not_iot_send_command_clean(): - from unittest.mock import MagicMock - v = a_vacbot(iotmq=False) - v.xmpp.send_command = MagicMock() - v.send_command(VacBotCommand('Clean')) - assert v.xmpp.send_command.called #test when iot is False it uses xmpp.send_command - - -def test_iot_send_command_clean(): - from unittest.mock import MagicMock - v = a_vacbot(iotmq=True) - v.iotmq.send_command = MagicMock() - v.send_command(VacBotCommand('Clean')) - assert v.iotmq.send_command.called #test when iot is True it uses iotmq.send_command - - -def test_handle_charge_state(): - v = a_vacbot() - assert_equals(None, v.clean_status) - - v._handle_ctl({'event': 'charge_state', 'type': 'going'}) - assert_equals('returning', v.charge_status) - - v._handle_ctl({'event': 'charge_state', 'type': 'slot_charging'}) - assert_equals('charging', v.charge_status) - - v._handle_ctl({'event': 'charge_state', 'type': 'idle'}) - assert_equals('idle', v.charge_status) - - v._handle_ctl({'event': 'charge_state', 'ret': 'fail', 'errno': '9'}) #Seen in IOT - "but on charger, but turned off" - assert_equals('idle', v.charge_status) - - v._handle_ctl({'event': 'charge_state', 'ret': 'fail', 'errno': '8'}) #Seen in IOT - could be "already charging" - assert_equals('charging', v.charge_status) - - v._handle_ctl({'event': 'charge_state', 'ret': 'fail', 'errno': '5'}) #Seen in IOT - could be "busy with another command" - assert_equals('idle', v.charge_status) - - v._handle_ctl({'event': 'charge_state', 'ret': 'fail', 'errno': '3'}) #Seen in IOT - could be "Bot in stuck state, example dust bin out" - assert_equals('idle', v.charge_status) - - v._handle_ctl({'event': 'charge_state', 'type': 'a_type_not_supported_by_sucks'}) - assert_equals('a_type_not_supported_by_sucks', v.charge_status) - - -def test_vacuum_states(): - # Vacuum state usually mirrors the latest charge or clean report, but there are some edge cases where it doesn't - # work that way. This test ensures the edge cases are handled correctly. - v = a_vacbot() - assert_equals(None, v.vacuum_status) - - v._handle_ctl({'event': 'clean_report', 'type': 'auto', 'speed': 'strong'}) - assert_equals('auto', v.vacuum_status) - - # Ignore the "idle" charge state in most cases, as it can be reported during a cleaning (such as during initialization) - v._handle_ctl({'event': 'clean_report', 'type': 'auto'}) - v._handle_ctl({'event': 'charge_state', 'type': 'idle'}) - assert_equals('auto', v.vacuum_status) - - # However, we do honor the idle state when our current state is charging, as that can happen in some certain combination of events - v._handle_ctl({'event': 'charge_state', 'type': 'slot_charging'}) - v._handle_ctl({'event': 'charge_state', 'type': 'idle'}) - assert_equals('idle', v.vacuum_status) - -def test_handle_battery_info(): - v = a_vacbot() - assert_equals(None, v.battery_status) - - v._handle_ctl({'event': 'battery_info', 'power': '100'}) - assert_equals(1.0, v.battery_status) - - v._handle_ctl({'event': 'battery_info', 'power': '095'}) - assert_equals(0.95, v.battery_status) - - v._handle_ctl({'event': 'battery_info', 'power': '000'}) - assert_equals(0.0, v.battery_status) - - -def test_lifespan_reports(): - v = a_vacbot() - assert_equals({}, v.components) - - # Note: The "total" values don't seem to have any meaning - - v._handle_ctl({'event': 'life_span', 'type': 'side_brush', 'total': '100', 'val': '50'}) - assert_equals({'side_brush': 0.5}, v.components) - - v._handle_ctl({'event': 'life_span', 'type': 'brush', 'total': '200', 'val': '1'}) - assert_equals({'side_brush': 0.5, 'main_brush': 0.01}, v.components) - - v._handle_ctl({'event': 'life_span', 'type': 'side_brush', 'total': '100', 'val': '0'}) - assert_equals({'side_brush': 0, 'main_brush': 0.01}, v.components) - - v._handle_ctl({'event': 'life_span', 'type': 'a_weird_component', 'total': '100', 'val': '87'}) - assert_equals({'side_brush': 0, 'main_brush': 0.01, 'a_weird_component': 0.87}, v.components) - - v._handle_ctl({'event': 'life_span', 'type': 'side_brush', 'total': '100', 'left': '120'}) - assert_equals(2.0, v.components['side_brush']) #test left (2 hours / 120 mins) instead of val - - -def test_is_cleaning(): - v = a_vacbot() - - assert_false(v.is_cleaning) - - v._handle_ctl({'event': 'clean_report', 'type': 'auto', 'speed': 'strong'}) - assert_true(v.is_cleaning) - - v._handle_ctl({'event': 'clean_report', 'type': 'stop'}) - assert_false(v.is_cleaning) - - v._handle_ctl({'event': 'clean_report', 'type': 'edge', 'speed': 'normal'}) - assert_true(v.is_cleaning) - - v._handle_ctl({'event': 'charge_state', 'type': 'going'}) - assert_false(v.is_cleaning) - - v = a_vacbot(iotmq=True) - v._handle_ctl({'event': 'clean_report', 'type': 'spot_area', 'speed':'normal','st':'h'}) - assert_false(v.is_cleaning) #test iot and state paused - - v = a_vacbot(iotmq=True) - v._handle_ctl({'event': 'clean_report', 'type': 'spot_area', 'speed':'normal','st':'r'}) - assert_true(v.is_cleaning) #test iot and state running - -def test_is_charging(): - v = a_vacbot() - - assert_false(v.is_charging) - - v._handle_ctl({'event': 'clean_report', 'type': 'auto', 'speed': 'strong'}) - assert_false(v.is_charging) - - v._handle_ctl({'event': 'charge_state', 'type': 'going'}) - assert_false(v.is_charging) - - v._handle_ctl({'event': 'charge_state', 'type': 'slot_charging'}) - assert_true(v.is_charging) - - v._handle_ctl({'event': 'clean_report', 'type': 'edge', 'speed': 'normal'}) - assert_false(v.is_charging) - - - - -def test_send_ping_no_monitor(): - #Test XMPP Ping - v = a_vacbot() - mock = v.xmpp.send_ping = Mock() - v.send_ping() - - # On four failed pings, vacuum state gets set to 'offline' - mock.side_effect = XMPPError() - v.send_ping() - v.send_ping() - v.send_ping() - assert_equals(None, v.vacuum_status) - v.send_ping() - assert_equals('offline', v.vacuum_status) - - # On a successful ping after the offline state, state gets reset to None, indicating that it is unknown - mock.side_effect = None - v.send_ping() - assert_equals(None, v.vacuum_status) - - #Test MQTT Ping - v = a_vacbot(iotmq=True) - mock = v.iotmq.send_ping = Mock() - v.send_ping() - - # On four failed pings, vacuum state gets set to 'offline' - mock.return_value = False - v.send_ping() - v.send_ping() - v.send_ping() - assert_equals(None, v.vacuum_status) - v.send_ping() - assert_equals('offline', v.vacuum_status) - - # On a successful ping after the offline state, state gets reset to None, indicating that it is unknown - mock.return_value = True - v.send_ping() - assert_equals(None, v.vacuum_status) - - -def test_send_ping_with_monitor(): - #Test XMPP Ping - v = a_vacbot(monitor=True) - - ping_mock = v.xmpp.send_ping = Mock() - request_statuses_mock = v.request_all_statuses = Mock() - - # First ping should try to fetch statuses - v.send_ping() - assert_equals(1, request_statuses_mock.call_count) - - # Nothing blowing up is success - - # On four failed pings, vacuum state gets set to 'offline' - ping_mock.side_effect = XMPPError() - v.send_ping() - v.send_ping() - v.send_ping() - assert_equals(None, v.vacuum_status) - v.send_ping() - assert_equals('offline', v.vacuum_status) - - # On a successful ping after the offline state, a request for initial statuses is made - ping_mock.side_effect = None - request_statuses_mock.reset_mock() - v.send_ping() - assert_equals(1, request_statuses_mock.call_count) - - #Test MQTT Ping - v = a_vacbot(iotmq=True, monitor=True) - - ping_mock = v.iotmq.send_ping = Mock() - request_statuses_mock = v.request_all_statuses = Mock() - - # First ping should try to fetch statuses - v.send_ping() - assert_equals(1, request_statuses_mock.call_count) - - # Nothing blowing up is success - - # On four failed pings, vacuum state gets set to 'offline' - ping_mock.return_value = False - v.send_ping() - v.send_ping() - v.send_ping() - assert_equals(None, v.vacuum_status) - v.send_ping() - assert_equals('offline', v.vacuum_status) - - # On a successful ping after the offline state, a request for initial statuses is made - ping_mock.return_value = True - request_statuses_mock.reset_mock() - v.send_ping() - assert_equals(1, request_statuses_mock.call_count) - - -def test_status_event_subscription(): - v = a_vacbot() - - mock = Mock() - v.statusEvents.subscribe(mock) - v._handle_ctl({'event': 'clean_report', 'type': 'auto', 'speed': 'strong'}) - mock.assert_called_once_with('auto') - - mock = Mock() - v.statusEvents.subscribe(mock) - v._handle_ctl({'event': 'charge_state', 'type': 'going'}) - mock.assert_called_once_with('returning') - - # Test unsubscribe - mock = Mock() - subscription = v.statusEvents.subscribe(mock) - v._handle_ctl({'event': 'charge_state', 'type': 'going'}) - assert_equals(1, mock.call_count) - subscription.unsubscribe() - v._handle_ctl({'event': 'charge_state', 'type': 'slot_charging'}) - assert_equals(1, mock.call_count) - -def test_battery_event_subscription(): - v = a_vacbot() - - mock = Mock() - v.batteryEvents.subscribe(mock) - v._handle_ctl({'event': 'battery_info', 'power': '095'}) - mock.assert_called_once_with(0.95) - - # Test unsubscribe - mock = Mock() - subscription = v.batteryEvents.subscribe(mock) - v._handle_ctl({'event': 'battery_info', 'power': '095'}) - assert_equals(1, mock.call_count) - subscription.unsubscribe() - v._handle_ctl({'event': 'battery_info', 'power': '090'}) - assert_equals(1, mock.call_count) - -def test_lifespan_event_subscription(): - v = a_vacbot() - - mock = Mock() - v.lifespanEvents.subscribe(mock) - v._handle_ctl({'event': 'life_span', 'type': 'side_brush', 'total': '100', 'val': '50'}) - mock.assert_called_once_with({'type': 'side_brush', 'lifespan': 0.5}) - - # Test unsubscribe - mock = Mock() - subscription = v.lifespanEvents.subscribe(mock) - v._handle_ctl({'event': 'life_span', 'type': 'side_brush', 'total': '100', 'val': '50'}) - assert_equals(1, mock.call_count) - subscription.unsubscribe() - v._handle_ctl({'event': 'life_span', 'type': 'side_brush', 'total': '100', 'val': '25'}) - assert_equals(1, mock.call_count) - -def test_error_event_subscription(): - v = a_vacbot() - - mock = Mock() - v.errorEvents.subscribe(mock) - v._handle_ctl({'event': 'error', 'error': 'an_error_name'}) - v._handle_ctl({'event': 'error', 'errs': 'an_error_name2'}) #added for testing errs - assert_equals(2, mock.call_count) - #mock.assert_called_once_with('an_error_name') - - - # Test unsubscribe - mock = Mock() - subscription = v.errorEvents.subscribe(mock) - v._handle_ctl({'event': 'error', 'error': 'an_error_name'}) - assert_equals(1, mock.call_count) - subscription.unsubscribe() - v._handle_ctl({'event': 'error', 'error': 'an_error_name'}) - assert_equals(1, mock.call_count) - -def test_handle_unknown_ctl(): - v = a_vacbot() - v._handle_ctl({'event': 'weird_and_unknown_event', 'type': 'pretty_weird'}) - # as long as it doesn't blow up, that's fine - - -# as-yet unhandled messages: -# -# -# -# -# -# -# -# plus errors! - -def test_bot_address(): - v = a_vacbot(bot={"did": "E0000000001234567890", "class": "126", "nick": "bob", "iotmq":False}) - assert_equals('E0000000001234567890@126.ecorobot.net/atom', v._vacuum_address()) - - -def test_bot_address_iot(): - v = a_vacbot(bot={"did": "E0000000001234567890", "class": "126", "nick": "bob", "iotmq":True}) - assert_equals('E0000000001234567890', v._vacuum_address()) - - -def test_model_variation(): - v = a_vacbot(bot={"did": "E0000000001234567890", "class": "141", "nick": "bob","iotmq":False}) - assert_equals('E0000000001234567890@141.ecorobot.net/atom', v._vacuum_address()) - - - -def a_vacbot(bot=None, iotmq=False, monitor=False): - if bot is None: - bot = {"did": "E0000000001234567890", "class": "126", "nick": "bob", "iotmq": iotmq} - return VacBot('20170101abcdefabcdefa', 'ecouser.net', 'abcdef12', 'A1b2C3d4efghijklmNOPQrstuvwxyz12', - bot, 'na', monitor=monitor) - -def test_str_to_bool(): - assert_raises(ValueError, str_to_bool_or_cert, None) #Value error if str_to_bool can't convert - assert_equals(True, str_to_bool_or_cert("True")) - assert_equals(False, str_to_bool_or_cert("False")) - assert_equals( - os.path.abspath(os.path.join(".", "tests", "test_vacbot.py")), - str_to_bool_or_cert(os.path.abspath(os.path.join(".","tests","test_vacbot.py"))) - ) - assert_raises(ValueError, str_to_bool_or_cert ,(os.path.abspath(os.path.join(".","tests")))) - \ No newline at end of file