add-ideas/govoplan-calendar
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

Release v0.1.5

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-07 15:49:06 +02:00
parent 3bcb04ff76
commit b328a6d649
21 changed files with 5718 additions and 714 deletions
Download Patch File Download Diff File Expand all files Collapse all files

435
tests/test_caldav.py Normal file
View File

@@ -0,0 +1,435 @@
from __future__ import annotations
import unittest
from datetime import datetime, timezone
from types import SimpleNamespace
from unittest.mock import patch
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from govoplan_access.backend.db import models as access_models # noqa: F401 - populate users/accounts tables
from govoplan_calendar.backend.caldav import CalDAVObject, CalDAVPreconditionFailed, CalDAVReportResult, CalDAVWriteResult, parse_multistatus
from govoplan_calendar.backend.db.models import CalendarEvent, CalendarSyncCredential, CalendarSyncSource
from govoplan_calendar.backend.schemas import CalendarCalDavSourceCreateRequest, CalendarCollectionCreateRequest, CalendarEventCreateRequest, CalendarEventUpdateRequest
from govoplan_calendar.backend.service import (
CALDAV_INTERNAL_CREDENTIAL_PREFIX,
CalendarError,
caldav_client_for_source,
create_calendar,
create_caldav_source,
create_event,
delete_event,
list_freebusy,
sync_caldav_source,
sync_due_caldav_sources,
update_event,
)
from govoplan_core.db.base import Base
from govoplan_tenancy.backend.db.models import Tenant
class FakeCalDAVClient:
def __init__(
self,
*,
list_result: CalDAVReportResult | None = None,
sync_result: CalDAVReportResult | None = None,
objects: dict[str, str] | None = None,
put_etags: list[str | None] | None = None,
fail_precondition: bool = False,
) -> None:
self.list_result = list_result or CalDAVReportResult()
self.sync_result = sync_result
self.objects = objects or {}
self.put_etags = put_etags or ['"etag-written"']
self.fail_precondition = fail_precondition
self.fetched: list[str] = []
self.puts: list[dict[str, object]] = []
self.deletes: list[dict[str, object]] = []
def propfind_collection(self) -> CalDAVReportResult:
return CalDAVReportResult(sync_token="prop-token", ctag="ctag-1")
def list_objects(self) -> CalDAVReportResult:
return self.list_result
def sync_collection(self, sync_token: str) -> CalDAVReportResult:
if self.sync_result is None:
raise AssertionError(f"Unexpected sync_collection call with {sync_token}")
return self.sync_result
def fetch_object(self, href: str) -> str:
self.fetched.append(href)
return self.objects[href]
def put_object(self, href: str, ics: str, *, etag: str | None = None, create: bool = False, overwrite: bool = False) -> CalDAVWriteResult:
if self.fail_precondition:
raise CalDAVPreconditionFailed("changed")
self.puts.append({"href": href, "ics": ics, "etag": etag, "create": create, "overwrite": overwrite})
result_etag = self.put_etags.pop(0) if self.put_etags else None
return CalDAVWriteResult(href=href, etag=result_etag, status=201 if create else 204)
def delete_object(self, href: str, *, etag: str | None = None, overwrite: bool = False) -> CalDAVWriteResult:
if self.fail_precondition:
raise CalDAVPreconditionFailed("changed")
self.deletes.append({"href": href, "etag": etag, "overwrite": overwrite})
return CalDAVWriteResult(href=href, status=204)
class CalDAVParsingTests(unittest.TestCase):
def test_parse_multistatus_extracts_objects_sync_token_and_deletions(self) -> None:
result = parse_multistatus(
b"""<?xml version="1.0" encoding="utf-8"?>
<D:multistatus xmlns:D="DAV:" xmlns:C="urn:ietf:params:xml:ns:caldav">
<D:sync-token>token-2</D:sync-token>
<D:response>
<D:href>/cal/event-1.ics</D:href>
<D:propstat>
<D:prop>
<D:getetag>"etag-1"</D:getetag>
<C:calendar-data>BEGIN:VCALENDAR
BEGIN:VEVENT
UID:event-1@example.test
DTSTART:20260708T090000Z
SUMMARY:One
END:VEVENT
END:VCALENDAR</C:calendar-data>
</D:prop>
<D:status>HTTP/1.1 200 OK</D:status>
</D:propstat>
</D:response>
<D:response>
<D:href>/cal/event-2.ics</D:href>
<D:propstat>
<D:prop><D:getetag/></D:prop>
<D:status>HTTP/1.1 404 Not Found</D:status>
</D:propstat>
</D:response>
</D:multistatus>
"""
)
self.assertEqual(result.sync_token, "token-2")
self.assertEqual(result.objects[0].href, "/cal/event-1.ics")
self.assertEqual(result.objects[0].etag, '"etag-1"')
self.assertIn("BEGIN:VEVENT", result.objects[0].calendar_data or "")
self.assertTrue(result.objects[1].deleted)
class CalDAVSyncTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(bind=self.engine)
self.Session = sessionmaker(bind=self.engine)
def tearDown(self) -> None:
Base.metadata.drop_all(bind=self.engine)
def test_source_creation_encrypts_credential_and_resolves_client_secret(self) -> None:
session = self.Session()
session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote"))
source = create_caldav_source(
session,
tenant_id="tenant-1",
user_id=None,
payload=CalendarCalDavSourceCreateRequest(
calendar_id=calendar.id,
collection_url="https://dav.example.test/cal",
auth_type="basic",
username="ada",
password="secret",
),
)
session.commit()
credential = session.query(CalendarSyncCredential).one()
client = caldav_client_for_source(session, source)
self.assertTrue(source.credential_ref.startswith(CALDAV_INTERNAL_CREDENTIAL_PREFIX))
self.assertNotEqual(credential.secret_encrypted, "secret")
self.assertEqual(client.username, "ada")
self.assertEqual(client.password, "secret")
def test_due_sync_runs_due_sources_and_reschedules(self) -> None:
session = self.Session()
session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote"))
source = create_caldav_source(
session,
tenant_id="tenant-1",
user_id=None,
payload=CalendarCalDavSourceCreateRequest(calendar_id=calendar.id, collection_url="https://dav.example.test/cal"),
)
source.next_sync_at = datetime(2026, 7, 7, 8, 0, tzinfo=timezone.utc)
session.commit()
fake = FakeCalDAVClient(list_result=CalDAVReportResult(sync_token="token-1"))
results = sync_due_caldav_sources(
session,
tenant_id="tenant-1",
now=datetime(2026, 7, 7, 8, 1, tzinfo=timezone.utc),
client_factory=lambda _source: fake,
)
session.commit()
self.assertEqual(results[0].status, "ok")
self.assertEqual(source.sync_token, "token-1")
next_sync_at = source.next_sync_at if source.next_sync_at.tzinfo else source.next_sync_at.replace(tzinfo=timezone.utc)
self.assertGreater(next_sync_at, datetime(2026, 7, 7, 8, 1, tzinfo=timezone.utc))
def test_create_and_update_event_puts_caldav_resource_with_etag(self) -> None:
session = self.Session()
session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote"))
create_caldav_source(session, tenant_id="tenant-1", user_id=None, payload=CalendarCalDavSourceCreateRequest(calendar_id=calendar.id, collection_url="https://dav.example.test/cal"))
session.commit()
fake = FakeCalDAVClient(put_etags=['"etag-1"', '"etag-2"'])
with patch("govoplan_calendar.backend.service.caldav_client_for_source", return_value=fake):
event = create_event(
session,
tenant_id="tenant-1",
user_id=None,
payload=CalendarEventCreateRequest(
calendar_id=calendar.id,
uid="event-1@example.test",
summary="Planning",
start_at=datetime(2026, 7, 8, 9, 0, tzinfo=timezone.utc),
end_at=datetime(2026, 7, 8, 10, 0, tzinfo=timezone.utc),
),
)
update_event(session, tenant_id="tenant-1", user_id=None, event_id=event.id, payload=CalendarEventUpdateRequest(summary="Updated"))
session.commit()
self.assertEqual(len(fake.puts), 2)
self.assertTrue(fake.puts[0]["create"])
self.assertEqual(fake.puts[1]["etag"], '"etag-1"')
self.assertFalse(fake.puts[1]["create"])
self.assertEqual(event.source_kind, "caldav")
self.assertEqual(event.etag, '"etag-2"')
self.assertIn("SUMMARY:Updated", event.raw_ics or "")
def test_update_event_reports_remote_etag_conflict(self) -> None:
session = self.Session()
session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote"))
create_caldav_source(session, tenant_id="tenant-1", user_id=None, payload=CalendarCalDavSourceCreateRequest(calendar_id=calendar.id, collection_url="https://dav.example.test/cal"))
event = caldav_event(calendar.id, summary="Remote", href="https://dav.example.test/cal/event-1.ics", etag='"etag-1"')
session.add(event)
session.commit()
fake = FakeCalDAVClient(fail_precondition=True)
with patch("govoplan_calendar.backend.service.caldav_client_for_source", return_value=fake):
with self.assertRaisesRegex(CalendarError, "changed remotely"):
update_event(session, tenant_id="tenant-1", user_id=None, event_id=event.id, payload=CalendarEventUpdateRequest(summary="Updated"))
def test_delete_one_component_puts_remaining_resource_instead_of_deleting_object(self) -> None:
session = self.Session()
session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote"))
create_caldav_source(session, tenant_id="tenant-1", user_id=None, payload=CalendarCalDavSourceCreateRequest(calendar_id=calendar.id, collection_url="https://dav.example.test/cal"))
href = "https://dav.example.test/cal/series.ics"
master = caldav_event(calendar.id, summary="Master", href=href, etag='"etag-1"')
override = caldav_event(calendar.id, summary="Override", href=href, etag='"etag-1"', recurrence_id="20260715T090000Z")
session.add_all([master, override])
session.commit()
fake = FakeCalDAVClient(put_etags=['"etag-2"'])
with patch("govoplan_calendar.backend.service.caldav_client_for_source", return_value=fake):
delete_event(session, tenant_id="tenant-1", event_id=override.id)
session.commit()
self.assertEqual(len(fake.puts), 1)
self.assertEqual(fake.deletes, [])
self.assertIn("SUMMARY:Master", fake.puts[0]["ics"])
self.assertNotIn("SUMMARY:Override", fake.puts[0]["ics"])
self.assertEqual(master.etag, '"etag-2"')
self.assertIsNotNone(override.deleted_at)
def test_freebusy_expands_recurring_events_and_skips_transparent_items(self) -> None:
session = self.Session()
session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Local"))
create_event(
session,
tenant_id="tenant-1",
user_id=None,
payload=CalendarEventCreateRequest(
calendar_id=calendar.id,
uid="busy@example.test",
summary="Busy",
start_at=datetime(2026, 7, 8, 9, 0, tzinfo=timezone.utc),
end_at=datetime(2026, 7, 8, 10, 0, tzinfo=timezone.utc),
rrule={"FREQ": "WEEKLY", "COUNT": "2"},
),
)
create_event(
session,
tenant_id="tenant-1",
user_id=None,
payload=CalendarEventCreateRequest(
calendar_id=calendar.id,
uid="free@example.test",
summary="Free",
transparency="TRANSPARENT",
start_at=datetime(2026, 7, 8, 11, 0, tzinfo=timezone.utc),
end_at=datetime(2026, 7, 8, 12, 0, tzinfo=timezone.utc),
),
)
session.commit()
busy = list_freebusy(
session,
tenant_id="tenant-1",
start_at=datetime(2026, 7, 1, tzinfo=timezone.utc),
end_at=datetime(2026, 7, 31, tzinfo=timezone.utc),
)
self.assertEqual([item["uid"] for item in busy], ["busy@example.test", "busy@example.test"])
self.assertEqual(busy[0]["start_at"], datetime(2026, 7, 8, 9, 0, tzinfo=timezone.utc))
def test_full_sync_imports_multi_vevent_resource_and_fetches_missing_calendar_data(self) -> None:
session = self.Session()
tenant = Tenant(id="tenant-1", slug="tenant-1", name="Tenant")
session.add(tenant)
calendar = create_calendar(
session,
tenant_id="tenant-1",
user_id=None,
payload=CalendarCollectionCreateRequest(name="Remote", metadata={}),
)
source = create_caldav_source(
session,
tenant_id="tenant-1",
user_id=None,
payload=CalendarCalDavSourceCreateRequest(calendar_id=calendar.id, collection_url="https://dav.example.test/cal"),
)
session.commit()
href = "/cal/series.ics"
client = FakeCalDAVClient(
list_result=CalDAVReportResult(
objects=[CalDAVObject(href=href, etag='"etag-1"', calendar_data=None)],
sync_token="token-1",
ctag="ctag-2",
),
objects={href: recurring_resource()},
)
source, stats = sync_caldav_source(session, tenant_id="tenant-1", user_id=None, source_id=source.id, client=client)
session.commit()
events = session.query(CalendarEvent).filter(CalendarEvent.calendar_id == calendar.id, CalendarEvent.deleted_at.is_(None)).order_by(CalendarEvent.recurrence_id.asc()).all()
self.assertEqual(stats.created, 2)
self.assertEqual(stats.fetched, 1)
self.assertTrue(stats.full_sync)
self.assertEqual(source.sync_token, "token-1")
self.assertEqual(source.ctag, "ctag-2")
self.assertEqual(client.fetched, [href])
self.assertEqual({event.summary for event in events}, {"Series", "Moved instance"})
self.assertEqual({event.source_href for event in events}, {"https://dav.example.test/cal/series.ics"})
self.assertEqual(calendar.metadata_["source_kind"], "caldav")
def test_sync_token_deletion_soft_deletes_remote_resource_events(self) -> None:
session = self.Session()
tenant = Tenant(id="tenant-1", slug="tenant-1", name="Tenant")
session.add(tenant)
calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote"))
source = create_caldav_source(
session,
tenant_id="tenant-1",
user_id=None,
payload=CalendarCalDavSourceCreateRequest(calendar_id=calendar.id, collection_url="https://dav.example.test/cal"),
)
source.sync_token = "token-1"
session.add(
CalendarEvent(
tenant_id="tenant-1",
calendar_id=calendar.id,
uid="event-1@example.test",
recurrence_id=None,
summary="Remote",
status="CONFIRMED",
transparency="OPAQUE",
classification="PUBLIC",
start_at=datetime(2026, 7, 8, 9, 0, tzinfo=timezone.utc),
all_day=False,
attendees=[],
categories=[],
rdate=[],
exdate=[],
reminders=[],
attachments=[],
related_to=[],
source_kind="caldav",
source_href="https://dav.example.test/cal/event-1.ics",
icalendar={},
metadata_={"caldav": {"source_id": source.id}},
)
)
session.commit()
client = FakeCalDAVClient(sync_result=CalDAVReportResult(objects=[CalDAVObject(href="/cal/event-1.ics", deleted=True)], sync_token="token-2"))
source, stats = sync_caldav_source(session, tenant_id="tenant-1", user_id=None, source_id=source.id, client=client)
session.commit()
event = session.query(CalendarEvent).filter(CalendarEvent.uid == "event-1@example.test").one()
self.assertTrue(stats.used_sync_token)
self.assertEqual(stats.deleted, 1)
self.assertEqual(source.sync_token, "token-2")
self.assertIsNotNone(event.deleted_at)
def recurring_resource() -> str:
return """BEGIN:VCALENDAR
VERSION:2.0
BEGIN:VEVENT
UID:series@example.test
DTSTART;TZID=Europe/Berlin:20260708T090000
DTEND;TZID=Europe/Berlin:20260708T100000
RRULE:FREQ=WEEKLY;COUNT=2
SUMMARY:Series
END:VEVENT
BEGIN:VEVENT
UID:series@example.test
RECURRENCE-ID;TZID=Europe/Berlin:20260715T090000
DTSTART;TZID=Europe/Berlin:20260715T110000
DTEND;TZID=Europe/Berlin:20260715T120000
SUMMARY:Moved instance
END:VEVENT
END:VCALENDAR
"""
def caldav_event(calendar_id: str, *, summary: str, href: str, etag: str, recurrence_id: str | None = None) -> CalendarEvent:
return CalendarEvent(
tenant_id="tenant-1",
calendar_id=calendar_id,
uid="series@example.test" if recurrence_id else "event-1@example.test",
recurrence_id=recurrence_id,
summary=summary,
status="CONFIRMED",
transparency="OPAQUE",
classification="PUBLIC",
start_at=datetime(2026, 7, 8, 9, 0, tzinfo=timezone.utc),
end_at=datetime(2026, 7, 8, 10, 0, tzinfo=timezone.utc),
all_day=False,
attendees=[],
categories=[],
rdate=[],
exdate=[],
reminders=[],
attachments=[],
related_to=[],
source_kind="caldav",
source_href=href,
etag=etag,
icalendar={},
metadata_={},
)
if __name__ == "__main__":
unittest.main()

