Test changes and additions.

test/test_ad2.py

@@ -10,6 +10,9 @@ from alarmdecoder.devices import USBDevice
 from alarmdecoder.messages import Message, RFMessage, LRRMessage, ExpanderMessage
 from alarmdecoder.event.event import Event, EventHandler
 from alarmdecoder.zonetracking import Zonetracker
+from alarmdecoder.panels import ADEMCO, DSC
+from alarmdecoder.messages.lrr import LRR_EVENT_TYPE, LRR_EVENT_STATUS
+from alarmdecoder.states import FireState
 
 
 class TestAlarmDecoder(TestCase):
@@ -66,6 +69,7 @@ class TestAlarmDecoder(TestCase):
     def tearDown(self):
         pass
 
+    ### Library events
     def on_panic(self, sender, *args, **kwargs):
         self._panicked = kwargs['status']
 
@@ -123,6 +127,7 @@ class TestAlarmDecoder(TestCase):
     def on_zone_restore(self, sender, *args, **kwargs):
         self._zone_restored = kwargs['zone']
 
+    ### Tests
     def test_open(self):
         self._decoder.open()
         self._device.open.assert_any_calls()
@@ -183,108 +188,132 @@ class TestAlarmDecoder(TestCase):
         self.assertTrue(self._expander_message_received)
 
     def test_relay_message(self):
-        self._decoder.open()
         msg = self._decoder._handle_message(b'!REL:12,01,01')
         self.assertIsInstance(msg, ExpanderMessage)
-        self.assertEqual(self._relay_changed, True)
+        self.assertTrue(self._relay_changed)
 
     def test_rfx_message(self):
         msg = self._decoder._handle_message(b'!RFX:0180036,80')
         self.assertIsInstance(msg, RFMessage)
         self.assertTrue(self._rfx_message_received)
 
-    def test_panic(self):
-        self._decoder.open()
-
+    def test_panic_v1(self):
+        # LRR v1
         msg = self._decoder._handle_message(b'!LRR:012,1,ALARM_PANIC')
-        self.assertEquals(self._panicked, True)
+        self.assertIsInstance(msg, LRRMessage)
+        self.assertTrue(self._panicked)
 
         msg = self._decoder._handle_message(b'!LRR:012,1,CANCEL')
-        self.assertEquals(self._panicked, False)
         self.assertIsInstance(msg, LRRMessage)
+        self.assertFalse(self._panicked)
 
-    def test_config_message(self):
-        self._decoder.open()
+    def test_panic_v2(self):
+        # LRR v2
+        msg = self._decoder._handle_message(b'!LRR:099,1,CID_1123,ff') # Panic
+        self.assertIsInstance(msg, LRRMessage)
+        self.assertTrue(self._panicked)
 
-        msg = self._decoder._handle_message(b'!CONFIG>ADDRESS=18&CONFIGBITS=ff00&LRR=N&EXP=NNNNN&REL=NNNN&MASK=ffffffff&DEDUPLICATE=N')
+        msg = self._decoder._handle_message(b'!LRR:001,1,CID_1406,ff') # Cancel
+        self.assertIsInstance(msg, LRRMessage)
+        self.assertFalse(self._panicked)
+
+    def test_config_message(self):
+        msg = self._decoder._handle_message(b'!CONFIG>MODE=A&CONFIGBITS=ff04&ADDRESS=18&LRR=N&COM=N&EXP=NNNNN&REL=NNNN&MASK=ffffffff&DEDUPLICATE=N')
+        self.assertEquals(self._decoder.mode, ADEMCO)
         self.assertEquals(self._decoder.address, 18)
-        self.assertEquals(self._decoder.configbits, int('ff00', 16))
+        self.assertEquals(self._decoder.configbits, int('ff04', 16))
         self.assertEquals(self._decoder.address_mask, int('ffffffff', 16))
         self.assertEquals(self._decoder.emulate_zone, [False for x in range(5)])
         self.assertEquals(self._decoder.emulate_relay, [False for x in range(4)])
