1089 lines
37 KiB
Python
1089 lines
37 KiB
Python
# backend/tests/test_schedule_permissions.py
|
||
"""
|
||
Тесты для системы прав доступа календарей django-scheduler.
|
||
"""
|
||
|
||
from datetime import timedelta
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
from django.contrib.auth import get_user_model
|
||
from django.contrib.contenttypes.models import ContentType
|
||
from django.utils import timezone
|
||
from rest_framework import status
|
||
from rest_framework.test import APIClient
|
||
from schedule.models import (
|
||
Calendar,
|
||
CalendarRelation,
|
||
Event,
|
||
EventRelation,
|
||
Occurrence,
|
||
Rule,
|
||
)
|
||
|
||
from scheduling.models import CalendarBinding
|
||
|
||
User = get_user_model()
|
||
|
||
|
||
@pytest.fixture
|
||
def api_client():
|
||
"""API клиент для тестов."""
|
||
return APIClient()
|
||
|
||
|
||
@pytest.fixture
|
||
def user1(db):
|
||
"""Первый тестовый пользователь."""
|
||
return User.objects.create_user(
|
||
username="user1",
|
||
email="user1@example.com",
|
||
password="testpass123",
|
||
first_name="User",
|
||
last_name="One",
|
||
phone_number="+79991234567", # Правильное имя параметра
|
||
is_active=True,
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def user2(db):
|
||
"""Второй тестовый пользователь."""
|
||
return User.objects.create_user(
|
||
username="user2",
|
||
email="user2@example.com",
|
||
password="testpass123",
|
||
first_name="User",
|
||
last_name="Two",
|
||
phone_number="+79991234568",
|
||
is_active=True,
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def user3(db):
|
||
"""Третий тестовый пользователь."""
|
||
return User.objects.create_user(
|
||
username="user3",
|
||
email="user3@example.com",
|
||
password="testpass123",
|
||
first_name="User",
|
||
last_name="Three",
|
||
phone_number="+79991234569",
|
||
is_active=True,
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def admin_user(db):
|
||
"""Администратор для проверки scope календарей."""
|
||
return User.objects.create_user(
|
||
username="admin",
|
||
email="admin@example.com",
|
||
password="testpass123",
|
||
first_name="Admin",
|
||
last_name="User",
|
||
phone_number="+79991234570",
|
||
is_staff=True,
|
||
is_superuser=True,
|
||
email_verified=True,
|
||
)
|
||
|
||
|
||
@pytest.mark.django_db
|
||
@pytest.mark.parametrize(
|
||
("field", "handler_name"),
|
||
[
|
||
("notify_event_created", "notify_event_created"),
|
||
("notify_event_changed", "notify_event_changed"),
|
||
("notify_event_cancelled", "notify_event_cancelled"),
|
||
],
|
||
)
|
||
def test_disabled_calendar_notification_rule_stops_dispatch(
|
||
monkeypatch, field, handler_name
|
||
):
|
||
from scheduling.notifications import handlers
|
||
|
||
calendar = Calendar.objects.create(name="Silent", slug=f"silent-{field}")
|
||
CalendarBinding.objects.create(calendar=calendar, **{field: False})
|
||
event = SimpleNamespace(calendar=calendar)
|
||
monkeypatch.setattr(
|
||
handlers,
|
||
"get_event_recipients",
|
||
lambda _event: pytest.fail("Recipients must not be loaded for a disabled rule"),
|
||
)
|
||
|
||
getattr(handlers, handler_name)(event)
|
||
|
||
|
||
@pytest.mark.django_db
|
||
class TestCalendarPermissions:
|
||
"""Тесты прав доступа к календарям."""
|
||
|
||
def test_create_calendar_creates_owner_relation(self, api_client, user1):
|
||
"""При создании календаря автоматически создается owner relation."""
|
||
api_client.force_authenticate(user=user1)
|
||
|
||
response = api_client.post(
|
||
"/api/v1/schedule/calendars/",
|
||
{
|
||
"name": "Test Calendar",
|
||
},
|
||
)
|
||
|
||
assert response.status_code == status.HTTP_201_CREATED
|
||
calendar_id = response.data["id"]
|
||
|
||
# Проверяем, что создался CalendarRelation с owner
|
||
ct = ContentType.objects.get_for_model(User)
|
||
relation = CalendarRelation.objects.get(
|
||
calendar_id=calendar_id, content_type=ct, object_id=user1.id
|
||
)
|
||
|
||
assert relation.distinction == "owner"
|
||
assert relation.inheritable is True
|
||
|
||
def test_get_calendars_shows_only_accessible(self, api_client, user1, user2):
|
||
"""Пользователь видит только свои календари."""
|
||
api_client.force_authenticate(user=user1)
|
||
|
||
# user1 создает календарь
|
||
response = api_client.post(
|
||
"/api/v1/schedule/calendars/",
|
||
{
|
||
"name": "User1 Calendar",
|
||
},
|
||
)
|
||
assert response.status_code == status.HTTP_201_CREATED
|
||
|
||
# user2 создает свой календарь
|
||
api_client.force_authenticate(user=user2)
|
||
response = api_client.post(
|
||
"/api/v1/schedule/calendars/",
|
||
{
|
||
"name": "User2 Calendar",
|
||
},
|
||
)
|
||
assert response.status_code == status.HTTP_201_CREATED
|
||
|
||
# user1 видит только свой календарь
|
||
api_client.force_authenticate(user=user1)
|
||
response = api_client.get("/api/v1/schedule/calendars/")
|
||
assert response.status_code == status.HTTP_200_OK
|
||
|
||
if isinstance(response.data, dict) and "results" in response.data:
|
||
calendars = response.data["results"]
|
||
else:
|
||
calendars = response.data
|
||
|
||
assert len(calendars) == 1
|
||
assert calendars[0]["name"] == "User1 Calendar"
|
||
|
||
def test_owner_can_make_calendar_available_to_entire_company(
|
||
self, api_client, user1, user2
|
||
):
|
||
api_client.force_authenticate(user=user1)
|
||
create_response = api_client.post(
|
||
"/api/v1/schedule/calendars/",
|
||
{"name": "Company Calendar"},
|
||
)
|
||
calendar_id = create_response.data["id"]
|
||
|
||
api_client.force_authenticate(user=user2)
|
||
assert api_client.get("/api/v1/schedule/calendars/").data == []
|
||
|
||
api_client.force_authenticate(user=user1)
|
||
access_response = api_client.patch(
|
||
f"/api/v1/schedule/calendars/{calendar_id}/access-scope/",
|
||
{"access_scope": "all"},
|
||
)
|
||
assert access_response.status_code == status.HTTP_200_OK
|
||
assert access_response.data["access_scope"] == "all"
|
||
|
||
event_response = api_client.post(
|
||
"/api/v1/schedule/events/",
|
||
{
|
||
"title": "Company event",
|
||
"calendar": calendar_id,
|
||
"start": timezone.now().isoformat(),
|
||
"end": (timezone.now() + timedelta(hours=1)).isoformat(),
|
||
},
|
||
)
|
||
assert event_response.status_code == status.HTTP_201_CREATED
|
||
|
||
api_client.force_authenticate(user=user2)
|
||
list_response = api_client.get("/api/v1/schedule/calendars/")
|
||
assert [calendar["id"] for calendar in list_response.data] == [calendar_id]
|
||
assert list_response.data[0]["user_role"] == "viewer"
|
||
assert list_response.data[0]["can_create_events"] is False
|
||
events_response = api_client.get(
|
||
f"/api/v1/schedule/events/?calendar={calendar_id}"
|
||
)
|
||
assert [event["id"] for event in events_response.data["results"]] == [
|
||
event_response.data["id"]
|
||
]
|
||
|
||
forbidden_response = api_client.patch(
|
||
f"/api/v1/schedule/calendars/{calendar_id}/access-scope/",
|
||
{"access_scope": "restricted"},
|
||
)
|
||
assert forbidden_response.status_code == status.HTTP_403_FORBIDDEN
|
||
|
||
api_client.force_authenticate(user=user1)
|
||
restricted_response = api_client.patch(
|
||
f"/api/v1/schedule/calendars/{calendar_id}/access-scope/",
|
||
{"access_scope": "restricted"},
|
||
)
|
||
assert restricted_response.status_code == status.HTTP_200_OK
|
||
|
||
api_client.force_authenticate(user=user2)
|
||
assert api_client.get("/api/v1/schedule/calendars/").data == []
|
||
|
||
def test_add_participant_to_calendar(self, api_client, user1, user2):
|
||
"""Owner может добавить участника к календарю."""
|
||
api_client.force_authenticate(user=user1)
|
||
|
||
# Создаем календарь
|
||
response = api_client.post(
|
||
"/api/v1/schedule/calendars/",
|
||
{
|
||
"name": "Shared Calendar",
|
||
},
|
||
)
|
||
calendar_id = response.data["id"]
|
||
|
||
# Добавляем user2 как viewer
|
||
response = api_client.post(
|
||
f"/api/v1/schedule/calendars/{calendar_id}/add-participant/",
|
||
{"user_id": user2.id, "role": "viewer"},
|
||
)
|
||
|
||
assert response.status_code == status.HTTP_201_CREATED
|
||
assert response.data["user"]["id"] == user2.id
|
||
assert response.data["distinction"] == "viewer"
|
||
|
||
# Теперь user2 должен видеть календарь
|
||
api_client.force_authenticate(user=user2)
|
||
response = api_client.get("/api/v1/schedule/calendars/")
|
||
|
||
if isinstance(response.data, dict) and "results" in response.data:
|
||
calendars = response.data["results"]
|
||
else:
|
||
calendars = response.data
|
||
|
||
assert len(calendars) == 1
|
||
assert calendars[0]["id"] == calendar_id
|
||
|
||
def test_cannot_add_inactive_participant_to_calendar(
|
||
self, api_client, user1, user2
|
||
):
|
||
"""Уволенного сотрудника нельзя заново добавить в календарь."""
|
||
api_client.force_authenticate(user=user1)
|
||
calendar_response = api_client.post(
|
||
"/api/v1/schedule/calendars/",
|
||
{"name": "Active participants only"},
|
||
)
|
||
user2.is_active = False
|
||
user2.save(update_fields=["is_active"])
|
||
|
||
response = api_client.post(
|
||
f"/api/v1/schedule/calendars/{calendar_response.data['id']}/add-participant/",
|
||
{"user_id": user2.id, "role": "viewer"},
|
||
)
|
||
|
||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||
assert "user_id" in response.data
|
||
|
||
def test_non_owner_cannot_add_participant(self, api_client, user1, user2, user3):
|
||
"""Не-owner не может добавлять участников."""
|
||
api_client.force_authenticate(user=user1)
|
||
|
||
# user1 создает календарь
|
||
response = api_client.post(
|
||
"/api/v1/schedule/calendars/",
|
||
{
|
||
"name": "Test Calendar",
|
||
},
|
||
)
|
||
calendar_id = response.data["id"]
|
||
|
||
# Добавляем user2 как editor
|
||
api_client.post(
|
||
f"/api/v1/schedule/calendars/{calendar_id}/add-participant/",
|
||
{"user_id": user2.id, "role": "editor"},
|
||
)
|
||
|
||
# user2 пытается добавить user3 - должно быть запрещено
|
||
api_client.force_authenticate(user=user2)
|
||
response = api_client.post(
|
||
f"/api/v1/schedule/calendars/{calendar_id}/add-participant/",
|
||
{"user_id": user3.id, "role": "viewer"},
|
||
)
|
||
|
||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||
|
||
def test_remove_participant_from_calendar(self, api_client, user1, user2):
|
||
"""Owner может удалить участника из календаря."""
|
||
api_client.force_authenticate(user=user1)
|
||
|
||
# Создаем календарь и добавляем user2
|
||
response = api_client.post(
|
||
"/api/v1/schedule/calendars/",
|
||
{
|
||
"name": "Test Calendar",
|
||
},
|
||
)
|
||
calendar_id = response.data["id"]
|
||
|
||
api_client.post(
|
||
f"/api/v1/schedule/calendars/{calendar_id}/add-participant/",
|
||
{"user_id": user2.id, "role": "viewer"},
|
||
)
|
||
|
||
# Удаляем user2
|
||
response = api_client.delete(
|
||
f"/api/v1/schedule/calendars/{calendar_id}/remove-participant/{user2.id}/"
|
||
)
|
||
|
||
assert response.status_code == status.HTTP_204_NO_CONTENT
|
||
|
||
# user2 больше не видит календарь
|
||
api_client.force_authenticate(user=user2)
|
||
response = api_client.get("/api/v1/schedule/calendars/")
|
||
|
||
if isinstance(response.data, dict) and "results" in response.data:
|
||
calendars = response.data["results"]
|
||
else:
|
||
calendars = response.data
|
||
|
||
assert len(calendars) == 0
|
||
|
||
def test_cannot_remove_owner(self, api_client, user1):
|
||
"""Нельзя удалить owner из календаря."""
|
||
api_client.force_authenticate(user=user1)
|
||
|
||
# Создаем календарь
|
||
response = api_client.post(
|
||
"/api/v1/schedule/calendars/",
|
||
{
|
||
"name": "Test Calendar",
|
||
},
|
||
)
|
||
calendar_id = response.data["id"]
|
||
|
||
# Пытаемся удалить себя (owner)
|
||
response = api_client.delete(
|
||
f"/api/v1/schedule/calendars/{calendar_id}/remove-participant/{user1.id}/"
|
||
)
|
||
|
||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||
|
||
def test_list_participants(self, api_client, user1, user2, user3):
|
||
"""Получение списка участников календаря."""
|
||
api_client.force_authenticate(user=user1)
|
||
|
||
# Создаем календарь
|
||
response = api_client.post(
|
||
"/api/v1/schedule/calendars/",
|
||
{
|
||
"name": "Test Calendar",
|
||
},
|
||
)
|
||
calendar_id = response.data["id"]
|
||
|
||
# Добавляем участников
|
||
api_client.post(
|
||
f"/api/v1/schedule/calendars/{calendar_id}/add-participant/",
|
||
{"user_id": user2.id, "role": "editor"},
|
||
)
|
||
api_client.post(
|
||
f"/api/v1/schedule/calendars/{calendar_id}/add-participant/",
|
||
{"user_id": user3.id, "role": "viewer"},
|
||
)
|
||
|
||
# Получаем список участников
|
||
response = api_client.get(
|
||
f"/api/v1/schedule/calendars/{calendar_id}/participants/"
|
||
)
|
||
|
||
assert response.status_code == status.HTTP_200_OK
|
||
assert len(response.data) == 3 # owner + 2 участника
|
||
|
||
# Проверяем роли
|
||
roles = {p["user"]["id"]: p["distinction"] for p in response.data}
|
||
assert roles[user1.id] == "owner"
|
||
assert roles[user2.id] == "editor"
|
||
assert roles[user3.id] == "viewer"
|
||
|
||
def test_calendar_notification_settings_keep_defaults_and_allow_editor(
|
||
self, api_client, user1, user2, user3
|
||
):
|
||
api_client.force_authenticate(user=user1)
|
||
create_response = api_client.post(
|
||
"/api/v1/schedule/calendars/",
|
||
{"name": "Notification settings"},
|
||
)
|
||
calendar_id = create_response.data["id"]
|
||
api_client.post(
|
||
f"/api/v1/schedule/calendars/{calendar_id}/add-participant/",
|
||
{"user_id": user2.id, "role": "editor"},
|
||
)
|
||
api_client.post(
|
||
f"/api/v1/schedule/calendars/{calendar_id}/add-participant/",
|
||
{"user_id": user3.id, "role": "viewer"},
|
||
)
|
||
|
||
defaults_response = api_client.get(
|
||
f"/api/v1/schedule/calendars/{calendar_id}/notification-settings/"
|
||
)
|
||
assert defaults_response.status_code == status.HTTP_200_OK
|
||
assert defaults_response.data == {
|
||
"notify_event_created": True,
|
||
"notify_event_changed": True,
|
||
"notify_event_cancelled": True,
|
||
"can_manage": True,
|
||
}
|
||
|
||
api_client.force_authenticate(user=user2)
|
||
update_response = api_client.patch(
|
||
f"/api/v1/schedule/calendars/{calendar_id}/notification-settings/",
|
||
{"notify_event_changed": False},
|
||
format="json",
|
||
)
|
||
assert update_response.status_code == status.HTTP_200_OK
|
||
assert update_response.data["notify_event_changed"] is False
|
||
|
||
api_client.force_authenticate(user=user3)
|
||
forbidden_response = api_client.patch(
|
||
f"/api/v1/schedule/calendars/{calendar_id}/notification-settings/",
|
||
{"notify_event_created": False},
|
||
format="json",
|
||
)
|
||
assert forbidden_response.status_code == status.HTTP_403_FORBIDDEN
|
||
|
||
def test_owner_can_change_roles_but_cannot_remove_last_owner_role(
|
||
self, api_client, user1, user2
|
||
):
|
||
api_client.force_authenticate(user=user1)
|
||
create_response = api_client.post(
|
||
"/api/v1/schedule/calendars/",
|
||
{"name": "Role settings"},
|
||
)
|
||
calendar_id = create_response.data["id"]
|
||
api_client.post(
|
||
f"/api/v1/schedule/calendars/{calendar_id}/add-participant/",
|
||
{"user_id": user2.id, "role": "viewer"},
|
||
)
|
||
|
||
update_response = api_client.patch(
|
||
f"/api/v1/schedule/calendars/{calendar_id}/participant-role/{user2.id}/",
|
||
{"role": "editor"},
|
||
format="json",
|
||
)
|
||
assert update_response.status_code == status.HTTP_200_OK
|
||
assert update_response.data["distinction"] == "editor"
|
||
|
||
last_owner_response = api_client.patch(
|
||
f"/api/v1/schedule/calendars/{calendar_id}/participant-role/{user1.id}/",
|
||
{"role": "viewer"},
|
||
format="json",
|
||
)
|
||
assert last_owner_response.status_code == status.HTTP_400_BAD_REQUEST
|
||
assert "role" in last_owner_response.data
|
||
|
||
api_client.force_authenticate(user=user2)
|
||
forbidden_response = api_client.patch(
|
||
f"/api/v1/schedule/calendars/{calendar_id}/participant-role/{user1.id}/",
|
||
{"role": "viewer"},
|
||
format="json",
|
||
)
|
||
assert forbidden_response.status_code == status.HTTP_403_FORBIDDEN
|
||
|
||
def test_admin_list_scope_defaults_to_accessible_only(
|
||
self, api_client, admin_user, user1
|
||
):
|
||
"""По умолчанию админ видит свои доступные календари, а не весь реестр."""
|
||
api_client.force_authenticate(user=admin_user)
|
||
admin_calendar_response = api_client.post(
|
||
"/api/v1/schedule/calendars/", {"name": "Admin Calendar"}
|
||
)
|
||
assert admin_calendar_response.status_code == status.HTTP_201_CREATED
|
||
admin_calendar_id = admin_calendar_response.data["id"]
|
||
|
||
api_client.force_authenticate(user=user1)
|
||
other_calendar_response = api_client.post(
|
||
"/api/v1/schedule/calendars/", {"name": "Other Calendar"}
|
||
)
|
||
assert other_calendar_response.status_code == status.HTTP_201_CREATED
|
||
other_calendar_id = other_calendar_response.data["id"]
|
||
|
||
api_client.force_authenticate(user=admin_user)
|
||
default_response = api_client.get("/api/v1/schedule/calendars/")
|
||
default_calendars = default_response.data
|
||
assert {cal["id"] for cal in default_calendars} == {admin_calendar_id}
|
||
|
||
full_response = api_client.get("/api/v1/schedule/calendars/?scope=all")
|
||
full_calendars = full_response.data
|
||
assert {cal["id"] for cal in full_calendars} == {
|
||
admin_calendar_id,
|
||
other_calendar_id,
|
||
}
|
||
|
||
|
||
@pytest.mark.django_db
|
||
class TestRecurringEvents:
|
||
"""Тесты повторяющихся событий."""
|
||
|
||
def test_admin_event_scope_matches_calendar_scope(
|
||
self, api_client, admin_user, user1
|
||
):
|
||
"""У admin default list/occurrences не должны тянуть чужие календари."""
|
||
api_client.force_authenticate(user=admin_user)
|
||
admin_calendar_response = api_client.post(
|
||
"/api/v1/schedule/calendars/", {"name": "Admin Calendar"}
|
||
)
|
||
admin_calendar_id = admin_calendar_response.data["id"]
|
||
|
||
api_client.force_authenticate(user=user1)
|
||
other_calendar_response = api_client.post(
|
||
"/api/v1/schedule/calendars/", {"name": "Other Calendar"}
|
||
)
|
||
other_calendar_id = other_calendar_response.data["id"]
|
||
|
||
start = timezone.now().replace(microsecond=0, second=0)
|
||
range_end = start + timedelta(hours=2)
|
||
Event.objects.create(
|
||
calendar_id=admin_calendar_id,
|
||
title="Admin event",
|
||
start=start,
|
||
end=start + timedelta(hours=1),
|
||
creator=admin_user,
|
||
)
|
||
Event.objects.create(
|
||
calendar_id=other_calendar_id,
|
||
title="Other event",
|
||
start=start,
|
||
end=start + timedelta(hours=1),
|
||
creator=user1,
|
||
)
|
||
|
||
api_client.force_authenticate(user=admin_user)
|
||
default_events = api_client.get(
|
||
"/api/v1/schedule/events/",
|
||
{
|
||
"start": start.isoformat(),
|
||
"end": range_end.isoformat(),
|
||
},
|
||
)
|
||
assert default_events.status_code == status.HTTP_200_OK
|
||
default_events_list = (
|
||
default_events.data["results"]
|
||
if isinstance(default_events.data, dict)
|
||
else default_events.data
|
||
)
|
||
assert {event["title"] for event in default_events_list} == {"Admin event"}
|
||
|
||
all_events = api_client.get(
|
||
"/api/v1/schedule/events/",
|
||
{
|
||
"scope": "all",
|
||
"start": start.isoformat(),
|
||
"end": range_end.isoformat(),
|
||
},
|
||
)
|
||
assert all_events.status_code == status.HTTP_200_OK
|
||
all_events_list = (
|
||
all_events.data["results"]
|
||
if isinstance(all_events.data, dict)
|
||
else all_events.data
|
||
)
|
||
assert {event["title"] for event in all_events_list} == {
|
||
"Admin event",
|
||
"Other event",
|
||
}
|
||
|
||
default_occurrences = api_client.get(
|
||
"/api/v1/schedule/events/occurrences/",
|
||
{
|
||
"start": start.isoformat(),
|
||
"end": range_end.isoformat(),
|
||
},
|
||
)
|
||
assert default_occurrences.status_code == status.HTTP_200_OK
|
||
assert {event["title"] for event in default_occurrences.data} == {"Admin event"}
|
||
|
||
all_occurrences = api_client.get(
|
||
"/api/v1/schedule/events/occurrences/",
|
||
{
|
||
"scope": "all",
|
||
"start": start.isoformat(),
|
||
"end": range_end.isoformat(),
|
||
},
|
||
)
|
||
assert all_occurrences.status_code == status.HTTP_200_OK
|
||
assert {event["title"] for event in all_occurrences.data} == {
|
||
"Admin event",
|
||
"Other event",
|
||
}
|
||
|
||
def test_admin_accessible_scope_excludes_unrelated_department_occurrences(
|
||
self, api_client, admin_user, user1
|
||
):
|
||
"""Календарь отдела не становится доступным только из-за его типа."""
|
||
department_calendar = Calendar.objects.create(
|
||
name="Sales",
|
||
slug="department-sales",
|
||
)
|
||
CalendarBinding.objects.create(
|
||
calendar=department_calendar,
|
||
type=CalendarBinding.BindingType.DEPARTMENT,
|
||
)
|
||
rule = Rule.objects.create(
|
||
name="Department daily",
|
||
description="Department recurrence",
|
||
frequency="DAILY",
|
||
params="",
|
||
)
|
||
start = timezone.now().replace(microsecond=0, second=0)
|
||
range_end = start + timedelta(days=2)
|
||
Event.objects.create(
|
||
calendar=department_calendar,
|
||
title="Sales recurring event",
|
||
start=start,
|
||
end=start + timedelta(hours=1),
|
||
creator=user1,
|
||
rule=rule,
|
||
end_recurring_period=range_end,
|
||
)
|
||
|
||
api_client.force_authenticate(user=admin_user)
|
||
accessible_occurrences = api_client.get(
|
||
"/api/v1/schedule/events/occurrences/",
|
||
{
|
||
"start": start.isoformat(),
|
||
"end": range_end.isoformat(),
|
||
},
|
||
)
|
||
assert accessible_occurrences.status_code == status.HTTP_200_OK
|
||
assert accessible_occurrences.data == []
|
||
|
||
all_occurrences = api_client.get(
|
||
"/api/v1/schedule/events/occurrences/",
|
||
{
|
||
"scope": "all",
|
||
"start": start.isoformat(),
|
||
"end": range_end.isoformat(),
|
||
},
|
||
)
|
||
assert all_occurrences.status_code == status.HTTP_200_OK
|
||
assert {event["title"] for event in all_occurrences.data} == {
|
||
"Sales recurring event",
|
||
}
|
||
|
||
def test_create_rule(self, api_client, user1):
|
||
"""Создание правила повторения."""
|
||
api_client.force_authenticate(user=user1)
|
||
|
||
response = api_client.post(
|
||
"/api/v1/schedule/rules/",
|
||
{
|
||
"name": "Weekly",
|
||
"description": "Every week on Monday",
|
||
"frequency": "WEEKLY",
|
||
"params": "byweekday:0", # Monday
|
||
},
|
||
)
|
||
|
||
assert response.status_code == status.HTTP_201_CREATED
|
||
assert response.data["frequency"] == "WEEKLY"
|
||
assert response.data["frequency_display"] == "Еженедельно" # Django локализация
|
||
|
||
def test_create_recurring_event(self, api_client, user1):
|
||
"""Создание события с повторением."""
|
||
api_client.force_authenticate(user=user1)
|
||
|
||
# Создаем календарь
|
||
cal_response = api_client.post(
|
||
"/api/v1/schedule/calendars/",
|
||
{
|
||
"name": "Test Calendar",
|
||
},
|
||
)
|
||
calendar_id = cal_response.data["id"]
|
||
|
||
# Создаем правило
|
||
rule_response = api_client.post(
|
||
"/api/v1/schedule/rules/",
|
||
{
|
||
"name": "Daily",
|
||
"description": "Every day",
|
||
"frequency": "DAILY",
|
||
"params": "",
|
||
},
|
||
)
|
||
rule_id = rule_response.data["id"]
|
||
|
||
# Создаем событие с правилом
|
||
from datetime import timedelta
|
||
|
||
from django.utils import timezone
|
||
|
||
start = timezone.now()
|
||
end = start + timedelta(hours=1)
|
||
end_recurring = start + timedelta(days=30)
|
||
|
||
response = api_client.post(
|
||
"/api/v1/schedule/events/",
|
||
{
|
||
"title": "Daily Meeting",
|
||
"description": "Team standup",
|
||
"start": start.isoformat(),
|
||
"end": end.isoformat(),
|
||
"calendar": calendar_id,
|
||
"rule": rule_id,
|
||
"end_recurring_period": end_recurring.isoformat(),
|
||
},
|
||
)
|
||
|
||
assert response.status_code == status.HTTP_201_CREATED
|
||
assert response.data["rule"] == rule_id
|
||
assert response.data["end_recurring_period"] is not None
|
||
|
||
def test_get_occurrences(self, api_client, user1):
|
||
"""Получение материализованных вхождений для recurring события."""
|
||
api_client.force_authenticate(user=user1)
|
||
|
||
# Создаем календарь
|
||
cal_response = api_client.post(
|
||
"/api/v1/schedule/calendars/",
|
||
{
|
||
"name": "Test Calendar",
|
||
},
|
||
)
|
||
calendar_id = cal_response.data["id"]
|
||
|
||
# Создаем правило - еженедельно
|
||
rule_response = api_client.post(
|
||
"/api/v1/schedule/rules/",
|
||
{
|
||
"name": "Weekly",
|
||
"description": "Every week",
|
||
"frequency": "WEEKLY",
|
||
"params": "",
|
||
},
|
||
)
|
||
rule_id = rule_response.data["id"]
|
||
|
||
# Создаем событие
|
||
from datetime import timedelta
|
||
|
||
from django.utils import timezone
|
||
|
||
start = timezone.now()
|
||
end = start + timedelta(hours=1)
|
||
end_recurring = start + timedelta(weeks=4)
|
||
|
||
api_client.post(
|
||
"/api/v1/schedule/events/",
|
||
{
|
||
"title": "Weekly Meeting",
|
||
"start": start.isoformat(),
|
||
"end": end.isoformat(),
|
||
"calendar": calendar_id,
|
||
"rule": rule_id,
|
||
"end_recurring_period": end_recurring.isoformat(),
|
||
},
|
||
)
|
||
|
||
# Получаем occurrences
|
||
response = api_client.get(
|
||
"/api/v1/schedule/events/occurrences/",
|
||
{
|
||
"calendar": calendar_id,
|
||
"start": start.isoformat(),
|
||
"end": (start + timedelta(weeks=5)).isoformat(),
|
||
},
|
||
)
|
||
|
||
assert response.status_code == status.HTTP_200_OK
|
||
# Должно быть ~4 вхождения (еженедельно в течение 4 недель)
|
||
assert len(response.data) >= 4
|
||
|
||
|
||
@pytest.mark.django_db
|
||
class TestiCalendarExport:
|
||
"""Тесты экспорта в iCalendar формат."""
|
||
|
||
def test_export_calendar_to_ics(self, api_client, user1):
|
||
"""Экспорт календаря в .ics файл."""
|
||
api_client.force_authenticate(user=user1)
|
||
|
||
# Создаем календарь
|
||
cal_response = api_client.post(
|
||
"/api/v1/schedule/calendars/",
|
||
{
|
||
"name": "Test Calendar",
|
||
},
|
||
)
|
||
calendar_id = cal_response.data["id"]
|
||
|
||
# Добавляем событие
|
||
from datetime import timedelta
|
||
|
||
from django.utils import timezone
|
||
|
||
start = timezone.now()
|
||
end = start + timedelta(hours=2)
|
||
|
||
api_client.post(
|
||
"/api/v1/schedule/events/",
|
||
{
|
||
"title": "Test Event",
|
||
"description": "Test Description",
|
||
"start": start.isoformat(),
|
||
"end": end.isoformat(),
|
||
"calendar": calendar_id,
|
||
},
|
||
)
|
||
|
||
# Экспортируем календарь
|
||
response = api_client.get(
|
||
f"/api/v1/schedule/calendars/{calendar_id}/export-ical/"
|
||
)
|
||
|
||
assert response.status_code == status.HTTP_200_OK
|
||
assert response["Content-Type"] == "text/calendar"
|
||
assert "Content-Disposition" in response
|
||
assert ".ics" in response["Content-Disposition"]
|
||
|
||
# Проверяем содержимое
|
||
content = response.content.decode("utf-8")
|
||
assert "BEGIN:VCALENDAR" in content
|
||
assert "BEGIN:VEVENT" in content
|
||
assert "Test Event" in content
|
||
assert "END:VEVENT" in content
|
||
assert "END:VCALENDAR" in content
|
||
|
||
|
||
@pytest.mark.django_db
|
||
class TestEventRelations:
|
||
"""Тесты участников событий."""
|
||
|
||
def test_add_participant_to_event(self, api_client, user1, user2):
|
||
"""Добавление участника к событию."""
|
||
api_client.force_authenticate(user=user1)
|
||
|
||
# Создаем календарь
|
||
cal_response = api_client.post(
|
||
"/api/v1/schedule/calendars/",
|
||
{
|
||
"name": "Test Calendar",
|
||
},
|
||
)
|
||
calendar_id = cal_response.data["id"]
|
||
|
||
# Создаем событие
|
||
from datetime import timedelta
|
||
|
||
from django.utils import timezone
|
||
|
||
start = timezone.now()
|
||
end = start + timedelta(hours=1)
|
||
|
||
event_response = api_client.post(
|
||
"/api/v1/schedule/events/",
|
||
{
|
||
"title": "Meeting",
|
||
"start": start.isoformat(),
|
||
"end": end.isoformat(),
|
||
"calendar": calendar_id,
|
||
},
|
||
)
|
||
event_id = event_response.data["id"]
|
||
|
||
# Добавляем user2 как участника
|
||
ct = ContentType.objects.get_for_model(User)
|
||
response = api_client.post(
|
||
"/api/v1/schedule/relations/",
|
||
{
|
||
"event": event_id,
|
||
"content_type": ct.id,
|
||
"object_id": user2.id,
|
||
"distinction": "attendee",
|
||
},
|
||
)
|
||
|
||
assert response.status_code == status.HTTP_201_CREATED
|
||
assert response.data["event"] == event_id
|
||
|
||
def test_list_event_participants(self, api_client, user1, user2, user3):
|
||
"""Получение списка участников события."""
|
||
api_client.force_authenticate(user=user1)
|
||
|
||
# Создаем календарь и событие
|
||
cal_response = api_client.post(
|
||
"/api/v1/schedule/calendars/",
|
||
{
|
||
"name": "Test Calendar",
|
||
},
|
||
)
|
||
calendar_id = cal_response.data["id"]
|
||
|
||
from datetime import timedelta
|
||
|
||
from django.utils import timezone
|
||
|
||
start = timezone.now()
|
||
end = start + timedelta(hours=1)
|
||
|
||
event_response = api_client.post(
|
||
"/api/v1/schedule/events/",
|
||
{
|
||
"title": "Team Meeting",
|
||
"start": start.isoformat(),
|
||
"end": end.isoformat(),
|
||
"calendar": calendar_id,
|
||
},
|
||
)
|
||
event_id = event_response.data["id"]
|
||
|
||
# Добавляем участников
|
||
ct = ContentType.objects.get_for_model(User)
|
||
api_client.post(
|
||
"/api/v1/schedule/relations/",
|
||
{
|
||
"event": event_id,
|
||
"content_type": ct.id,
|
||
"object_id": user2.id,
|
||
"distinction": "attendee",
|
||
},
|
||
)
|
||
api_client.post(
|
||
"/api/v1/schedule/relations/",
|
||
{
|
||
"event": event_id,
|
||
"content_type": ct.id,
|
||
"object_id": user3.id,
|
||
"distinction": "attendee",
|
||
},
|
||
)
|
||
|
||
# Получаем список участников
|
||
response = api_client.get("/api/v1/schedule/relations/", {"event": event_id})
|
||
|
||
assert response.status_code == status.HTTP_200_OK
|
||
|
||
if isinstance(response.data, dict) and "results" in response.data:
|
||
relations = response.data["results"]
|
||
else:
|
||
relations = response.data
|
||
|
||
assert len(relations) == 2
|
||
|
||
|
||
@pytest.mark.django_db
|
||
class TestScheduleObjectBoundaries:
|
||
"""Низкоуровневые endpoints не обходят доступ к событию."""
|
||
|
||
def test_occurrences_relations_and_rules_respect_event_access(
|
||
self, api_client, user1, user2
|
||
):
|
||
calendar = Calendar.objects.create(
|
||
name="Private calendar",
|
||
slug="private-calendar",
|
||
)
|
||
user_ct = ContentType.objects.get_for_model(User)
|
||
CalendarRelation.objects.create(
|
||
calendar=calendar,
|
||
content_type=user_ct,
|
||
object_id=user1.id,
|
||
distinction="owner",
|
||
inheritable=True,
|
||
)
|
||
rule = Rule.objects.create(
|
||
name="Private daily",
|
||
description="Private recurrence",
|
||
frequency="DAILY",
|
||
params="",
|
||
)
|
||
start = timezone.now().replace(microsecond=0)
|
||
event = Event.objects.create(
|
||
calendar=calendar,
|
||
creator=user1,
|
||
rule=rule,
|
||
title="Private event",
|
||
start=start,
|
||
end=start + timedelta(hours=1),
|
||
end_recurring_period=start + timedelta(days=3),
|
||
)
|
||
occurrence = Occurrence.objects.create(
|
||
event=event,
|
||
title=event.title,
|
||
description="",
|
||
start=start,
|
||
end=start + timedelta(hours=1),
|
||
original_start=start,
|
||
original_end=start + timedelta(hours=1),
|
||
)
|
||
|
||
api_client.force_authenticate(user=user2)
|
||
|
||
occurrence_response = api_client.get("/api/v1/schedule/occurrences/")
|
||
assert occurrence_response.status_code == status.HTTP_200_OK
|
||
occurrence_results = (
|
||
occurrence_response.data.get("results", [])
|
||
if isinstance(occurrence_response.data, dict)
|
||
else occurrence_response.data
|
||
)
|
||
assert all(item["id"] != occurrence.id for item in occurrence_results)
|
||
|
||
relation_response = api_client.get(
|
||
"/api/v1/schedule/relations/", {"event": event.id}
|
||
)
|
||
assert relation_response.status_code == status.HTTP_200_OK
|
||
relation_results = (
|
||
relation_response.data.get("results", [])
|
||
if isinstance(relation_response.data, dict)
|
||
else relation_response.data
|
||
)
|
||
assert relation_results == []
|
||
|
||
forbidden_relation = api_client.post(
|
||
"/api/v1/schedule/relations/",
|
||
{
|
||
"event": event.id,
|
||
"content_type": user_ct.id,
|
||
"object_id": user2.id,
|
||
"distinction": "attendee",
|
||
},
|
||
)
|
||
assert forbidden_relation.status_code == status.HTTP_403_FORBIDDEN
|
||
|
||
forbidden_rule_update = api_client.patch(
|
||
f"/api/v1/schedule/rules/{rule.id}/",
|
||
{"description": "Leaked update"},
|
||
format="json",
|
||
)
|
||
assert forbidden_rule_update.status_code == status.HTTP_403_FORBIDDEN
|
||
|
||
EventRelation.objects.create(
|
||
event=event,
|
||
content_type=user_ct,
|
||
object_id=user2.id,
|
||
distinction="attendee",
|
||
)
|
||
|
||
assert (
|
||
api_client.get(f"/api/v1/schedule/events/{event.id}/").status_code
|
||
== status.HTTP_200_OK
|
||
)
|
||
visible_occurrences = api_client.get(
|
||
"/api/v1/schedule/occurrences/", {"event": event.id}
|
||
)
|
||
visible_results = (
|
||
visible_occurrences.data.get("results", [])
|
||
if isinstance(visible_occurrences.data, dict)
|
||
else visible_occurrences.data
|
||
)
|
||
assert any(item["id"] == occurrence.id for item in visible_results)
|