From bc647b4f2b3f55fadd0c730a3fe6040db0439903 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 8 Dec 2025 23:03:14 +0000 Subject: [PATCH 1/3] Add get_nvr_events and inject_events_into_camera functions These functions extend the event detection capabilities to support NVRs with non-standard notification methods (record, email, beep) in addition to the existing 'center' and 'HTTP' methods. - get_nvr_events: Fetches events from NVR with broader notification support - inject_events_into_camera: Injects discovered events into HikCamera - HikCamera.inject_events: New method to inject events into camera's state Based on home-assistant/core#158279 --- pyhik/__init__.py | 6 ++ pyhik/constants.py | 4 ++ pyhik/hikvision.py | 148 ++++++++++++++++++++++++++++++++++++++++++++- 3 files changed, 157 insertions(+), 1 deletion(-) diff --git a/pyhik/__init__.py b/pyhik/__init__.py index e69de29..ca8bee6 100644 --- a/pyhik/__init__.py +++ b/pyhik/__init__.py @@ -0,0 +1,6 @@ +"""pyhik - Python library for Hikvision camera/NVR events.""" + +from pyhik.hikvision import HikCamera, get_nvr_events, inject_events_into_camera +from pyhik.constants import __version__ + +__all__ = ['HikCamera', 'get_nvr_events', 'inject_events_into_camera', '__version__'] diff --git a/pyhik/constants.py b/pyhik/constants.py index fa00c90..dc3fc71 100755 --- a/pyhik/constants.py +++ b/pyhik/constants.py @@ -67,3 +67,7 @@ CONTEXT_TRIG = 'TRIGGERS' CONTEXT_ALERT = 'ALERTS' CONTEXT_MOTION = 'MOTION' + +# Notification methods that indicate the event is active/configured +# Expanded from original list of just 'center' and 'HTTP' to support NVRs +VALID_NOTIFICATION_METHODS = {'center', 'HTTP', 'record', 'email', 'beep'} diff --git a/pyhik/hikvision.py b/pyhik/hikvision.py index ed27b0e..364cd1c 100755 --- a/pyhik/hikvision.py +++ b/pyhik/hikvision.py @@ -37,7 +37,7 @@ DEFAULT_PORT, DEFAULT_HEADERS, XML_NAMESPACE, SENSOR_MAP, CAM_DEVICE, NVR_DEVICE, CONNECT_TIMEOUT, READ_TIMEOUT, CONTEXT_INFO, CONTEXT_TRIG, CONTEXT_MOTION, CONTEXT_ALERT, CHANNEL_NAMES, ID_TYPES, - __version__) + VALID_NOTIFICATION_METHODS, __version__) _LOGGING = logging.getLogger(__name__) @@ -677,3 +677,149 @@ def update_attributes(self, event, channel, attr): except KeyError: _LOGGING.debug('Error updating attributes for: (%s, %s)', event, channel) + + def inject_events(self, events): + """Inject discovered events into the camera's event_states. + + This allows the camera to track events that wouldn't normally be + detected, such as those from NVRs with non-standard notification + methods. + + Args: + events: Dict mapping event type names to lists of channel numbers. + """ + for event_name, channels in events.items(): + for channel in channels: + # Only add if not already present + if event_name not in self.event_states: + self.event_states[event_name] = [] + + # Check if this channel is already tracked + channel_exists = any( + sensor[1] == channel for sensor in self.event_states[event_name] + ) + if not channel_exists: + # Add the event state: [is_active, channel, count, last_update_time] + self.event_states[event_name].append( + [False, channel, 0, datetime.datetime.now()] + ) + + +def get_nvr_events(host, port=DEFAULT_PORT, usr=None, pwd=None, verify_ssl=True): + """Fetch events from NVR with broader notification method support. + + This function extends the standard event detection by also accepting + 'record', 'email', and 'beep' notification methods, which are commonly + used on NVRs but ignored by the standard get_event_triggers method. + + Args: + host: The host URL (e.g., 'http://192.168.1.64'). + port: The port number (default 80). + usr: Username for authentication. + pwd: Password for authentication. + verify_ssl: Whether to verify SSL certificates. + + Returns: + Dict mapping event type names to lists of channel numbers. + """ + root_url = '{}:{}'.format(host, port) + events = {} + + session = requests.Session() + session.verify = verify_ssl + session.auth = HTTPDigestAuth(usr, pwd) + session.headers.update(DEFAULT_HEADERS) + + urls = [ + '%s/ISAPI/Event/triggers' % root_url, + '%s/Event/triggers' % root_url, + ] + + response = None + for url in urls: + try: + response = session.get(url, timeout=CONNECT_TIMEOUT) + if response.status_code == requests.codes.ok: + break + except (requests.exceptions.RequestException, + requests.exceptions.ConnectionError): + continue + + if response is None or response.status_code != requests.codes.ok: + _LOGGING.warning('Unable to fetch event triggers from NVR') + session.close() + return events + + try: + tree = ET.fromstring(response.text) + except ET.ParseError as err: + _LOGGING.error('Failed to parse event triggers XML: %s', err) + session.close() + return events + + # Find namespace from root tag + namespace = '' + root_tag = tree.tag + if root_tag.startswith('{'): + namespace = root_tag.split('}')[0] + '}' + + # Try different XML structures (camera vs NVR) + event_triggers = tree.findall('.//{0}EventTrigger'.format(namespace)) + + for trigger in event_triggers: + # Get event type + event_type_elem = trigger.find('{0}eventType'.format(namespace)) + if event_type_elem is None or not event_type_elem.text: + continue + + event_type = event_type_elem.text.lower() + + # Skip videoloss as it's used for watchdog + if event_type == 'videoloss': + continue + + # Get channel number + channel_num = 0 + for channel_name in CHANNEL_NAMES: + channel_elem = trigger.find('{0}{1}'.format(namespace, channel_name)) + if channel_elem is not None and channel_elem.text: + try: + channel_num = int(channel_elem.text) + break + except ValueError: + continue + + # Check if any valid notification method is configured + notification_list = trigger.find( + '{0}EventTriggerNotificationList'.format(namespace)) + has_valid_notification = False + + if notification_list is not None: + for notification in notification_list: + method_elem = notification.find( + '{0}notificationMethod'.format(namespace)) + if method_elem is not None and method_elem.text: + if method_elem.text.lower() in {m.lower() for m in VALID_NOTIFICATION_METHODS}: + has_valid_notification = True + break + + if has_valid_notification: + # Map to friendly name + friendly_name = SENSOR_MAP.get(event_type) + if friendly_name: + events.setdefault(friendly_name, []).append(channel_num) + + session.close() + return events + + +def inject_events_into_camera(camera, events): + """Inject discovered events into the pyhik camera's event_states. + + This allows the camera to track events that wouldn't normally be detected. + + Args: + camera: A HikCamera instance. + events: Dict mapping event type names to lists of channel numbers. + """ + camera.inject_events(events) From 9c698da8399f56cdc8b98a68b1a612143579ccc4 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 8 Dec 2025 23:05:44 +0000 Subject: [PATCH 2/3] Add tests for get_nvr_events and inject_events_into_camera Tests cover: - Parsing NVR events with various notification methods (record, email, beep, center) - Skipping videoloss events (used for watchdog) - Handling connection errors gracefully - Handling non-200 responses - Handling invalid XML - Injecting events into camera event_states - Preventing duplicate channel events --- test/test_hikvision.py | 215 ++++++++++++++++++++++++++++++++++++++++- 1 file changed, 214 insertions(+), 1 deletion(-) diff --git a/test/test_hikvision.py b/test/test_hikvision.py index 21d4687..21ab249 100644 --- a/test/test_hikvision.py +++ b/test/test_hikvision.py @@ -5,7 +5,7 @@ import unittest from unittest.mock import MagicMock, patch, PropertyMock -from pyhik.hikvision import HikCamera +from pyhik.hikvision import HikCamera, get_nvr_events, inject_events_into_camera from pyhik.constants import (CONNECT_TIMEOUT) XML = """ @@ -82,5 +82,218 @@ def change_get_response(url, data,timeout): self.assertFalse(device.current_motion_detection_state) +NVR_EVENTS_XML = """ + + + 1 + VMD + 1 + + + 1 + record + + + + + 2 + linedetection + 2 + + + 1 + email + + + + + 3 + fielddetection + 3 + + + 1 + beep + + + + + 4 + VMD + 4 + + + 1 + center + + + + + 5 + videoloss + 1 + + + 1 + center + + + + + 6 + facedetection + 1 + + + 1 + unknown + + + +""" + + +class GetNvrEventsTestCase(unittest.TestCase): + @patch("pyhik.hikvision.requests.Session") + def test_get_nvr_events_parses_events(self, mock_session_class): + """Test that get_nvr_events correctly parses events with various notification methods.""" + session = mock_session_class.return_value + response = MagicMock() + response.status_code = requests.codes.ok + response.text = NVR_EVENTS_XML + session.get.return_value = response + + events = get_nvr_events("http://localhost", usr="admin", pwd="password") + + # Should find Motion events on channels 1 and 4 (VMD with record and center) + self.assertIn("Motion", events) + self.assertEqual(sorted(events["Motion"]), [1, 4]) + + # Should find Line Crossing on channel 2 (email notification) + self.assertIn("Line Crossing", events) + self.assertEqual(events["Line Crossing"], [2]) + + # Should find Field Detection on channel 3 (beep notification) + self.assertIn("Field Detection", events) + self.assertEqual(events["Field Detection"], [3]) + + # Should NOT include videoloss (skipped) + self.assertNotIn("Video Loss", events) + + # Should NOT include facedetection (unknown notification method) + self.assertNotIn("Face Detection", events) + + session.close.assert_called_once() + + @patch("pyhik.hikvision.requests.Session") + def test_get_nvr_events_handles_connection_error(self, mock_session_class): + """Test that get_nvr_events handles connection errors gracefully.""" + session = mock_session_class.return_value + session.get.side_effect = requests.exceptions.ConnectionError("Connection refused") + + events = get_nvr_events("http://localhost", usr="admin", pwd="password") + + self.assertEqual(events, {}) + session.close.assert_called_once() + + @patch("pyhik.hikvision.requests.Session") + def test_get_nvr_events_handles_bad_response(self, mock_session_class): + """Test that get_nvr_events handles non-200 responses.""" + session = mock_session_class.return_value + response = MagicMock() + response.status_code = requests.codes.unauthorized + session.get.return_value = response + + events = get_nvr_events("http://localhost", usr="admin", pwd="password") + + self.assertEqual(events, {}) + session.close.assert_called_once() + + @patch("pyhik.hikvision.requests.Session") + def test_get_nvr_events_handles_invalid_xml(self, mock_session_class): + """Test that get_nvr_events handles invalid XML gracefully.""" + session = mock_session_class.return_value + response = MagicMock() + response.status_code = requests.codes.ok + response.text = "not valid xml" + session.get.return_value = response + + events = get_nvr_events("http://localhost", usr="admin", pwd="password") + + self.assertEqual(events, {}) + session.close.assert_called_once() + + +class InjectEventsTestCase(unittest.TestCase): + def test_inject_events_adds_new_events(self): + """Test that inject_events adds new events to camera event_states.""" + camera = MagicMock() + camera.event_states = {} + + events = { + "Motion": [1, 2], + "Line Crossing": [3] + } + + inject_events_into_camera(camera, events) + + camera.inject_events.assert_called_once_with(events) + + @patch("pyhik.hikvision.requests.Session") + @patch("pyhik.hikvision.HikCamera.get_device_info") + @patch("pyhik.hikvision.HikCamera.get_event_triggers") + def test_inject_events_method(self, mock_triggers, mock_info, mock_session): + """Test that HikCamera.inject_events correctly adds events.""" + mock_info.return_value = {"deviceName": "Test", "deviceID": "12345678901"} + mock_triggers.return_value = {} + session = mock_session.return_value + session.get.return_value = MagicMock(status_code=requests.codes.not_found) + + camera = HikCamera(host="localhost") + camera.event_states = {} + + # Inject events + events = { + "Motion": [1, 2], + "Line Crossing": [3] + } + camera.inject_events(events) + + # Verify events were added + self.assertIn("Motion", camera.event_states) + self.assertEqual(len(camera.event_states["Motion"]), 2) + self.assertEqual(camera.event_states["Motion"][0][1], 1) # channel 1 + self.assertEqual(camera.event_states["Motion"][1][1], 2) # channel 2 + self.assertFalse(camera.event_states["Motion"][0][0]) # not active + + self.assertIn("Line Crossing", camera.event_states) + self.assertEqual(len(camera.event_states["Line Crossing"]), 1) + self.assertEqual(camera.event_states["Line Crossing"][0][1], 3) # channel 3 + + @patch("pyhik.hikvision.requests.Session") + @patch("pyhik.hikvision.HikCamera.get_device_info") + @patch("pyhik.hikvision.HikCamera.get_event_triggers") + def test_inject_events_does_not_duplicate(self, mock_triggers, mock_info, mock_session): + """Test that inject_events doesn't add duplicate channel events.""" + mock_info.return_value = {"deviceName": "Test", "deviceID": "12345678901"} + mock_triggers.return_value = {} + session = mock_session.return_value + session.get.return_value = MagicMock(status_code=requests.codes.not_found) + + camera = HikCamera(host="localhost") + camera.event_states = { + "Motion": [[False, 1, 0, None]] # Already has channel 1 + } + + # Try to inject event for same channel + events = {"Motion": [1, 2]} + camera.inject_events(events) + + # Should only have 2 entries (original + channel 2, not duplicate of 1) + self.assertEqual(len(camera.event_states["Motion"]), 2) + channels = [sensor[1] for sensor in camera.event_states["Motion"]] + self.assertEqual(sorted(channels), [1, 2]) + + if __name__ == "__main__": unittest.main() From dc52d46d885b837bf1c7be475ab755bef21c9325 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 17 Dec 2025 16:17:31 +0000 Subject: [PATCH 3/3] Refactor: Add notification_methods parameter to get_event_triggers Instead of a separate get_nvr_events function, extend the existing get_event_triggers method to accept a notification_methods parameter. This allows users to specify which notification methods to accept (defaults to {'center', 'HTTP'} for backwards compatibility). For NVRs, users can pass VALID_NOTIFICATION_METHODS which includes 'record', 'email', and 'beep' in addition to the defaults. Changes: - Add notification_methods parameter to get_event_triggers() - Remove duplicate get_nvr_events function - Export VALID_NOTIFICATION_METHODS constant for easy use - Update tests to cover new parameter functionality --- pyhik/__init__.py | 11 +++- pyhik/hikvision.py | 130 +++++--------------------------------- test/test_hikvision.py | 139 +++++++++++++++++++++++------------------ 3 files changed, 101 insertions(+), 179 deletions(-) diff --git a/pyhik/__init__.py b/pyhik/__init__.py index ca8bee6..bac8386 100644 --- a/pyhik/__init__.py +++ b/pyhik/__init__.py @@ -1,6 +1,11 @@ """pyhik - Python library for Hikvision camera/NVR events.""" -from pyhik.hikvision import HikCamera, get_nvr_events, inject_events_into_camera -from pyhik.constants import __version__ +from pyhik.hikvision import HikCamera, inject_events_into_camera +from pyhik.constants import __version__, VALID_NOTIFICATION_METHODS -__all__ = ['HikCamera', 'get_nvr_events', 'inject_events_into_camera', '__version__'] +__all__ = [ + 'HikCamera', + 'inject_events_into_camera', + 'VALID_NOTIFICATION_METHODS', + '__version__' +] diff --git a/pyhik/hikvision.py b/pyhik/hikvision.py index 364cd1c..45024ad 100755 --- a/pyhik/hikvision.py +++ b/pyhik/hikvision.py @@ -37,7 +37,7 @@ DEFAULT_PORT, DEFAULT_HEADERS, XML_NAMESPACE, SENSOR_MAP, CAM_DEVICE, NVR_DEVICE, CONNECT_TIMEOUT, READ_TIMEOUT, CONTEXT_INFO, CONTEXT_TRIG, CONTEXT_MOTION, CONTEXT_ALERT, CHANNEL_NAMES, ID_TYPES, - VALID_NOTIFICATION_METHODS, __version__) + __version__) _LOGGING = logging.getLogger(__name__) @@ -367,12 +367,22 @@ def get_device_info(self): _LOGGING.error('There was a problem: %s', err) return None - def get_event_triggers(self): + def get_event_triggers(self, notification_methods=None): """ Returns dict of supported events. Key = Event Type List = Channels that have that event activated + + Args: + notification_methods: Set of notification method strings to accept. + Defaults to {'center', 'HTTP'}. For NVRs, you may want to include + additional methods like 'record', 'email', 'beep'. """ + if notification_methods is None: + notification_methods = {'center', 'HTTP'} + # Normalize to lowercase for comparison + notification_methods_lower = {m.lower() for m in notification_methods} + events = {} nvrflag = False event_xml = [] @@ -445,11 +455,9 @@ def get_event_triggers(self): for notifytrigger in etnotify: ntype = notifytrigger.find( self.element_query('notificationMethod', CONTEXT_TRIG)) - if ntype.text == 'center' or ntype.text == 'HTTP': - """ - If we got this far we found an event that we want - to track. - """ + if ntype is not None and ntype.text and \ + ntype.text.lower() in notification_methods_lower: + # Found an event with a valid notification method events.setdefault(ettype.text, []) \ .append(etchannel_num) @@ -705,114 +713,6 @@ def inject_events(self, events): ) -def get_nvr_events(host, port=DEFAULT_PORT, usr=None, pwd=None, verify_ssl=True): - """Fetch events from NVR with broader notification method support. - - This function extends the standard event detection by also accepting - 'record', 'email', and 'beep' notification methods, which are commonly - used on NVRs but ignored by the standard get_event_triggers method. - - Args: - host: The host URL (e.g., 'http://192.168.1.64'). - port: The port number (default 80). - usr: Username for authentication. - pwd: Password for authentication. - verify_ssl: Whether to verify SSL certificates. - - Returns: - Dict mapping event type names to lists of channel numbers. - """ - root_url = '{}:{}'.format(host, port) - events = {} - - session = requests.Session() - session.verify = verify_ssl - session.auth = HTTPDigestAuth(usr, pwd) - session.headers.update(DEFAULT_HEADERS) - - urls = [ - '%s/ISAPI/Event/triggers' % root_url, - '%s/Event/triggers' % root_url, - ] - - response = None - for url in urls: - try: - response = session.get(url, timeout=CONNECT_TIMEOUT) - if response.status_code == requests.codes.ok: - break - except (requests.exceptions.RequestException, - requests.exceptions.ConnectionError): - continue - - if response is None or response.status_code != requests.codes.ok: - _LOGGING.warning('Unable to fetch event triggers from NVR') - session.close() - return events - - try: - tree = ET.fromstring(response.text) - except ET.ParseError as err: - _LOGGING.error('Failed to parse event triggers XML: %s', err) - session.close() - return events - - # Find namespace from root tag - namespace = '' - root_tag = tree.tag - if root_tag.startswith('{'): - namespace = root_tag.split('}')[0] + '}' - - # Try different XML structures (camera vs NVR) - event_triggers = tree.findall('.//{0}EventTrigger'.format(namespace)) - - for trigger in event_triggers: - # Get event type - event_type_elem = trigger.find('{0}eventType'.format(namespace)) - if event_type_elem is None or not event_type_elem.text: - continue - - event_type = event_type_elem.text.lower() - - # Skip videoloss as it's used for watchdog - if event_type == 'videoloss': - continue - - # Get channel number - channel_num = 0 - for channel_name in CHANNEL_NAMES: - channel_elem = trigger.find('{0}{1}'.format(namespace, channel_name)) - if channel_elem is not None and channel_elem.text: - try: - channel_num = int(channel_elem.text) - break - except ValueError: - continue - - # Check if any valid notification method is configured - notification_list = trigger.find( - '{0}EventTriggerNotificationList'.format(namespace)) - has_valid_notification = False - - if notification_list is not None: - for notification in notification_list: - method_elem = notification.find( - '{0}notificationMethod'.format(namespace)) - if method_elem is not None and method_elem.text: - if method_elem.text.lower() in {m.lower() for m in VALID_NOTIFICATION_METHODS}: - has_valid_notification = True - break - - if has_valid_notification: - # Map to friendly name - friendly_name = SENSOR_MAP.get(event_type) - if friendly_name: - events.setdefault(friendly_name, []).append(channel_num) - - session.close() - return events - - def inject_events_into_camera(camera, events): """Inject discovered events into the pyhik camera's event_states. diff --git a/test/test_hikvision.py b/test/test_hikvision.py index 21ab249..dc7d25b 100644 --- a/test/test_hikvision.py +++ b/test/test_hikvision.py @@ -5,8 +5,8 @@ import unittest from unittest.mock import MagicMock, patch, PropertyMock -from pyhik.hikvision import HikCamera, get_nvr_events, inject_events_into_camera -from pyhik.constants import (CONNECT_TIMEOUT) +from pyhik.hikvision import HikCamera, inject_events_into_camera +from pyhik.constants import CONNECT_TIMEOUT, VALID_NOTIFICATION_METHODS XML = """ {} @@ -82,7 +82,8 @@ def change_get_response(url, data,timeout): self.assertFalse(device.current_motion_detection_state) -NVR_EVENTS_XML = """ +# XML for testing get_event_triggers with various notification methods +EVENT_TRIGGERS_XML = """ 1 @@ -130,98 +131,114 @@ def change_get_response(url, data,timeout): 5 - videoloss - 1 - - - 1 - center - - - - - 6 - facedetection - 1 + VMD + 5 1 - unknown + HTTP """ -class GetNvrEventsTestCase(unittest.TestCase): +class GetEventTriggersTestCase(unittest.TestCase): @patch("pyhik.hikvision.requests.Session") - def test_get_nvr_events_parses_events(self, mock_session_class): - """Test that get_nvr_events correctly parses events with various notification methods.""" - session = mock_session_class.return_value + @patch("pyhik.hikvision.HikCamera.get_device_info") + def test_default_notification_methods(self, mock_info, mock_session): + """Test that get_event_triggers defaults to center and HTTP only.""" + mock_info.return_value = {"deviceName": "Test", "deviceID": "12345678901"} + session = mock_session.return_value response = MagicMock() response.status_code = requests.codes.ok - response.text = NVR_EVENTS_XML + response.text = EVENT_TRIGGERS_XML session.get.return_value = response - events = get_nvr_events("http://localhost", usr="admin", pwd="password") - - # Should find Motion events on channels 1 and 4 (VMD with record and center) - self.assertIn("Motion", events) - self.assertEqual(sorted(events["Motion"]), [1, 4]) - - # Should find Line Crossing on channel 2 (email notification) - self.assertIn("Line Crossing", events) - self.assertEqual(events["Line Crossing"], [2]) + camera = HikCamera(host="localhost") + # Call get_event_triggers with default (no args) + events = camera.get_event_triggers() - # Should find Field Detection on channel 3 (beep notification) - self.assertIn("Field Detection", events) - self.assertEqual(events["Field Detection"], [3]) + # Should only find VMD on channels 4 and 5 (center and HTTP) + self.assertIn("VMD", events) + self.assertEqual(sorted(events["VMD"]), [4, 5]) - # Should NOT include videoloss (skipped) - self.assertNotIn("Video Loss", events) + # Should NOT find events with record, email, beep notification methods + self.assertNotIn("linedetection", events) + self.assertNotIn("fielddetection", events) - # Should NOT include facedetection (unknown notification method) - self.assertNotIn("Face Detection", events) + @patch("pyhik.hikvision.requests.Session") + @patch("pyhik.hikvision.HikCamera.get_device_info") + def test_custom_notification_methods(self, mock_info, mock_session): + """Test that get_event_triggers accepts custom notification methods.""" + mock_info.return_value = {"deviceName": "Test", "deviceID": "12345678901"} + session = mock_session.return_value + response = MagicMock() + response.status_code = requests.codes.ok + response.text = EVENT_TRIGGERS_XML + session.get.return_value = response - session.close.assert_called_once() + camera = HikCamera(host="localhost") + # Call get_event_triggers with expanded notification methods + events = camera.get_event_triggers( + notification_methods={'center', 'HTTP', 'record', 'email', 'beep'} + ) - @patch("pyhik.hikvision.requests.Session") - def test_get_nvr_events_handles_connection_error(self, mock_session_class): - """Test that get_nvr_events handles connection errors gracefully.""" - session = mock_session_class.return_value - session.get.side_effect = requests.exceptions.ConnectionError("Connection refused") + # Should find VMD on channels 1, 4, 5 (record, center, HTTP) + self.assertIn("VMD", events) + self.assertEqual(sorted(events["VMD"]), [1, 4, 5]) - events = get_nvr_events("http://localhost", usr="admin", pwd="password") + # Should find linedetection on channel 2 (email) + self.assertIn("linedetection", events) + self.assertEqual(events["linedetection"], [2]) - self.assertEqual(events, {}) - session.close.assert_called_once() + # Should find fielddetection on channel 3 (beep) + self.assertIn("fielddetection", events) + self.assertEqual(events["fielddetection"], [3]) @patch("pyhik.hikvision.requests.Session") - def test_get_nvr_events_handles_bad_response(self, mock_session_class): - """Test that get_nvr_events handles non-200 responses.""" - session = mock_session_class.return_value + @patch("pyhik.hikvision.HikCamera.get_device_info") + def test_valid_notification_methods_constant(self, mock_info, mock_session): + """Test using VALID_NOTIFICATION_METHODS constant.""" + mock_info.return_value = {"deviceName": "Test", "deviceID": "12345678901"} + session = mock_session.return_value response = MagicMock() - response.status_code = requests.codes.unauthorized + response.status_code = requests.codes.ok + response.text = EVENT_TRIGGERS_XML session.get.return_value = response - events = get_nvr_events("http://localhost", usr="admin", pwd="password") + camera = HikCamera(host="localhost") + # Use the exported constant + events = camera.get_event_triggers( + notification_methods=VALID_NOTIFICATION_METHODS + ) - self.assertEqual(events, {}) - session.close.assert_called_once() + # Should find all events + self.assertIn("VMD", events) + self.assertEqual(sorted(events["VMD"]), [1, 4, 5]) + self.assertIn("linedetection", events) + self.assertIn("fielddetection", events) @patch("pyhik.hikvision.requests.Session") - def test_get_nvr_events_handles_invalid_xml(self, mock_session_class): - """Test that get_nvr_events handles invalid XML gracefully.""" - session = mock_session_class.return_value + @patch("pyhik.hikvision.HikCamera.get_device_info") + def test_case_insensitive_notification_methods(self, mock_info, mock_session): + """Test that notification method matching is case insensitive.""" + mock_info.return_value = {"deviceName": "Test", "deviceID": "12345678901"} + session = mock_session.return_value response = MagicMock() response.status_code = requests.codes.ok - response.text = "not valid xml" + response.text = EVENT_TRIGGERS_XML session.get.return_value = response - events = get_nvr_events("http://localhost", usr="admin", pwd="password") + camera = HikCamera(host="localhost") + # Use uppercase - should still match lowercase in XML + events = camera.get_event_triggers( + notification_methods={'CENTER', 'http', 'RECORD'} + ) - self.assertEqual(events, {}) - session.close.assert_called_once() + # Should find VMD on channels 1, 4, 5 + self.assertIn("VMD", events) + self.assertEqual(sorted(events["VMD"]), [1, 4, 5]) class InjectEventsTestCase(unittest.TestCase):