-        self.assertEquals(self._decoder.emulate_lrr, False)
-        self.assertEquals(self._decoder.deduplicate, False)
-        self.assertEqual(self._got_config, True)
+        self.assertFalse(self._decoder.emulate_lrr)
+        self.assertFalse(self._decoder.emulate_com)
+        self.assertFalse(self._decoder.deduplicate)
+        self.assertTrue(self._got_config)
 
     def test_power_changed_event(self):
         msg = self._decoder._handle_message(b'[0000000100000000----],000,[f707000600e5800c0c020000],"                                "')
-        self.assertEquals(self._power_changed, False)   # Not set first time we hit it.
+        self.assertFalse(self._power_changed)   # Not set first time we hit it.
 
         msg = self._decoder._handle_message(b'[0000000000000000----],000,[f707000600e5800c0c020000],"                                "')
-        self.assertEquals(self._power_changed, False)
+        self.assertFalse(self._power_changed)
 
         msg = self._decoder._handle_message(b'[0000000100000000----],000,[f707000600e5800c0c020000],"                                "')
-        self.assertEquals(self._power_changed, True)
+        self.assertTrue(self._power_changed)
 
     def test_alarm_event(self):
         msg = self._decoder._handle_message(b'[0000000000100000----],000,[f707000600e5800c0c020000],"                                "')
-        self.assertEquals(self._alarmed, False)   # Not set first time we hit it.
+        self.assertFalse(self._alarmed)   # Not set first time we hit it.
 
         msg = self._decoder._handle_message(b'[0000000000000000----],000,[f707000600e5800c0c020000],"                                "')
-        self.assertEquals(self._alarmed, False)
-        self.assertEquals(self._alarm_restored, True)
+        self.assertFalse(self._alarmed)
+        self.assertTrue(self._alarm_restored)
 
         msg = self._decoder._handle_message(b'[0000000000100000----],000,[f707000600e5800c0c020000],"                                "')
-        self.assertEquals(self._alarmed, True)
+        self.assertTrue(self._alarmed)
 
     def test_zone_bypassed_event(self):
-        msg = self._decoder._handle_message(b'[0000001000000000----],000,[f707000600e5800c0c020000],"                                "')
-        self.assertEquals(self._bypassed, False)   # Not set first time we hit it.
-
         msg = self._decoder._handle_message(b'[0000000000000000----],000,[f707000600e5800c0c020000],"                                "')
-        self.assertEquals(self._bypassed, False)
+        self.assertFalse(self._bypassed)
 
         msg = self._decoder._handle_message(b'[0000001000000000----],000,[f707000600e5800c0c020000],"                                "')
-        self.assertEquals(self._bypassed, True)
+        self.assertTrue(self._bypassed)
 
     def test_armed_away_event(self):
         msg = self._decoder._handle_message(b'[0100000000000000----],000,[f707000600e5800c0c020000],"                                "')
-        self.assertEquals(self._armed, False)   # Not set first time we hit it.
+        self.assertFalse(self._armed)   # Not set first time we hit it.
+
+        msg = self._decoder._handle_message(b'[0100000000000000----],000,[f707000600e5800c0c020000],"                                "')
+        self.assertFalse(self._armed)
 
         msg = self._decoder._handle_message(b'[0000000000000000----],000,[f707000600e5800c0c020000],"                                "')
-        self.assertEquals(self._armed, False)
+        self.assertFalse(self._armed)
 
         msg = self._decoder._handle_message(b'[0100000000000000----],000,[f707000600e5800c0c020000],"                                "')
-        self.assertEquals(self._armed, True)
+        self.assertTrue(self._armed)
 
         self._armed = False
-
         msg = self._decoder._handle_message(b'[0010000000000000----],000,[f707000600e5800c0c020000],"                                "')
