jmg
/
alarmdecoder
1
0
Fork 0
Code Issues 0 Pull Requests 0 Releases 0 Wiki Activity
A clone of: https://github.com/nutechsoftware/alarmdecoder This is requires as they dropped support for older firmware releases w/o building in backward compatibility code, and they had previously hardcoded pyserial to a python2 only version.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
330 Commits
2 Branches
2.3 MiB
 
Tree: 5244e91fb8
Branches Tags
${ item.name }
Create branch ${ searchTerm }
from '5244e91fb8'
${ noResults }
alarmdecoder/test/test_ad2.py

318 lines
14 KiB
Raw Blame History 

  1. import time

  2. 

  3. from builtins import bytes

  4. 

  5. from unittest import TestCase

  6. from mock import Mock, MagicMock, patch

  7. 

  8. from alarmdecoder.decoder import AlarmDecoder

  9. from alarmdecoder.devices import USBDevice

  10. from alarmdecoder.messages import Message, RFMessage, LRRMessage, ExpanderMessage

  11. from alarmdecoder.event.event import Event, EventHandler

  12. from alarmdecoder.zonetracking import Zonetracker

  13. 

  14. 

  15. class TestAlarmDecoder(TestCase):

  16.     def setUp(self):

  17.         self._panicked = False

  18.         self._relay_changed = False

  19.         self._power_changed = False

  20.         self._alarmed = False

  21.         self._bypassed = False

  22.         self._battery = False

  23.         self._fire = False

  24.         self._armed = False

  25.         self._got_config = False

  26.         self._message_received = False

  27.         self._rfx_message_received = False

  28.         self._lrr_message_received = False

  29.         self._expander_message_received = False

  30.         self._sending_received_status = None

  31.         self._alarm_restored = False

  32.         self._on_boot_received = False

  33.         self._zone_faulted = None

  34.         self._zone_restored = None

  35. 

  36.         self._device = Mock(spec=USBDevice)

  37.         self._device.on_open = EventHandler(Event(), self._device)

  38.         self._device.on_close = EventHandler(Event(), self._device)

  39.         self._device.on_read = EventHandler(Event(), self._device)

  40.         self._device.on_write = EventHandler(Event(), self._device)

  41. 

  42.         self._decoder = AlarmDecoder(self._device)

  43.         self._decoder.on_panic += self.on_panic

  44.         self._decoder.on_relay_changed += self.on_relay_changed

  45.         self._decoder.on_power_changed += self.on_power_changed

  46.         self._decoder.on_alarm += self.on_alarm

  47.         self._decoder.on_alarm_restored += self.on_alarm_restored

  48.         self._decoder.on_bypass += self.on_bypass

  49.         self._decoder.on_low_battery += self.on_battery

  50.         self._decoder.on_fire += self.on_fire

  51.         self._decoder.on_arm += self.on_arm

  52.         self._decoder.on_disarm += self.on_disarm

  53.         self._decoder.on_config_received += self.on_config

  54.         self._decoder.on_message += self.on_message

  55.         self._decoder.on_rfx_message += self.on_rfx_message

  56.         self._decoder.on_lrr_message += self.on_lrr_message

  57.         self._decoder.on_expander_message += self.on_expander_message

  58.         self._decoder.on_sending_received += self.on_sending_received

  59.         self._decoder.on_boot += self.on_boot

  60.         self._decoder.on_zone_fault += self.on_zone_fault

  61.         self._decoder.on_zone_restore += self.on_zone_restore

  62. 

  63.         self._decoder.address_mask = int('ffffffff', 16)

  64.         self._decoder.open()

  65. 

  66.     def tearDown(self):

  67.         pass

  68. 

  69.     def on_panic(self, sender, *args, **kwargs):

  70.         self._panicked = kwargs['status']

  71. 

  72.     def on_relay_changed(self, sender, *args, **kwargs):

  73.         self._relay_changed = True

  74. 

  75.     def on_power_changed(self, sender, *args, **kwargs):

  76.         self._power_changed = kwargs['status']

  77. 

  78.     def on_alarm(self, sender, *args, **kwargs):

  79.         self._alarmed = True

  80. 

  81.     def on_alarm_restored(self, sender, *args, **kwargs):

  82.         self._alarm_restored = True

  83. 

  84.     def on_bypass(self, sender, *args, **kwargs):

  85.         self._bypassed = kwargs['status']

  86. 

  87.     def on_battery(self, sender, *args, **kwargs):

  88.         self._battery = kwargs['status']

  89. 

  90.     def on_fire(self, sender, *args, **kwargs):

  91.         self._fire = kwargs['status']

  92. 

  93.     def on_arm(self, sender, *args, **kwargs):

  94.         self._armed = True

  95. 

  96.     def on_disarm(self, sender, *args, **kwargs):

  97.         self._armed = False

  98. 

  99.     def on_config(self, sender, *args, **kwargs):

  100.         self._got_config = True

  101. 

  102.     def on_message(self, sender, *args, **kwargs):

  103.         self._message_received = True

  104. 

  105.     def on_rfx_message(self, sender, *args, **kwargs):

  106.         self._rfx_message_received = True

  107. 

  108.     def on_lrr_message(self, sender, *args, **kwargs):

  109.         self._lrr_message_received = True

  110. 

  111.     def on_expander_message(self, sender, *args, **kwargs):

  112.         self._expander_message_received = True

  113. 

  114.     def on_sending_received(self, sender, *args, **kwargs):

  115.         self._sending_received_status = kwargs['status']

  116. 

  117.     def on_boot(self, sender, *args, **kwargs):

  118.         self._on_boot_received = True

  119. 

  120.     def on_zone_fault(self, sender, *args, **kwargs):

  121.         self._zone_faulted = kwargs['zone']

  122. 

  123.     def on_zone_restore(self, sender, *args, **kwargs):

  124.         self._zone_restored = kwargs['zone']

  125. 

  126.     def test_open(self):

  127.         self._decoder.open()

  128.         self._device.open.assert_any_calls()

  129. 

  130.     def test_close(self):

  131.         self._decoder.open()

  132. 

  133.         self._decoder.close()

  134.         self._device.close.assert_any_calls()

  135. 

  136.     def test_send(self):

  137.         self._decoder.send('test')

  138.         self._device.write.assert_called_with(b'test')

  139. 

  140.     def test_get_config(self):

  141.         self._decoder.get_config()

  142.         self._device.write.assert_called_with(b"C\r")

  143. 

  144.     def test_save_config(self):

  145.         self._decoder.save_config()

  146.         self._device.write.assert_any_calls()

  147. 

  148.     def test_reboot(self):

  149.         self._decoder.reboot()

  150.         self._device.write.assert_called_with(b'=')

  151. 

  152.     def test_fault(self):

  153.         self._decoder.fault_zone(1)

  154.         self._device.write.assert_called_with(bytes("L{0:02}{1}\r".format(1, 1), 'utf-8'))

  155. 

  156.     def test_fault_wireproblem(self):

  157.         self._decoder.fault_zone(1, simulate_wire_problem=True)

  158.         self._device.write.assert_called_with(bytes("L{0:02}{1}\r".format(1, 2), 'utf-8'))

  159. 

  160.     def test_clear_zone(self):

  161.         self._decoder.clear_zone(1)

  162.         self._device.write.assert_called_with(bytes("L{0:02}0\r".format(1), 'utf-8'))

  163. 

  164.     def test_message(self):

  165.         msg = self._decoder._handle_message(b'[0000000000000000----],000,[f707000600e5800c0c020000],"                                "')

  166.         self.assertIsInstance(msg, Message)

  167. 

  168.         self._decoder._on_read(self, data=b'[0000000000000000----],000,[f707000600e5800c0c020000],"                                "')

  169.         self.assertTrue(self._message_received)

  170. 

  171.     def test_message_kpm(self):

  172.         msg = self._decoder._handle_message(b'!KPM:[0000000000000000----],000,[f707000600e5800c0c020000],"                                "')

  173.         self.assertIsInstance(msg, Message)

  174. 

  175.         self._decoder._on_read(self, data=b'[0000000000000000----],000,[f707000600e5800c0c020000],"                                "')

  176.         self.assertTrue(self._message_received)

  177. 

  178.     def test_expander_message(self):

  179.         msg = self._decoder._handle_message(b'!EXP:07,01,01')

  180.         self.assertIsInstance(msg, ExpanderMessage)

  181. 

  182.         self._decoder._on_read(self, data=b'!EXP:07,01,01')

  183.         self.assertTrue(self._expander_message_received)

  184. 

  185.     def test_relay_message(self):

  186.         self._decoder.open()

  187.         msg = self._decoder._handle_message(b'!REL:12,01,01')

  188.         self.assertIsInstance(msg, ExpanderMessage)

  189.         self.assertEqual(self._relay_changed, True)

  190. 

  191.     def test_rfx_message(self):

  192.         msg = self._decoder._handle_message(b'!RFX:0180036,80')

  193.         self.assertIsInstance(msg, RFMessage)

  194.         self.assertTrue(self._rfx_message_received)

  195. 

  196.     def test_panic(self):

  197.         self._decoder.open()

  198. 

  199.         msg = self._decoder._handle_message(b'!LRR:012,1,ALARM_PANIC')

  200.         self.assertEquals(self._panicked, True)

  201. 

  202.         msg = self._decoder._handle_message(b'!LRR:012,1,CANCEL')

  203.         self.assertEquals(self._panicked, False)

  204.         self.assertIsInstance(msg, LRRMessage)

  205. 

  206.     def test_config_message(self):

  207.         self._decoder.open()

  208. 

  209.         msg = self._decoder._handle_message(b'!CONFIG>ADDRESS=18&CONFIGBITS=ff00&LRR=N&EXP=NNNNN&REL=NNNN&MASK=ffffffff&DEDUPLICATE=N')

  210.         self.assertEquals(self._decoder.address, 18)

  211.         self.assertEquals(self._decoder.configbits, int('ff00', 16))

  212.         self.assertEquals(self._decoder.address_mask, int('ffffffff', 16))

  213.         self.assertEquals(self._decoder.emulate_zone, [False for x in range(5)])

  214.         self.assertEquals(self._decoder.emulate_relay, [False for x in range(4)])

  215.         self.assertEquals(self._decoder.emulate_lrr, False)

  216.         self.assertEquals(self._decoder.deduplicate, False)

  217.         self.assertEqual(self._got_config, True)

  218. 

  219.     def test_power_changed_event(self):

  220.         msg = self._decoder._handle_message(b'[0000000100000000----],000,[f707000600e5800c0c020000],"                                "')

  221.         self.assertEquals(self._power_changed, False)   # Not set first time we hit it.

  222. 

  223.         msg = self._decoder._handle_message(b'[0000000000000000----],000,[f707000600e5800c0c020000],"                                "')

  224.         self.assertEquals(self._power_changed, False)

  225. 

  226.         msg = self._decoder._handle_message(b'[0000000100000000----],000,[f707000600e5800c0c020000],"                                "')

  227.         self.assertEquals(self._power_changed, True)

  228. 

  229.     def test_alarm_event(self):

  230.         msg = self._decoder._handle_message(b'[0000000000100000----],000,[f707000600e5800c0c020000],"                                "')

  231.         self.assertEquals(self._alarmed, False)   # Not set first time we hit it.

  232. 

  233.         msg = self._decoder._handle_message(b'[0000000000000000----],000,[f707000600e5800c0c020000],"                                "')

  234.         self.assertEquals(self._alarmed, False)

  235.         self.assertEquals(self._alarm_restored, True)

  236. 

  237.         msg = self._decoder._handle_message(b'[0000000000100000----],000,[f707000600e5800c0c020000],"                                "')

  238.         self.assertEquals(self._alarmed, True)

  239. 

  240.     def test_zone_bypassed_event(self):

  241.         msg = self._decoder._handle_message(b'[0000001000000000----],000,[f707000600e5800c0c020000],"                                "')

  242.         self.assertEquals(self._bypassed, False)   # Not set first time we hit it.

  243. 

  244.         msg = self._decoder._handle_message(b'[0000000000000000----],000,[f707000600e5800c0c020000],"                                "')

  245.         self.assertEquals(self._bypassed, False)

  246. 

  247.         msg = self._decoder._handle_message(b'[0000001000000000----],000,[f707000600e5800c0c020000],"                                "')

  248.         self.assertEquals(self._bypassed, True)

  249. 

  250.     def test_armed_away_event(self):

  251.         msg = self._decoder._handle_message(b'[0100000000000000----],000,[f707000600e5800c0c020000],"                                "')

  252.         self.assertEquals(self._armed, False)   # Not set first time we hit it.

  253. 

  254.         msg = self._decoder._handle_message(b'[0000000000000000----],000,[f707000600e5800c0c020000],"                                "')

  255.         self.assertEquals(self._armed, False)

  256. 

  257.         msg = self._decoder._handle_message(b'[0100000000000000----],000,[f707000600e5800c0c020000],"                                "')

  258.         self.assertEquals(self._armed, True)

  259. 

  260.         self._armed = False

  261. 

  262.         msg = self._decoder._handle_message(b'[0010000000000000----],000,[f707000600e5800c0c020000],"                                "')

  263.         self.assertEquals(self._armed, False)   # Not set first time we hit it.

  264. 

  265.         msg = self._decoder._handle_message(b'[0000000000000000----],000,[f707000600e5800c0c020000],"                                "')

  266.         self.assertEquals(self._armed, False)

  267. 

  268.         msg = self._decoder._handle_message(b'[0010000000000000----],000,[f707000600e5800c0c020000],"                                "')

  269.         self.assertEquals(self._armed, True)

  270. 

  271.     def test_battery_low_event(self):

  272.         msg = self._decoder._handle_message(b'[0000000000010000----],000,[f707000600e5800c0c020000],"                                "')

  273.         self.assertEquals(self._battery, True)

  274. 

  275.         # force the timeout to expire.

  276.         with patch.object(time, 'time', return_value=self._decoder._battery_status[1] + 35):

  277.             msg = self._decoder._handle_message(b'[0000000000000000----],000,[f707000600e5800c0c020000],"                                "')

  278.             self.assertEquals(self._battery, False)

  279. 

  280.     def test_fire_alarm_event(self):

  281.         msg = self._decoder._handle_message(b'[0000000000000100----],000,[f707000600e5800c0c020000],"                                "')

  282.         self.assertEquals(self._fire, True)

  283. 

  284.         # force the timeout to expire.

  285.         with patch.object(time, 'time', return_value=self._decoder._battery_status[1] + 35):

  286.             msg = self._decoder._handle_message(b'[0000000000000000----],000,[f707000600e5800c0c020000],"                                "')

  287.             self.assertEquals(self._fire, False)

  288. 

  289.     def test_hit_for_faults(self):

  290.         self._decoder._handle_message(b'[0000000000000000----],000,[f707000600e5800c0c020000],"Hit * for faults                "')

  291. 

  292.         self._decoder._device.write.assert_called_with(b'*')

  293. 

  294.     def test_sending_received(self):

  295.         self._decoder._on_read(self, data=b'!Sending.done')

  296.         self.assertTrue(self._sending_received_status)

  297. 

  298.         self._decoder._on_read(self, data=b'!Sending.....done')

  299.         self.assertFalse(self._sending_received_status)

  300. 

  301.     def test_boot(self):

  302.         self._decoder._on_read(self, data=b'!Ready')

  303.         self.assertTrue(self._on_boot_received)

  304. 

  305.     def test_zone_fault_and_restore(self):

  306.         self._decoder._on_read(self, data=b'[00010001000000000A--],003,[f70000051003000008020000000000],"FAULT 03                        "')

  307.         self.assertEquals(self._zone_faulted, 3)

  308. 

  309.         self._decoder._on_read(self, data=b'[00010001000000000A--],004,[f70000051003000008020000000000],"FAULT 04                        "')

  310.         self.assertEquals(self._zone_faulted, 4)

  311. 

  312.         self._decoder._on_read(self, data=b'[00010001000000000A--],005,[f70000051003000008020000000000],"FAULT 05                        "')

  313.         self.assertEquals(self._zone_faulted, 5)

  314. 

  315.         self._decoder._on_read(self, data=b'[00010001000000000A--],004,[f70000051003000008020000000000],"FAULT 04                        "')

  316.         self.assertEquals(self._zone_restored, 3)