studio/sbbb
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

🐛(data) remove Calendar and CalendarShare models

Browse Source 
The only source of truth for those is now in the caldav server
This commit is contained in:
Sylvain Zimmer
2026-02-09 20:48:11 +01:00
parent 3a0f64e791
commit 3051100f8a
24 changed files with 408 additions and 883 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/backend/core/admin.py
View File

@@ -93,16 +93,6 @@ class UserAdmin(auth_admin.UserAdmin):
search_fields = ("id", "sub", "admin_email", "email", "full_name")
@admin.register(models.Calendar)
class CalendarAdmin(admin.ModelAdmin):
"""Admin class for Calendar model."""
list_display = ("name", "owner", "is_default", "is_visible", "created_at")
list_filter = ("is_default", "is_visible")
search_fields = ("name", "owner__email", "caldav_path")
readonly_fields = ("id", "created_at", "updated_at")
@admin.register(models.CalendarSubscriptionToken)
class CalendarSubscriptionTokenAdmin(admin.ModelAdmin):
"""Admin class for CalendarSubscriptionToken model."""

47
src/backend/core/api/serializers.py
View File

@@ -114,49 +114,12 @@ class UserMeSerializer(UserSerializer):
read_only_fields = UserSerializer.Meta.read_only_fields
# CalDAV serializers
class CalendarSerializer(serializers.ModelSerializer):
"""Serializer for Calendar model."""
class CalendarCreateSerializer(serializers.Serializer): # pylint: disable=abstract-method
"""Serializer for creating a Calendar (CalDAV only, no Django model)."""
class Meta:
model = models.Calendar
fields = [
"id",
"name",
"color",
"description",
"is_default",
"is_visible",
"caldav_path",
"created_at",
"updated_at",
]
read_only_fields = [
"id",
"is_default",
"caldav_path",
"created_at",
"updated_at",
]
class CalendarCreateSerializer(serializers.ModelSerializer):
"""Serializer for creating a Calendar."""
class Meta:
model = models.Calendar
fields = ["name", "color", "description"]
class CalendarShareSerializer(serializers.ModelSerializer):
"""Serializer for CalendarShare model."""
shared_with_email = serializers.EmailField(write_only=True)
class Meta:
model = models.CalendarShare
fields = ["id", "shared_with_email", "permission", "is_visible", "created_at"]
read_only_fields = ["id", "created_at"]
name = serializers.CharField(max_length=255)
color = serializers.CharField(max_length=7, required=False, default="#3174ad")
description = serializers.CharField(required=False, default="")
class CalendarSubscriptionTokenSerializer(serializers.ModelSerializer):

249
src/backend/core/api/viewsets.py
View File