-        self.assertEquals(self._armed, False)   # Not set first time we hit it.
+        self.assertTrue(self._armed)
 
         msg = self._decoder._handle_message(b'[0000000000000000----],000,[f707000600e5800c0c020000],"                                "')
-        self.assertEquals(self._armed, False)
-
-        msg = self._decoder._handle_message(b'[0010000000000000----],000,[f707000600e5800c0c020000],"                                "')
-        self.assertEquals(self._armed, True)
+        self.assertFalse(self._armed)
 
     def test_battery_low_event(self):
         msg = self._decoder._handle_message(b'[0000000000010000----],000,[f707000600e5800c0c020000],"                                "')
-        self.assertEquals(self._battery, True)
+        self.assertTrue(self._battery)
 
         # force the timeout to expire.
         with patch.object(time, 'time', return_value=self._decoder._battery_status[1] + 35):
             msg = self._decoder._handle_message(b'[0000000000000000----],000,[f707000600e5800c0c020000],"                                "')
-            self.assertEquals(self._battery, False)
+            self.assertFalse(self._battery)
 
     def test_fire_alarm_event(self):
+        self._fire = FireState.NONE
+
         msg = self._decoder._handle_message(b'[0000000000000100----],000,[f707000600e5800c0c020000],"                                "')
-        self.assertEquals(self._fire, True)
+        self.assertEquals(self._fire, FireState.ALARM)
 
         # force the timeout to expire.
-        with patch.object(time, 'time', return_value=self._decoder._battery_status[1] + 35):
+        with patch.object(time, 'time', return_value=self._decoder._fire_status[1] + 35):
+            msg = self._decoder._handle_message(b'[0000000000000000----],000,[f707000600e5800c0c020000],"                                "')
+            self.assertEquals(self._fire, FireState.NONE)
+
+    def test_fire_lrr(self):
+        self._fire = FireState.NONE
+
+        msg = self._decoder._handle_message(b'!LRR:095,1,CID_1110,ff') # Fire: Non-specific
+
+        self.assertIsInstance(msg, LRRMessage)
+        self.assertEquals(self._fire, FireState.ALARM)
+
+        msg = self._decoder._handle_message(b'!LRR:001,1,CID_1406,ff') # Open/Close: Cancel
+        self.assertIsInstance(msg, LRRMessage)
+        self.assertEquals(self._fire, FireState.ACKNOWLEDGED)
+
+        # force the timeout to expire.
+        with patch.object(time, 'time', return_value=self._decoder._fire_status[1] + 35):
             msg = self._decoder._handle_message(b'[0000000000000000----],000,[f707000600e5800c0c020000],"                                "')
-            self.assertEquals(self._fire, False)
+            self.assertEquals(self._fire, FireState.NONE)
 
     def test_hit_for_faults(self):
         self._decoder._handle_message(b'[0000000000000000----],000,[f707000600e5800c0c020000],"Hit * for faults                "')
@@ -314,4 +343,3 @@ class TestAlarmDecoder(TestCase):
 
         self._decoder._on_read(self, data=b'[00010001000000000A--],004,[f70000051003000008020000000000],"FAULT 04                        "')
         self.assertEquals(self._zone_restored, 3)
-

test/test_devices.py

@@ -48,12 +48,14 @@ class TestUSBDevice(TestCase):
     def tearDown(self):
         self._device.close()
 
+    ### Library events
     def attached_event(self, sender, *args, **kwargs):
         self._attached = True
 
     def detached_event(self, sender, *args, **kwargs):
         self._detached = True
 
+    ### Tests
     def test_find_default_param(self):
         with patch.object(Ftdi, 'find_all', return_value=[(0, 0, 'AD2', 1, 'AD2')]):
             device = USBDevice.find()
@@ -69,8 +71,8 @@ class TestUSBDevice(TestCase):
             self.assertEqual(device.interface, 'AD2-2')
 
     def test_events(self):
