389 lines
17 KiB
Python
389 lines
17 KiB
Python
from django.test import TestCase
|
|
from django.urls import reverse
|
|
from django.contrib.auth import get_user_model
|
|
from notifications.models import Discussion, Message
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
class DiscussionListTests(TestCase):
|
|
def setUp(self):
|
|
self.user1 = User.objects.create_user(username='user1', password='pass')
|
|
self.user2 = User.objects.create_user(username='user2', password='pass')
|
|
self.user3 = User.objects.create_user(username='user3', password='pass')
|
|
|
|
def test_discussion_list_filtering_on_inbox(self):
|
|
# Discussion 1: created by user1
|
|
d1 = Discussion.objects.create(title="Discussion 1", created_by=self.user1)
|
|
d1.participants.add(self.user1, self.user2, self.user3)
|
|
|
|
# Discussion 2: created by user2, user1 wrote a message
|
|
d2 = Discussion.objects.create(title="Discussion 2", created_by=self.user2)
|
|
d2.participants.add(self.user1, self.user2, self.user3)
|
|
Message.objects.create(discussion=d2, author=self.user1, content="Hello from user1")
|
|
|
|
# Discussion 3: created by user2, user1 is tagged
|
|
d3 = Discussion.objects.create(title="Discussion 3", created_by=self.user2)
|
|
d3.participants.add(self.user1, self.user2, self.user3)
|
|
Message.objects.create(discussion=d3, author=self.user2, content=f"Hello @[User 1](u:{self.user1.pk})")
|
|
|
|
# Discussion 4: created by user2, user3 participant but user1 is NOT author and NOT tagged
|
|
d4 = Discussion.objects.create(title="Discussion 4", created_by=self.user2)
|
|
d4.participants.add(self.user1, self.user2, self.user3)
|
|
|
|
# Discussion 5: auto-created intervention discussion with created_by=user1, but user1 never posted a message and was never tagged
|
|
from django.contrib.contenttypes.models import ContentType
|
|
ct = ContentType.objects.get_for_model(User)
|
|
d5 = Discussion.objects.create(title="Discussion 5", created_by=self.user1, content_type=ct, object_id=1)
|
|
Message.objects.create(discussion=d5, author=self.user2, content="Hello user3")
|
|
|
|
self.client.force_login(self.user1)
|
|
response = self.client.get(reverse('notifications:inbox') + '?tab=discussions')
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.context['active_tab'], 'discussions')
|
|
|
|
discussions_in_context = [item['discussion'] for item in response.context['discussion_items']]
|
|
self.assertIn(d1, discussions_in_context)
|
|
self.assertIn(d2, discussions_in_context)
|
|
self.assertIn(d3, discussions_in_context)
|
|
self.assertNotIn(d4, discussions_in_context)
|
|
self.assertNotIn(d5, discussions_in_context)
|
|
|
|
def test_discussion_list_url_redirect(self):
|
|
self.client.force_login(self.user1)
|
|
response = self.client.get(reverse('notifications:discussion_list'))
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertIn('tab=discussions', response.url)
|
|
|
|
def test_discussion_create(self):
|
|
self.client.force_login(self.user1)
|
|
response = self.client.post(reverse('notifications:discussion_create'), {
|
|
'title': 'Sujet test',
|
|
'content': f'Bonjour @[User 2](u:{self.user2.pk})',
|
|
})
|
|
self.assertEqual(response.status_code, 302)
|
|
discussion = Discussion.objects.get(title='Sujet test')
|
|
self.assertEqual(discussion.created_by, self.user1)
|
|
self.assertIn(self.user1, discussion.participants.all())
|
|
self.assertIn(self.user2, discussion.participants.all())
|
|
self.assertEqual(discussion.messages.count(), 1)
|
|
|
|
|
|
class InboxTests(TestCase):
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(username='test_inbox_user', password='pass')
|
|
from notifications.models import Notification, NotificationRecipient, Discussion
|
|
# Create a notification for user
|
|
notif = Notification.objects.create(
|
|
title="Test notification",
|
|
content="Notification content",
|
|
notification_type=Notification.TYPE_INFO,
|
|
)
|
|
NotificationRecipient.objects.create(
|
|
notification=notif,
|
|
user=self.user,
|
|
)
|
|
# Create a discussion for user
|
|
d = Discussion.objects.create(title="Test discussion", created_by=self.user)
|
|
d.participants.add(self.user)
|
|
|
|
def test_inbox_renders_both_notifications_and_discussions(self):
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse('notifications:inbox'))
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.context['active_tab'], 'notifications')
|
|
self.assertEqual(response.context['unread_count'], 1)
|
|
self.assertEqual(len(response.context['notifications']), 1)
|
|
self.assertEqual(len(response.context['discussion_items']), 1)
|
|
self.assertContains(response, 'Test notification')
|
|
self.assertContains(response, 'Test discussion')
|
|
self.assertContains(response, 'id="pane-notifications"')
|
|
self.assertContains(response, 'id="pane-discussions"')
|
|
self.assertContains(response, 'id="tab-notifications-btn"')
|
|
self.assertContains(response, 'id="tab-discussions-btn"')
|
|
|
|
def test_inbox_tab_discussions_parameter(self):
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse('notifications:inbox') + '?tab=discussions')
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.context['active_tab'], 'discussions')
|
|
|
|
|
|
class NotificationReminderTests(TestCase):
|
|
def setUp(self):
|
|
from common.models import UserConfig
|
|
self.user = User.objects.create_user(username='testuser', email='test@example.com', password='pass')
|
|
self.config, _ = UserConfig.objects.get_or_create(user=self.user)
|
|
|
|
def test_notification_settings_view(self):
|
|
self.client.force_login(self.user)
|
|
# GET settings page
|
|
response = self.client.get(reverse('notifications:settings'))
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
# POST updated settings
|
|
response = self.client.post(reverse('notifications:settings'), {
|
|
'preferred_language': 'nl',
|
|
'notification_reminder_enabled': 'on',
|
|
'notification_reminder_emails': 'custom@example.com, other@example.com',
|
|
'notification_reminder_delay_hours': 12,
|
|
})
|
|
self.assertEqual(response.status_code, 302)
|
|
|
|
self.config.refresh_from_db()
|
|
self.assertEqual(self.config.preferred_language, 'nl')
|
|
self.assertTrue(self.config.notification_reminder_enabled)
|
|
self.assertEqual(self.config.notification_reminder_emails, 'custom@example.com, other@example.com')
|
|
self.assertEqual(self.config.notification_reminder_delay_hours, 12)
|
|
|
|
from django.test import override_settings
|
|
|
|
@override_settings(EMAIL_OVERRIDE=None)
|
|
def test_send_unread_notification_reminders_command(self):
|
|
from datetime import timedelta
|
|
from django.utils import timezone
|
|
from django.core import mail
|
|
from django.core.management import call_command
|
|
from notifications.models import Notification, NotificationRecipient
|
|
|
|
# Configure user for reminders
|
|
self.config.notification_reminder_enabled = True
|
|
self.config.notification_reminder_delay_hours = 2
|
|
self.config.save()
|
|
|
|
# Create unread notification created 3 hours ago (> 2h delay)
|
|
past_time = timezone.now() - timedelta(hours=3)
|
|
notif = Notification.objects.create(
|
|
title="Notification urgente non lue",
|
|
content="Détail de la notification",
|
|
notification_type=Notification.TYPE_ALERT,
|
|
)
|
|
recipient = NotificationRecipient.objects.create(
|
|
notification=notif,
|
|
user=self.user,
|
|
)
|
|
# Backdate notification creation
|
|
Notification.objects.filter(pk=notif.pk).update(created_at=past_time)
|
|
|
|
# Run management command
|
|
call_command('send_unread_notification_reminders')
|
|
|
|
# Check that 1 email was sent
|
|
self.assertEqual(len(mail.outbox), 1)
|
|
sent_email = mail.outbox[0]
|
|
self.assertIn("Notification urgente non lue", sent_email.body)
|
|
self.assertIn('test@example.com', sent_email.to)
|
|
|
|
# Verify reminder_sent_at is set
|
|
recipient.refresh_from_db()
|
|
self.assertIsNotNone(recipient.reminder_sent_at)
|
|
|
|
@override_settings(EMAIL_OVERRIDE=None)
|
|
def test_send_unread_notification_reminders_multilingual(self):
|
|
from datetime import timedelta
|
|
from django.utils import timezone
|
|
from django.core import mail
|
|
from django.core.management import call_command
|
|
from notifications.models import Notification, NotificationRecipient
|
|
|
|
self.config.notification_reminder_enabled = True
|
|
self.config.notification_reminder_delay_hours = 1
|
|
self.config.preferred_language = 'nl'
|
|
self.config.save()
|
|
|
|
past_time = timezone.now() - timedelta(hours=2)
|
|
notif = Notification.objects.create(
|
|
title="Dringende melding",
|
|
content="Details",
|
|
notification_type=Notification.TYPE_ALERT,
|
|
)
|
|
recipient = NotificationRecipient.objects.create(
|
|
notification=notif,
|
|
user=self.user,
|
|
)
|
|
Notification.objects.filter(pk=notif.pk).update(created_at=past_time)
|
|
|
|
mail.outbox.clear()
|
|
call_command('send_unread_notification_reminders')
|
|
|
|
self.assertEqual(len(mail.outbox), 1)
|
|
sent_email = mail.outbox[0]
|
|
# In Dutch, subject contains 'Herinnering'
|
|
self.assertIn("Herinnering", sent_email.subject)
|
|
|
|
|
|
class WebPushTests(TestCase):
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(username='pushuser', password='password123')
|
|
self.client.force_login(self.user)
|
|
|
|
def test_device_info_parsing(self):
|
|
from notifications.webpush import parse_device_info
|
|
ua_win_chrome = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
|
|
res = parse_device_info(ua_win_chrome)
|
|
self.assertEqual(res['browser'], 'Google Chrome')
|
|
self.assertEqual(res['os'], 'Windows')
|
|
self.assertEqual(res['device_name'], 'Google Chrome sur Windows')
|
|
|
|
ua_ios_safari = "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1"
|
|
res_ios = parse_device_info(ua_ios_safari)
|
|
self.assertEqual(res_ios['browser'], 'Safari')
|
|
self.assertEqual(res_ios['os'], 'iOS')
|
|
|
|
def test_service_worker_route_and_headers(self):
|
|
response = self.client.get('/service-worker.js')
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn('application/javascript', response['Content-Type'])
|
|
self.assertEqual(response['Service-Worker-Allowed'], '/')
|
|
|
|
def test_api_vapid_public_key(self):
|
|
response = self.client.get(reverse('notifications:api_webpush_vapid_key'))
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.json()
|
|
self.assertIn('publicKey', data)
|
|
self.assertTrue(len(data['publicKey']) > 10)
|
|
|
|
def test_api_subscribe_and_unsubscribe(self):
|
|
from notifications.models import WebPushSubscription
|
|
payload = {
|
|
'endpoint': 'https://fcm.googleapis.com/fcm/send/test-sub-12345',
|
|
'p256dh': 'BNcRdreALRFXTkOOUHK1EtK2wtaz5Ry4YfYCA_0QTpQtUbVlUls0VJXg7A8u-Ts1XbjhazAkj7I99e8QcYP7DkM',
|
|
'auth': 'tBHItJI5svbpez7KI4CCXg',
|
|
'user_agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0'
|
|
}
|
|
|
|
# 1. Subscribe
|
|
response = self.client.post(
|
|
reverse('notifications:api_webpush_subscribe'),
|
|
data=payload,
|
|
content_type='application/json'
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(WebPushSubscription.objects.filter(user=self.user).count(), 1)
|
|
sub = WebPushSubscription.objects.get(user=self.user)
|
|
self.assertEqual(sub.endpoint, payload['endpoint'])
|
|
self.assertEqual(sub.browser_name, 'Google Chrome')
|
|
self.assertEqual(sub.os_name, 'Windows')
|
|
|
|
# 2. Delete device via delete route
|
|
del_resp = self.client.post(reverse('notifications:api_webpush_delete_device', args=[sub.pk]))
|
|
self.assertEqual(del_resp.status_code, 302)
|
|
self.assertEqual(WebPushSubscription.objects.filter(user=self.user).count(), 0)
|
|
|
|
# 3. Subscribe again and unsubscribe via API
|
|
self.client.post(
|
|
reverse('notifications:api_webpush_subscribe'),
|
|
data=payload,
|
|
content_type='application/json'
|
|
)
|
|
self.assertEqual(WebPushSubscription.objects.filter(user=self.user).count(), 1)
|
|
|
|
unsub_resp = self.client.post(
|
|
reverse('notifications:api_webpush_unsubscribe'),
|
|
data={'endpoint': payload['endpoint']},
|
|
content_type='application/json'
|
|
)
|
|
self.assertEqual(unsub_resp.status_code, 200)
|
|
self.assertEqual(WebPushSubscription.objects.filter(user=self.user).count(), 0)
|
|
|
|
def test_send_webpush_mocked(self):
|
|
import base64
|
|
import os
|
|
from unittest.mock import patch, MagicMock
|
|
from cryptography.hazmat.primitives.asymmetric import ec
|
|
from cryptography.hazmat.primitives import serialization
|
|
from notifications.models import WebPushSubscription
|
|
from notifications.webpush import send_webpush_to_user
|
|
|
|
fake_client_priv = ec.generate_private_key(ec.SECP256R1())
|
|
fake_pub = fake_client_priv.public_key().public_bytes(
|
|
encoding=serialization.Encoding.X962,
|
|
format=serialization.PublicFormat.UncompressedPoint
|
|
)
|
|
fake_auth = os.urandom(16)
|
|
sub = WebPushSubscription.objects.create(
|
|
user=self.user,
|
|
endpoint='https://fcm.googleapis.com/fcm/send/mock-123',
|
|
p256dh=base64.urlsafe_b64encode(fake_pub).decode('utf-8').rstrip('='),
|
|
auth=base64.urlsafe_b64encode(fake_auth).decode('utf-8').rstrip('='),
|
|
device_name='Test Device'
|
|
)
|
|
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 201
|
|
|
|
with patch('notifications.webpush.requests.post', return_value=mock_resp) as mock_post:
|
|
send_webpush_to_user(
|
|
user=self.user,
|
|
title="Alerte Intervention",
|
|
body="Nouvelle intervention assignée",
|
|
action_url="/interventions/1/",
|
|
async_send=False
|
|
)
|
|
mock_post.assert_called_once()
|
|
called_endpoint = mock_post.call_args[0][0]
|
|
self.assertEqual(called_endpoint, sub.endpoint)
|
|
headers = mock_post.call_args[1]['headers']
|
|
self.assertIn('Authorization', headers)
|
|
self.assertEqual(headers['Content-Encoding'], 'aes128gcm')
|
|
|
|
def test_send_webpush_410_auto_cleanup(self):
|
|
import base64
|
|
import os
|
|
from unittest.mock import patch, MagicMock
|
|
from cryptography.hazmat.primitives.asymmetric import ec
|
|
from cryptography.hazmat.primitives import serialization
|
|
from notifications.models import WebPushSubscription
|
|
from notifications.webpush import send_webpush_to_user
|
|
|
|
fake_client_priv = ec.generate_private_key(ec.SECP256R1())
|
|
fake_pub = fake_client_priv.public_key().public_bytes(
|
|
encoding=serialization.Encoding.X962,
|
|
format=serialization.PublicFormat.UncompressedPoint
|
|
)
|
|
fake_auth = os.urandom(16)
|
|
sub = WebPushSubscription.objects.create(
|
|
user=self.user,
|
|
endpoint='https://fcm.googleapis.com/fcm/send/expired-123',
|
|
p256dh=base64.urlsafe_b64encode(fake_pub).decode('utf-8').rstrip('='),
|
|
auth=base64.urlsafe_b64encode(fake_auth).decode('utf-8').rstrip('='),
|
|
)
|
|
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 410
|
|
|
|
with patch('notifications.webpush.requests.post', return_value=mock_resp):
|
|
send_webpush_to_user(
|
|
user=self.user,
|
|
title="Test",
|
|
async_send=False
|
|
)
|
|
# Subscription should be automatically deleted from DB
|
|
self.assertFalse(WebPushSubscription.objects.filter(pk=sub.pk).exists())
|
|
|
|
def test_notify_user_triggers_webpush(self):
|
|
from unittest.mock import patch
|
|
from notifications.models import WebPushSubscription
|
|
from notifications.services import notify_user
|
|
|
|
WebPushSubscription.objects.create(
|
|
user=self.user,
|
|
endpoint='https://fcm.googleapis.com/fcm/send/auto-123',
|
|
p256dh='fake-key',
|
|
auth='fake-auth',
|
|
)
|
|
|
|
with patch('notifications.webpush.send_webpush_to_user') as mock_send:
|
|
notify_user(
|
|
recipient=self.user,
|
|
title="Nouvelle tâche",
|
|
content="Veuillez vérifier l'intervention.",
|
|
action_url="/interventions/99/",
|
|
)
|
|
mock_send.assert_called_once_with(
|
|
user=self.user,
|
|
title="Nouvelle tâche",
|
|
body="Veuillez vérifier l'intervention.",
|
|
action_url="/interventions/99/",
|
|
)
|
|
|