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

316 lines
13 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.         self._expander_message_received = False

  28.         self._sending_received_status = None

  29.         self._alarm_restored = False

  30.         self._on_boot_received = False

  31.         self._zone_faulted = None

  32.         self._zone_restored = None

  33. 

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

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

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

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

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

  39. 

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

  41.         self._decoder.on_panic += self.on_panic

  42.         self._decoder.on_relay_changed += self.on_relay_changed

  43.         self._decoder.on_power_changed += self.on_power_changed

  44.         self._decoder.on_alarm += self.on_alarm

  45.         self._decoder.on_alarm_restored += self.on_alarm_restored

  46.         self._decoder.on_bypass += self.on_bypass

  47.         self._decoder.on_low_battery += self.on_battery

  48.         self._decoder.on_fire += self.on_fire

  49.         self._decoder.on_arm += self.on_arm

  50.         self._decoder.on_disarm += self.on_disarm

  51.         self._decoder.on_config_received += self.on_config

  52.         self._decoder.on_message += self.on_message

  53.         self._decoder.on_rfx_message += self.on_rfx_message

  54.         self._decoder.on_lrr_message += self.on_lrr_message

  55.         self._decoder.on_expander_message += self.on_expander_message

  56.         self._decoder.on_sending_received += self.on_sending_received

  57.         self._decoder.on_boot += self.on_boot

  58.         self._decoder.on_zone_fault += self.on_zone_fault

  59.         self._decoder.on_zone_restore += self.on_zone_restore

  60. 

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

  62.         self._decoder.open()

  63. 

  64.     def tearDown(self):

  65.         pass

  66. 

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

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

  69. 

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

  71.         self._relay_changed = True

  72. 

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

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

  75. 

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

  77.         self._alarmed = True

  78. 

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

  80.         self._alarm_restored = True

  81. 

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

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

  84. 

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

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

  87. 

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

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

  90. 

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

  92.         self._armed = True

  93. 

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

  95.         self._armed = False

  96. 

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

  98.         self._got_config = True

  99. 

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

  101.         self._message_received = True

  102. 

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

  104.         self._rfx_message_received = True

  105. 

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

  107.         self._lrr_message_received = True

  108. 

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

  110.         self._expander_message_received = True

  111. 

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

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

  114. 

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

  116.         self._on_boot_received = True

  117. 

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

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

  120. 

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

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

  123. 

  124.     def test_open(self):

  125.         self._decoder.open()

  126.         self._device.open.assert_any_calls()

  127. 

  128.     def test_close(self):

  129.         self._decoder.open()

  130. 

  131.         self._decoder.close()

  132.         self._device.close.assert_any_calls()

  133. 

  134.     def test_send(self):

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

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

  137. 

  138.     def test_get_config(self):

  139.         self._decoder.get_config()

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

  141. 

  142.     def test_save_config(self):

  143.         self._decoder.save_config()

  144.         self._device.write.assert_any_calls()

  145. 

  146.     def test_reboot(self):

  147.         self._decoder.reboot()

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

  149. 

  150.     def test_fault(self):

  151.         self._decoder.fault_zone(1)

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

  153. 

  154.     def test_fault_wireproblem(self):

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

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

  157. 

  158.     def test_clear_zone(self):

  159.         self._decoder.clear_zone(1)

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

  161. 

  162.     def test_message(self):

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

  164.         self.assertIsInstance(msg, Message)

  165. 

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

  167.         self.assertTrue(self._message_received)

  168. 

  169.     def test_message_kpm(self):

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

  171.         self.assertIsInstance(msg, Message)

  172. 

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

  174.         self.assertTrue(self._message_received)

  175. 

  176.     def test_expander_message(self):

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

  178.         self.assertIsInstance(msg, ExpanderMessage)

  179. 

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

  181.         self.assertTrue(self._expander_message_received)

  182. 

  183.     def test_relay_message(self):

  184.         self._decoder.open()

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

  186.         self.assertIsInstance(msg, ExpanderMessage)

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

  188. 

  189.     def test_rfx_message(self):

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

  191.         self.assertIsInstance(msg, RFMessage)

  192.         self.assertTrue(self._rfx_message_received)

  193. 

  194.     def test_panic(self):

  195.         self._decoder.open()

  196. 

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

  198.         self.assertEqual(self._panicked, True)

  199. 

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

  201.         self.assertEqual(self._panicked, False)

  202.         self.assertIsInstance(msg, LRRMessage)

  203. 

  204.     def test_config_message(self):

  205.         self._decoder.open()

  206. 

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

  208.         self.assertEqual(self._decoder.address, 18)

  209.         self.assertEqual(self._decoder.configbits, int('ff00', 16))

  210.         self.assertEqual(self._decoder.address_mask, int('ffffffff', 16))

  211.         self.assertEqual(self._decoder.emulate_zone, [False for x in range(5)])

  212.         self.assertEqual(self._decoder.emulate_relay, [False for x in range(4)])

  213.         self.assertEqual(self._decoder.emulate_lrr, False)

  214.         self.assertEqual(self._decoder.deduplicate, False)

  215. 

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

  217. 

  218.     def test_power_changed_event(self):

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

  220.         self.assertEqual(self._power_changed, False)   # Not set first time we hit it.

  221. 

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

  223.         self.assertEqual(self._power_changed, False)

  224. 

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

  226.         self.assertEqual(self._power_changed, True)

  227. 

  228.     def test_alarm_event(self):

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

  230.         self.assertEqual(self._alarmed, False)   # Not set first time we hit it.

  231. 

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

  233.         self.assertEqual(self._alarmed, False)

  234.         self.assertEqual(self._alarm_restored, True)

  235. 

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

  237.         self.assertEqual(self._alarmed, True)

  238. 

  239.     def test_zone_bypassed_event(self):

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

  241.         self.assertEqual(self._bypassed, False)   # Not set first time we hit it.

  242. 

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

  244.         self.assertEqual(self._bypassed, False)

  245. 

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

  247.         self.assertEqual(self._bypassed, True)

  248. 

  249.     def test_armed_away_event(self):

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

  251.         self.assertEqual(self._armed, False)   # Not set first time we hit it.

  252. 

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

  254.         self.assertEqual(self._armed, False)

  255. 

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

  257.         self.assertEqual(self._armed, True)

  258. 

  259.         self._armed = False

  260. 

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

  262.         self.assertEqual(self._armed, False)   # Not set first time we hit it.

  263. 

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

  265.         self.assertEqual(self._armed, False)

  266. 

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

  268.         self.assertEqual(self._armed, True)

  269. 

  270.     def test_battery_low_event(self):

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

  272.         self.assertEqual(self._battery, True)

  273. 

  274.         # force the timeout to expire.

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

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

  277.             self.assertEqual(self._battery, False)

  278. 

  279.     def test_fire_alarm_event(self):

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

  281.         self.assertEqual(self._fire, True)

  282. 

  283.         # force the timeout to expire.

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

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

  286.             self.assertEqual(self._fire, False)

  287. 

  288.     def test_hit_for_faults(self):

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

  290. 

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

  292. 

  293.     def test_sending_received(self):

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

  295.         self.assertTrue(self._sending_received_status)

  296. 

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

  298.         self.assertFalse(self._sending_received_status)

  299. 

  300.     def test_boot(self):

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

  302.         self.assertTrue(self._on_boot_received)

  303. 

  304.     def test_zone_fault_and_restore(self):

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

  306.         self.assertEqual(self._zone_faulted, 3)

  307. 

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

  309.         self.assertEqual(self._zone_faulted, 4)

  310. 

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

  312.         self.assertEqual(self._zone_faulted, 5)

  313. 

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

  315.         self.assertEqual(self._zone_restored, 3)