[feature] Add disarm_failed counter as partition attribute (#61)

This counter will increase everytime we encounter a `DISARM_FAILED`
error for the partition, and will reset when the partition is
successfully disarmed.

apps/qolsysgw/mqtt/updater.py

@@ -361,6 +361,7 @@ def update_attributes(self):
                 'last_error_type': self._partition.last_error_type,
                 'last_error_desc': self._partition.last_error_desc,
                 'last_error_at': self._partition.last_error_at,
+                'disarm_failed': self._partition.disarm_failed,
             }),
         )
 

apps/qolsysgw/qolsys/partition.py

@@ -31,6 +31,7 @@ def __init__(self, partition_id: int, name: str, status: str,
         self._last_error_type = None
         self._last_error_desc = None
         self._last_error_at = None
+        self._disarm_failed = 0
 
     @property
     def id(self):
@@ -68,6 +69,10 @@ def last_error_desc(self):
     def last_error_at(self):
         return self._last_error_at
 
+    @property
+    def disarm_failed(self):
+        return self._disarm_failed
+
     @status.setter
     def status(self, value):
         new_value = value.upper()
@@ -80,6 +85,10 @@ def status(self, value):
             self.notify(change=self.NOTIFY_UPDATE_STATUS,
                         prev_value=prev_value, new_value=new_value)
 
+        # If the panel is disarmed, we can reset the failed disarm attempts
+        if new_value.upper() == 'DISARM':
+            self.disarm_failed = 0
+
         self.alarm_type = None
 
     @secure_arm.setter
@@ -106,6 +115,16 @@ def alarm_type(self, value):
             self.notify(change=self.NOTIFY_UPDATE_ALARM_TYPE,
                         prev_value=prev_value, new_value=value)
 
+    @disarm_failed.setter
+    def disarm_failed(self, value):
+        new_value = int(value)
+
+        if self._disarm_failed != new_value:
+            LOGGER.debug(f"Partition '{self.id}' ({self.name}) disarm failed updated to '{value}'")
+            self._disarm_failed = new_value
+
+            self.notify(change=self.NOTIFY_UPDATE_ATTRIBUTES)
+
     def triggered(self, alarm_type: str = None):
         self.status = 'ALARM'
         self.alarm_type = alarm_type
@@ -115,6 +134,10 @@ def errored(self, error_type: str, error_description: str):
         self._last_error_desc = error_description
         self._last_error_at = datetime.now(timezone.utc).isoformat()
 
+        # If this is a failed disarm attempt, let's increase the counter
+        if error_type.upper() == 'DISARM_FAILED':
+            self._disarm_failed += 1
+
         self.notify(change=self.NOTIFY_UPDATE_ATTRIBUTES)
 
     def zone(self, zone_id, default=None):

tests/end-to-end/test_qolsysgw.py

@@ -203,6 +203,7 @@ async def _check_initial_state(self, ctx):
                     'last_error_at': None,
                     'last_error_type': None,
                     'last_error_desc': None,
+                    'disarm_failed': 0,
                     'secure_arm': False,
                     'supported_features': 63,
                 },
@@ -339,6 +340,7 @@ async def _check_initial_state(self, ctx):
                     'last_error_at': None,
                     'last_error_type': None,
                     'last_error_desc': None,
+                    'disarm_failed': 0,
                     'secure_arm': False,
                     'supported_features': 63,
                 },
@@ -475,6 +477,7 @@ async def _check_panel_events(self, ctx):
                     'last_error_at': ISODATE,
                     'last_error_type': 'DISARM_FAILED',
                     'last_error_desc': 'Invalid usercode',
+                    'disarm_failed': 1,
                     'secure_arm': False,
                     'supported_features': 63,
                 },
@@ -611,6 +614,7 @@ async def _check_panel_events(self, ctx):
                     'last_error_at': None,
                     'last_error_type': None,
                     'last_error_desc': None,
+                    'disarm_failed': 0,
                     'secure_arm': True,
                     'supported_features': 63,
                 },

tests/integration/test_qolsys_events.py

