Allow to rollback to non-persistant connections

Sometimes, we need to rollback to old non-persistant connections
to allow other software to connect to the MAX! cube to pair
new devices or perform other operations not yet supported.

Author: unaiur

maxcube/commander.py

@@ -26,6 +26,7 @@ class Commander(object):
     def __init__(self, host: str, port: int):
         self.__host: str = host
         self.__port: int = port
+        self.use_persistent_connection = True
         self.__connection: Connection = None
         self.__unsolicited_messages: List[Message] = []
 
@@ -54,6 +55,8 @@ def update(self) -> List[Message]:
                 self.__connect(deadline)
         else:
             self.__connect(deadline)
+        if not self.use_persistent_connection:
+            self.disconnect()
         return self.get_unsolicited_messages()
 
     def send_radio_msg(self, hex_radio_msg: str) -> bool:
@@ -101,6 +104,10 @@ def __call(self, msg: Message, deadline: Deadline) -> Message:
             else:
                 raise
 
+        finally:
+            if not self.use_persistent_connection:
+                self.disconnect()
+
     def __is_connected(self) -> bool:
         return self.__connection is not None
 

maxcube/cube.py

@@ -44,6 +44,14 @@ def __init__(self, host: str, port: int = DEFAULT_PORT):
         self.update()
         self.log()
 
+    @property
+    def use_persistent_connection(self) -> bool:
+        return self.__commander.use_persistent_connection
+
+    @use_persistent_connection.setter
+    def use_persistent_connection(self, value: bool) -> None:
+        self.__commander.use_persistent_connection = value
+
     def disconnect(self):
         self.__commander.disconnect()
 

tests/test_commander.py

@@ -14,6 +14,7 @@
 S_CMD_SUCCESS = Message('S', '00,0,31')
 S_CMD_ERROR = Message('S', '100,1,31')
 S_CMD_THROTTLE_ERROR = Message('S', '100,1,0')
+Q_CMD = Message('q')
 
 TEST_TIMEOUT = Timeout('test', 1.0)
 
@@ -35,16 +36,17 @@ def testDisconnectIsNoopIfAlreadyDisconnected(self, ClassMock):
         self.connection.recv.assert_not_called()
         self.connection.close.assert_not_called()
 
-    def testUpdateOpensNewConnectionIfDisconnected(self, ClassMock):
+    def testUpdateOpensNewNonPersistantConnectionAndClosesIt(self, ClassMock):
         messages = [Message('H'), Message('L')]
         self.init(ClassMock)
+        self.commander.use_persistent_connection = False
         self.connection.recv.side_effect = messages
 
         self.assertEqual(messages, self.commander.update())
 
-        self.connection.send.assert_not_called()
+        self.connection.send.assert_called_once_with(Q_CMD)
         self.assertEqual(2, self.connection.recv.call_count)
-        self.connection.close.assert_not_called()
+        self.connection.close.assert_called_once()
 
     def testUpdateSendsCommandAfterAfterTimeout(self, ClassMock):
         messages = [Message('H'), None]
@@ -78,6 +80,21 @@ def testSendRadioMsgAutoconnects(self, ClassMock = None):
         self.assertEqual(2, self.connection.recv.call_count)
         self.connection.close.assert_not_called()
 
+    def testSendRadioMsgOpensNewNonPersistentConnectionAndClosesIt(self, ClassMock = None):
+        self.init(ClassMock)
+        self.commander.use_persistent_connection = False
+        self.connection.recv.side_effect = [
+            L_CMD_SUCCESS, # connection preambule
+            S_CMD_SUCCESS
+        ]
+
+        self.assertTrue(self.commander.send_radio_msg(S_CMD_HEX))
+        self.connection.send.assert_has_calls([
+            call(S_CMD), call(Q_CMD)
+        ])
+        self.assertEqual(2, self.connection.recv.call_count)
+        self.connection.close.assert_called_once()
+
     def testSendRadioMsgReusesConnection(self, ClassMock):
         self.testSendRadioMsgAutoconnects()
 

tests/test_cube.py

@@ -154,6 +154,13 @@ def test_disconnect(self, ClassMock):
         self.cube.disconnect()
         self.commander.disconnect.assert_called_once()
 
+    def test_use_persistent_connection(self, ClassMock):
+        self.init(ClassMock, INIT_RESPONSE_1)
+        self.commander.use_persistent_connection = True
+        self.assertTrue(self.cube.use_persistent_connection)
+        self.cube.use_persistent_connection = False
+        self.assertFalse(self.commander.use_persistent_connection)
+
     def test_is_thermostat(self, _):
         device = MaxDevice()
         device.type = MAX_CUBE