@@ -8,12 +8,11 @@ from urllib.parse import unquote
from django.conf import settings
from django.core.cache import cache
from django.db import models as db
from django.utils.text import slugify
import rest_framework as drf
from rest_framework import mixins, status, viewsets
from rest_framework import response as drf_response
from rest_framework import status, viewsets
from rest_framework.decorators import action
from rest_framework.parsers import MultiPartParser
from rest_framework.permissions import AllowAny, IsAuthenticated
@@ -262,149 +261,105 @@ class ConfigView(drf.views.APIView):
return theme_customization
# CalDAV ViewSets
class CalendarViewSet(
mixins.ListModelMixin,
mixins.RetrieveModelMixin,
mixins.CreateModelMixin,
mixins.UpdateModelMixin,
mixins.DestroyModelMixin,
viewsets.GenericViewSet,
):
"""
ViewSet for managing user calendars.
# Regex for CalDAV path validation (shared with SubscriptionTokenViewSet)
# Pattern: /calendars/<email-or-encoded>/<calendar-id>/
CALDAV_PATH_PATTERN = re.compile(
r"^/calendars/[^/]+/[a-zA-Z0-9-]+/$",
)
list: Get all calendars accessible by the user (owned + shared)
retrieve: Get a specific calendar
create: Create a new calendar
update: Update calendar properties
destroy: Delete a calendar
def _verify_caldav_access(user, caldav_path):
"""Verify that the user has access to the CalDAV calendar.
Checks that:
1. The path matches the expected pattern (prevents path injection)
2. The user's email matches the email in the path
"""
if not CALDAV_PATH_PATTERN.match(caldav_path):
return False
parts = caldav_path.strip("/").split("/")
if len(parts) >= 2 and parts[0] == "calendars":
path_email = unquote(parts[1])
return path_email.lower() == user.email.lower()
return False
def _normalize_caldav_path(caldav_path):
"""Normalize CalDAV path to consistent format.
Strips the CalDAV API prefix (e.g. /api/v1.0/caldav/) if present,
so that paths like /api/v1.0/caldav/calendars/user@ex.com/uuid/
become /calendars/user@ex.com/uuid/.
"""
if not caldav_path.startswith("/"):
caldav_path = "/" + caldav_path
# Strip CalDAV API prefix — keep from /calendars/ onwards
calendars_idx = caldav_path.find("/calendars/")
if calendars_idx > 0:
caldav_path = caldav_path[calendars_idx:]
if not caldav_path.endswith("/"):
caldav_path = caldav_path + "/"
return caldav_path
class CalendarViewSet(viewsets.GenericViewSet):
"""ViewSet for calendar operations.
create: Create a new calendar (CalDAV only, no Django record).
import_events: Import events from an ICS file.
"""
permission_classes = [IsAuthenticated]
serializer_class = serializers.CalendarSerializer
serializer_class = serializers.CalendarCreateSerializer
def get_queryset(self):
"""Return calendars owned by or shared with the current user."""
user = self.request.user
shared_ids = models.CalendarShare.objects.filter(shared_with=user).values_list(
"calendar_id", flat=True
)
return (
models.Calendar.objects.filter(db.Q(owner=user) | db.Q(id__in=shared_ids))
.distinct()
.order_by("-is_default", "name")
)
def create(self, request):
"""Create a new calendar via CalDAV.
def get_serializer_class(self):
if self.action == "create":
return serializers.CalendarCreateSerializer
return serializers.CalendarSerializer
POST /api/v1.0/calendars/
Body: { name, color?, description? }
Returns: { caldav_path }
"""
serializer = serializers.CalendarCreateSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
def perform_create(self, serializer):
"""Create a new calendar via CalendarService."""
service = CalendarService()
calendar = service.create_calendar(
user=self.request.user,
caldav_path = service.create_calendar(
user=request.user,
name=serializer.validated_data["name"],
color=serializer.validated_data.get("color", "#3174ad"),
)
# Update the serializer instance with the created calendar
serializer.instance = calendar
def perform_destroy(self, instance):
"""Delete calendar. Prevent deletion of default calendar."""
if instance.is_default:
raise ValueError("Cannot delete the default calendar.")
if instance.owner != self.request.user:
raise PermissionError("You can only delete your own calendars.")
instance.delete()
@action(detail=True, methods=["patch"])
def toggle_visibility(self, request, **kwargs):
"""Toggle calendar visibility."""
calendar = self.get_object()
# Check if it's a shared calendar
share = models.CalendarShare.objects.filter(
calendar=calendar, shared_with=request.user
).first()
if share:
share.is_visible = not share.is_visible
share.save()
is_visible = share.is_visible
elif calendar.owner == request.user:
calendar.is_visible = not calendar.is_visible
calendar.save()
is_visible = calendar.is_visible
else:
return drf_response.Response(
{"error": "Permission denied"}, status=status.HTTP_403_FORBIDDEN
)
return drf_response.Response({"is_visible": is_visible})
@action(
detail=True,
methods=["post"],
serializer_class=serializers.CalendarShareSerializer,
)
def share(self, request, **kwargs):
"""Share calendar with another user."""
calendar = self.get_object()
if calendar.owner != request.user:
return drf_response.Response(
{"error": "Only the owner can share this calendar"},
status=status.HTTP_403_FORBIDDEN,
)
serializer = serializers.CalendarShareSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
email = serializer.validated_data["shared_with_email"]
try:
user_to_share = models.User.objects.get(email=email)
except models.User.DoesNotExist:
return drf_response.Response(
{"error": "User not found"}, status=status.HTTP_404_NOT_FOUND
)
share, created = models.CalendarShare.objects.get_or_create(
calendar=calendar,
shared_with=user_to_share,
defaults={
"permission": serializer.validated_data.get("permission", "read")
},
)
if not created:
share.permission = serializer.validated_data.get(
"permission", share.permission
)
share.save()
return drf_response.Response(
serializers.CalendarShareSerializer(share).data,
status=status.HTTP_201_CREATED if created else status.HTTP_200_OK,
{"caldav_path": caldav_path},
status=status.HTTP_201_CREATED,
)
@action(
detail=True,
detail=False,
methods=["post"],
parser_classes=[MultiPartParser],
url_path="import_events",
url_path="import-events",
url_name="import-events",
)
def import_events(self, request, **kwargs):
"""Import events from an ICS file into this calendar."""
calendar = self.get_object()
"""Import events from an ICS file into a calendar.
# Only the owner can import events
if calendar.owner != request.user:
POST /api/v1.0/calendars/import-events/
Body (multipart): file=<ics>, caldav_path=/calendars/user@.../uuid/
"""
caldav_path = request.data.get("caldav_path", "")
if not caldav_path:
return drf_response.Response(
{"error": "Only the owner can import events"},
{"error": "caldav_path is required"},
status=status.HTTP_400_BAD_REQUEST,
)
caldav_path = _normalize_caldav_path(caldav_path)
# Verify user access
if not _verify_caldav_access(request.user, caldav_path):
return drf_response.Response(
{"error": "You don't have access to this calendar"},
status=status.HTTP_403_FORBIDDEN,
)
@@ -426,7 +381,7 @@ class CalendarViewSet(
ics_data = uploaded_file.read()
service = ICSImportService()
result = service.import_events(request.user, calendar, ics_data)
result = service.import_events(request.user, caldav_path, ics_data)
response_data = {
"total_events": result.total_events,
@@ -457,48 +412,6 @@ class SubscriptionTokenViewSet(viewsets.GenericViewSet):
permission_classes = [IsAuthenticated]
serializer_class = serializers.CalendarSubscriptionTokenSerializer
# Regex for CalDAV path validation
# Pattern: /calendars/<email>/<calendar-id>/
# Calendar ID: alphanumeric with hyphens only (prevents path traversal like ../)
# This blocks injection attacks while allowing UUIDs and test identifiers
CALDAV_PATH_PATTERN = re.compile(
r"^/calendars/[^/]+/[a-zA-Z0-9-]+/$",
)
def _verify_caldav_access(self, user, caldav_path):
"""
Verify that the user has access to the CalDAV calendar.
We verify by checking:
1. The path matches the expected pattern (prevents path injection)
2. The user's email matches the email in the path
CalDAV paths follow the pattern: /calendars/<user_email>/<calendar_id>/
"""
# Format validation to prevent path injection attacks (e.g., ../, query params)
if not self.CALDAV_PATH_PATTERN.match(caldav_path):
logger.warning(
"Invalid CalDAV path format rejected: %s",
caldav_path[:100], # Truncate for logging
)
return False
# Extract and verify email from path
# Path format: /calendars/user@example.com/calendar-id/
parts = caldav_path.strip("/").split("/")
if len(parts) >= 2 and parts[0] == "calendars":
path_email = unquote(parts[1])
return path_email.lower() == user.email.lower()
return False
def _normalize_caldav_path(self, caldav_path):
"""Normalize CalDAV path to consistent format."""
if not caldav_path.startswith("/"):
caldav_path = "/" + caldav_path
if not caldav_path.endswith("/"):
caldav_path = caldav_path + "/"
return caldav_path
def create(self, request):
"""
Create or get existing subscription token.
@@ -516,7 +429,7 @@ class SubscriptionTokenViewSet(viewsets.GenericViewSet):
calendar_name = create_serializer.validated_data.get("calendar_name", "")
# Verify user has access to this calendar
if not self._verify_caldav_access(request.user, caldav_path):
if not _verify_caldav_access(request.user, caldav_path):
return drf_response.Response(
{"error": "You don't have access to this calendar"},
status=status.HTTP_403_FORBIDDEN,
@@ -555,10 +468,10 @@ class SubscriptionTokenViewSet(viewsets.GenericViewSet):
status=status.HTTP_400_BAD_REQUEST,
)
caldav_path = self._normalize_caldav_path(caldav_path)
caldav_path = _normalize_caldav_path(caldav_path)
# Verify user has access to this calendar
if not self._verify_caldav_access(request.user, caldav_path):
if not _verify_caldav_access(request.user, caldav_path):
return drf_response.Response(
{"error": "You don't have access to this calendar"},
status=status.HTTP_403_FORBIDDEN,

17
src/backend/core/factories.py
View File

@@ -28,23 +28,6 @@ class UserFactory(factory.django.DjangoModelFactory):
password = make_password("password")
class CalendarFactory(factory.django.DjangoModelFactory):
"""A factory to create calendars for testing purposes."""
class Meta:
model = models.Calendar
owner = factory.SubFactory(UserFactory)
name = factory.Faker("sentence", nb_words=3)
color = factory.Faker("hex_color")
description = factory.Faker("paragraph")
is_default = False
is_visible = True
caldav_path = factory.LazyAttribute(
lambda obj: f"/calendars/{obj.owner.email}/{fake.uuid4()}"
)
class CalendarSubscriptionTokenFactory(factory.django.DjangoModelFactory):
"""A factory to create calendar subscription tokens for testing purposes."""

19
src/backend/core/migrations/0004_remove_calendarshare_calendar_and_more.py Normal file
View File

@@ -0,0 +1,19 @@
# Generated by Django 5.2.9 on 2026-02-09 18:23
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
('core', '0003_calendarsubscriptiontoken_token_active_idx'),
]
operations = [
migrations.DeleteModel(
name='CalendarShare',
),
migrations.DeleteModel(
name='Calendar',
),
]

