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

Add test MQTTPing

Browse Source 
Add test MQTTPing & Fix tests
This commit is contained in:
Brian Martin
2019-01-17 09:13:47 -05:00
parent c72be5509a
commit e2b0ea7b55
3 changed files with 81 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+2
View File
@@ -15,3 +15,5 @@ cover/
# Ignore sucks.egg-info
sucks.egg-info/
.noseids
nosetests.xml
sucks/__init__.py
+24 -14
View File
@@ -379,6 +379,11 @@ class VacBot():
self.lifespanEvents = EventEmitter()
self.errorEvents = EventEmitter()
#Set none for clients to start
self.mqtt = None
self.iot = None
self.xmpp = None
if vacuum['iot']:
self.iot = EcoVacsIOT(user, domain, resource, secret, continent, vacuum)
self.iot.subscribe_to_ctls(self._handle_ctl)
@@ -481,6 +486,7 @@ class VacBot():
if event['ret'] == 'fail' and event['errno'] == '8': #Already charging
status = 'slot_charging'
else:
status = 'idle' #Fall back to Idle status
_LOGGER.error("Unknown charging status '" + event['errno'] + "'") #Log this so we can identify more errors
try:
@@ -515,6 +521,14 @@ class VacBot():
try:
if not self.vacuum['iot']:
self.xmpp.send_ping(self._vacuum_address())
elif self.vacuum['iot']:
if not self.mqtt.send_ping():
raise RuntimeError()
#self.xmpp.send_ping(EcoVacsAPI.REALM) #IOT vacuums are using the realm instead
#Some devices may utilize this, but it appears to
# just be an oversight in the app communidcations. IOT should probably be using MQTT pings (which are automatic when connected)
except XMPPError as err:
_LOGGER.warning("Ping did not reach VacBot. Will retry.")
@@ -525,21 +539,12 @@ class VacBot():
self.vacuum_status = 'offline'
self.statusEvents.notify(self.vacuum_status)
try:
if self.vacuum['iot']:
self.mqtt.send_ping()
#self.xmpp.send_ping(EcoVacsAPI.REALM) #IOT vacuums are using the realm instead
#Some devices may utilize this, but it appears to
# just be an oversight in the app communidcations. IOT should probably be using MQTT pings (which are automatic when connected)
except MQTTException as err:
except RuntimeError as err:
_LOGGER.warning("Ping did not reach VacBot. Will retry.")
_LOGGER.debug("*** Error type: " + err.etype)
_LOGGER.debug("*** Error condition: " + err.condition)
self._failed_pings += 1
if self._failed_pings >= 4:
self.vacuum_status = 'offline'
self.statusEvents.notify(self.vacuum_status)
self.statusEvents.notify(self.vacuum_status)
else:
self._failed_pings = 0
@@ -806,9 +811,14 @@ class EcoVacsMQTT(ClientMQTT):
_LOGGER.debug("*** MQTT sending ping ***")
rc = self._send_simple_command(MQTTPublish.paho.PINGREQ)
if rc == MQTTPublish.paho.MQTT_ERR_SUCCESS:
_LOGGER.debug("*** MQTT ping acknowledged ***")
return rc
_LOGGER.debug("*** MQTT ping acknowledged ***")
print(rc)
return True
else:
print(rc)
return False
def connect_and_wait_until_ready(self):
tests/test_vacbot.py
+55 -3
View File
@@ -4,6 +4,8 @@ from sucks import *
from unittest.mock import Mock
from sleekxmpp.exceptions import XMPPError
from paho.mqtt.client import MQTT_ERR_UNKNOWN as MQTTError
def test_handle_clean_report():
@@ -46,6 +48,9 @@ def test_handle_charge_state():
v._handle_ctl({'event': 'charge_state', 'ret': 'fail', 'errno': '8'}) #Seen in IOT when 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
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)
@@ -135,8 +140,8 @@ def test_is_charging():
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()
@@ -154,8 +159,28 @@ def test_send_ping_no_monitor():
v.send_ping()
assert_equals(None, v.vacuum_status)
#Test MQTT Ping
v = a_vacbot(iot=True)
mock = v.mqtt.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()
@@ -182,6 +207,33 @@ def test_send_ping_with_monitor():
v.send_ping()
assert_equals(1, request_statuses_mock.call_count)
#Test MQTT Ping
v = a_vacbot(iot=True, monitor=True)
ping_mock = v.mqtt.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()
@@ -283,8 +335,8 @@ def test_model_variation():
def a_vacbot(bot=None, monitor=False):
def a_vacbot(bot=None, iot=False, monitor=False):
if bot is None:
bot = {"did": "E0000000001234567890", "class": "126", "nick": "bob", "iot": False}
bot = {"did": "E0000000001234567890", "class": "126", "nick": "bob", "iot": iot}
return VacBot('20170101abcdefabcdefa', 'ecouser.net', 'abcdef12', 'A1b2C3d4efghijklmNOPQrstuvwxyz12',
bot, 'na', monitor=monitor)