get rid of sucks stuff to isolate custom component with its changes
This commit is contained in:
bittles
@@ -1,9 +0,0 @@
|
||||
language: python
|
||||
python:
|
||||
- "3.4"
|
||||
- "3.5"
|
||||
- "3.6"
|
||||
install:
|
||||
- pip install -e .[dev]
|
||||
|
||||
script: nosetests
|
||||
@@ -1,152 +0,0 @@
|
||||
Linux: [](https://travis-ci.org/wpietri/sucks)
|
||||
Windows: [](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!
|
||||
@@ -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"
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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"]
|
||||
@@ -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:
|
||||
@@ -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)
|
||||
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']
|
||||
@@ -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
|
||||
@@ -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('<auth mechanism="PLAIN" xmlns="urn:ietf:params:xml:ns:xmpp-sasl">([-A-Za-z0-9+/=]+)</auth>',
|
||||
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(
|
||||
'<iq id="(\d+)" to="ROBOTID@126.ecorobot.net/atom" from="USERID@ecouser.net/RESOURCEID" type="set"><query xmlns="com:ctl">(<ctl .*>)</query></iq>',
|
||||
'id=\\1 command=\\2', line)
|
||||
|
||||
# translate server responses
|
||||
|
||||
line = re.sub(
|
||||
'<iq to="USERID@ecouser.net/RESOURCEID" type="result" id="(\d+)" from="ROBOTID@126.ecorobot.net/atom"/>',
|
||||
'id=\\1 result =empty', line)
|
||||
line = re.sub(
|
||||
'<iq to="USERID@ecouser.net/RESOURCEID" type="set" id="(\d+)" from="ROBOTID@126.ecorobot.net/atom"><query xmlns="com:ctl"><ctl id="(\d+)" ret="([^"]+)"/></query></iq>',
|
||||
'id=\\1 id=\\2 result=\\3', line)
|
||||
line = re.sub(
|
||||
'<iq to="USERID@ecouser.net/RESOURCEID" type="set" id="(\d+)" from="ROBOTID@126.ecorobot.net/atom"><query xmlns="com:ctl">(<ctl .*)</query></iq>',
|
||||
'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")
|
||||
@@ -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',
|
||||
],
|
||||
},
|
||||
)
|
||||
@@ -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 "$@"
|
||||
-1096
File diff suppressed because it is too large
Load Diff
-244
@@ -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()
|
||||
@@ -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')
|
||||
@@ -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'<ctl td="CustomCommand" type="customtype" />')
|
||||
|
||||
|
||||
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'<ctl td="CustomCommand"><customtag customvar="customvalue" /></ctl>')
|
||||
|
||||
|
||||
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'<ctl td="CustomCommand"><customtag customvar="customvalue1" /><customtag customvar="customvalue2" /></ctl>')
|
||||
|
||||
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'<ctl arg1="value1" td="CustomCommand"><customtag customvar="customvalue1" /><customtag customvar="customvalue2" /></ctl>')
|
||||
|
||||
|
||||
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'<ctl td="CustomCommand" />')
|
||||
|
||||
|
||||
def test_clean_command():
|
||||
c = Clean()
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="Clean"><clean act="s" speed="standard" type="auto" /></ctl>') # protocol has attribs in other order
|
||||
|
||||
c = Clean('edge', 'high')
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="Clean"><clean act="s" speed="strong" type="border" /></ctl>') # protocol has attribs in other order
|
||||
|
||||
c = Clean(iotmq=True)
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="Clean"><clean act="s" speed="standard" type="auto" /></ctl>') # 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'<ctl td="Clean"><clean act="s" mid="0" speed="standard" type="SpotArea" /></ctl>') #Test namedarea clean
|
||||
|
||||
c = SpotArea('start', area='0')
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="Clean"><clean act="s" mid="0" speed="standard" type="SpotArea" /></ctl>') #Test namedarea keyword clean
|
||||
|
||||
c = SpotArea('start', '', '-602,1812,800,723')
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="Clean"><clean act="s" deep="1" p="-602,1812,800,723" speed="standard" type="SpotArea" /></ctl>') #Test customarea clean
|
||||
|
||||
c = SpotArea('start', '', '-602,1812,800,723', '2')
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="Clean"><clean act="s" deep="2" p="-602,1812,800,723" speed="standard" type="SpotArea" /></ctl>') #Test customarea clean with deep 2
|
||||
|
||||
c = SpotArea('start', '', map_position='-602,1812,800,723')
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="Clean"><clean act="s" deep="1" p="-602,1812,800,723" speed="standard" type="SpotArea" /></ctl>') #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'<ctl td="Clean"><clean act="s" deep="2" p="-602,1812,800,723" speed="standard" type="SpotArea" /></ctl>') #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'<ctl td="Clean"><clean act="s" mid="0" speed="standard" type="SpotArea" /></ctl>') #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'<ctl td="Clean"><clean act="s" mid="0" speed="standard" type="SpotArea" /></ctl>') #Test all keywords specified, should default to only mid
|
||||
|
||||
|
||||
def test_edge_command():
|
||||
c = Edge()
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="Clean"><clean act="s" speed="strong" type="border" /></ctl>') # protocol has attribs in other order
|
||||
|
||||
|
||||
def test_spot_command():
|
||||
c = Spot()
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="Clean"><clean act="s" speed="strong" type="spot" /></ctl>') # protocol has attribs in other order
|
||||
|
||||
|
||||
def test_charge_command():
|
||||
c = Charge()
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="Charge"><charge type="go" /></ctl>')
|
||||
|
||||
|
||||
def test_stop_command():
|
||||
c = Stop()
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="Clean"><clean act="s" speed="standard" type="stop" /></ctl>')
|
||||
|
||||
|
||||
def test_play_sound_command():
|
||||
c = PlaySound()
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl sid="0" td="PlaySound" />')
|
||||
|
||||
|
||||
def test_play_sound_command_with_sid():
|
||||
c = PlaySound(sid="1")
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl sid="1" td="PlaySound" />')
|
||||
|
||||
|
||||
def test_get_clean_state_command():
|
||||
c = GetCleanState()
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="GetCleanState" />')
|
||||
|
||||
|
||||
def test_get_charge_state_command():
|
||||
c = GetChargeState()
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="GetChargeState" />')
|
||||
|
||||
|
||||
def test_get_battery_state_command():
|
||||
c = GetBatteryState()
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="GetBatteryInfo" />')
|
||||
|
||||
|
||||
def test_move_command():
|
||||
c = Move(action='left')
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="Move"><move action="SpinLeft" /></ctl>')
|
||||
c = Move(action='right')
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="Move"><move action="SpinRight" /></ctl>')
|
||||
c = Move(action='turn_around')
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="Move"><move action="TurnAround" /></ctl>')
|
||||
c = Move(action='forward')
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="Move"><move action="forward" /></ctl>')
|
||||
c = Move(action='backward')
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="Move"><move action="backward" /></ctl>')
|
||||
c = Move(action='stop')
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="Move"><move action="stop" /></ctl>')
|
||||
|
||||
|
||||
def test_get_lifepsan_command():
|
||||
c = GetLifeSpan('main_brush')
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="GetLifeSpan" type="Brush" />')
|
||||
|
||||
c = GetLifeSpan('side_brush')
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="GetLifeSpan" type="SideBrush" />')
|
||||
|
||||
c = GetLifeSpan('filter')
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="GetLifeSpan" type="DustCaseHeap" />')
|
||||
|
||||
def test_set_time_command():
|
||||
c = SetTime('1234', 'GMT-5')
|
||||
assert_equals(ElementTree.tostring(c.to_xml()),
|
||||
b'<ctl td="SetTime"><time t="1234" tz="GMT-5" /></ctl>')
|
||||
@@ -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')
|
||||
@@ -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"<ctl ts='1547824270099' td='CleanReport'><clean type='auto' speed='standard' st='h' rsn='a' a='' l='' sts=''/></ctl>"
|
||||
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'] = '<ctl td="CleanReport"> <clean type="auto" /> </ctl>'
|
||||
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'], '<ctl><charge type="go" /></ctl>')
|
||||
|
||||
|
||||
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":"<ctl ret='ok'><clean type='auto' speed='standard' st='h' t='1159' a='15' s='0' tr=''/></ctl>","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':"<ctl ret='ok'><clean type='auto' speed='standard' st='h' t='1159' a='15' s='0' tr=''/></ctl>",'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, "<ctl ts='1547824270099' td='CleanReport'><clean type='auto' speed='standard' st='h' rsn='a' a='' l='' sts=''/></ctl>"),
|
||||
{'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, "<ctl ts='1547824270099' td='CleanReport'><clean type='auto' speed='strong' st='h' rsn='a' a='' l='' sts=''/></ctl>"),
|
||||
{'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, "<ctl ts='1547824270099'><clean type='auto' speed='strong' st='h' rsn='a' a='' l='' sts=''/></ctl>"),
|
||||
{'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, "<ctl ts='1547823289924' td='BatteryInfo'><battery power='64'/></ctl>"),
|
||||
{'event': 'battery_info', 'ts':'1547823289924', 'power': '64'})
|
||||
|
||||
assert_dict_equal(
|
||||
x._ctl_to_dict_mqtt(test_topic, "<ctl ts='1547823289924'><battery power='64'/></ctl>"),
|
||||
{'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, "<ctl ts='1547823129670' td='SleepStatus' st='1'/>"),
|
||||
{'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, "<ctl ts='1547822982581' td='errors' old='' new='102'/>"),
|
||||
{'event': 'errors', 'ts':'1547822982581','old':'','new':'102'})
|
||||
assert_dict_equal(
|
||||
x._ctl_to_dict_mqtt(test_topic, "<ctl ts='1547822982581' td='errors' old='102' new=''/>"),
|
||||
{'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, "<ctl td='Pos' t='p' p='7,-10' a='-42' valid='0'/>"),
|
||||
{'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, "<ctl ts='1547822871328' td='DustCaseST' st='1'/>"),
|
||||
{'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, "<ctl ts='1547823592934' td='MapSt' st='relocGoChgStart' method='' info=''/>"),
|
||||
{'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, "<ctl ret='ok' type='Brush' left='9876' total='18000'/>"),
|
||||
{'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, '<ctl td="CustomCommand"><customtag customvar="customvalue1" /></ctl>'),
|
||||
{'event': 'custom_command', 'customvar': 'customvalue1'})
|
||||
|
||||
|
||||
def test_xml_to_dict_api():
|
||||
x = make_ecovacs_iotmq()
|
||||
message = {}
|
||||
|
||||
cmd = VacBotCommand("Clean")
|
||||
message['resp'] = "<ctl ret='ok'><clean type='auto' speed='standard' st='h' t='1159' a='15' s='0' tr=''/></ctl>"
|
||||
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'] = "<ctl ret='ok'><clean type='auto' speed='strong' st='h' t='1159' a='15' s='0' tr=''/></ctl>"
|
||||
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'] = "<ctl ret='ok'><battery power='82'/></ctl>"
|
||||
assert_dict_equal(
|
||||
x._ctl_to_dict_api(cmd,message['resp']),
|
||||
{'event': 'battery_info', 'power': '82'})
|
||||
|
||||
cmd = VacBotCommand("GetLifeSpan")
|
||||
message['resp'] = "<ctl ret='ok' type='Brush' left='9876' total='18000'/>"
|
||||
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'] = "<ctl ts='1547823651958' td='ChargeState'><charge type='Going' h='' r='a' s='' g='0'/></ctl>"
|
||||
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'] = "<ctl td='TestCommand'><test type='command'/></ctl>"
|
||||
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'] = "<ctl ret='fail' errno='8'/>"
|
||||
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)
|
||||
@@ -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('<query xmlns="com:ctl"><ctl td="CleanReport"> <clean type="auto" /> </ctl></query>'))
|
||||
|
||||
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('<ctl td="CleanReport"> <clean type="auto" /> </ctl>')),
|
||||
{'event': 'clean_report', 'type': 'auto'})
|
||||
assert_dict_equal(
|
||||
x._ctl_to_dict(make_ctl('<ctl td="CleanReport"> <clean type="auto" speed="strong" /> </ctl>')),
|
||||
{'event': 'clean_report', 'type': 'auto', 'speed': 'strong'})
|
||||
|
||||
assert_dict_equal(
|
||||
x._ctl_to_dict(make_ctl('<ctl td="BatteryInfo"><battery power="095"/></ctl>')),
|
||||
{'event': 'battery_info', 'power': '095'})
|
||||
|
||||
assert_dict_equal(
|
||||
x._ctl_to_dict(make_ctl('# <ctl td="LifeSpan" type="Brush" val="099" total="365"/>')),
|
||||
{'event': 'life_span', 'type': 'brush', 'val': '099', 'total': '365'})
|
||||
|
||||
assert_dict_equal(
|
||||
x._ctl_to_dict(make_ctl('<ctl td="LifeSpan" type="DustCaseHeap" val="-050" total="365"/>')),
|
||||
{'event': 'life_span', 'type': 'dust_case_heap', 'val': '-050', 'total': '365'})
|
||||
|
||||
assert_equals(x._ctl_to_dict(make_ctl('<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('<query xmlns="com:ctl">' + string + '</query>')[0]
|
||||
@@ -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:
|
||||
#
|
||||
# <ctl td="LifeSpan" type="Brush" val="099" total="365"/>
|
||||
# <ctl td="LifeSpan" type="DustCaseHeap" val="098" total="365"/>
|
||||
# <ctl td="LifeSpan" type="SideBrush" val="098" total="365"/>
|
||||
# <ctl td="Sched2"/>
|
||||
# <ctl td="Sched2" id="30800321"/>
|
||||
#
|
||||
# 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"))))
|
||||
|
||||
Reference in New Issue
Block a user