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

324 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 ..ad2 import Overseer, AD2

  7. from ..devices import USBDevice

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

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

  10. from ..zonetracking import Zonetracker

  11. 

  12. class TestOverseer(TestCase):

  13.     def setUp(self):

  14.         self._attached = False

  15.         self._detached = False

  16. 

  17.         with patch.object(USBDevice, 'find_all', return_value=[(0, 0, 'AD2', 1, 'AD2')]):

  18.             self._overseer = Overseer()

  19. 

  20.     def tearDown(self):

  21.         self._overseer.stop()

  22. 

  23.     def attached_event(self, sender, args):

  24.         self._attached = True

  25. 

  26.     def detached_event(self, sender, args):

  27.         self._detached = True

  28. 

  29.     def test_find_all(self):

  30.         with patch.object(USBDevice, 'find_all', return_value=[(0, 0, 'AD2', 1, 'AD2')]):

  31.             devices = Overseer.find_all()

  32. 

  33.             self.assertEquals(devices[0][2], 'AD2')

  34. 

  35.     def test_create_default_param(self):

  36.         with patch.object(USBDevice, 'find_all', return_value=[(0, 0, 'AD2', 1, 'AD2')]):

  37.             device = Overseer.create()

  38. 

  39.             self.assertEquals(device._device.interface, ('AD2', 0))

  40. 

  41.     def test_create_with_param(self):

  42.         with patch.object(USBDevice, 'find_all', return_value=[(0, 0, 'AD2-1', 1, 'AD2'), (0, 0, 'AD2-2', 1, 'AD2')]):

  43.             device = Overseer.create((0, 0, 'AD2-1', 1, 'AD2'))

  44.             self.assertEquals(device._device.interface, ('AD2-1', 0))

  45. 

  46.             device = Overseer.create((0, 0, 'AD2-2', 1, 'AD2'))

  47.             self.assertEquals(device._device.interface, ('AD2-2', 0))

  48. 

  49.     def test_events(self):

  50.         self.assertEquals(self._attached, False)

  51.         self.assertEquals(self._detached, False)

  52. 

  53.         # this is ugly, but it works.

  54.         self._overseer.stop()

  55.         self._overseer._detect_thread = Overseer.DetectThread(self._overseer)

  56.         self._overseer.on_attached += self.attached_event

  57.         self._overseer.on_detached += self.detached_event

  58. 

  59.         with patch.object(USBDevice, 'find_all', return_value=[(0, 0, 'AD2-1', 1, 'AD2'), (0, 0, 'AD2-2', 1, 'AD2')]):

  60.             self._overseer.start()

  61. 

  62.             with patch.object(USBDevice, 'find_all', return_value=[(0, 0, 'AD2-2', 1, 'AD2')]):

  63.                 Overseer.find_all()

  64.                 time.sleep(1)

  65.                 self._overseer.stop()

  66. 

  67.         self.assertEquals(self._attached, True)

  68.         self.assertEquals(self._detached, True)

  69. 

  70. class TestAD2(TestCase):

  71.     def setUp(self):

  72.         self._panicked = False

  73.         self._relay_changed = False

  74.         self._power_changed = False

  75.         self._alarmed = False

  76.         self._bypassed = False

  77.         self._battery = (False, 0)

  78.         self._fire = (False, 0)

  79.         self._armed = False

  80.         self._got_config = False

  81.         self._message_received = False

  82.         self._rfx_message_received = False

  83.         self._lrr_message_received = False

  84. 

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

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

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

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

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

  90. 

  91.         self._ad2 = AD2(self._device)

  92. 

  93.         self._ad2._zonetracker = Mock(spec=Zonetracker)

  94.         self._ad2._zonetracker.on_fault = EventHandler(Event(), self._ad2._zonetracker)

  95.         self._ad2._zonetracker.on_restore = EventHandler(Event(), self._ad2._zonetracker)

  96. 

  97.         self._ad2.on_panic += self.on_panic

  98.         self._ad2.on_relay_changed += self.on_relay_changed

  99.         self._ad2.on_power_changed += self.on_power_changed

  100.         self._ad2.on_alarm += self.on_alarm

  101.         self._ad2.on_bypass += self.on_bypass

  102.         self._ad2.on_low_battery += self.on_battery

  103.         self._ad2.on_fire += self.on_fire

  104.         self._ad2.on_arm += self.on_arm

  105.         self._ad2.on_disarm += self.on_disarm

  106.         self._ad2.on_config_received += self.on_config

  107.         self._ad2.on_message += self.on_message

  108.         self._ad2.on_rfx_message += self.on_rfx_message

  109.         self._ad2.on_lrr_message += self.on_lrr_message

  110. 

  111.         self._ad2.address_mask = int('ffffffff', 16)

  112.         self._ad2.open()

  113. 

  114.     def tearDown(self):

  115.         pass

  116. 

  117.     def on_panic(self, sender, args):

  118.         self._panicked = args

  119. 

  120.     def on_relay_changed(self, sender, args):

  121.         self._relay_changed = True

  122. 

  123.     def on_power_changed(self, sender, args):

  124.         self._power_changed = args

  125. 

  126.     def on_alarm(self, sender, args):

  127.         self._alarmed = args

  128. 

  129.     def on_bypass(self, sender, args):

  130.         self._bypassed = args

  131. 

  132.     def on_battery(self, sender, args):

  133.         self._battery = args

  134. 

  135.     def on_fire(self, sender, args):

  136.         self._fire = args

  137. 

  138.     def on_arm(self, sender, args):

  139.         self._armed = True

  140. 

  141.     def on_disarm(self, sender, args):

  142.         self._armed = False

  143. 

  144.     def on_config(self, sender, args):

  145.         self._got_config = True

  146. 

  147.     def on_message(self, sender, args):

  148.         self._message_received = True

  149. 

  150.     def on_rfx_message(self, sender, args):

  151.         self._rfx_message_received = True

  152. 

  153.     def on_lrr_message(self, sender, args):

  154.         self._lrr_message_received = True

  155. 

  156.     def test_open(self):

  157.         self._ad2.open()

  158.         self._device.open.assert_any_calls()

  159. 

  160.     def test_close(self):

  161.         self._ad2.open()

  162. 

  163.         self._ad2.close()

  164.         self._device.close.assert_any_calls()

  165. 

  166.     def test_send(self):

  167.         self._ad2.send('test')

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

  169. 

  170.     def test_get_config(self):

  171.         self._ad2.get_config()

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

  173. 

  174.     def test_save_config(self):

  175.         self._ad2.save_config()

  176.         self._device.write.assert_any_calls()

  177. 

  178.     def test_reboot(self):

  179.         self._ad2.reboot()

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

  181. 

  182.     def test_fault(self):

  183.         self._ad2.fault_zone(1)

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

  185. 

  186.     def test_fault_wireproblem(self):

  187.         self._ad2.fault_zone(1, simulate_wire_problem=True)

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

  189. 

  190.     def test_clear_zone(self):

  191.         self._ad2.clear_zone(1)

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

  193. 

  194.     def test_message(self):

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

  196.         self.assertIsInstance(msg, Message)

  197. 

  198.         self._ad2._on_read(self, '[0000000000000000----],000,[f707000600e5800c0c020000],"                                "')

  199.         self.assertTrue(self._message_received)

  200. 

  201.     def test_message_kpe(self):

  202.         msg = self._ad2._handle_message('!KPE:[0000000000000000----],000,[f707000600e5800c0c020000],"                                "')

  203.         self.assertIsInstance(msg, Message)

  204. 

  205.         self._ad2._on_read(self, '[0000000000000000----],000,[f707000600e5800c0c020000],"                                "')

  206.         self.assertTrue(self._message_received)

  207. 

  208.     def test_expander_message(self):

  209.         msg = self._ad2._handle_message('!EXP:07,01,01')

  210.         self.assertIsInstance(msg, ExpanderMessage)

  211. 

  212.     def test_relay_message(self):

  213.         self._ad2.open()

  214.         msg = self._ad2._handle_message('!REL:12,01,01')

  215.         self.assertIsInstance(msg, ExpanderMessage)

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

  217. 

  218.     def test_rfx_message(self):

  219.         msg = self._ad2._handle_message('!RFX:0180036,80')

  220.         self.assertIsInstance(msg, RFMessage)

  221.         self.assertTrue(self._rfx_message_received)

  222. 

  223.     def test_panic(self):

  224.         self._ad2.open()

  225. 

  226.         msg = self._ad2._handle_message('!LRR:012,1,ALARM_PANIC')

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

  228. 

  229.         msg = self._ad2._handle_message('!LRR:012,1,CANCEL')

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

  231.         self.assertIsInstance(msg, LRRMessage)

  232. 

  233.     def test_config_message(self):

  234.         self._ad2.open()

  235. 

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

  237.         self.assertEquals(self._ad2.address, 18)

  238.         self.assertEquals(self._ad2.configbits, int('ff00', 16))

  239.         self.assertEquals(self._ad2.address_mask, int('ffffffff', 16))

  240.         self.assertEquals(self._ad2.emulate_zone, [False for x in range(5)])

  241.         self.assertEquals(self._ad2.emulate_relay, [False for x in range(4)])

  242.         self.assertEquals(self._ad2.emulate_lrr, False)

  243.         self.assertEquals(self._ad2.deduplicate, False)

  244. 

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

  246. 

  247.     def test_power_changed_event(self):

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

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

  250. 

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

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

  253. 

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

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

  256. 

  257.     def test_alarm_event(self):

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

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

  260. 

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

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

  263. 

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

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

  266. 

  267.     def test_zone_bypassed_event(self):

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

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

  270. 

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

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

  273. 

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

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

  276. 

  277.     def test_armed_away_event(self):

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

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

  280. 

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

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

  283. 

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

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

  286. 

  287.         self._armed = False

  288. 

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

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

  291. 

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

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

  294. 

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

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

  297. 

  298.     def test_battery_low_event(self):

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

  300.         self.assertEquals(self._battery[0], True)

  301. 

  302.         # force the timeout to expire.

  303.         with patch.object(time, 'time', return_value=self._battery[1] + 35):

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

  305.             self.assertEquals(self._battery[0], False)

  306. 

  307.     def test_fire_alarm_event(self):

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

  309.         self.assertEquals(self._fire[0], True)

  310. 

  311.         # force the timeout to expire.

  312.         with patch.object(time, 'time', return_value=self._fire[1] + 35):

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

  314.             self.assertEquals(self._fire[0], False)

  315. 

  316.     def test_hit_for_faults(self):

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

  318. 

  319.         self._ad2._device.write.assert_called_with('*')

  320. 

  321.     def test_zonetracker_update(self):

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

  323.         self._ad2._zonetracker.update.assert_called_with(msg)