from datetime import timedelta from unittest.mock import patch import pytest from channels.db import database_sync_to_async from django.contrib.contenttypes.models import ContentType from django.utils import timezone from schedule.models import Calendar, CalendarRelation, Event, EventRelation, Rule from employees.models import Employee from scheduling.realtime import ( build_calendar_snapshot, get_calendar_recipient_ids, parse_calendar_subscription, send_calendar_update, ) pytestmark = pytest.mark.django_db def _calendar_with_owner(user): calendar = Calendar.objects.create(name="Realtime", slug=f"realtime-{user.id}") CalendarRelation.objects.create( calendar=calendar, content_type=ContentType.objects.get_for_model(Employee), object_id=user.id, distinction="owner", inheritable=True, ) return calendar def test_calendar_recipient_ids_include_calendar_event_and_admin_users(user, user2): admin = Employee.objects.create_user( email="calendar-admin@example.com", password="testpass123", phone_number="+79991234569", is_staff=True, ) Employee.objects.filter(id__in=[user.id, user2.id, admin.id]).update(is_active=True) user.refresh_from_db() user2.refresh_from_db() admin.refresh_from_db() calendar = _calendar_with_owner(user) event = Event.objects.create( calendar=calendar, title="Общая встреча", start=timezone.now(), end=timezone.now() + timedelta(hours=1), ) EventRelation.objects.create( event=event, content_type=ContentType.objects.get_for_model(Employee), object_id=user2.id, distinction="participant", ) assert get_calendar_recipient_ids(calendar, event=event) == { user.id, user2.id, admin.id, } def test_calendar_sender_uses_existing_user_groups(): delivered = [] class ChannelLayer: async def group_send(self, group, message): delivered.append((group, message)) with ( patch("scheduling.realtime.get_channel_layer", return_value=ChannelLayer()), patch( "scheduling.realtime.transaction.on_commit", side_effect=lambda callback: callback(), ), ): send_calendar_update( action="updated", entity="event", calendar_id=7, event_id=42, recipient_ids={10, 11}, changed_fields=["start"], ) assert {group for group, _ in delivered} == {"user_10", "user_11"} assert delivered[0][1]["type"] == "calendar_update" assert delivered[0][1]["data"] == { "action": "updated", "entity": "event", "calendar_id": 7, "event_id": 42, "changed_fields": ["start"], } def test_event_lifecycle_emits_calendar_invalidation(user): calendar = _calendar_with_owner(user) with patch("scheduling.realtime_signals.send_calendar_update") as send_update: event = Event.objects.create( calendar=calendar, title="Создание", start=timezone.now(), end=timezone.now() + timedelta(hours=1), ) assert send_update.call_args.kwargs["action"] == "created" send_update.reset_mock() event.title = "Изменение" event.save(update_fields=["title"]) assert send_update.call_args.kwargs["action"] == "updated" assert send_update.call_args.kwargs["event_id"] == event.id send_update.reset_mock() event_id = event.id event.delete() assert send_update.call_args.kwargs["action"] == "deleted" assert send_update.call_args.kwargs["event_id"] == event_id def test_calendar_snapshot_contains_visible_regular_event(user): calendar = _calendar_with_owner(user) event = Event.objects.create( calendar=calendar, title="WebSocket snapshot", start=timezone.now(), end=timezone.now() + timedelta(hours=1), ) subscription = parse_calendar_subscription( { "subscription_id": "calendar-main", "start": (timezone.now() - timedelta(days=1)).isoformat(), "end": (timezone.now() + timedelta(days=1)).isoformat(), "calendar_id": calendar.id, "scope": "accessible", }, user=user, ) snapshot = build_calendar_snapshot(user=user, subscription=subscription) assert snapshot["subscription_id"] == "calendar-main" assert snapshot["calendar_id"] == calendar.id assert snapshot["events"] == [ { "id": event.id, "title": "WebSocket snapshot", "description": "", "start": event.start.isoformat(), "end": event.end.isoformat(), "calendar": calendar.id, "calendar_name": calendar.name, "color_event": None, "rule": None, "end_recurring_period": None, "can_edit": True, "can_delete": True, "is_recurring": False, } ] def test_calendar_snapshot_expands_recurring_events_and_enforces_access(user, user2): calendar = _calendar_with_owner(user) rule = Rule.objects.create( name="Daily realtime", description="Two occurrences", frequency="DAILY", params="count:2", ) event = Event.objects.create( calendar=calendar, title="Повтор", start=timezone.now(), end=timezone.now() + timedelta(hours=1), rule=rule, ) content = { "subscription_id": "calendar-recurring", "start": (timezone.now() - timedelta(days=1)).isoformat(), "end": (timezone.now() + timedelta(days=3)).isoformat(), "scope": "accessible", } owner_subscription = parse_calendar_subscription(content, user=user) owner_snapshot = build_calendar_snapshot( user=user, subscription=owner_subscription, ) assert len(owner_snapshot["events"]) == 2 assert {item["event_id"] for item in owner_snapshot["events"]} == {event.id} assert all(item["is_recurring"] for item in owner_snapshot["events"]) outsider_subscription = parse_calendar_subscription(content, user=user2) outsider_snapshot = build_calendar_snapshot( user=user2, subscription=outsider_subscription, ) assert outsider_snapshot["events"] == [] @pytest.mark.asyncio @pytest.mark.django_db(transaction=True) async def test_consumer_serves_calendar_snapshot(ws_communicator, user): @database_sync_to_async def create_event(): calendar = _calendar_with_owner(user) event = Event.objects.create( calendar=calendar, title="Realtime range", start=timezone.now(), end=timezone.now() + timedelta(hours=1), ) return calendar.id, event.id calendar_id, event_id = await create_event() communicator = await ws_communicator(user=user) connected, _ = await communicator.connect() assert connected await communicator.send_json_to( { "action": "calendar_subscribe", "subscription_id": "calendar-test", "start": (timezone.now() - timedelta(days=1)).isoformat(), "end": (timezone.now() + timedelta(days=1)).isoformat(), "calendar_id": calendar_id, "scope": "accessible", } ) response = await communicator.receive_json_from(timeout=5) assert response["type"] == "calendar_snapshot" assert response["reason"] == "subscribed" assert response["data"]["subscription_id"] == "calendar-test" assert [event["id"] for event in response["data"]["events"]] == [event_id] @database_sync_to_async def rename_event(): event = Event.objects.get(pk=event_id) event.title = "Realtime range updated" with patch("scheduling.realtime_signals.send_calendar_update"): event.save(update_fields=["title"]) await rename_event() from channels.layers import get_channel_layer await get_channel_layer().group_send( f"user_{user.id}", { "type": "calendar_update", "event": "event_updated", "data": { "action": "updated", "entity": "event", "calendar_id": calendar_id, "event_id": event_id, }, }, ) update = await communicator.receive_json_from(timeout=5) refreshed = await communicator.receive_json_from(timeout=5) assert update["type"] == "calendar_update" assert refreshed["type"] == "calendar_snapshot" assert refreshed["reason"] == "updated" assert refreshed["data"]["events"][0]["title"] == "Realtime range updated" await communicator.disconnect() @pytest.mark.asyncio @pytest.mark.django_db(transaction=True) async def test_consumer_forwards_calendar_update(ws_communicator, user): communicator = await ws_communicator(user=user) connected, _ = await communicator.connect() assert connected from channels.layers import get_channel_layer await get_channel_layer().group_send( f"user_{user.id}", { "type": "calendar_update", "event": "event_updated", "data": { "action": "updated", "entity": "event", "calendar_id": 7, "event_id": 42, }, }, ) response = await communicator.receive_json_from(timeout=5) assert response == { "type": "calendar_update", "event": "event_updated", "data": { "action": "updated", "entity": "event", "calendar_id": 7, "event_id": 42, }, } await communicator.disconnect()