Corp/backend/tests/realtime/test_calendar_updates.py

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()