102
src/backend/core/models.py
View File

@@ -1,27 +1,16 @@
"""
Declare and configure the models for the calendars core application
"""
# pylint: disable=too-many-lines
import uuid
from datetime import timedelta
from logging import getLogger
from django.conf import settings
from django.contrib.auth import models as auth_models
from django.contrib.auth.base_user import AbstractBaseUser
from django.contrib.postgres.indexes import GistIndex
from django.contrib.sites.models import Site
from django.core import mail, validators
from django.core.cache import cache
from django.core.exceptions import ValidationError
from django.core.mail import send_mail
from django.db import models, transaction
from django.db.models.expressions import RawSQL
from django.template.loader import render_to_string
from django.utils import timezone
from django.db import models
from django.utils.functional import cached_property
from django.utils.translation import get_language, override
from django.utils.translation import gettext_lazy as _
from timezone_field import TimeZoneField
@@ -225,14 +214,6 @@ class User(AbstractBaseUser, BaseModel, auth_models.PermissionsMixin):
def __str__(self):
return self.email or self.admin_email or str(self.id)
def save(self, *args, **kwargs):
"""
If it's a new user, give its user access to the items to which s.he was invited.
"""
is_adding = self._state.adding
super().save(*args, **kwargs)
def email_user(self, subject, message, from_email=None, **kwargs):
"""Email this user."""
if not self.email:
@@ -321,87 +302,6 @@ class BaseAccess(BaseModel):
}
class Calendar(models.Model):
"""
Represents a calendar owned by a user.
This model tracks calendars stored in the CalDAV server and links them to Django users.
"""
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
owner = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.CASCADE,
related_name="calendars",
)
name = models.CharField(max_length=255)
color = models.CharField(max_length=7, default="#3174ad") # Hex color
description = models.TextField(blank=True, default="")
is_default = models.BooleanField(default=False)
is_visible = models.BooleanField(default=True)
# CalDAV server reference - the calendar path in the CalDAV server
caldav_path = models.CharField(max_length=512, unique=True)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
"""Meta options for Calendar model."""
ordering = ["-is_default", "name"]
constraints = [
models.UniqueConstraint(
fields=["owner"],
condition=models.Q(is_default=True),
name="unique_default_calendar_per_user",
)
]
def __str__(self):
return f"{self.name} ({self.owner.email})"
class CalendarShare(models.Model):
"""
Represents a calendar shared with another user.
"""
PERMISSION_READ = "read"
PERMISSION_WRITE = "write"
PERMISSION_CHOICES = [
(PERMISSION_READ, "Read only"),
(PERMISSION_WRITE, "Read and write"),
]
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
calendar = models.ForeignKey(
Calendar,
on_delete=models.CASCADE,
related_name="shares",
)
shared_with = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.CASCADE,
related_name="shared_calendars",
)
permission = models.CharField(
max_length=10,
choices=PERMISSION_CHOICES,
default=PERMISSION_READ,
)
is_visible = models.BooleanField(default=True)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
"""Meta options for CalendarShare model."""
unique_together = ["calendar", "shared_with"]
def __str__(self):
return f"{self.calendar.name} shared with {self.shared_with.email}"
class CalendarSubscriptionToken(models.Model):
"""
Stores subscription tokens for iCal export.

74
src/backend/core/services/caldav_service.py
View File

@@ -10,7 +10,6 @@ from django.utils import timezone
from caldav import DAVClient
from caldav.lib.error import NotFoundError
from core.models import Calendar
logger = logging.getLogger(__name__)
@@ -385,72 +384,33 @@ class CalendarService:
def __init__(self):
self.caldav = CalDAVClient()
def create_default_calendar(self, user) -> Calendar:
"""
Create a default calendar for a user.
"""
def create_default_calendar(self, user) -> str:
"""Create a default calendar for a user. Returns the caldav_path."""
calendar_id = str(uuid4())
calendar_name = "Mon calendrier"
return self.caldav.create_calendar(user, calendar_name, calendar_id)
# Create calendar in CalDAV server
caldav_path = self.caldav.create_calendar(user, calendar_name, calendar_id)
# Create local Calendar record
calendar = Calendar.objects.create(
owner=user,
name=calendar_name,
caldav_path=caldav_path,
is_default=True,
color="#3174ad",
)
return calendar
def create_calendar(self, user, name: str, color: str = "#3174ad") -> Calendar:
"""
Create a new calendar for a user.
"""
def create_calendar( # pylint: disable=unused-argument
self, user, name: str, color: str = "#3174ad"
) -> str:
"""Create a new calendar for a user. Returns the caldav_path."""
calendar_id = str(uuid4())
return self.caldav.create_calendar(user, name, calendar_id)
# Create calendar in CalDAV server
caldav_path = self.caldav.create_calendar(user, name, calendar_id)
def get_events(self, user, caldav_path: str, start=None, end=None) -> list:
"""Get events from a calendar. Returns parsed event data."""
return self.caldav.get_events(user, caldav_path, start, end)
# Create local Calendar record
calendar = Calendar.objects.create(
owner=user,
name=name,
caldav_path=caldav_path,
is_default=False,
color=color,
)
return calendar
def get_user_calendars(self, user):
"""
Get all calendars accessible by a user (owned + shared).
"""
owned = Calendar.objects.filter(owner=user)
shared = Calendar.objects.filter(shares__shared_with=user)
return owned.union(shared)
def get_events(self, user, calendar: Calendar, start=None, end=None) -> list:
"""
Get events from a calendar.
Returns parsed event data.
"""
return self.caldav.get_events(user, calendar.caldav_path, start, end)
def create_event(self, user, calendar: Calendar, event_data: dict) -> str:
def create_event(self, user, caldav_path: str, event_data: dict) -> str:
"""Create a new event."""
return self.caldav.create_event(user, calendar.caldav_path, event_data)
return self.caldav.create_event(user, caldav_path, event_data)
def update_event(
self, user, calendar: Calendar, event_uid: str, event_data: dict
self, user, caldav_path: str, event_uid: str, event_data: dict
) -> None:
"""Update an existing event."""
self.caldav.update_event(user, calendar.caldav_path, event_uid, event_data)
self.caldav.update_event(user, caldav_path, event_uid, event_data)
def delete_event(self, user, calendar: Calendar, event_uid: str) -> None:
def delete_event(self, user, caldav_path: str, event_uid: str) -> None:
"""Delete an event."""
self.caldav.delete_event(user, calendar.caldav_path, event_uid)
self.caldav.delete_event(user, caldav_path, event_uid)

16
src/backend/core/services/import_service.py
View File

@@ -38,18 +38,26 @@ class ICSImportService:
def __init__(self):
self.base_url = settings.CALDAV_URL.rstrip("/")
def import_events(self, user, calendar, ics_data: bytes) -> ImportResult:
def import_events(self, user, caldav_path: str, ics_data: bytes) -> ImportResult:
"""Import events from ICS data into a calendar.
Sends the raw ICS bytes to SabreDAV's ?import endpoint which
handles all ICS parsing, splitting by UID, VALARM repair, and
per-event insertion.
Args:
user: The authenticated user performing the import.
caldav_path: CalDAV path of the calendar
(e.g. /calendars/user@example.com/uuid/).
ics_data: Raw ICS file content.
"""
result = ImportResult()
# caldav_path already includes the base URI prefix
# e.g. /api/v1.0/caldav/calendars/user@example.com/uuid/
url = f"{self.base_url}{calendar.caldav_path}?import"
# Ensure caldav_path includes the base URI prefix that SabreDAV expects
base_uri = "/api/v1.0/caldav/"
if not caldav_path.startswith(base_uri):
caldav_path = base_uri.rstrip("/") + caldav_path
url = f"{self.base_url}{caldav_path}?import"
outbound_api_key = settings.CALDAV_OUTBOUND_API_KEY
if not outbound_api_key:

4
src/backend/core/signals.py
View File

@@ -23,10 +23,6 @@ def provision_default_calendar(sender, instance, created, **kwargs): # pylint:
if not created:
return
# Check if user already has a default calendar
if instance.calendars.filter(is_default=True).exists():
return
# Skip calendar creation if CalDAV server is not configured
if not settings.CALDAV_URL:
return

2
src/backend/core/tests/authentication/test_backends.py
View File

@@ -500,7 +500,7 @@ def test_authentication_session_tokens(
status=200,
)
with django_assert_num_queries(7):
with django_assert_num_queries(5):
user = klass.authenticate(
request,
code="test-code",

4
src/backend/core/tests/test_caldav_scheduling.py
View File

@@ -95,7 +95,7 @@ class TestCalDAVScheduling:
# Create calendar for organizer
service = CalendarService()
calendar = service.create_calendar(
caldav_path = service.create_calendar(
organizer, name="Test Calendar", color="#ff0000"
)
@@ -126,7 +126,7 @@ class TestCalDAVScheduling:
try:
# Create an event with an attendee
client = service.caldav._get_client(organizer) # pylint: disable=protected-access
calendar_url = f"{settings.CALDAV_URL}{calendar.caldav_path}"
calendar_url = f"{settings.CALDAV_URL}{caldav_path}"
# Add custom callback URL header to the client
# The CalDAV server will use this URL for the callback

14
src/backend/core/tests/test_caldav_service.py
View File

@@ -59,12 +59,10 @@ class TestCalDAVClient:
user = factories.UserFactory(email="test@example.com")
service = CalendarService()
# Create a calendar
calendar = service.create_calendar(user, name="My Calendar", color="#ff0000")
# Create a calendar — returns caldav_path string
caldav_path = service.create_calendar(user, name="My Calendar", color="#ff0000")
# Verify calendar was created
assert calendar is not None
assert calendar.owner == user
assert calendar.name == "My Calendar"
assert calendar.color == "#ff0000"
assert calendar.caldav_path is not None
# Verify caldav_path was returned
assert caldav_path is not None
assert isinstance(caldav_path, str)
assert "calendars/" in caldav_path

263
src/backend/core/tests/test_import_events.py
View File

@@ -1,6 +1,7 @@
"""Tests for the ICS import events feature.""" # pylint: disable=too-many-lines
import json
import uuid
from datetime import datetime
from datetime import timezone as dt_tz
from unittest.mock import MagicMock, patch
@@ -244,6 +245,11 @@ END:VEVENT
END:VCALENDAR"""
def _make_caldav_path(user):
"""Build a caldav_path string for a user (test helper)."""
return f"/calendars/{user.email}/{uuid.uuid4()}/"
def _make_sabredav_response( # noqa: PLR0913 # pylint: disable=too-many-arguments,too-many-positional-arguments
status_code=200,
total_events=0,
@@ -278,10 +284,10 @@ class TestICSImportService:
)
user = factories.UserFactory()
calendar = factories.CalendarFactory(owner=user)
caldav_path = _make_caldav_path(user)
service = ICSImportService()
result = service.import_events(user, calendar, ICS_SINGLE_EVENT)
result = service.import_events(user, caldav_path, ICS_SINGLE_EVENT)
assert result.total_events == 1
assert result.imported_count == 1
@@ -301,10 +307,10 @@ class TestICSImportService:
)
user = factories.UserFactory()
calendar = factories.CalendarFactory(owner=user)
caldav_path = _make_caldav_path(user)
service = ICSImportService()
result = service.import_events(user, calendar, ICS_MULTIPLE_EVENTS)
result = service.import_events(user, caldav_path, ICS_MULTIPLE_EVENTS)
assert result.total_events == 3
assert result.imported_count == 3
@@ -321,10 +327,10 @@ class TestICSImportService:
)
user = factories.UserFactory()
calendar = factories.CalendarFactory(owner=user)
caldav_path = _make_caldav_path(user)
service = ICSImportService()
result = service.import_events(user, calendar, ICS_EMPTY)
result = service.import_events(user, caldav_path, ICS_EMPTY)
assert result.total_events == 0
assert result.imported_count == 0
@@ -340,10 +346,10 @@ class TestICSImportService:
mock_post.return_value.text = '{"error": "Failed to parse ICS file"}'
user = factories.UserFactory()
calendar = factories.CalendarFactory(owner=user)
caldav_path = _make_caldav_path(user)
service = ICSImportService()
result = service.import_events(user, calendar, ICS_INVALID)
result = service.import_events(user, caldav_path, ICS_INVALID)
assert result.imported_count == 0
assert len(result.errors) >= 1
@@ -356,10 +362,10 @@ class TestICSImportService:
)
user = factories.UserFactory()
calendar = factories.CalendarFactory(owner=user)
caldav_path = _make_caldav_path(user)
service = ICSImportService()
result = service.import_events(user, calendar, ICS_WITH_TIMEZONE)
result = service.import_events(user, caldav_path, ICS_WITH_TIMEZONE)
assert result.total_events == 1
assert result.imported_count == 1
@@ -386,10 +392,10 @@ class TestICSImportService:
)
user = factories.UserFactory()
calendar = factories.CalendarFactory(owner=user)
caldav_path = _make_caldav_path(user)
service = ICSImportService()
result = service.import_events(user, calendar, ICS_MULTIPLE_EVENTS)
result = service.import_events(user, caldav_path, ICS_MULTIPLE_EVENTS)
assert result.total_events == 3
assert result.imported_count == 2
@@ -406,10 +412,10 @@ class TestICSImportService:
)
user = factories.UserFactory()
calendar = factories.CalendarFactory(owner=user)
caldav_path = _make_caldav_path(user)
service = ICSImportService()
result = service.import_events(user, calendar, ICS_ALL_DAY_EVENT)
result = service.import_events(user, caldav_path, ICS_ALL_DAY_EVENT)
assert result.total_events == 1
assert result.imported_count == 1
@@ -422,10 +428,10 @@ class TestICSImportService:
)
user = factories.UserFactory()
calendar = factories.CalendarFactory(owner=user)
caldav_path = _make_caldav_path(user)
service = ICSImportService()
result = service.import_events(user, calendar, ICS_VALARM_NO_ACTION)
result = service.import_events(user, caldav_path, ICS_VALARM_NO_ACTION)
assert result.total_events == 1
assert result.imported_count == 1
@@ -438,10 +444,10 @@ class TestICSImportService:
)
user = factories.UserFactory()
calendar = factories.CalendarFactory(owner=user)
caldav_path = _make_caldav_path(user)
service = ICSImportService()
result = service.import_events(user, calendar, ICS_RECURRING_WITH_EXCEPTION)
result = service.import_events(user, caldav_path, ICS_RECURRING_WITH_EXCEPTION)
# Two VEVENTs with same UID = one logical event
assert result.total_events == 1
@@ -464,10 +470,10 @@ class TestICSImportService:
)
user = factories.UserFactory()
calendar = factories.CalendarFactory(owner=user)
caldav_path = _make_caldav_path(user)
service = ICSImportService()
result = service.import_events(user, calendar, ICS_NO_DTSTART)
result = service.import_events(user, caldav_path, ICS_NO_DTSTART)
assert result.total_events == 1
assert result.imported_count == 0
@@ -476,20 +482,20 @@ class TestICSImportService:
@patch("core.services.import_service.requests.post")
def test_import_passes_calendar_path(self, mock_post):
"""The import URL should include the calendar's caldav_path."""
"""The import URL should include the caldav_path."""
mock_post.return_value = _make_sabredav_response(
total_events=1, imported_count=1
)
user = factories.UserFactory()
calendar = factories.CalendarFactory(owner=user)
caldav_path = _make_caldav_path(user)
service = ICSImportService()
service.import_events(user, calendar, ICS_SINGLE_EVENT)
service.import_events(user, caldav_path, ICS_SINGLE_EVENT)
call_args = mock_post.call_args
url = call_args.args[0] if call_args.args else call_args.kwargs.get("url", "")
assert calendar.caldav_path in url
assert caldav_path in url
assert "?import" in url
@patch("core.services.import_service.requests.post")
@@ -500,10 +506,10 @@ class TestICSImportService:
)
user = factories.UserFactory()
calendar = factories.CalendarFactory(owner=user)
caldav_path = _make_caldav_path(user)
service = ICSImportService()
service.import_events(user, calendar, ICS_SINGLE_EVENT)
service.import_events(user, caldav_path, ICS_SINGLE_EVENT)
call_kwargs = mock_post.call_args.kwargs
headers = call_kwargs["headers"]
@@ -524,10 +530,10 @@ class TestICSImportService:
)
user = factories.UserFactory()
calendar = factories.CalendarFactory(owner=user)
caldav_path = _make_caldav_path(user)
service = ICSImportService()
result = service.import_events(user, calendar, ICS_MULTIPLE_EVENTS)
result = service.import_events(user, caldav_path, ICS_MULTIPLE_EVENTS)
assert result.total_events == 3
assert result.imported_count == 1
@@ -541,10 +547,10 @@ class TestICSImportService:
mock_post.side_effect = req.ConnectionError("Connection refused")
user = factories.UserFactory()
calendar = factories.CalendarFactory(owner=user)
caldav_path = _make_caldav_path(user)
service = ICSImportService()
result = service.import_events(user, calendar, ICS_SINGLE_EVENT)
result = service.import_events(user, caldav_path, ICS_SINGLE_EVENT)
assert result.imported_count == 0
assert len(result.errors) >= 1
@@ -553,23 +559,19 @@ class TestICSImportService:
class TestImportEventsAPI:
"""API endpoint tests for the import_events action."""
def _get_url(self, calendar_id):
return f"/api/v1.0/calendars/{calendar_id}/import_events/"
IMPORT_URL = "/api/v1.0/calendars/import-events/"
def test_import_events_requires_authentication(self):
"""Unauthenticated requests should be rejected."""
calendar = factories.CalendarFactory()
client = APIClient()
response = client.post(self._get_url(calendar.id))
response = client.post(self.IMPORT_URL)
assert response.status_code == 401
def test_import_events_forbidden_for_non_owner(self):
"""Non-owners should not be able to access the calendar."""
owner = factories.UserFactory()
other_user = factories.UserFactory()
calendar = factories.CalendarFactory(owner=owner)
def test_import_events_forbidden_for_wrong_user(self):
"""Users cannot import to a calendar they don't own."""
owner = factories.UserFactory(email="owner@example.com")
other_user = factories.UserFactory(email="other@example.com")
caldav_path = f"/calendars/{owner.email}/some-uuid/"
client = APIClient()
client.force_login(other_user)
@@ -578,29 +580,45 @@ class TestImportEventsAPI:
"events.ics", ICS_SINGLE_EVENT, content_type="text/calendar"
)
response = client.post(
self._get_url(calendar.id), {"file": ics_file}, format="multipart"
self.IMPORT_URL,
{"file": ics_file, "caldav_path": caldav_path},
format="multipart",
)
assert response.status_code == 403
# Calendar not in queryset for non-owner, so 404 (not 403)
assert response.status_code == 404
def test_import_events_missing_caldav_path(self):
"""Request without caldav_path should return 400."""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
ics_file = SimpleUploadedFile(
"events.ics", ICS_SINGLE_EVENT, content_type="text/calendar"
)
response = client.post(self.IMPORT_URL, {"file": ics_file}, format="multipart")
assert response.status_code == 400
assert "caldav_path" in response.json()["error"]
def test_import_events_missing_file(self):
"""Request without a file should return 400."""
user = factories.UserFactory()
calendar = factories.CalendarFactory(owner=user)
user = factories.UserFactory(email="nofile@example.com")
caldav_path = f"/calendars/{user.email}/some-uuid/"
client = APIClient()
client.force_login(user)
response = client.post(self._get_url(calendar.id), format="multipart")
response = client.post(
self.IMPORT_URL,
{"caldav_path": caldav_path},
format="multipart",
)
assert response.status_code == 400
assert "No file provided" in response.json()["error"]
def test_import_events_file_too_large(self):
"""Files exceeding MAX_FILE_SIZE should be rejected."""
user = factories.UserFactory()
calendar = factories.CalendarFactory(owner=user)
user = factories.UserFactory(email="largefile@example.com")
caldav_path = f"/calendars/{user.email}/some-uuid/"
client = APIClient()
client.force_login(user)
@@ -611,9 +629,10 @@ class TestImportEventsAPI:
content_type="text/calendar",
)
response = client.post(
self._get_url(calendar.id), {"file": large_file}, format="multipart"
self.IMPORT_URL,
{"file": large_file, "caldav_path": caldav_path},
format="multipart",
)
assert response.status_code == 400
assert "too large" in response.json()["error"]
@@ -628,8 +647,8 @@ class TestImportEventsAPI:
errors=[],
)
user = factories.UserFactory()
calendar = factories.CalendarFactory(owner=user)
user = factories.UserFactory(email="success@example.com")
caldav_path = f"/calendars/{user.email}/some-uuid/"
client = APIClient()
client.force_login(user)
@@ -638,7 +657,9 @@ class TestImportEventsAPI:
"events.ics", ICS_MULTIPLE_EVENTS, content_type="text/calendar"
)
response = client.post(
self._get_url(calendar.id), {"file": ics_file}, format="multipart"
self.IMPORT_URL,
{"file": ics_file, "caldav_path": caldav_path},
format="multipart",
)
assert response.status_code == 200
@@ -659,8 +680,8 @@ class TestImportEventsAPI:
errors=["Planning session"],
)
user = factories.UserFactory()
calendar = factories.CalendarFactory(owner=user)
user = factories.UserFactory(email="partial@example.com")
caldav_path = f"/calendars/{user.email}/some-uuid/"
client = APIClient()
client.force_login(user)
@@ -669,7 +690,9 @@ class TestImportEventsAPI:
"events.ics", ICS_MULTIPLE_EVENTS, content_type="text/calendar"
)
response = client.post(
self._get_url(calendar.id), {"file": ics_file}, format="multipart"
self.IMPORT_URL,
{"file": ics_file, "caldav_path": caldav_path},
format="multipart",
)
assert response.status_code == 200
@@ -688,17 +711,17 @@ class TestImportEventsE2E:
"""End-to-end tests that import ICS events through the real SabreDAV server."""
def _create_calendar(self, user):
"""Create a real calendar in both Django and SabreDAV."""
"""Create a real calendar in SabreDAV. Returns the caldav_path."""
service = CalendarService()
return service.create_calendar(user, name="Import Test", color="#3174ad")
def test_import_single_event_e2e(self):
"""Import a single event and verify it exists in SabreDAV."""
user = factories.UserFactory(email="import-single@example.com")
calendar = self._create_calendar(user)
caldav_path = self._create_calendar(user)
import_service = ICSImportService()
result = import_service.import_events(user, calendar, ICS_SINGLE_EVENT)
result = import_service.import_events(user, caldav_path, ICS_SINGLE_EVENT)
assert result.total_events == 1
assert result.imported_count == 1
@@ -709,7 +732,7 @@ class TestImportEventsE2E:
caldav = CalDAVClient()
events = caldav.get_events(
user,
calendar.caldav_path,
caldav_path,
start=datetime(2026, 2, 10, tzinfo=dt_tz.utc),
end=datetime(2026, 2, 11, tzinfo=dt_tz.utc),
)
@@ -720,10 +743,10 @@ class TestImportEventsE2E:
def test_import_multiple_events_e2e(self):
"""Import multiple events and verify they all exist in SabreDAV."""
user = factories.UserFactory(email="import-multi@example.com")
calendar = self._create_calendar(user)
caldav_path = self._create_calendar(user)
import_service = ICSImportService()
result = import_service.import_events(user, calendar, ICS_MULTIPLE_EVENTS)
result = import_service.import_events(user, caldav_path, ICS_MULTIPLE_EVENTS)
assert result.total_events == 3
assert result.imported_count == 3
@@ -733,7 +756,7 @@ class TestImportEventsE2E:
caldav = CalDAVClient()
events = caldav.get_events(
user,
calendar.caldav_path,
caldav_path,
start=datetime(2026, 2, 10, tzinfo=dt_tz.utc),
end=datetime(2026, 2, 12, tzinfo=dt_tz.utc),
)
@@ -744,10 +767,10 @@ class TestImportEventsE2E:
def test_import_all_day_event_e2e(self):
"""Import an all-day event and verify it exists in SabreDAV."""
user = factories.UserFactory(email="import-allday@example.com")
calendar = self._create_calendar(user)
caldav_path = self._create_calendar(user)
import_service = ICSImportService()
result = import_service.import_events(user, calendar, ICS_ALL_DAY_EVENT)
result = import_service.import_events(user, caldav_path, ICS_ALL_DAY_EVENT)
assert result.total_events == 1
assert result.imported_count == 1
@@ -757,7 +780,7 @@ class TestImportEventsE2E:
caldav = CalDAVClient()
events = caldav.get_events(
user,
calendar.caldav_path,
caldav_path,
start=datetime(2026, 2, 14, tzinfo=dt_tz.utc),
end=datetime(2026, 2, 17, tzinfo=dt_tz.utc),
)
@@ -767,10 +790,10 @@ class TestImportEventsE2E:
def test_import_with_timezone_e2e(self):
"""Import an event with timezone info and verify it in SabreDAV."""
user = factories.UserFactory(email="import-tz@example.com")
calendar = self._create_calendar(user)
caldav_path = self._create_calendar(user)
import_service = ICSImportService()
result = import_service.import_events(user, calendar, ICS_WITH_TIMEZONE)
result = import_service.import_events(user, caldav_path, ICS_WITH_TIMEZONE)
assert result.total_events == 1
assert result.imported_count == 1
@@ -780,7 +803,7 @@ class TestImportEventsE2E:
caldav = CalDAVClient()
events = caldav.get_events(
user,
calendar.caldav_path,
caldav_path,
start=datetime(2026, 2, 10, tzinfo=dt_tz.utc),
end=datetime(2026, 2, 11, tzinfo=dt_tz.utc),
)
@@ -790,7 +813,7 @@ class TestImportEventsE2E:
def test_import_via_api_e2e(self):
"""Import events via the API endpoint hitting real SabreDAV."""
user = factories.UserFactory(email="import-api@example.com")
calendar = self._create_calendar(user)
caldav_path = self._create_calendar(user)
client = APIClient()
client.force_login(user)
@@ -799,8 +822,8 @@ class TestImportEventsE2E:
"events.ics", ICS_MULTIPLE_EVENTS, content_type="text/calendar"
)
response = client.post(
f"/api/v1.0/calendars/{calendar.id}/import_events/",
{"file": ics_file},
"/api/v1.0/calendars/import-events/",
{"file": ics_file, "caldav_path": caldav_path},
format="multipart",
)
@@ -814,7 +837,7 @@ class TestImportEventsE2E:
caldav = CalDAVClient()
events = caldav.get_events(
user,
calendar.caldav_path,
caldav_path,
start=datetime(2026, 2, 10, tzinfo=dt_tz.utc),
end=datetime(2026, 2, 12, tzinfo=dt_tz.utc),
)
@@ -828,11 +851,11 @@ class TestImportEventsE2E:
plugin used the wrong callback signature for that event.
"""
user = factories.UserFactory(email="import-attendee@example.com")
calendar = self._create_calendar(user)
caldav_path = self._create_calendar(user)
# Import event with attendees
import_service = ICSImportService()
result = import_service.import_events(user, calendar, ICS_WITH_ATTENDEES)
result = import_service.import_events(user, caldav_path, ICS_WITH_ATTENDEES)
assert result.total_events == 1
assert result.imported_count == 1
@@ -842,7 +865,7 @@ class TestImportEventsE2E:
caldav = CalDAVClient()
caldav.update_event(
user,
calendar.caldav_path,
caldav_path,
"attendee-event-1",
{"title": "Updated review meeting"},
)
@@ -850,7 +873,7 @@ class TestImportEventsE2E:
# Verify update was applied
events = caldav.get_events(
user,
calendar.caldav_path,
caldav_path,
start=datetime(2026, 2, 10, tzinfo=dt_tz.utc),
end=datetime(2026, 2, 11, tzinfo=dt_tz.utc),
)
@@ -865,11 +888,11 @@ class TestImportEventsE2E:
binds values as PARAM_STR instead of PARAM_LOB.
"""
user = factories.UserFactory(email="import-escapes@example.com")
calendar = self._create_calendar(user)
caldav_path = self._create_calendar(user)
import_service = ICSImportService()
result = import_service.import_events(
user, calendar, ICS_WITH_NEWLINES_IN_DESCRIPTION
user, caldav_path, ICS_WITH_NEWLINES_IN_DESCRIPTION
)
assert result.total_events == 1
@@ -880,7 +903,7 @@ class TestImportEventsE2E:
caldav = CalDAVClient()
events = caldav.get_events(
user,
calendar.caldav_path,
caldav_path,
start=datetime(2026, 2, 10, tzinfo=dt_tz.utc),
end=datetime(2026, 2, 11, tzinfo=dt_tz.utc),
)
@@ -890,17 +913,17 @@ class TestImportEventsE2E:
def test_import_same_file_twice_no_duplicates_e2e(self):
"""Importing the same ICS file twice should not create duplicates."""
user = factories.UserFactory(email="import-dedup@example.com")
calendar = self._create_calendar(user)
caldav_path = self._create_calendar(user)
import_service = ICSImportService()
# First import
result1 = import_service.import_events(user, calendar, ICS_MULTIPLE_EVENTS)
result1 = import_service.import_events(user, caldav_path, ICS_MULTIPLE_EVENTS)
assert result1.imported_count == 3
assert not result1.errors
# Second import of the same file — all should be duplicates
result2 = import_service.import_events(user, calendar, ICS_MULTIPLE_EVENTS)
result2 = import_service.import_events(user, caldav_path, ICS_MULTIPLE_EVENTS)
assert result2.duplicate_count == 3
assert result2.imported_count == 0
assert result2.skipped_count == 0
@@ -909,7 +932,7 @@ class TestImportEventsE2E:
caldav = CalDAVClient()
events = caldav.get_events(
user,
calendar.caldav_path,
caldav_path,
start=datetime(2026, 2, 10, tzinfo=dt_tz.utc),
end=datetime(2026, 2, 12, tzinfo=dt_tz.utc),
)
@@ -918,21 +941,21 @@ class TestImportEventsE2E:
def test_import_dead_recurring_event_skipped_silently_e2e(self):
"""A recurring event whose EXDATE excludes all instances is skipped, not an error."""
user = factories.UserFactory(email="import-dead-recur@example.com")
calendar = self._create_calendar(user)
caldav_path = self._create_calendar(user)
import_service = ICSImportService()
result = import_service.import_events(user, calendar, ICS_DEAD_RECURRING)
result = import_service.import_events(user, caldav_path, ICS_DEAD_RECURRING)
assert result.total_events == 1
assert result.imported_count == 0
assert result.skipped_count == 1
assert not result.errors
def _get_raw_event(self, user, calendar, uid):
def _get_raw_event(self, user, caldav_path, uid):
"""Fetch the raw ICS data of a single event from SabreDAV by UID."""
caldav_client = CalDAVClient()
client = caldav_client._get_client(user) # pylint: disable=protected-access
cal_url = f"{caldav_client.base_url}{calendar.caldav_path}"
cal_url = f"{caldav_client.base_url}{caldav_path}"
cal = client.calendar(url=cal_url)
event = cal.event_by_uid(uid)
return event.data
@@ -940,11 +963,11 @@ class TestImportEventsE2E:
def test_import_strips_binary_attachments_e2e(self):
"""Binary attachments should be stripped during import."""
user = factories.UserFactory(email="import-strip-attach@example.com")
calendar = self._create_calendar(user)
caldav_path = self._create_calendar(user)
import_service = ICSImportService()
result = import_service.import_events(
user, calendar, ICS_WITH_BINARY_ATTACHMENT
user, caldav_path, ICS_WITH_BINARY_ATTACHMENT
)
assert result.total_events == 1
@@ -952,7 +975,7 @@ class TestImportEventsE2E:
assert not result.errors
# Verify event exists and binary attachment was stripped
raw = self._get_raw_event(user, calendar, "attach-binary-1")
raw = self._get_raw_event(user, caldav_path, "attach-binary-1")
assert "Event with inline attachment" in raw
assert "iVBORw0KGgo" not in raw
assert "ATTACH" not in raw
@@ -960,28 +983,30 @@ class TestImportEventsE2E:
def test_import_keeps_url_attachments_e2e(self):
"""URL-based attachments should NOT be stripped during import."""
user = factories.UserFactory(email="import-keep-url-attach@example.com")
calendar = self._create_calendar(user)
caldav_path = self._create_calendar(user)
import_service = ICSImportService()
result = import_service.import_events(user, calendar, ICS_WITH_URL_ATTACHMENT)
result = import_service.import_events(
user, caldav_path, ICS_WITH_URL_ATTACHMENT
)
assert result.total_events == 1
assert result.imported_count == 1
assert not result.errors
# Verify URL attachment is preserved in raw ICS
raw = self._get_raw_event(user, calendar, "attach-url-1")
raw = self._get_raw_event(user, caldav_path, "attach-url-1")
assert "https://example.com/doc.pdf" in raw
assert "ATTACH" in raw
def test_import_truncates_large_description_e2e(self):
"""Descriptions exceeding IMPORT_MAX_DESCRIPTION_BYTES should be truncated."""
user = factories.UserFactory(email="import-trunc-desc@example.com")
calendar = self._create_calendar(user)
caldav_path = self._create_calendar(user)
import_service = ICSImportService()
result = import_service.import_events(
user, calendar, ICS_WITH_LARGE_DESCRIPTION
user, caldav_path, ICS_WITH_LARGE_DESCRIPTION
)
assert result.total_events == 1
@@ -989,7 +1014,7 @@ class TestImportEventsE2E:
assert not result.errors
# Verify description was truncated (default 100KB limit, original 200KB)
raw = self._get_raw_event(user, calendar, "large-desc-1")
raw = self._get_raw_event(user, caldav_path, "large-desc-1")
assert "Event with huge description" in raw
# Raw ICS should be much smaller than the 200KB original
assert len(raw) < 150000
@@ -1005,15 +1030,15 @@ class TestCalendarSanitizerE2E:
"""E2E tests for CalendarSanitizerPlugin on normal CalDAV PUT operations."""
def _create_calendar(self, user):
"""Create a real calendar in both Django and SabreDAV."""
"""Create a real calendar in SabreDAV. Returns the caldav_path."""
service = CalendarService()
return service.create_calendar(user, name="Sanitizer Test", color="#3174ad")
def _get_raw_event(self, user, calendar, uid):
def _get_raw_event(self, user, caldav_path, uid):
"""Fetch the raw ICS data of a single event from SabreDAV by UID."""
caldav_client = CalDAVClient()
client = caldav_client._get_client(user) # pylint: disable=protected-access
cal_url = f"{caldav_client.base_url}{calendar.caldav_path}"
cal_url = f"{caldav_client.base_url}{caldav_path}"
cal = client.calendar(url=cal_url)
event = cal.event_by_uid(uid)
return event.data
@@ -1021,14 +1046,12 @@ class TestCalendarSanitizerE2E:
def test_caldav_put_strips_binary_attachment_e2e(self):
"""A normal CalDAV PUT with binary attachment should be sanitized."""
user = factories.UserFactory(email="sanitizer-put-attach@example.com")
calendar = self._create_calendar(user)
caldav_path = self._create_calendar(user)
caldav = CalDAVClient()
caldav.create_event_raw(
user, calendar.caldav_path, ICS_WITH_BINARY_ATTACHMENT.decode()
)
caldav.create_event_raw(user, caldav_path, ICS_WITH_BINARY_ATTACHMENT.decode())
raw = self._get_raw_event(user, calendar, "attach-binary-1")
raw = self._get_raw_event(user, caldav_path, "attach-binary-1")
assert "Event with inline attachment" in raw
assert "iVBORw0KGgo" not in raw
assert "ATTACH" not in raw
@@ -1036,28 +1059,24 @@ class TestCalendarSanitizerE2E:
def test_caldav_put_keeps_url_attachment_e2e(self):
"""A normal CalDAV PUT with URL attachment should preserve it."""
user = factories.UserFactory(email="sanitizer-put-url@example.com")
calendar = self._create_calendar(user)
caldav_path = self._create_calendar(user)
caldav = CalDAVClient()
caldav.create_event_raw(
user, calendar.caldav_path, ICS_WITH_URL_ATTACHMENT.decode()
)
caldav.create_event_raw(user, caldav_path, ICS_WITH_URL_ATTACHMENT.decode())
raw = self._get_raw_event(user, calendar, "attach-url-1")
raw = self._get_raw_event(user, caldav_path, "attach-url-1")
assert "https://example.com/doc.pdf" in raw
assert "ATTACH" in raw
def test_caldav_put_truncates_large_description_e2e(self):
"""A normal CalDAV PUT with oversized description should be truncated."""
user = factories.UserFactory(email="sanitizer-put-desc@example.com")
calendar = self._create_calendar(user)
caldav_path = self._create_calendar(user)
caldav = CalDAVClient()
caldav.create_event_raw(
user, calendar.caldav_path, ICS_WITH_LARGE_DESCRIPTION.decode()
)
caldav.create_event_raw(user, caldav_path, ICS_WITH_LARGE_DESCRIPTION.decode())
raw = self._get_raw_event(user, calendar, "large-desc-1")
raw = self._get_raw_event(user, caldav_path, "large-desc-1")
assert "Event with huge description" in raw
assert len(raw) < 150000
assert "..." in raw
@@ -1065,23 +1084,21 @@ class TestCalendarSanitizerE2E:
def test_caldav_put_rejects_oversized_event_e2e(self):
"""A CalDAV PUT exceeding max-resource-size should be rejected (HTTP 507)."""
user = factories.UserFactory(email="sanitizer-put-oversize@example.com")
calendar = self._create_calendar(user)
caldav_path = self._create_calendar(user)
caldav = CalDAVClient()
with pytest.raises(Exception) as exc_info:
caldav.create_event_raw(
user, calendar.caldav_path, ICS_OVERSIZED_EVENT.decode()
)
caldav.create_event_raw(user, caldav_path, ICS_OVERSIZED_EVENT.decode())
# SabreDAV returns 507 Insufficient Storage
assert "507" in str(exc_info.value) or "Insufficient" in str(exc_info.value)
def test_import_rejects_oversized_event_e2e(self):
"""Import of an event exceeding max-resource-size should skip it."""
user = factories.UserFactory(email="sanitizer-import-oversize@example.com")
calendar = self._create_calendar(user)
caldav_path = self._create_calendar(user)
import_service = ICSImportService()
result = import_service.import_events(user, calendar, ICS_OVERSIZED_EVENT)
result = import_service.import_events(user, caldav_path, ICS_OVERSIZED_EVENT)
assert result.total_events == 1
assert result.imported_count == 0