tests: Add backend test for expiration feature.

apps/m3u/tests/test_expiration_notifications.py

@@ -0,0 +1,216 @@
+"""
+Tests for evaluate_profile_expiration_notification.
+
+Covers all four branches:
+  - no exp_date         → returns None, touches nothing
+  - already expired     → creates/updates expired notification, removes warning
+  - expiring within 7d  → creates/updates warning notification, removes expired
+  - not expiring soon   → removes any stale notifications, returns None
+"""
+from datetime import timedelta
+from unittest.mock import patch, MagicMock
+
+from django.test import SimpleTestCase
+from django.utils import timezone
+
+
+def _make_profile(exp_date, profile_id=1, profile_name="Test Profile",
+                  account_id=10, account_name="Test Account"):
+    """Return a minimal mock M3UAccountProfile."""
+    profile = MagicMock()
+    profile.id = profile_id
+    profile.name = profile_name
+    profile.exp_date = exp_date
+    profile.m3u_account.id = account_id
+    profile.m3u_account.name = account_name
+    return profile
+
+
+class EvaluateProfileExpirationNotificationTests(SimpleTestCase):
+
+    def setUp(self):
+        # These three names are local imports inside evaluate_profile_expiration_notification,
+        # so we must patch them at their source modules rather than on apps.m3u.tasks.
+        self.mock_sn = MagicMock()
+        self.mock_send_ws = patch("core.utils.send_websocket_notification").start()
+        self.mock_dismissed = patch("core.utils.send_notification_dismissed").start()
+        patch("core.models.SystemNotification", self.mock_sn).start()
+
+    def tearDown(self):
+        patch.stopall()
+
+    def _run(self, profile):
+        from apps.m3u.tasks import evaluate_profile_expiration_notification
+        return evaluate_profile_expiration_notification(profile)
+
+    # ------------------------------------------------------------------ #
+    # No expiration date
+    # ------------------------------------------------------------------ #
+
+    def test_no_exp_date_returns_none(self):
+        profile = _make_profile(exp_date=None)
+        result = self._run(profile)
+        self.assertIsNone(result)
+        self.mock_sn.objects.update_or_create.assert_not_called()
+        self.mock_send_ws.assert_not_called()
+
+    # ------------------------------------------------------------------ #
+    # Already expired
+    # ------------------------------------------------------------------ #
+
+    @patch("apps.m3u.tasks.timezone")
+    def test_expired_creates_expired_notification(self, mock_tz):
+        now = timezone.now()
+        mock_tz.now.return_value = now
+        mock_tz.timedelta = timedelta
+
+        profile = _make_profile(exp_date=now - timedelta(days=1))
+        # No existing warning notification to delete
+        self.mock_sn.objects.filter.return_value.values_list.return_value = []
+        notification = MagicMock()
+        self.mock_sn.objects.update_or_create.return_value = (notification, True)
+
+        result = self._run(profile)
+
+        self.assertEqual(result, f"xc-exp-expired-{profile.id}")
+        self.mock_sn.objects.update_or_create.assert_called_once()
+        call_kwargs = self.mock_sn.objects.update_or_create.call_args
+        self.assertEqual(call_kwargs.kwargs["notification_key"], f"xc-exp-expired-{profile.id}")
+        self.assertTrue(call_kwargs.kwargs["defaults"]["admin_only"])
+        self.mock_send_ws.assert_called_once_with(notification)
+
+    @patch("apps.m3u.tasks.timezone")
+    def test_expired_removes_stale_warning_notification(self, mock_tz):
+        now = timezone.now()
+        mock_tz.now.return_value = now
+        mock_tz.timedelta = timedelta
+
+        profile = _make_profile(exp_date=now - timedelta(hours=1))
+        warning_key = f"xc-exp-warning-{profile.id}"
+        # Simulate an existing warning notification
+        self.mock_sn.objects.filter.return_value.values_list.return_value = [warning_key]
+        self.mock_sn.objects.update_or_create.return_value = (MagicMock(), False)
+
+        self._run(profile)
+
+        self.mock_dismissed.assert_any_call(warning_key)
+
+    # ------------------------------------------------------------------ #
+    # Expiring within 7 days
+    # ------------------------------------------------------------------ #
+
+    @patch("apps.m3u.tasks.timezone")
+    def test_warning_window_creates_warning_notification(self, mock_tz):
+        now = timezone.now()
+        mock_tz.now.return_value = now
+        mock_tz.timedelta = timedelta
+
+        profile = _make_profile(exp_date=now + timedelta(days=3))
+        self.mock_sn.objects.filter.return_value.values_list.return_value = []
+        notification = MagicMock()
+        self.mock_sn.objects.update_or_create.return_value = (notification, True)
+
+        result = self._run(profile)
+
+        self.assertEqual(result, f"xc-exp-warning-{profile.id}")
+        call_kwargs = self.mock_sn.objects.update_or_create.call_args
+        self.assertEqual(call_kwargs.kwargs["notification_key"], f"xc-exp-warning-{profile.id}")
+        self.assertTrue(call_kwargs.kwargs["defaults"]["admin_only"])
+        self.mock_send_ws.assert_called_once_with(notification)
+
+    @patch("apps.m3u.tasks.timezone")
+    def test_warning_message_says_today_when_same_day(self, mock_tz):
+        now = timezone.now()
+        mock_tz.now.return_value = now
+        mock_tz.timedelta = timedelta
+
+        profile = _make_profile(exp_date=now + timedelta(hours=2))
+        self.mock_sn.objects.filter.return_value.values_list.return_value = []
+        self.mock_sn.objects.update_or_create.return_value = (MagicMock(), True)
+
+        self._run(profile)
+
+        defaults = self.mock_sn.objects.update_or_create.call_args.kwargs["defaults"]
+        self.assertIn("today", defaults["message"])
+
+    @patch("apps.m3u.tasks.timezone")
+    def test_warning_message_says_1_day(self, mock_tz):
+        now = timezone.now()
+        mock_tz.now.return_value = now
+        mock_tz.timedelta = timedelta
+
+        profile = _make_profile(exp_date=now + timedelta(hours=30))
+        self.mock_sn.objects.filter.return_value.values_list.return_value = []
+        self.mock_sn.objects.update_or_create.return_value = (MagicMock(), True)
+
+        self._run(profile)
+
+        defaults = self.mock_sn.objects.update_or_create.call_args.kwargs["defaults"]
+        self.assertIn("in 1 day", defaults["message"])
+
+    @patch("apps.m3u.tasks.timezone")
+    def test_warning_removes_stale_expired_notification(self, mock_tz):
+        now = timezone.now()
+        mock_tz.now.return_value = now
+        mock_tz.timedelta = timedelta
+
+        profile = _make_profile(exp_date=now + timedelta(days=5))
+        expired_key = f"xc-exp-expired-{profile.id}"
+        self.mock_sn.objects.filter.return_value.values_list.return_value = [expired_key]
+        self.mock_sn.objects.update_or_create.return_value = (MagicMock(), False)
+
+        self._run(profile)
+
+        self.mock_dismissed.assert_any_call(expired_key)
+
+    # ------------------------------------------------------------------ #
+    # Not expiring soon (> 7 days away)
+    # ------------------------------------------------------------------ #
+
+    @patch("apps.m3u.tasks.timezone")
+    def test_not_expiring_soon_returns_none(self, mock_tz):
+        now = timezone.now()
+        mock_tz.now.return_value = now
+        mock_tz.timedelta = timedelta
+
+        profile = _make_profile(exp_date=now + timedelta(days=30))
+        self.mock_sn.objects.filter.return_value.values_list.return_value = []
+
+        result = self._run(profile)
+
+        self.assertIsNone(result)
+        self.mock_sn.objects.update_or_create.assert_not_called()
+        self.mock_send_ws.assert_not_called()
+
+    @patch("apps.m3u.tasks.timezone")
+    def test_not_expiring_soon_removes_stale_notifications(self, mock_tz):
+        now = timezone.now()
+        mock_tz.now.return_value = now
+        mock_tz.timedelta = timedelta
+
+        profile = _make_profile(exp_date=now + timedelta(days=30))
+        warning_key = f"xc-exp-warning-{profile.id}"
+        self.mock_sn.objects.filter.return_value.values_list.return_value = [warning_key]
+
+        self._run(profile)
+
+        self.mock_dismissed.assert_called_once_with(warning_key)
+
+    # ------------------------------------------------------------------ #
+    # Boundary: exactly at the 7-day warning threshold
+    # ------------------------------------------------------------------ #
+
+    @patch("apps.m3u.tasks.timezone")
+    def test_exactly_7_days_away_triggers_warning(self, mock_tz):
+        now = timezone.now()
+        mock_tz.now.return_value = now
+        mock_tz.timedelta = timedelta
+
+        # exp_date == now + 7 days  →  exp <= warning_threshold  →  warning
+        profile = _make_profile(exp_date=now + timedelta(days=7))
+        self.mock_sn.objects.filter.return_value.values_list.return_value = []
+        self.mock_sn.objects.update_or_create.return_value = (MagicMock(), True)
+
+        result = self._run(profile)
+
+        self.assertEqual(result, f"xc-exp-warning-{profile.id}")