-        self.assertEqual(self._attached, False)
-        self.assertEqual(self._detached, False)
+        self.assertFalse(self._attached)
+        self.assertFalse(self._detached)
 
         # this is ugly, but it works.
         with patch.object(USBDevice, 'find_all', return_value=[(0, 0, 'AD2-1', 1, 'AD2'), (0, 0, 'AD2-2', 1, 'AD2')]):
@@ -81,8 +83,8 @@ class TestUSBDevice(TestCase):
                 time.sleep(1)
                 USBDevice.stop_detection()
 
-        self.assertEqual(self._attached, True)
-        self.assertEqual(self._detached, True)
+        self.assertTrue(self._attached)
+        self.assertTrue(self._detached)
 
     def test_find_all(self):
         with patch.object(USBDevice, 'find_all', return_value=[]) as mock:
@@ -149,6 +151,7 @@ class TestSerialDevice(TestCase):
     def tearDown(self):
         self._device.close()
 
+    ### Tests
     def test_open(self):
         self._device.interface = '/dev/ttyS0'
 
@@ -249,6 +252,7 @@ class TestSocketDevice(TestCase):
     def tearDown(self):
         self._device.close()
 
+    ### Tests
     def test_open(self):
         with patch.object(socket.socket, '__init__', return_value=None):
             with patch.object(socket.socket, 'connect', return_value=None) as mock:
@@ -411,12 +415,14 @@ if have_pyftdi:
         def tearDown(self):
             self._device.close()
 
+        ### Library events
         def attached_event(self, sender, *args, **kwargs):
             self._attached = True
 
         def detached_event(self, sender, *args, **kwargs):
             self._detached = True
 
+        ### Tests
         def test_find_default_param(self):
             with patch.object(Ftdi, 'find_all', return_value=[(0, 0, 'AD2', 1, 'AD2')]):
                 device = USBDevice.find()
@@ -432,8 +438,8 @@ if have_pyftdi:
                 self.assertEquals(device.interface, 'AD2-2')
 
         def test_events(self):
-            self.assertEquals(self._attached, False)
-            self.assertEquals(self._detached, False)
+            self.assertFalse(self._attached)
+            self.assertFalse(self._detached)
 
             # this is ugly, but it works.
             with patch.object(USBDevice, 'find_all', return_value=[(0, 0, 'AD2-1', 1, 'AD2'), (0, 0, 'AD2-2', 1, 'AD2')]):
@@ -444,8 +450,8 @@ if have_pyftdi:
                     time.sleep(1)
                     USBDevice.stop_detection()
 
-            self.assertEquals(self._attached, True)
-            self.assertEquals(self._detached, True)
+            self.assertTrue(self._attached)
+            self.assertTrue(self._detached)
 
         def test_find_all(self):
             with patch.object(USBDevice, 'find_all', return_value=[]) as mock:

test/test_messages.py

@@ -1,7 +1,9 @@
 from unittest import TestCase
 
 from alarmdecoder.messages import Message, ExpanderMessage, RFMessage, LRRMessage
+from alarmdecoder.messages.lrr import LRR_EVENT_TYPE, LRR_CID_EVENT, LRR_EVENT_STATUS
 from alarmdecoder.util import InvalidMessageError
+from alarmdecoder.panels import ADEMCO
 
 
 class TestMessages(TestCase):
@@ -11,10 +13,32 @@ class TestMessages(TestCase):
     def tearDown(self):
         pass
 
+    ### Tests
     def test_message_parse(self):