152
tests/test_ical.py
View File

@@ -1,8 +1,10 @@
from __future__ import annotations
import unittest
from datetime import datetime, timezone
from types import SimpleNamespace
from govoplan_calendar.backend.ical import event_to_ics, parse_vevent
from govoplan_calendar.backend.ical import expand_event_occurrences, event_to_ics, parse_vevent
class ICalendarParsingTests(unittest.TestCase):
@@ -48,6 +50,154 @@ END:VCALENDAR
self.assertTrue(event["all_day"])
self.assertEqual(event["start_at"].hour, 0)
def test_parse_advanced_vevent_fields_and_valarm(self) -> None:
event = parse_vevent(
"""BEGIN:VCALENDAR
VERSION:2.0
METHOD:REQUEST
BEGIN:VEVENT
UID:event-advanced@example.test
DTSTAMP:20260707T080000Z
CREATED:20260707T070000Z
LAST-MODIFIED:20260707T073000Z
DTSTART;TZID=Europe/Berlin:20260708T090000
DURATION:PT1H30M
SUMMARY:Planning
DESCRIPTION:Line one\\nLine two
LOCATION:Room\\; 1
PRIORITY:5
URL:https://example.test/events/event-advanced
GEO:52.5200;13.4050
COMMENT:First comment
RESOURCES:Projector,Room
REQUEST-STATUS:2.0;Success
ORGANIZER;CN=Orga:mailto:orga@example.test
ATTENDEE;CN=Ada;ROLE=REQ-PARTICIPANT;PARTSTAT=ACCEPTED:mailto:ada@example.test
RRULE:FREQ=WEEKLY;COUNT=2
EXDATE;TZID=Europe/Berlin:20260715T090000
RDATE;TZID=Europe/Berlin:20260722T090000
ATTACH;FMTTYPE=application/pdf:https://example.test/a.pdf
RELATED-TO;RELTYPE=PARENT:parent@example.test
X-GOVOPLAN-CUSTOM:kept
BEGIN:VALARM
ACTION:DISPLAY
DESCRIPTION:Reminder
TRIGGER:-PT15M
END:VALARM
END:VEVENT
END:VCALENDAR
"""
)
self.assertEqual(event["start_at"], datetime(2026, 7, 8, 7, 0, tzinfo=timezone.utc))
self.assertEqual(event["end_at"], datetime(2026, 7, 8, 8, 30, tzinfo=timezone.utc))
self.assertEqual(event["duration_seconds"], 5400)
self.assertEqual(event["description"], "Line one\nLine two")
self.assertEqual(event["location"], "Room; 1")
self.assertEqual(event["timezone"], "Europe/Berlin")
self.assertEqual(event["organizer"]["params"]["CN"], ["Orga"])
self.assertEqual(event["attendees"][0]["params"]["PARTSTAT"], ["ACCEPTED"])
self.assertEqual(event["icalendar"]["method"], "REQUEST")
self.assertEqual(event["icalendar"]["standard"]["priority"]["value"], "5")
self.assertEqual(event["icalendar"]["standard"]["url"]["value"], "https://example.test/events/event-advanced")
self.assertEqual(event["reminders"][0]["trigger"]["value"], "-PT15M")
self.assertEqual(event["reminders"][0]["trigger"]["decoded"], -900)
property_names = [item["name"] for item in event["icalendar"]["properties"]]
self.assertIn("REQUEST-STATUS", property_names)
self.assertIn("X-GOVOPLAN-CUSTOM", property_names)
def test_generated_ics_preserves_unmapped_properties_and_alarms(self) -> None:
event = SimpleNamespace(
uid="event-export@example.test",
start_at=datetime(2026, 7, 8, 7, 0, tzinfo=timezone.utc),
end_at=datetime(2026, 7, 8, 8, 0, tzinfo=timezone.utc),
duration_seconds=None,
all_day=False,
timezone="Europe/Berlin",
summary="Generated",
sequence=4,
status="CONFIRMED",
transparency="OPAQUE",
classification="PUBLIC",
description=None,
location=None,
categories=["GovOPlaN"],
rrule={"FREQ": "WEEKLY", "COUNT": "2"},
organizer={"value": "mailto:orga@example.test", "params": {"CN": ["Orga"]}},
attendees=[{"value": "mailto:ada@example.test", "params": {"CN": ["Ada"], "ROLE": ["REQ-PARTICIPANT"]}}],
rdate=[],
exdate=[],
reminders=[],
attachments=[{"name": "ATTACH", "value": "https://example.test/a.pdf", "params": {"FMTTYPE": ["application/pdf"]}}],
related_to=[{"name": "RELATED-TO", "value": "parent@example.test", "params": {"RELTYPE": ["PARENT"]}}],
icalendar={
"method": "PUBLISH",
"properties": [
{"name": "URL", "value": "https://example.test/events/event-export", "params": {}},
{"name": "X-GOVOPLAN-CUSTOM", "value": "kept", "params": {}},
{"name": "SUMMARY", "value": "Stale summary", "params": {}},
],
"alarms": [
{
"component": "VALARM",
"properties": [
{"name": "ACTION", "value": "DISPLAY", "params": {}},
{"name": "DESCRIPTION", "value": "Reminder", "params": {}},
{"name": "TRIGGER", "value": "-PT15M", "params": {}},
],
}
],
},
)
ics = event_to_ics(event)
self.assertIn("METHOD:PUBLISH", ics)
self.assertIn("DTSTART;TZID=Europe/Berlin:20260708T090000", ics)
self.assertIn("SUMMARY:Generated", ics)
self.assertNotIn("SUMMARY:Stale summary", ics)
self.assertIn("RRULE:FREQ=WEEKLY;COUNT=2", ics)
self.assertIn("ATTACH;FMTTYPE=application/pdf:https://example.test/a.pdf", ics)
self.assertIn("RELATED-TO;RELTYPE=PARENT:parent@example.test", ics)
self.assertIn("URL:https://example.test/events/event-export", ics)
self.assertIn("X-GOVOPLAN-CUSTOM:kept", ics)
self.assertIn("BEGIN:VALARM", ics)
self.assertIn("TRIGGER:-PT15M", ics)
def test_expand_event_occurrences_applies_rrule_rdate_and_exdate(self) -> None:
parsed = parse_vevent(
"""BEGIN:VCALENDAR
VERSION:2.0
BEGIN:VEVENT
UID:event-recur@example.test
DTSTART;TZID=Europe/Berlin:20260708T090000
DTEND;TZID=Europe/Berlin:20260708T100000
SUMMARY:Recurring
RRULE:FREQ=WEEKLY;COUNT=3
EXDATE;TZID=Europe/Berlin:20260715T090000
RDATE;TZID=Europe/Berlin:20260729T090000
END:VEVENT
END:VCALENDAR
"""
)
event = SimpleNamespace(**parsed)
occurrences = expand_event_occurrences(
event,
datetime(2026, 7, 1, tzinfo=timezone.utc),
datetime(2026, 7, 31, 23, 59, tzinfo=timezone.utc),
)
self.assertEqual(
[item["start_at"] for item in occurrences],
[
datetime(2026, 7, 8, 7, 0, tzinfo=timezone.utc),
datetime(2026, 7, 22, 7, 0, tzinfo=timezone.utc),
datetime(2026, 7, 29, 7, 0, tzinfo=timezone.utc),
],
)
self.assertEqual(occurrences[0]["end_at"], datetime(2026, 7, 8, 8, 0, tzinfo=timezone.utc))
def test_generated_ics_contains_required_vevent_fields(self) -> None:
class Event:
uid = "event-3@example.test"

