diff --git a/tests/test_comnav_0_106.py b/tests/test_comnav_0_106.py index bb3840b..c27c714 100644 --- a/tests/test_comnav_0_106.py +++ b/tests/test_comnav_0_106.py @@ -104,7 +104,7 @@ def test_comnav_0_106_communication(mock_post): assert len(uobj.areas) == 1 assert uobj.areas[0]['name'] == 'Area 1' assert uobj.areas[0]['bank'] == 0 - assert uobj.areas[0]['sequence'] == 244 + assert uobj.areas[0]['sequence'] == 1 assert uobj.areas[0]['status'] == 'Ready' assert isinstance(uobj.zones, dict) @@ -158,7 +158,7 @@ def test_comnav_0_106_communication(mock_post): assert mock_post.call_count == 1 assert mock_post.call_args_list[0][0][0] == \ 'http://zerowire/user/seq.xml' - assert uobj.areas[0]['sequence'] == 244 + assert uobj.areas[0]['sequence'] == 1 # Update our sequence file so that it reflects a change # This update will report an area update and 2 zones (bank 0 and 4) @@ -206,8 +206,8 @@ def test_comnav_0_106_communication(mock_post): assert len(uobj.areas) == 1 assert uobj.areas[0]['name'] == 'Area 1' assert uobj.areas[0]['bank'] == 0 - # Our sequence got bumped - assert uobj.areas[0]['sequence'] == 245 + # Our sequence got bumped because of a status change + assert uobj.areas[0]['sequence'] == 2 assert uobj.areas[0]['status'] == 'Not Ready' # Reset our mock object @@ -222,4 +222,4 @@ def test_comnav_0_106_communication(mock_post): assert mock_post.call_count == 1 assert mock_post.call_args_list[0][0][0] == \ 'http://zerowire/user/seq.xml' - assert uobj.areas[0]['sequence'] == 245 + assert uobj.areas[0]['sequence'] == 2 diff --git a/tests/test_comnav_0_108-zone-check.py b/tests/test_comnav_0_108-zone-check.py new file mode 100644 index 0000000..0a22776 --- /dev/null +++ b/tests/test_comnav_0_108-zone-check.py @@ -0,0 +1,191 @@ +# -*- coding: utf-8 -*- +# +# Copyright (C) 2021 Chris Caron +# All rights reserved. +# +# This code is licensed under the MIT License. +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files(the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions : +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. + +import mock +import requests +from os.path import join +from os.path import dirname + +from ultrasync import UltraSync +from ultrasync.common import NX595EVendor + +# Disable logging for a cleaner testing output +import logging +logging.disable(logging.CRITICAL) + +# Reference Directory +ULTRASYNC_TEST_VAR_DIR = \ + join(dirname(__file__), 'var', NX595EVendor.COMNAV, '0.108-zone-test') + + +@mock.patch('requests.Session.post') +def test_comnav_0_108_zone_filter(mock_post): + """ + Test ComNav v0.108 Zone Filtering + + """ + + # A area response object + arobj = mock.Mock() + + # Simulate a valid login return + with open(join(ULTRASYNC_TEST_VAR_DIR, 'area.htm'), 'rb') as f: + arobj.content = f.read() + arobj.status_code = requests.codes.ok + + # A zone response object + zrobj = mock.Mock() + + # Simulate initial zone configuration + with open(join(ULTRASYNC_TEST_VAR_DIR, 'zones.htm'), 'rb') as f: + zrobj.content = f.read() + zrobj.status_code = requests.codes.ok + + # A sequence response object + seq_obj = mock.Mock() + + # Simulate initial sequence configuration + with open(join(ULTRASYNC_TEST_VAR_DIR, 'seq.xml'), 'rb') as f: + seq_obj.content = f.read() + seq_obj.status_code = requests.codes.ok + + # A zone state response object + zst_obj = mock.Mock() + + # Simulate initial zone fetch configuration + with open(join(ULTRASYNC_TEST_VAR_DIR, 'zstate.xml'), 'rb') as f: + zst_obj.content = f.read() + zst_obj.status_code = requests.codes.ok + + # An area state response object + ast_obj = mock.Mock() + + # Simulate initial area status fetch configuration + with open(join(ULTRASYNC_TEST_VAR_DIR, 'status.xml'), 'rb') as f: + ast_obj.content = f.read() + ast_obj.status_code = requests.codes.ok + + # Assign our response object to our mocked instance of requests + mock_post.side_effect = (arobj, zrobj) + + uobj = UltraSync() + + # Perform a login which under the hood queries both area.htm and zones.htm + # (in that order) + assert uobj.login() + assert uobj.vendor is NX595EVendor.COMNAV + assert uobj.version == '0.108' + assert uobj.release == 'O' + + assert isinstance(uobj.areas, dict) + # 2 Areas defined + assert len(uobj.areas) == 2 + assert uobj.areas[0]['name'] == 'AREA 1' + assert uobj.areas[0]['bank'] == 0 + assert uobj.areas[0]['sequence'] == 1 + assert uobj.areas[0]['status'] == 'Ready' + + assert uobj.areas[1]['name'] == 'AREA 2' + assert uobj.areas[1]['bank'] == 1 + assert uobj.areas[1]['sequence'] == 1 + assert uobj.areas[1]['status'] == 'Ready' + + assert isinstance(uobj.zones, dict) + # our total zones defined (we want to be sure we don't + # include the '%2D' or '-') + assert len(uobj.zones) == 17 + + # A call to login.cgi (which fetches area.html) and then zones.htm + assert mock_post.call_count == 2 + assert mock_post.call_args_list[0][0][0] == \ + 'http://zerowire/login.cgi' + assert mock_post.call_args_list[1][0][0] == \ + 'http://zerowire/user/zones.htm' + + # Reset our mock object + mock_post.reset_mock() + + # Update our side effects + mock_post.side_effect = (seq_obj, ast_obj, zst_obj) + + # Perform Updated Query + uobj.update(max_age_sec=0) + + # Only 1 query made because seq.xml file unchanged + assert mock_post.call_count == 1 + assert mock_post.call_args_list[0][0][0] == \ + 'http://zerowire/user/seq.xml' + assert uobj.areas[0]['sequence'] == 1 + + # Update our sequence file so that it reflects a change + with open(join(ULTRASYNC_TEST_VAR_DIR, 'seq.w.update.xml'), 'rb') as f: + seq_obj.content = f.read() + + # Reset our mock object + mock_post.reset_mock() + + # Update our side effects + mock_post.side_effect = (seq_obj, ast_obj, zst_obj) + + # Perform Updated Query + uobj.update(max_age_sec=0) + + assert mock_post.call_count == 3 + assert mock_post.call_args_list[0][0][0] == \ + 'http://zerowire/user/seq.xml' + assert mock_post.call_args_list[1][0][0] == \ + 'http://zerowire/user/zstate.xml' + assert mock_post.call_args_list[2][0][0] == \ + 'http://zerowire/user/status.xml' + + assert isinstance(uobj.areas, dict) + assert len(uobj.areas) == 2 + assert uobj.areas[0]['name'] == 'AREA 1' + assert uobj.areas[0]['bank'] == 0 + # Our sequence does not change because our status did not + # change from one state to another + assert uobj.areas[0]['sequence'] == 1 + assert uobj.areas[0]['status'] == 'Ready' + + assert uobj.areas[1]['name'] == 'AREA 2' + assert uobj.areas[1]['bank'] == 1 + # Our sequence does not change because our status did not + # change from one state to another + assert uobj.areas[1]['sequence'] == 1 + assert uobj.areas[1]['status'] == 'Ready' + + # Reset our mock object + mock_post.reset_mock() + + # Update our side effects + mock_post.side_effect = seq_obj + + uobj.details(max_age_sec=0) + + # Only 1 query made because seq.xml file unchanged + assert mock_post.call_count == 1 + assert mock_post.call_args_list[0][0][0] == \ + 'http://zerowire/user/seq.xml' + assert uobj.areas[0]['sequence'] == 1 diff --git a/tests/test_comnav_0_108.py b/tests/test_comnav_0_108.py index 293b47e..fa57db6 100644 --- a/tests/test_comnav_0_108.py +++ b/tests/test_comnav_0_108.py @@ -104,7 +104,7 @@ def test_comnav_0_108_communication(mock_post): assert len(uobj.areas) == 1 assert uobj.areas[0]['name'] == 'Home' assert uobj.areas[0]['bank'] == 0 - assert uobj.areas[0]['sequence'] == 178 + assert uobj.areas[0]['sequence'] == 1 assert uobj.areas[0]['status'] == 'Ready' assert isinstance(uobj.zones, dict) @@ -157,7 +157,7 @@ def test_comnav_0_108_communication(mock_post): assert mock_post.call_count == 1 assert mock_post.call_args_list[0][0][0] == \ 'http://zerowire/user/seq.xml' - assert uobj.areas[0]['sequence'] == 178 + assert uobj.areas[0]['sequence'] == 1 # Update our sequence file so that it reflects a change with open(join(ULTRASYNC_TEST_VAR_DIR, 'seq.w.update.xml'), 'rb') as f: @@ -185,7 +185,7 @@ def test_comnav_0_108_communication(mock_post): assert uobj.areas[0]['name'] == 'Home' assert uobj.areas[0]['bank'] == 0 # Our sequence got bumped - assert uobj.areas[0]['sequence'] == 179 + assert uobj.areas[0]['sequence'] == 1 assert uobj.areas[0]['status'] == 'Ready' # Reset our mock object @@ -200,4 +200,4 @@ def test_comnav_0_108_communication(mock_post): assert mock_post.call_count == 1 assert mock_post.call_args_list[0][0][0] == \ 'http://zerowire/user/seq.xml' - assert uobj.areas[0]['sequence'] == 179 + assert uobj.areas[0]['sequence'] == 1 diff --git a/tests/test_comnav_0_108_burglar_alarm_on.py b/tests/test_comnav_0_108_burglar_alarm_on.py index bc30e9c..8ecdb43 100644 --- a/tests/test_comnav_0_108_burglar_alarm_on.py +++ b/tests/test_comnav_0_108_burglar_alarm_on.py @@ -104,7 +104,7 @@ def test_comnav_0_108_active_burglary(mock_post): assert len(uobj.areas) == 1 assert uobj.areas[0]['name'] == 'Home' assert uobj.areas[0]['bank'] == 0 - assert uobj.areas[0]['sequence'] == 208 + assert uobj.areas[0]['sequence'] == 1 assert uobj.areas[0]['status'] == 'Burglar Alarm' assert isinstance(uobj.zones, dict) diff --git a/tests/test_xgen_general.py b/tests/test_xgen_general.py index 94d7750..4f10ef9 100644 --- a/tests/test_xgen_general.py +++ b/tests/test_xgen_general.py @@ -120,12 +120,12 @@ def test_xgen_general_communication(mock_post): assert len(uobj.areas) == 2 assert uobj.areas[0]['name'] == 'Area 1' assert uobj.areas[0]['bank'] == 0 - assert uobj.areas[0]['sequence'] == 126 + assert uobj.areas[0]['sequence'] == 1 assert uobj.areas[0]['status'] == 'Ready' assert uobj.areas[1]['name'] == 'Area 2' assert uobj.areas[1]['bank'] == 1 - assert uobj.areas[1]['sequence'] == 0 + assert uobj.areas[1]['sequence'] == 1 assert uobj.areas[1]['status'] == 'Ready' assert isinstance(uobj.zones, dict) @@ -205,7 +205,7 @@ def test_xgen_general_communication(mock_post): assert uobj.areas[1]['name'] == 'Area 2' assert uobj.areas[1]['bank'] == 1 # Our sequence got bumped - assert uobj.areas[0]['sequence'] == 127 + assert uobj.areas[0]['sequence'] == 1 assert uobj.areas[0]['status'] == 'Ready' # Reset our mock object @@ -220,7 +220,7 @@ def test_xgen_general_communication(mock_post): assert mock_post.call_count == 1 assert mock_post.call_args_list[0][0][0] == \ 'http://zerowire/user/seq.json' - assert uobj.areas[0]['sequence'] == 127 + assert uobj.areas[0]['sequence'] == 1 # Reset our mock object mock_post.reset_mock() diff --git a/tests/test_zerowire_armed.py b/tests/test_zerowire_armed.py index 506d97b..cd25e37 100644 --- a/tests/test_zerowire_armed.py +++ b/tests/test_zerowire_armed.py @@ -80,7 +80,7 @@ def test_zerowire_armed_communication(mock_post): assert len(uobj.areas) == 1 assert uobj.areas[0]['name'] == 'Area 1' assert uobj.areas[0]['bank'] == 0 - assert uobj.areas[0]['sequence'] == 203 + assert uobj.areas[0]['sequence'] == 1 assert uobj.areas[0]['status'] == 'Exit Delay 1' assert isinstance(uobj.zones, dict) diff --git a/tests/test_zerowire_general.py b/tests/test_zerowire_general.py index b64086f..532a85b 100644 --- a/tests/test_zerowire_general.py +++ b/tests/test_zerowire_general.py @@ -104,7 +104,7 @@ def test_zerowire_general_communication(mock_post): assert len(uobj.areas) == 1 assert uobj.areas[0]['name'] == 'Area 1' assert uobj.areas[0]['bank'] == 0 - assert uobj.areas[0]['sequence'] == 149 + assert uobj.areas[0]['sequence'] == 1 assert uobj.areas[0]['status'] == 'Ready' assert isinstance(uobj.zones, dict) @@ -176,7 +176,7 @@ def test_zerowire_general_communication(mock_post): assert uobj.areas[0]['name'] == 'Area 1' assert uobj.areas[0]['bank'] == 0 # Our sequence got bumped - assert uobj.areas[0]['sequence'] == 150 + assert uobj.areas[0]['sequence'] == 1 assert uobj.areas[0]['status'] == 'Ready' # Reset our mock object @@ -191,4 +191,4 @@ def test_zerowire_general_communication(mock_post): assert mock_post.call_count == 1 assert mock_post.call_args_list[0][0][0] == \ 'http://zerowire/user/seq.json' - assert uobj.areas[0]['sequence'] == 150 + assert uobj.areas[0]['sequence'] == 1 diff --git a/tests/var/comnav/0.108-zone-test/area.htm b/tests/var/comnav/0.108-zone-test/area.htm new file mode 100644 index 0000000..cb2fb2e --- /dev/null +++ b/tests/var/comnav/0.108-zone-test/area.htm @@ -0,0 +1,71 @@ + + + + + NX-595E :: Secure Network + + + + + + + + + + +
+ + + +
+ + + + diff --git a/tests/var/comnav/0.108-zone-test/seq.w.update.xml b/tests/var/comnav/0.108-zone-test/seq.w.update.xml new file mode 100644 index 0000000..093144d --- /dev/null +++ b/tests/var/comnav/0.108-zone-test/seq.w.update.xml @@ -0,0 +1,4 @@ + + 71 + 202,0,0,225,2,16,0,0,1,0,0,8,0,0 + diff --git a/tests/var/comnav/0.108-zone-test/seq.xml b/tests/var/comnav/0.108-zone-test/seq.xml new file mode 100644 index 0000000..bd65ce4 --- /dev/null +++ b/tests/var/comnav/0.108-zone-test/seq.xml @@ -0,0 +1,4 @@ + + 70 + 201,0,0,225,2,16,0,0,1,0,0,8,0,0 + diff --git a/tests/var/comnav/0.108-zone-test/status.xml b/tests/var/comnav/0.108-zone-test/status.xml new file mode 100644 index 0000000..eaff1a3 --- /dev/null +++ b/tests/var/comnav/0.108-zone-test/status.xml @@ -0,0 +1,23 @@ + + 0 + 70 + 0 + 0 + 3 + 0 + 0 + 0 + 0 + 0 + 0 + 0 + 0 + 0 + 0 + 0 + 0 + 2 + 0 + No System Faults + + diff --git a/tests/var/comnav/0.108-zone-test/zones.htm b/tests/var/comnav/0.108-zone-test/zones.htm new file mode 100644 index 0000000..7bab4be --- /dev/null +++ b/tests/var/comnav/0.108-zone-test/zones.htm @@ -0,0 +1,57 @@ + + + + + NX-595E :: Secure Network + + + + + + + + + + +
+ + + +
+ + + + diff --git a/tests/var/comnav/0.108-zone-test/zstate.xml b/tests/var/comnav/0.108-zone-test/zstate.xml new file mode 100644 index 0000000..7533130 --- /dev/null +++ b/tests/var/comnav/0.108-zone-test/zstate.xml @@ -0,0 +1,5 @@ + + 0 + 201 + 0,0,0,0 + diff --git a/ultrasync/main.py b/ultrasync/main.py index 97bea8e..fcce2ab 100644 --- a/ultrasync/main.py +++ b/ultrasync/main.py @@ -116,6 +116,7 @@ def __init__(self, *args, **kwargs): # Our areas get populated after we connect self.areas = {} + self._asequence = None # Track the time our information was polled from our panel self.__updated = None @@ -640,7 +641,7 @@ def _areas(self, response=None): if not match: # No match and/or bad login return False - sequence = json.loads(match.group('sequence')) + self._asequence = json.loads(match.group('sequence')) else: # self.vendor is NX595EVendor.COMNAV # It looks like this in the area.htm response @@ -651,7 +652,8 @@ def _areas(self, response=None): if not match: # No match and/or bad login return False - sequence = json.loads('[{}]'.format(match.group('sequence'))) + self._asequence = json.loads( + '[{}]'.format(match.group('sequence'))) # # Get our Area Status (Bank States) @@ -705,7 +707,8 @@ def _areas(self, response=None): area_names = json.loads('[{}]'.format(match.group('area_names'))) # Ensure we've defined all of our possible sequences and areas - sequence.extend([0] * (UltraSync.max_sequence_count - len(sequence))) + self._asequence.extend( + [0] * (UltraSync.max_sequence_count - len(self._asequence))) area_names.extend(['!'] * (UltraSync.max_area_count - len(area_names))) # Store our Areas ('%21' == '!'; these are un-used areas) @@ -713,10 +716,9 @@ def _areas(self, response=None): {x: {'name': unquote(y).strip() if unquote(y).strip() else 'Area {}'.format(x + 1), 'bank': x, - 'sequence': sequence[x], 'bank_state': bank_states[math.floor(x / 8) * 17: (math.floor(x / 8) * 17) + 17] - if self.vendor == NX595EVendor.COMNAV + if self.vendor is NX595EVendor.COMNAV else bank_states[0]} for x, y in enumerate(area_names) @@ -861,6 +863,12 @@ def xgen_process_areas(self): # Assign priority to 5 priority = 5 + # Update our sequence + sequence = UltraSync.next_sequence( + area.get('sequence', 0)) \ + if area.get('status') != status \ + else area.get('sequence', 0) + area.update({ 'priority': priority, 'states': { @@ -871,6 +879,7 @@ def xgen_process_areas(self): 'exit2': st_exit2, }, 'status': status, + 'sequence': sequence, }) return True @@ -1005,6 +1014,12 @@ def zerowire_process_areas(self): # Assign priority to 5 priority = 5 + # Update our sequence + sequence = UltraSync.next_sequence( + area.get('sequence', 0)) \ + if area.get('status') != status \ + else area.get('sequence', 0) + area.update({ 'priority': priority, 'states': { @@ -1015,6 +1030,7 @@ def zerowire_process_areas(self): 'exit2': st_exit2, }, 'status': status, + 'sequence': sequence, }) return True @@ -1124,6 +1140,12 @@ def comnav_process_areas(self): # Assign priority to 5 priority = 5 + # Update our sequence + sequence = UltraSync.next_sequence( + area.get('sequence', 0)) \ + if area.get('status') != status \ + else area.get('sequence', 0) + area.update({ 'priority': priority, 'states': { @@ -1134,6 +1156,7 @@ def comnav_process_areas(self): 'exit2': st_exit2, }, 'status': status, + 'sequence': sequence, }) return True @@ -1475,18 +1498,21 @@ def _zones(self): # this version zone_naming = True if float(self.version) > 0.106 else False - # Store our Zones ('%21' == '!'; these are un-used sensors) + # Store our Zones: + # The following are unused sensors: + # - '%21' == '!' + # - '%2D' == '-' self.zones = \ {x: {'name': unquote(y).strip() if unquote(y).strip() else 'Sensor {}'.format(x + 1), 'bank': x} for x, y in enumerate(zone_names) - if not zone_naming or (y != '%21' and y != '!' and y != '')} + if not zone_naming or (y not in ('%21', '!', '%2D', '-', ''))} # # Get our Master Status # - if self.vendor == NX595EVendor.ZEROWIRE: + if self.vendor is NX595EVendor.ZEROWIRE: # It looks like this in the zone.htm response: # var master = 1; # 1 if master, 0 if not match = re.search( @@ -1874,18 +1900,18 @@ def _zerowire_sequence(self): # Process Zones/Sensors first for bank, sequence in enumerate(self._zsequence): if sequence != response['zone'][bank]: - logger.debug('Zone {} sequence changed'.format(bank + 1)) + logger.debug('Zone Bank {}:{} changed'.format(bank, sequence)) # We need to update our status here self._zsequence[bank] = response['zone'][bank] self._zone_status_update(bank=bank) perform_zone_update = True # Process Area now - for bank, area in self.areas.items(): - if area['sequence'] != response['area'][bank]: - logger.debug('Area {} sequence changed'.format(bank + 1)) + for bank, sequence in enumerate(self._asequence): + if sequence != response['area'][bank]: + logger.debug('Area Bank {}:{} changed'.format(bank, sequence)) # We need to update our status here - area['sequence'] = response['area'][bank] + self._asequence[bank] = response['area'][bank] self._area_status_update(bank=bank) perform_area_update = True @@ -1948,22 +1974,22 @@ def _xgen_sequence(self): # Process Zones/Sensors first for bank, sequence in enumerate(self._zsequence): if sequence != response['zone'][bank]: - logger.debug('Zone {} sequence changed'.format(bank + 1)) + logger.debug('Zone Bank {}:{} changed'.format(bank, sequence)) # We need to update our status here self._zsequence[bank] = response['zone'][bank] self._zone_status_update(bank=bank) perform_zone_update = True # Process Area now - for bank, area in self.areas.items(): + for bank, sequence in enumerate(self._asequence): if bank >= len(response['area']): # We're done break - if area['sequence'] != response['area'][bank]: - logger.debug('Area {} sequence changed'.format(bank + 1)) + if sequence != response['area'][bank]: + logger.debug('Area Bank {}:{} changed'.format(bank, sequence)) # We need to update our status here - area['sequence'] = response['area'][bank] + self._asequence[bank] = response['area'][bank] self._area_status_update(bank=bank) perform_area_update = True @@ -2035,18 +2061,18 @@ def _comnav_sequence(self): # Process Zones/Sensors first for bank, sequence in enumerate(z_seq): if sequence != self._zsequence[bank]: - logger.debug('Zone {} sequence changed'.format(bank + 1)) + logger.debug('Zone Bank {}:{} changed'.format(bank, sequence)) # We need to update our sequence here self._zsequence[bank] = z_seq[bank] self._zone_status_update(bank=bank) perform_zone_update = True # Process Area now - for bank, area in self.areas.items(): - if area['sequence'] != a_seq[bank]: - logger.debug('Area {} sequence changed'.format(bank + 1)) + for bank, sequence in enumerate(a_seq): + if sequence != self._asequence[bank]: + logger.debug('Area Bank {}:{} changed'.format(bank, sequence)) # We need to update our status here - area['sequence'] = a_seq[bank] + self._asequence[bank] = a_seq[bank] self._area_status_update(bank=bank) perform_area_update = True