Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Use Remote Group When Adding Resp Link on Modem #453

Merged
merged 1 commit into from
Nov 8, 2021
Merged

Use Remote Group When Adding Resp Link on Modem #453

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 9 additions & 3 deletions insteon_mqtt/Modem.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1305,9 +1305,15 @@ def _db_update(self, local_group, is_controller, remote_addr, remote_group,
seq = CommandSeq(self, "Device db update complete", on_done,
name="ModemDBUpd")

# Group number in the db is the group number of the controller since
# that's the group number in the broadcast message we'll receive.
db_group = local_group
if not is_controller:
db_group = remote_group

# Create a new database entry for the modem and send it to the modem
# for updating.
entry = db.ModemEntry(remote_addr, local_group, is_controller,
entry = db.ModemEntry(remote_addr, db_group, is_controller,
local_data)
seq.add(self.db.add_on_device, entry)

Expand All @@ -1319,10 +1325,10 @@ def _db_update(self, local_group, is_controller, remote_addr, remote_group,
on_done = None
if is_controller:
seq.add(remote.db_add_resp_of, remote_group, self.addr,
local_group, two_way, refresh, remote_data=remote_data)
local_group, two_way, refresh, local_data=remote_data)
else:
seq.add(remote.db_add_ctrl_of, remote_group, self.addr,
local_group, two_way, refresh, remote_data=remote_data)
local_group, two_way, refresh, local_data=remote_data)

# Start the command sequence.
seq.run()
Expand Down
138 changes: 138 additions & 0 deletions tests/test_Modem.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,11 +3,13 @@
# Tests for: insteont_mqtt/Modem.py
#
#===========================================================================
import logging
import pytest
# from pprint import pprint
from unittest import mock
from unittest.mock import call
import insteon_mqtt as IM
from insteon_mqtt.device.base.Base import Base
import insteon_mqtt.message as Msg
import insteon_mqtt.util as util
import helpers as H
Expand All @@ -23,6 +25,42 @@ def test_device():
device = IM.Modem(protocol, stack, timed_call)
return device

@pytest.fixture
def test_device_2(tmpdir):
'''
Returns a generically configured device for testing
'''
protocol = H.main.MockProtocol()
modem = H.main.MockModem(tmpdir)
modem.db = IM.db.Modem(None, modem)
modem.scenes = IM.Scenes.SceneManager(modem, None)
addr = IM.Address(0x56, 0x78, 0xcd)
device = Base(protocol, modem, addr)
return device

@pytest.fixture
def test_entry_2():
addr = IM.Address('56.78.cd')
data = bytes([0xff, 0x00, 0x00])
group = 0x01
in_use = True
is_controller = True
is_last_rec = False
db_flags = Msg.DbFlags(in_use, is_controller, is_last_rec)
mem_loc = 1
return IM.db.DeviceEntry(addr, group, mem_loc, db_flags, data)

@pytest.fixture
def test_entry_multigroup():
addr = IM.Address('56.78.cd')
data = bytes([0xff, 0x00, 0x00])
group = 0x02
in_use = True
is_controller = True
is_last_rec = False
db_flags = Msg.DbFlags(in_use, is_controller, is_last_rec)
mem_loc = 1
return IM.db.DeviceEntry(addr, group, mem_loc, db_flags, data)

class Test_Base_Config():
def test_load_config_no_addr(self, test_device):
Expand Down Expand Up @@ -87,3 +125,103 @@ def test_load_config_step2_success_no_addr(self, test_device, tmpdir,
for record in caplog.records:
assert record.levelname != "ERROR"
assert test_device.addr == IM.Address('44.85.12')

def test_db_update(self, test_device, test_entry_2,
test_device_2, caplog):
test_device.add(test_device_2)
with mock.patch.object(IM.CommandSeq, 'add') as mocked:
with caplog.at_level(logging.DEBUG):
two_way = True
refresh = True
test_device._db_update(test_entry_2.group,
test_entry_2.is_controller,
test_entry_2.addr,
test_entry_2.group,
two_way,
refresh,
None,
bytes([0x00, 0x00, 0x00]),
test_entry_2.data)
assert mocked.call_count == 2
call_args = mocked.call_args_list
assert call_args[0].args[0] == test_device.db.add_on_device
assert call_args[0].args[1].addr == test_entry_2.addr
assert call_args[0].args[1].group == test_entry_2.group
assert call_args[0].args[1].is_controller == test_entry_2.is_controller
assert call_args[0].args[1].data == bytes([0x00, 0x00, 0x00])
assert call_args[1] == call(test_device_2.db_add_resp_of,
test_entry_2.group,
test_device.addr, test_entry_2.group,
False,
refresh, local_data=test_entry_2.data)

def test_db_update_resp(self, test_device, test_entry_2,
test_device_2, caplog):
test_device.add(test_device_2)
with mock.patch.object(IM.CommandSeq, 'add') as mocked:
with caplog.at_level(logging.DEBUG):
two_way = True
refresh = True
test_device._db_update(test_entry_2.group,
False,
test_entry_2.addr,
test_entry_2.group,
two_way,
refresh,
None,
bytes([0x00, 0x00, 0x00]),
test_entry_2.data)
assert mocked.call_count == 2
call_args = mocked.call_args_list
assert call_args[0].args[0] == test_device.db.add_on_device
assert call_args[0].args[1].addr == test_entry_2.addr
assert call_args[0].args[1].group == test_entry_2.group
assert call_args[0].args[1].is_controller == False
assert call_args[0].args[1].data == bytes([0x00, 0x00, 0x00])
assert call_args[1] == call(test_device_2.db_add_ctrl_of, test_entry_2.group,
test_device.addr, test_entry_2.group, False,
refresh, local_data=test_entry_2.data)

def test_db_update_no_remote(self, test_device, test_entry_2, caplog):
with mock.patch.object(IM.CommandSeq, 'add') as mocked:
with caplog.at_level(logging.DEBUG):
two_way = True
refresh = True
test_device._db_update(test_entry_2.group,
test_entry_2.is_controller,
test_entry_2.addr,
test_entry_2.group,
two_way,
refresh,
None,
bytes([0x00, 0x00, 0x00]),
test_entry_2.data)
calls = [call(test_device.db.add_on_device, test_entry_2.addr,
test_entry_2.group, test_entry_2.is_controller,
bytes([0x00, 0x00, 0x00]))]
assert mocked.call_count == 1
call_args = mocked.call_args_list
assert call_args[0].args[0] == test_device.db.add_on_device
assert call_args[0].args[1].addr == test_entry_2.addr
assert call_args[0].args[1].group == test_entry_2.group
assert call_args[0].args[1].is_controller == test_entry_2.is_controller
assert call_args[0].args[1].data == bytes([0x00, 0x00, 0x00])
assert "Modem db add CTRL can't find remote" in caplog.text

def test_db_add_resp_multigroup(self, test_device, test_device_2,
test_entry_multigroup):
test_device.add(test_device_2)
with mock.patch.object(IM.CommandSeq, 'add') as mocked:
test_device.db_add_resp_of(0x01, # local group for responder links on modem is always 0x01
test_device_2.addr,
test_entry_multigroup.group,
False,
False,
local_data=bytes([0x00, 0x00, 0x00]))
assert mocked.call_count == 1
call_args = mocked.call_args_list
assert call_args[0].args[0] == test_device.db.add_on_device
assert call_args[0].args[1].addr == test_entry_multigroup.addr
assert call_args[0].args[1].group == test_entry_multigroup.group
assert call_args[0].args[1].is_controller == False
assert call_args[0].args[1].data == bytes([0x00, 0x00, 0x00])