diff --git a/test/test_ad2.py b/test/test_ad2.py index be8d8bd..a357b3e 100644 --- a/test/test_ad2.py +++ b/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) - diff --git a/test/test_devices.py b/test/test_devices.py index 8c47719..604af8d 100644 --- a/test/test_devices.py +++ b/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: diff --git a/test/test_messages.py b/test/test_messages.py index 211dce9..0da13c6 100644 --- a/test/test_messages.py +++ b/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('') diff --git a/test/test_zonetracking.py b/test/test_zonetracking.py index ab7ef01..14cd53a 100644 --- a/test/test_zonetracking.py +++ b/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)