gronod/ha_ecovacs_bumper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add more iot tests

Browse Source 
Add more iot tests
This commit is contained in:
Brian Martin
2019-01-20 16:30:53 -05:00
parent 10993e678a
commit f3bcfb4ddc
3 changed files with 110 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sucks/__init__.py
+12 -1
View File
@@ -416,7 +416,12 @@ class VacBot():
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)
@@ -483,6 +488,10 @@ class VacBot():
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
@@ -632,7 +641,9 @@ class EcoVacsIOT():
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(action, self.api._EcoVacsAPI__call_portal_api(self.api, self.api.IOTDEVMANAGERAPI,'',c ))
self._handle_ctl(action,
self.api._EcoVacsAPI__call_portal_api(self.api, self.api.IOTDEVMANAGERAPI,'',c )
)
def _wrap_command(self, cmd, recipient):
tests/test_ecovacs_iot.py
+74 -39
View File
@@ -1,15 +1,21 @@
from re import compile
import requests_mock
import requests
from nose.tools import *
from sucks import *
from test_ecovacs_api import make_api
from tests.test_ecovacs_api import make_api
# There are few tests for the IOT 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_is_iot():
x = make_ecovacs_iot()
assert_equal(x.vacuum['iot'], True)
def test_wrap_command():
x = make_ecovacs_iot()
@@ -18,52 +24,81 @@ def test_wrap_command():
assert_equal(c['toId'], 'E0000000001234567890')
assert_equal(c['payload'], '<ctl><charge type="go" /></ctl>')
def test_is_iot():
def test_iotapi_response():
x = make_ecovacs_iot()
api = make_api()
x.api = api
with requests_mock.mock() as m:
#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(compile('iot/devmanager.do'),
json=resp)
cmd = VacBotCommand("GetCleanState")
c = x._wrap_command(cmd, x.vacuum['did'])
rtnval = api._EcoVacsAPI__call_portal_api(api.IOTDEVMANAGERAPI, '', 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 Timeout
r2 = m.post(compile('iot/devmanager.do'),exc=requests.exceptions.ReadTimeout)
cmd = VacBotCommand("GetCleanState")
c = x._wrap_command(cmd, x.vacuum['did'])
rtnval = api._EcoVacsAPI__call_portal_api(api.IOTDEVMANAGERAPI, '', c)
assert_equal(rtnval, {}) #Right now it sends back a blank object
def test_subscribe_to_ctls():
response = None
def save_response(value):
nonlocal response
response = value
x = make_ecovacs_iot()
# TODO - Error response from command
#'cmdName': 'Charge', 'payload': '<ctl><charge type="go" /></ctl>', 'payloadType': 'x', 'td': 'q', 'toId': '0e084f6c-0846-4342-a947-fe14c293301f', 'toRes': 'wC3g', 'toType': 'ls1ok3'}
# - Already charging on dock
#{'ret': 'ok', 'resp': "<ctl ret='fail' errno='8'/>", 'id': 'NLQy'}
x.subscribe_to_ctls(save_response)
message = {}
message['resp'] = '<ctl td="CleanReport"> <clean type="auto" /> </ctl>'
#Timeout
# {'ret': 'fail', 'errno': 500, 'debug': 'wait for response timed out'}
# def test_subscribe_to_ctls():
# response = None
# def save_response(value):
# nonlocal response
# response = value
# x = make_ecovacs_iot()
# 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'})
x.subscribe_to_ctls(save_response)
x._handle_ctl("Clean", message)
assert_dict_equal(response, {'event': 'clean_report', 'type': 'auto'})
# def test_xml_to_dict():
# x = make_ecovacs_iot()
def test_xml_to_dict():
x = make_ecovacs_iot()
message = {}
# 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'})
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(cmd,message['resp']),
{'event': 'clean_report', 'type': 'auto', 'speed': 'standard', 'st':'h','t':'1159','a':'15','s':'0','tr':''})
# assert_dict_equal(
# x._ctl_to_dict(make_ctl('<ctl td="BatteryInfo"><battery power="095"/></ctl>')),
# {'event': 'battery_info', 'power': '095'})
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(cmd,message['resp']),
{'event': 'clean_report', 'type': 'auto', 'speed': 'strong', 'st':'h','t':'1159','a':'15','s':'0','tr':''})
# 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'})
cmd = VacBotCommand("GetBatteryInfo")
message['resp'] = "<ctl ret='ok'><battery power='82'/></ctl>"
assert_dict_equal(
x._ctl_to_dict(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(cmd,message['resp']),
{'event': 'life_span','ret':'ok', 'type': 'brush', 'left': '9876', 'total': '18000'})
cmd = VacBotCommand("Charge")
message['resp'] = "<ctl ret='fail' errno='8'/>"
assert_dict_equal(
x._ctl_to_dict(cmd,message['resp']),
{'event': 'charge_state','ret':'fail', 'errno': '8'}) #Test fail from charge command
def make_ecovacs_iot():
tests/test_vacbot.py
+19 -2
View File
@@ -45,10 +45,16 @@ def test_handle_charge_state():
v._handle_ctl({'event': 'charge_state', 'type': 'idle'})
assert_equals('idle', v.charge_status)
v._handle_ctl({'event': 'charge_state', 'ret': 'fail', 'errno': '8'}) #Seen in IOT when already charging
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 randomly - not sure what this is yet
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'})
@@ -87,6 +93,14 @@ def test_handle_battery_info():
v._handle_ctl({'event': 'battery_info', 'power': '000'})
assert_equals(0.0, v.battery_status)
def test_handle_geterrors():
v = a_vacbot()
#v._handle_error
#ssert_equals({}, v.components)
def test_lifespan_reports():
v = a_vacbot()
assert_equals({}, v.components)
@@ -139,6 +153,9 @@ def test_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()