fix(telegram): send notifications for CONFIG_DETECTION_LABEL only

Author: roflcoopter

docs/src/pages/components-explorer/components/telegram/config.json

@@ -25,6 +25,32 @@
         "type": "map",
         "value": [
           {
+            "type": "map",
+            "value": [
+              {
+                "type": "string",
+                "name": {
+                  "type": "deprecated",
+                  "name": "detection_label",
+                  "value": "Config option 'detection_label' is deprecated and will be removed in a future version."
+                },
+                "description": "Deprecated. Use detection_labels instead.",
+                "deprecated": true,
+                "default": null
+              },
+              {
+                "type": "list",
+                "values": [
+                  {
+                    "type": "string"
+                  }
+                ],
+                "name": "detection_labels",
+                "description": "Labels of objects to send notifications for.",
+                "optional": true,
+                "default": null
+              }
+            ],
             "name": {
               "type": "CAMERA_IDENTIFIER"
             },
@@ -52,10 +78,28 @@
       },
       {
         "type": "string",
-        "name": "detection_label",
-        "description": "Label of the object to send notifications for.",
+        "name": {
+          "type": "deprecated",
+          "name": "detection_label",
+          "value": "Config option 'detection_label' is deprecated and will be removed in a future version."
+        },
+        "description": "Deprecated. Use detection_labels instead.",
+        "deprecated": true,
+        "default": null
+      },
+      {
+        "type": "list",
+        "values": [
+          {
+            "type": "string"
+          }
+        ],
+        "name": "detection_labels",
+        "description": "Labels of objects to send notifications for.",
         "optional": true,
-        "default": "person"
+        "default": [
+          "person"
+        ]
       },
       {
         "type": "boolean",

tests/components/telegram/__init__.py

@@ -0,0 +1 @@
+"""Tests for telegram component."""

tests/components/telegram/test_notifier.py

@@ -0,0 +1,300 @@
+"""Tests for TelegramEventNotifier detection_label filtering."""
+
+from __future__ import annotations
+
+import asyncio
+import datetime
+from typing import TYPE_CHECKING, Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from viseron.components.storage.models import TriggerTypes
+from viseron.components.telegram import TelegramEventNotifier
+from viseron.components.telegram.const import (
+    CONFIG_CAMERAS,
+    CONFIG_DETECTION_LABEL,
+    CONFIG_DETECTION_LABELS,
+    CONFIG_SEND_MESSAGE,
+    CONFIG_SEND_THUMBNAIL,
+    CONFIG_SEND_VIDEO,
+    CONFIG_TELEGRAM_BOT_TOKEN,
+    CONFIG_TELEGRAM_CHAT_IDS,
+    CONFIG_TELEGRAM_LOG_IDS,
+    CONFIG_TELEGRAM_USER_IDS,
+    DEFAULT_DETECTION_LABELS,
+    DEFAULT_SEND_MESSAGE,
+    DEFAULT_TELEGRAM_LOG_IDS,
+    DEFAULT_TELEGRAM_USER_IDS,
+)
+from viseron.domains.camera.recorder import EventRecorderData, Recording
+from viseron.domains.object_detector.detected_object import DetectedObject
+from viseron.events import Event
+from viseron.helpers.validators import UNDEFINED
+
+if TYPE_CHECKING:
+    from tests.conftest import MockViseron
+
+CAMERA_ID = "test_camera"
+
+
+def make_config(
+    detection_labels: list[str] | None = None,
+    cameras: dict | None = None,
+) -> dict:
+    """Build a minimal TelegramEventNotifier config."""
+    return {
+        CONFIG_TELEGRAM_BOT_TOKEN: "token",
+        CONFIG_TELEGRAM_CHAT_IDS: [123],
+        CONFIG_TELEGRAM_USER_IDS: DEFAULT_TELEGRAM_USER_IDS,
+        CONFIG_TELEGRAM_LOG_IDS: DEFAULT_TELEGRAM_LOG_IDS,
+        CONFIG_DETECTION_LABELS: detection_labels
+        if detection_labels is not None
+        else DEFAULT_DETECTION_LABELS,
+        CONFIG_SEND_THUMBNAIL: False,
+        CONFIG_SEND_VIDEO: False,
+        CONFIG_SEND_MESSAGE: DEFAULT_SEND_MESSAGE,
+        CONFIG_CAMERAS: cameras if cameras is not None else {CAMERA_ID: {}},
+    }
+
+
+def make_detected_object(label: str) -> DetectedObject:
+    """Build a minimal DetectedObject with the given label."""
+    return DetectedObject(
+        label=label,
+        confidence=0.9,
+        x1=0.1,
+        y1=0.1,
+        x2=0.5,
+        y2=0.5,
+        frame_res=(1920, 1080),
+    )
+
+
+def make_recording(objects: list[DetectedObject]) -> Recording:
+    """Build a minimal Recording with the given detected objects."""
+    return Recording(
+        id=1,
+        start_time=datetime.datetime(2025, 1, 1, tzinfo=datetime.timezone.utc),
+        start_timestamp=0.0,
+        end_time=datetime.datetime(2025, 1, 1, 0, 0, 10, tzinfo=datetime.timezone.utc),
+        end_timestamp=10.0,
+        date="2025-01-01",
+        thumbnail=None,
+        thumbnail_path=None,
+        clip_path=None,
+        objects=objects,
+        trigger_type=TriggerTypes.OBJECT,
+    )
+
+
+def make_event(recording: Recording) -> Event[EventRecorderData]:
+    """Build an Event wrapping the given recording."""
+    camera_mock = MagicMock()
+    camera_mock.identifier = CAMERA_ID
+    event_data = EventRecorderData(camera=camera_mock, recording=recording)
+    return Event(name="test", data=event_data, timestamp=0.0)
+
+
+@pytest.fixture
+def notifier(vis: MockViseron) -> Any:
+    """Return a TelegramEventNotifier with mocked internals."""
+    config = make_config()
+    with (
+        patch("viseron.components.telegram.Bot"),
+        patch("viseron.components.telegram.Application"),
+    ):
+        n = TelegramEventNotifier(vis, config)
+    # Replace bot with an AsyncMock so send_* calls can be awaited
+    n._bot = MagicMock()
+    n._bot.send_message = AsyncMock()
+    n._bot.send_video = AsyncMock()
+    n._bot.send_photo = AsyncMock()
+    return n
+
+
+def run_send_notifications(
+    notifier: TelegramEventNotifier, event: Event[EventRecorderData]
+):
+    """Run _send_notifications synchronously via a fresh event loop."""
+    loop = asyncio.new_event_loop()
+    try:
+        loop.run_until_complete(notifier._send_notifications(event))
+    finally:
+        loop.close()
+
+
+class TestDetectionLabelFiltering:
+    """Tests for detection_label filtering in _send_notifications."""
+
+    def test_sends_when_label_matches_global(self, notifier: Any):
+        """Notification sent when detected object matches global detection_labels."""
+        recording = make_recording([make_detected_object("person")])
+        event = make_event(recording)
+        notifier._config[CONFIG_SEND_MESSAGE] = True
+
+        run_send_notifications(notifier, event)
+
+        notifier._bot.send_message.assert_awaited_once()
+
+    def test_skips_when_label_does_not_match_global(self, notifier: Any):
+        """No notification sent when detected object does not match detection_labels."""
+        recording = make_recording([make_detected_object("car")])
+        event = make_event(recording)
+        notifier._config[CONFIG_DETECTION_LABELS] = ["person"]
+        notifier._config[CONFIG_SEND_MESSAGE] = True
+
+        run_send_notifications(notifier, event)
+
+        notifier._bot.send_message.assert_not_awaited()
+        notifier._bot.send_video.assert_not_awaited()
+        notifier._bot.send_photo.assert_not_awaited()
+
+    def test_sends_for_manual_recording_with_no_objects(self, notifier: Any):
+        """Notification always sent when recording has no detected objects."""
+        recording = make_recording([])
+        recording.trigger_type = TriggerTypes.MANUAL
+        event = make_event(recording)
+        notifier._config[CONFIG_DETECTION_LABELS] = ["person"]
+        notifier._config[CONFIG_SEND_MESSAGE] = True
+
+        run_send_notifications(notifier, event)
+
+        notifier._bot.send_message.assert_awaited_once()
+
+    def test_comma_separated_labels_match(self, notifier: Any):
+        """Notification sent when comma-separated detection_label matches."""
+        recording = make_recording([make_detected_object("cat")])
+        event = make_event(recording)
+        notifier._config[CONFIG_DETECTION_LABELS] = None
+        notifier._config[CONFIG_DETECTION_LABEL] = "person,cat"
+        notifier._config[CONFIG_SEND_MESSAGE] = True
+
+        run_send_notifications(notifier, event)
+
+        notifier._bot.send_message.assert_awaited_once()
+
+    def test_comma_separated_labels_no_match(self, notifier: Any):
+        """No notification when comma-separated detection_label has no match."""
+        recording = make_recording([make_detected_object("car")])
+        event = make_event(recording)
+        notifier._config[CONFIG_DETECTION_LABELS] = None
+        notifier._config[CONFIG_DETECTION_LABEL] = "person,cat"
+        notifier._config[CONFIG_SEND_MESSAGE] = True
+
+        run_send_notifications(notifier, event)
+
+        notifier._bot.send_message.assert_not_awaited()
+
+    def test_camera_level_detection_label_overrides_global(self, vis: MockViseron):
+        """Camera-level detection_labels overrides the global setting."""
+        cameras = {CAMERA_ID: {CONFIG_DETECTION_LABELS: ["car"]}}
+        config = make_config(detection_labels=["person"], cameras=cameras)
+        config[CONFIG_SEND_MESSAGE] = True
+        with (
+            patch("viseron.components.telegram.Bot"),
+            patch("viseron.components.telegram.Application"),
+        ):
+            notifier: Any = TelegramEventNotifier(vis, config)
+        notifier._bot = MagicMock()
+        notifier._bot.send_message = AsyncMock()
+        notifier._bot.send_video = AsyncMock()
+        notifier._bot.send_photo = AsyncMock()
+
+        # "car" matches camera-level override → should send
+        recording = make_recording([make_detected_object("car")])
+        run_send_notifications(notifier, make_event(recording))
+        notifier._bot.send_message.assert_awaited_once()
+
+        notifier._bot.send_message.reset_mock()
+
+        # "person" matches global but NOT camera-level → should NOT send
+        recording2 = make_recording([make_detected_object("person")])
+        run_send_notifications(notifier, make_event(recording2))
+        notifier._bot.send_message.assert_not_awaited()
+
+    def test_detection_labels_list(self, notifier: Any):
+        """Notification sent when detection_labels is a list with matching entry."""
+        recording = make_recording([make_detected_object("cat")])
+        event = make_event(recording)
+        notifier._config[CONFIG_DETECTION_LABELS] = ["person", "cat"]
+        notifier._config[CONFIG_SEND_MESSAGE] = True
+
+        run_send_notifications(notifier, event)
+
+        notifier._bot.send_message.assert_awaited_once()
+
+    def test_detection_labels_list_no_match(self, notifier: Any):
+        """No notification when detection_labels list has no matching entry."""
+        recording = make_recording([make_detected_object("car")])
+        event = make_event(recording)
+        notifier._config[CONFIG_DETECTION_LABELS] = ["person", "cat"]
+        notifier._config[CONFIG_SEND_MESSAGE] = True
+
+        run_send_notifications(notifier, event)
+
+        notifier._bot.send_message.assert_not_awaited()
+
+
+class TestGetEffectiveDetectionLabels:
+    """Tests for _get_effective_detection_labels."""
+
+    def test_undefined_camera_labels_falls_through_to_global(self, vis: MockViseron):
+        """When camera detection_labels is UNDEFINED, global labels used."""
+        cameras = {CAMERA_ID: {CONFIG_DETECTION_LABELS: UNDEFINED()}}
+        config = make_config(detection_labels=["car"], cameras=cameras)
+        with (
+            patch("viseron.components.telegram.Bot"),
+            patch("viseron.components.telegram.Application"),
+        ):
+            notifier: Any = TelegramEventNotifier(vis, config)
+
+        labels = notifier._get_effective_detection_labels(CAMERA_ID)
+
+        assert labels == ["car"]
+
+    def test_undefined_camera_labels_does_not_raise_on_notification(
+        self, vis: MockViseron
+    ):
+        """Detection with UNDEFINED camera labels does not raise TypeError."""
+        cameras = {CAMERA_ID: {CONFIG_DETECTION_LABELS: UNDEFINED()}}
+        config = make_config(detection_labels=["person"], cameras=cameras)
+        config[CONFIG_SEND_MESSAGE] = True
+        with (
+            patch("viseron.components.telegram.Bot"),
+            patch("viseron.components.telegram.Application"),
+        ):
+            notifier: Any = TelegramEventNotifier(vis, config)
+        notifier._bot = MagicMock()
+        notifier._bot.send_message = AsyncMock()
+
+        recording = make_recording([make_detected_object("person")])
+        event = make_event(recording)
+        run_send_notifications(notifier, event)
+
+        notifier._bot.send_message.assert_awaited_once()
+
+    def test_global_comma_deprecated_label_strips_empty_entries(self, notifier: Any):
+        """Global detection_label with surrounding commas strips empty entries."""
+        notifier._config[CONFIG_DETECTION_LABELS] = None
+        notifier._config[CONFIG_DETECTION_LABEL] = ",person,"
+
+        labels = notifier._get_effective_detection_labels(CAMERA_ID)
+
+        assert "" not in labels
+        assert labels == ["person"]
+
+    def test_camera_comma_deprecated_label_strips_empty_entries(self, vis: MockViseron):
+        """Camera detection_label with surrounding commas strips empty entries."""
+        cameras = {CAMERA_ID: {CONFIG_DETECTION_LABEL: ",car,"}}
+        config = make_config(cameras=cameras)
+        with (
+            patch("viseron.components.telegram.Bot"),
+            patch("viseron.components.telegram.Application"),
+        ):
+            notifier: Any = TelegramEventNotifier(vis, config)
+
+        labels = notifier._get_effective_detection_labels(CAMERA_ID)
+
+        assert "" not in labels
+        assert labels == ["car"]