88 lines
3.4 KiB
Python
88 lines
3.4 KiB
Python
# backend/modules/fleet/tests/test_pusher_service.py
|
|
from django.test import TestCase
|
|
from unittest.mock import Mock, patch
|
|
import uuid
|
|
|
|
from infrastructure.pusher_service import PusherService
|
|
|
|
|
|
class PusherServiceTests(TestCase):
|
|
@patch("infrastructure.pusher_service.PusherService.get_client")
|
|
@patch("infrastructure.pusher_service.PusherService.is_available", return_value=True)
|
|
def test_broadcast_notification_success(self, _is_available, mock_get_client):
|
|
mock_client = Mock()
|
|
mock_get_client.return_value = mock_client
|
|
mock_client.trigger.return_value = {"message": "ok"}
|
|
|
|
ok = PusherService.broadcast_notification(
|
|
user_id=42,
|
|
title="Test naslov",
|
|
message="Test poruka",
|
|
level="info",
|
|
notification_id=123,
|
|
)
|
|
|
|
self.assertTrue(ok)
|
|
mock_client.trigger.assert_called_once()
|
|
_, kwargs = mock_client.trigger.call_args
|
|
self.assertEqual(kwargs["channels"], "private-user-42")
|
|
self.assertEqual(kwargs["event_name"], "vehicle.notification.created")
|
|
self.assertEqual(kwargs["data"]["id"], 123)
|
|
|
|
@patch("infrastructure.pusher_service.PusherService.get_client")
|
|
@patch("infrastructure.pusher_service.PusherService.is_available", return_value=True)
|
|
def test_authenticate_channel_allows_own_private_channel(self, _is_available, mock_get_client):
|
|
mock_client = Mock()
|
|
mock_get_client.return_value = mock_client
|
|
mock_client.authenticate.return_value = {"auth": "signed-token"}
|
|
|
|
auth_data = PusherService.authenticate_channel(
|
|
socket_id="123.456",
|
|
channel_name="private-user-7",
|
|
user_id=7,
|
|
)
|
|
|
|
self.assertEqual(auth_data, {"auth": "signed-token"})
|
|
mock_client.authenticate.assert_called_once_with(
|
|
channel="private-user-7",
|
|
socket_id="123.456",
|
|
)
|
|
|
|
@patch("infrastructure.pusher_service.PusherService.get_client")
|
|
@patch("infrastructure.pusher_service.PusherService.is_available", return_value=True)
|
|
def test_authenticate_channel_rejects_other_users_channel(self, _is_available, mock_get_client):
|
|
mock_client = Mock()
|
|
mock_get_client.return_value = mock_client
|
|
|
|
auth_data = PusherService.authenticate_channel(
|
|
socket_id="123.456",
|
|
channel_name="private-user-999",
|
|
user_id=7,
|
|
)
|
|
|
|
self.assertIsNone(auth_data)
|
|
mock_client.authenticate.assert_not_called()
|
|
|
|
@patch("infrastructure.pusher_service.PusherService.get_client")
|
|
@patch("infrastructure.pusher_service.PusherService.is_available", return_value=True)
|
|
def test_broadcast_notification_serializes_uuid_values(self, _is_available, mock_get_client):
|
|
mock_client = Mock()
|
|
mock_get_client.return_value = mock_client
|
|
mock_client.trigger.return_value = {"message": "ok"}
|
|
|
|
notification_id = uuid.uuid4()
|
|
metadata_uuid = uuid.uuid4()
|
|
|
|
ok = PusherService.broadcast_notification(
|
|
user_id=10,
|
|
title="UUID test",
|
|
message="UUID payload",
|
|
level="info",
|
|
notification_id=notification_id,
|
|
metadata={"entity_id": metadata_uuid},
|
|
)
|
|
|
|
self.assertTrue(ok)
|
|
_, kwargs = mock_client.trigger.call_args
|
|
self.assertEqual(kwargs["data"]["id"], str(notification_id))
|
|
self.assertEqual(kwargs["data"]["metadata"]["entity_id"], str(metadata_uuid)) |