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.
255 Commits
2 Branches
2.3 MiB
 
Tree: 9eb6999b39
Branches Tags
${ item.name }
Create branch ${ searchTerm }
from '9eb6999b39'
${ noResults }
alarmdecoder/test/test_ad2.py

267 lines
11 KiB
Raw Blame History 

  1. import time

  2. 

  3. from unittest import TestCase

  4. from mock import Mock, MagicMock, patch

  5. 

  6. from alarmdecoder.decoder import AlarmDecoder

  7. from alarmdecoder.devices import USBDevice

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

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

  10. from alarmdecoder.zonetracking import Zonetracker

  11. 

  12. 

  13. class TestAlarmDecoder(TestCase):

  14.     def setUp(self):

  15.         self._panicked = False

  16.         self._relay_changed = False

  17.         self._power_changed = False

  18.         self._alarmed = False

  19.         self._bypassed = False

  20.         self._battery = False

  21.         self._fire = False

  22.         self._armed = False

  23.         self._got_config = False

  24.         self._message_received = False

  25.         self._rfx_message_received = False

  26.         self._lrr_message_received = False

  27. 

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

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

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

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

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

  33. 

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

  35. 

  36.         self._decoder._zonetracker = Mock(spec=Zonetracker)

  37.         self._decoder._zonetracker.on_fault = EventHandler(Event(), self._decoder._zonetracker)

  38.         self._decoder._zonetracker.on_restore = EventHandler(Event(), self._decoder._zonetracker)

  39. 

  40.         self._decoder.on_panic += self.on_panic

  41.         self._decoder.on_relay_changed += self.on_relay_changed

  42.         self._decoder.on_power_changed += self.on_power_changed

  43.         self._decoder.on_alarm += self.on_alarm

  44.         self._decoder.on_bypass += self.on_bypass

  45.         self._decoder.on_low_battery += self.on_battery

  46.         self._decoder.on_fire += self.on_fire

  47.         self._decoder.on_arm += self.on_arm

  48.         self._decoder.on_disarm += self.on_disarm

  49.         self._decoder.on_config_received += self.on_config

  50.         self._decoder.on_message += self.on_message

  51.         self._decoder.on_rfx_message += self.on_rfx_message

  52.         self._decoder.on_lrr_message += self.on_lrr_message

  53. 

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

  55.         self._decoder.open()

  56. 

  57.     def tearDown(self):

  58.         pass

  59. 

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

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

  62. 

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

  64.         self._relay_changed = True

  65. 

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

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

  68. 

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

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

  71. 

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

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

  74. 

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

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

  77. 

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

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

  80. 

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

  82.         self._armed = True

  83. 

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

  85.         self._armed = False

  86. 

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

  88.         self._got_config = True

  89. 

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

  91.         self._message_received = True

  92. 

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

  94.         self._rfx_message_received = True

  95. 

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

  97.         self._lrr_message_received = True

  98. 

  99.     def test_open(self):

  100.         self._decoder.open()

  101.         self._device.open.assert_any_calls()

  102. 

  103.     def test_close(self):

  104.         self._decoder.open()

  105. 

  106.         self._decoder.close()

  107.         self._device.close.assert_any_calls()

  108. 

  109.     def test_send(self):

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

  111.         self._device.write.assert_called_with('test')

  112. 

  113.     def test_get_config(self):

  114.         self._decoder.get_config()

  115.         self._device.write.assert_called_with("C\r")

  116. 

  117.     def test_save_config(self):

  118.         self._decoder.save_config()

  119.         self._device.write.assert_any_calls()

  120. 

  121.     def test_reboot(self):

  122.         self._decoder.reboot()

  123.         self._device.write.assert_called_with('=')

  124. 

  125.     def test_fault(self):

  126.         self._decoder.fault_zone(1)

  127.         self._device.write.assert_called_with("L{0:02}{1}\r".format(1, 1))

  128. 

  129.     def test_fault_wireproblem(self):

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

  131.         self._device.write.assert_called_with("L{0:02}{1}\r".format(1, 2))

  132. 

  133.     def test_clear_zone(self):

  134.         self._decoder.clear_zone(1)

  135.         self._device.write.assert_called_with("L{0:02}0\r".format(1))

  136. 

  137.     def test_message(self):

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

  139.         self.assertIsInstance(msg, Message)

  140. 

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

  142.         self.assertTrue(self._message_received)

  143. 

  144.     def test_message_kpe(self):

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

  146.         self.assertIsInstance(msg, Message)

  147. 

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

  149.         self.assertTrue(self._message_received)

  150. 

  151.     def test_expander_message(self):

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

  153.         self.assertIsInstance(msg, ExpanderMessage)

  154. 

  155.     def test_relay_message(self):

  156.         self._decoder.open()

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

  158.         self.assertIsInstance(msg, ExpanderMessage)

  159.         self.assertEquals(self._relay_changed, True)

  160. 

  161.     def test_rfx_message(self):

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

  163.         self.assertIsInstance(msg, RFMessage)

  164.         self.assertTrue(self._rfx_message_received)

  165. 

  166.     def test_panic(self):

  167.         self._decoder.open()

  168. 

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

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

  171. 

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

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

  174.         self.assertIsInstance(msg, LRRMessage)

  175. 

  176.     def test_config_message(self):

  177.         self._decoder.open()

  178. 

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

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

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

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

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

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

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

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

  187. 

  188.         self.assertEquals(self._got_config, True)

  189. 

  190.     def test_power_changed_event(self):

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

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

  193. 

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

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

  196. 

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

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

  199. 

  200.     def test_alarm_event(self):

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

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

  203. 

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

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

  206. 

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

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

  209. 

  210.     def test_zone_bypassed_event(self):

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

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

  213. 

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

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

  216. 

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

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

  219. 

  220.     def test_armed_away_event(self):

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

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

  223. 

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

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

  226. 

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

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

  229. 

  230.         self._armed = False

  231. 

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

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

  234. 

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

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

  237. 

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

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

  240. 

  241.     def test_battery_low_event(self):

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

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

  244. 

  245.         # force the timeout to expire.

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

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

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

  249. 

  250.     def test_fire_alarm_event(self):

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

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

  253. 

  254.         # force the timeout to expire.

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

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

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

  258. 

  259.     def test_hit_for_faults(self):

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

  261. 

  262.         self._decoder._device.write.assert_called_with('*')

  263. 

  264.     def test_zonetracker_update(self):

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

  266.         self._decoder._zonetracker.update.assert_called_with(msg)