@@ -1,5 +1,7 @@
 import json
 
+from types import SimpleNamespace
+
 from unittest import mock
 
 import testenv  # noqa: F401
@@ -107,6 +109,7 @@ async def _check_partition_mqtt_messages(self, gw, partition_flat_name,
                     'last_error_type': state.last_error_type,
                     'last_error_desc': state.last_error_desc,
                     'last_error_at': state.last_error_at,
+                    'disarm_failed': state.disarm_failed,
                 },
                 json.loads(mqtt_attributes['payload']),
             )
@@ -1146,14 +1149,19 @@ async def test_integration_event_alarm_auxiliary(self):
             alarm_type='AUXILIARY',
         )
 
-    async def _test_integration_event_error(self, error_type, error_desc):
-        panel, gw, _, _ = await self._ready_panel_and_gw(
-            partition_ids=[0],
-            zone_ids=[100],
-            partition_status={
-                0: 'ARM_STAY',
-            },
-        )
+    async def _test_integration_event_error(self, error_type, error_desc,
+                                            extra_expect=None, init_data=None):
+        if init_data:
+            panel = init_data.panel
+            gw = init_data.gw
+        else:
+            panel, gw, _, _ = await self._ready_panel_and_gw(
+                partition_ids=[0],
+                zone_ids=[100],
+                partition_status={
+                    0: 'ARM_STAY',
+                },
+            )
 
         event = {
             'event': 'ERROR',
@@ -1179,10 +1187,17 @@ async def _test_integration_event_error(self, error_type, error_desc):
             'last_error_desc': error_desc,
             'last_error_at': ISODATE,
         }
+        if extra_expect:
+            expected.update(extra_expect)
         actual = json.loads(attributes['payload'])
         actual_subset = {k: v for k, v in actual.items() if k in expected}
         self.assertDictEqual(expected, actual_subset)
 
+        return SimpleNamespace(
+            panel=panel,
+            gw=gw,
+        )
+
     async def test_integration_event_error_usercode(self):
         await self._test_integration_event_error(
             error_type='usercode',
@@ -1191,7 +1206,53 @@ async def test_integration_event_error_usercode(self):
         )
 
     async def test_integration_event_error_disarm_failed(self):
-        await self._test_integration_event_error(
-            error_type='DISARM_FAILED',
-            error_desc='Invalid usercode',
-        )
+        with self.subTest(msg='Disarm failed error is handled properly'):
+            init_data = await self._test_integration_event_error(
+                error_type='DISARM_FAILED',
+                error_desc='Invalid usercode',
+                extra_expect={
+                    'disarm_failed': 1,
+                },
+            )
+
+        with self.subTest(msg='Other disarm failed error keeps raising the counter'):
+            await self._test_integration_event_error(
+                error_type='DISARM_FAILED',
+                error_desc='Invalid usercode',
+                extra_expect={
+                    'disarm_failed': 2,
+                },
+                init_data=init_data,
+            )
+
+        with self.subTest(msg='Disarming the panel resets the counter'):
+            panel = init_data.panel
+            gw = init_data.gw
+
+            self.assertEqual(2, gw._state.partition(0).disarm_failed)
+
+            event = {
+                'event': 'ARMING',
+                'arming_type': 'DISARM',
+                'partition_id': 0,
+                'version': 1,
+                'requestID': '<request_id>',
+            }
+            await panel.writeline(event)
+
+            attributes = await gw.wait_for_next_mqtt_publish(
+                timeout=self._TIMEOUT,
+                filters={'topic': 'homeassistant/alarm_control_panel/'
+                                  'qolsys_panel/partition0/attributes'},
+            )
+
+            self.assertIsNotNone(attributes)
+
+            expected = {
+                'disarm_failed': 0,
+            }
+            actual = json.loads(attributes['payload'])
+            actual_subset = {k: v for k, v in actual.items() if k in expected}
+            self.assertDictEqual(expected, actual_subset)
+
+            self.assertEqual(0, gw._state.partition(0).disarm_failed)