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

Combine MQTT and IOT into IOTMQ

Browse Source 
Combined MQTT and IOT into IOTMQ
Updated tests - now with 99% coverage for __init__
This commit is contained in:
Brian Martin
2019-02-18 17:44:41 -05:00
parent 9919baad27
commit 597129dda0
8 changed files with 559 additions and 532 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_vacbot.py
+16 -73
View File
@@ -34,7 +34,7 @@ def test_handle_clean_report():
def test_not_iot_send_command_clean():
from unittest.mock import MagicMock
v = a_vacbot(iot=False)
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
@@ -42,10 +42,10 @@ def test_not_iot_send_command_clean():
def test_iot_send_command_clean():
from unittest.mock import MagicMock
v = a_vacbot(iot=True)
v.iot.send_command = MagicMock()
v = a_vacbot(iotmq=True)
v.iotmq.send_command = MagicMock()
v.send_command(VacBotCommand('Clean'))
assert v.iot.send_command.called #test when iot is True it uses iot.send_command
assert v.iotmq.send_command.called #test when iot is True it uses iotmq.send_command
def test_handle_charge_state():
@@ -149,11 +149,11 @@ def test_is_cleaning():
v._handle_ctl({'event': 'charge_state', 'type': 'going'})
assert_false(v.is_cleaning)
v = a_vacbot(iot=True)
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(iot=True)
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
@@ -198,8 +198,8 @@ def test_send_ping_no_monitor():
assert_equals(None, v.vacuum_status)
#Test MQTT Ping
v = a_vacbot(iot=True)
mock = v.mqtt.send_ping = Mock()
v = a_vacbot(iotmq=True)
mock = v.iotmq.send_ping = Mock()
v.send_ping()
# On four failed pings, vacuum state gets set to 'offline'
@@ -246,9 +246,9 @@ def test_send_ping_with_monitor():
assert_equals(1, request_statuses_mock.call_count)
#Test MQTT Ping
v = a_vacbot(iot=True, monitor=True)
v = a_vacbot(iotmq=True, monitor=True)
ping_mock = v.mqtt.send_ping = Mock()
ping_mock = v.iotmq.send_ping = Mock()
request_statuses_mock = v.request_all_statuses = Mock()
# First ping should try to fetch statuses
@@ -366,84 +366,27 @@ def test_handle_unknown_ctl():
# plus errors!
def test_bot_address():
v = a_vacbot(bot={"did": "E0000000001234567890", "class": "126", "nick": "bob", "iot":False})
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", "iot":True})
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","iot":False})
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, iot=False, monitor=False):
def a_vacbot(bot=None, iotmq=False, monitor=False):
if bot is None:
bot = {"did": "E0000000001234567890", "class": "126", "nick": "bob", "iot": iot}
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, None) #Value error if str_to_bool can't convert
def test_connect_and_wait():
from unittest.mock import MagicMock
v = a_vacbot(iot=False, monitor=True)
v.xmpp.connect_and_wait_until_ready = MagicMock()
v.send_ping = MagicMock()
v.xmpp.schedule = MagicMock()
v.connect_and_wait_until_ready()
assert v.xmpp.schedule.called #test when iot is False it uses xmpp.schedule
v = a_vacbot(iot=True)
v.mqtt.connect_and_wait_until_ready = MagicMock()
v.send_ping = MagicMock()
v.mqtt.schedule = MagicMock()
v.connect_and_wait_until_ready()
assert v.mqtt.schedule.called #test when iot is True it uses mqtt.schedule
def test_run():
from unittest.mock import MagicMock
v = a_vacbot(iot=False)
v.send_command = MagicMock()
v.run(VacBotCommand('Clean'))
assert v.send_command.called
def test_disconnect():
from unittest.mock import MagicMock
v = a_vacbot(iot=False)
v.xmpp.disconnect = MagicMock()
v.disconnect()
assert v.xmpp.disconnect.called
v = a_vacbot(iot=True)
v.mqtt.disconnect = MagicMock()
v.disconnect()
assert v.mqtt.disconnect.called
def test_refresh_all():
from unittest.mock import MagicMock
v = a_vacbot(iot=False)
v.refresh_statuses = MagicMock()
v.refresh_components = MagicMock
v.request_all_statuses()
assert v.refresh_components.called and v.refresh_statuses.called
def test_refresh_statuses():
from unittest.mock import MagicMock
v = a_vacbot(iot=False)
v.run = MagicMock()
v.refresh_statuses()
assert v.run.called
def test_refresh_components():
from unittest.mock import MagicMock
v = a_vacbot(iot=False)
v.run = MagicMock()
v.refresh_components()
assert v.run.called