-        msg = Message('[0000000000000000----],001,[f707000600e5800c0c020000],"FAULT 1                         "')
-
+        msg = Message('[00000000000000000A--],001,[f707000600e5800c0c020000],"FAULT 1                         "')
+
+        self.assertFalse(msg.ready)
+        self.assertFalse(msg.armed_away)
+        self.assertFalse(msg.armed_home)
+        self.assertFalse(msg.backlight_on)
+        self.assertFalse(msg.programming_mode)
+        self.assertEqual(msg.beeps, 0)
+        self.assertFalse(msg.zone_bypassed)
+        self.assertFalse(msg.ac_power)
+        self.assertFalse(msg.chime_on)
+        self.assertFalse(msg.alarm_event_occurred)
+        self.assertFalse(msg.alarm_sounding)
+        self.assertFalse(msg.battery_low)
+        self.assertFalse(msg.entry_delay_off)
+        self.assertFalse(msg.fire_alarm)
+        self.assertFalse(msg.check_zone)
+        self.assertFalse(msg.perimeter_only)
+        self.assertFalse(msg.system_fault)
+        self.assertFalse(msg.panel_type, ADEMCO)
         self.assertEqual(msg.numeric_code, '001')
+        self.assertEqual(msg.mask, int('07000600', 16))
+        self.assertEqual(msg.cursor_location, -1)
+        self.assertEqual(msg.text, 'FAULT 1                         ')
 
     def test_message_parse_fail(self):
         with self.assertRaises(InvalidMessageError):
@@ -24,6 +48,8 @@ class TestMessages(TestCase):
         msg = ExpanderMessage('!EXP:07,01,01')
 
         self.assertEqual(msg.address, 7)
+        self.assertEqual(msg.channel, 1)
+        self.assertEqual(msg.value, 1)
 
     def test_expander_message_parse_fail(self):
         with self.assertRaises(InvalidMessageError):
@@ -33,16 +59,34 @@ class TestMessages(TestCase):
         msg = RFMessage('!RFX:0180036,80')
 
         self.assertEqual(msg.serial_number, '0180036')
+        self.assertEqual(msg.value, int('80', 16))
 
     def test_rf_message_parse_fail(self):
         with self.assertRaises(InvalidMessageError):
             msg = RFMessage('')
 
-    def test_lrr_message_parse(self):
+    def test_lrr_message_parse_v1(self):
         msg = LRRMessage('!LRR:012,1,ARM_STAY')
 
+        self.assertEqual(msg.event_data, '012')
+        self.assertEqual(msg.partition, '1')
         self.assertEqual(msg.event_type, 'ARM_STAY')
 
+    def test_lrr_message_parse_v2(self):
+        msg = LRRMessage(b'!LRR:001,1,CID_3401,ff')
+        self.assertIsInstance(msg, LRRMessage)
+        self.assertEquals(msg.event_data, '001')
+        self.assertEquals(msg.partition, '1')
+        self.assertEquals(msg.event_prefix, 'CID')
+        self.assertEquals(msg.event_source, LRR_EVENT_TYPE.CID)
+        self.assertEquals(msg.event_status, LRR_EVENT_STATUS.RESTORE)
+        self.assertEquals(msg.event_code, LRR_CID_EVENT.OPENCLOSE_BY_USER)
+        self.assertEquals(msg.report_code, 'ff')
+
+    def test_lrr_event_code_override(self):
+        msg = LRRMessage(b'!LRR:001,1,CID_3400,01')
+        self.assertEquals(msg.event_code, LRR_CID_EVENT.OPENCLOSE_BY_USER)  # 400 -> 401
+
     def test_lrr_message_parse_fail(self):
         with self.assertRaises(InvalidMessageError):
             msg = LRRMessage('')

test/test_zonetracking.py

@@ -23,18 +23,21 @@ class TestZonetracking(TestCase):
     def tearDown(self):
         pass
 
+    ### Library events
     def fault_event(self, sender, *args, **kwargs):
         self._faulted = True
 
     def restore_event(self, sender, *args, **kwargs):
         self._restored = True
 
+    ### Util
     def _build_expander_message(self, msg):
         msg = ExpanderMessage(msg)
         zone = self._zonetracker.expander_to_zone(msg.address, msg.channel)
 
         return zone, msg
 
+    ### Tests
     def test_zone_fault(self):
         zone, msg = self._build_expander_message('!EXP:07,01,01')
         self._zonetracker.update(msg)