115
tests/test_service.py Normal file
View File

@@ -0,0 +1,115 @@
from __future__ import annotations
import unittest
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace
from unittest.mock import patch
from govoplan_calendar.backend.schemas import CalendarCollectionDeleteRequest, CalendarEventResponse
from govoplan_calendar.backend.service import delete_calendar, event_response
class FakeSession:
def __init__(self) -> None:
self.flushed = False
def flush(self) -> None:
self.flushed = True
class CalendarServiceResponseTests(unittest.TestCase):
def test_event_response_marks_naive_database_datetimes_as_utc(self) -> None:
event = SimpleNamespace(
id="event-1",
tenant_id="tenant-1",
calendar_id="calendar-1",
uid="event-1@example.test",
recurrence_id=None,
sequence=0,
summary="Planning",
description=None,
location=None,
status="CONFIRMED",
transparency="OPAQUE",
classification="PUBLIC",
start_at=datetime(2026, 7, 7, 18, 0),
end_at=datetime(2026, 7, 7, 19, 0),
duration_seconds=None,
all_day=False,
timezone=None,
organizer=None,
attendees=[],
categories=[],
rrule=None,
rdate=[],
exdate=[],
reminders=[],
attachments=[],
related_to=[],
source_kind="local",
source_href=None,
etag=None,
icalendar={},
created_at=datetime(2026, 7, 7, 17, 55),
updated_at=datetime(2026, 7, 7, 17, 55),
metadata_={},
)
payload = event_response(event)
response = CalendarEventResponse.model_validate(payload)
self.assertEqual(response.start_at.utcoffset(), timedelta(0))
self.assertEqual(response.end_at.utcoffset(), timedelta(0))
self.assertEqual(response.created_at.utcoffset(), timedelta(0))
self.assertEqual(response.updated_at.utcoffset(), timedelta(0))
self.assertEqual(response.start_at.tzinfo, timezone.utc)
class CalendarServiceDeleteTests(unittest.TestCase):
def test_delete_default_calendar_soft_deletes_calendar_and_events(self) -> None:
session = FakeSession()
event = SimpleNamespace(calendar_id="calendar-1", deleted_at=None)
calendar = SimpleNamespace(id="calendar-1", is_default=True, deleted_at=None, events=[event])
with patch("govoplan_calendar.backend.service.get_calendar", return_value=calendar):
delete_calendar(
session,
tenant_id="tenant-1",
calendar_id="calendar-1",
payload=CalendarCollectionDeleteRequest(event_action="delete"),
)
self.assertTrue(session.flushed)
self.assertFalse(calendar.is_default)
self.assertIsNotNone(calendar.deleted_at)
self.assertEqual(event.deleted_at, calendar.deleted_at)
self.assertEqual(event.calendar_id, "calendar-1")
def test_delete_calendar_can_move_events_to_another_calendar_and_make_it_default(self) -> None:
session = FakeSession()
event = SimpleNamespace(calendar_id="calendar-1", deleted_at=None)
calendar = SimpleNamespace(id="calendar-1", is_default=True, deleted_at=None, events=[event])
target = SimpleNamespace(id="calendar-2", is_default=False, deleted_at=None, events=[])
with (
patch("govoplan_calendar.backend.service.get_calendar", side_effect=[calendar, target]),
patch("govoplan_calendar.backend.service.clear_default_calendar") as clear_default_calendar,
):
delete_calendar(
session,
tenant_id="tenant-1",
calendar_id="calendar-1",
payload=CalendarCollectionDeleteRequest(event_action="move", target_calendar_id="calendar-2", make_target_default=True),
)
self.assertTrue(session.flushed)
clear_default_calendar.assert_called_once_with(session, tenant_id="tenant-1")
self.assertFalse(calendar.is_default)
self.assertTrue(target.is_default)
self.assertIsNotNone(calendar.deleted_at)
self.assertIsNone(event.deleted_at)
self.assertEqual(event.calendar_id, "calendar-2")
if __name__ == "__main__":
unittest.main()