Fix YEARLY recurring CalDAV events only showing on DTSTART year (#179)

* Fix YEARLY recurring CalDAV events only showing on DTSTART year (#170)

Recurring events with RRULE:FREQ=YEARLY only appeared in the calendar
on the year matching DTSTART, not in subsequent years. The list_events
query filtered by , which excludes
recurring events whose original dtend (e.g. 2019-07-22) falls before
the requested window (e.g. 2026).

Fix: split the query into two branches — non-recurring events still
require window overlap, but recurring events (with non-empty RRULE)
are fetched by dtstart < end_dt alone. A new helper,
_expand_rrule_occurrences(), uses dateutil.rrule to expand each
recurring event into individual occurrence dicts within the requested
date range, so YEARLY/WEEKLY/MONTHLY events render correctly across
all years.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* recurrence: compound UIDs, frontend fixes, python-dateutil req, tests

- Replace _expand_rrule_occurrences with _expand_rrule that emits stable
  compound UIDs ({base_uid}::{date_or_datetime}) so the frontend can
  distinguish occurrences from the same series. Non-recurring events
  pass through with is_recurrence=false and series_uid=uid.

- Add _resolve_base_uid() to extract the base series UID from compound
  UIDs — used by PUT/DELETE /api/calendar/events/{uid} and the
  manage_calendar tool so edits/deletes always target the base row.

- Update manage_calendar tool to import and use _resolve_base_uid.

- Frontend _updateEvent / _deleteEvent: detect compound UIDs and
  invalidate localStorage cache after success so stale sibling
  occurrences aren't shown.

- Add python-dateutil to requirements.txt as an explicit dependency.

- Add 14 regression tests in tests/test_calendar_recurrence.py
  covering _resolve_base_uid edge cases, _expand_rrule with
  yearly/weekly/monthly/all-day/bad-rrule, unique UIDs, and
  metadata inheritance.

- Merge upstream's cleaner SQLAlchemy or_/and_ query pattern.

* recurrence: overlapping malformed-RRULE, exclusive end, multi-day crossings

Fix three edge cases in _expand_rrule:

1. Malformed-RRULE fallback now checks window overlap. list_events
   fetches recurring rows with only dtstart < end_dt, so a broken
   old recurring event could appear in unrelated future windows.
   Now fallback returns [] unless the base event's dtstart/dtend
   actually intersect [start, end).

2. Exclusive end boundary. rule.between(start, end, inc=True) was
   inclusive on end, but the route contract and non-recurring SQL
   filter both use [start, end). Added occ_start >= end guard.

3. Multi-day crossings. A recurring occurrence that starts before
   the window but ends inside it was missed (only occ_start was
   checked). Now expands from start - duration and filters by
   occ_start < end AND occ_end > start, matching non-recurring
   overlap behavior.

Tests: +4 tests for these cases (18 total)

---------

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
AzaelMew
2026-06-01 06:42:44 +02:00
committed by GitHub
parent 8df5ed2a96
commit 7023468cea
5 changed files with 505 additions and 13 deletions

View File

@@ -23,6 +23,10 @@ youtube-transcript-api
markdown
# Calendar .ics import/export (routes/calendar_routes.py).
icalendar
# Recurrence rule expansion for calendar events (routes/calendar_routes.py).
# Imported directly as dateutil.rrule — make it explicit even though caldav
# pulls it in transitively.
python-dateutil
# CalDAV sync (src/caldav_sync.py). Handles PROPFIND discovery + REPORT
# fetch across Radicale, Nextcloud, Apple, Fastmail; we'd be reinventing
# the protocol without it.

View File

@@ -3,10 +3,13 @@
import logging
import uuid
from datetime import datetime, date, timedelta
from typing import Optional
from typing import Optional, List, Tuple
from fastapi import APIRouter, HTTPException, Request, UploadFile, File
from pydantic import BaseModel
from sqlalchemy import or_, and_
from dateutil.rrule import rrulestr, rruleset
from dateutil.rrule import DAILY, WEEKLY, MONTHLY, YEARLY
from core.database import SessionLocal, CalendarCal, CalendarEvent
from src.auth_helpers import get_current_user
@@ -60,6 +63,23 @@ def _get_or_404_event(db, uid: str, owner: str) -> CalendarEvent:
raise HTTPException(404, "Event not found")
return ev
def _resolve_base_uid(uid: str) -> str:
"""Extract the base series UID from a compound occurrence UID.
Compound UIDs have the form ``{base_uid}::{date_suffix}``.
For plain UIDs (no ``::``), returns the UID unchanged.
"""
if not uid:
raise ValueError("empty uid")
idx = uid.find("::")
if idx == -1:
return uid # plain UID — no suffix
base = uid[:idx]
if not base:
raise ValueError("malformed compound UID: missing base before ::")
return base
# ── Pydantic models ──
class EventCreate(BaseModel):
@@ -387,6 +407,95 @@ def _event_to_dict(ev: CalendarEvent) -> dict:
}
# ── Recurrence expansion ──
def _expand_rrule(
ev: CalendarEvent, start: datetime, end: datetime
) -> List[dict]:
"""Expand a single recurring CalendarEvent into occurrence dicts.
Each occurrence gets a stable compound UID of the form
``{base_uid}::{date_or_datetime}`` so the frontend can tell
occurrences apart while the series UID is still recoverable
for edit/delete targeting.
Non-recurring events (empty rrule) are returned as a single-item
list — the caller doesn't need to branch.
"""
duration = ev.dtend - ev.dtstart
if not ev.rrule or not ev.rrule.strip():
# Non-recurring — return the base event as-is. list_events
# already filters non-recurring rows with the overlap check
# in SQL, so we don't re-check here.
d = _event_to_dict(ev)
d["is_recurrence"] = False
d["series_uid"] = ev.uid
return [d]
# Parse the rrule, applying it to the base dtstart.
try:
rule = rrulestr(ev.rrule, dtstart=ev.dtstart)
except Exception as ex:
logger.warning(
"Failed to parse rrule=%r for event %s: %s", ev.rrule, ev.uid, ex
)
d = _event_to_dict(ev)
d["is_recurrence"] = False
d["series_uid"] = ev.uid
# Malformed RRULE rows are fetched by the recurring SQL branch
# with only dtstart < end_dt — the base event may not actually
# overlap the window. Only return if it does.
if ev.dtstart < end and ev.dtend > start:
return [d]
return []
# Expand from start - duration so multi-day / overnight occurrences
# that start before the window but end inside it are captured
# (matching non-recurring overlap semantics: dtstart < end AND
# dtend > start).
expand_start = start - duration
occurrences = rule.between(expand_start, end, inc=True)
if not occurrences:
return []
results = []
base = _event_to_dict(ev)
for occ_start in occurrences:
occ_end = occ_start + duration
# Overlap filter: occurrence must intersect [start, end).
# This enforces exclusive-end semantics (occ_start >= end is
# excluded) and includes multi-day crossings (occ_end > start).
if occ_start >= end or occ_end <= start:
continue
# Build the compound uid: {base_uid}::{date} or ::{datetime}
if ev.all_day:
occ_uid = f"{ev.uid}::{occ_start.strftime('%Y-%m-%d')}"
else:
occ_uid = f"{ev.uid}::{occ_start.strftime('%Y-%m-%dT%H:%M')}"
d = dict(base)
d["uid"] = occ_uid
d["series_uid"] = ev.uid
d["is_recurrence"] = True
if ev.all_day:
d["dtstart"] = occ_start.strftime("%Y-%m-%d")
d["dtend"] = occ_end.strftime("%Y-%m-%d")
else:
suffix = "Z" if getattr(ev, "is_utc", False) else ""
d["dtstart"] = occ_start.isoformat() + suffix
d["dtend"] = occ_end.isoformat() + suffix
d["is_utc"] = bool(getattr(ev, "is_utc", False))
results.append(d)
return results
# ── Routes ──
def setup_calendar_routes() -> APIRouter:
@@ -535,11 +644,29 @@ def setup_calendar_routes() -> APIRouter:
db = SessionLocal()
try:
# Scope events to calendars owned by the caller.
# Non-recurring events must overlap the query window; recurring
# events (with RRULE) whose base dtstart is before the window end
# are fetched so their actual occurrences can be expanded
# server-side and appear in every year they repeat, not just the
# DTSTART year.
q = db.query(CalendarEvent).join(CalendarCal).filter(
CalendarEvent.dtstart < end_dt,
CalendarEvent.dtend > start_dt,
CalendarEvent.status != "cancelled",
CalendarCal.owner == owner,
or_(
# Non-recurring: event times must overlap the query window
and_(
or_(CalendarEvent.rrule == "", CalendarEvent.rrule.is_(None)),
CalendarEvent.dtstart < end_dt,
CalendarEvent.dtend > start_dt,
),
# Recurring: dtstart before window end — RRULE expansion
# generates the actual occurrences within the window
and_(
CalendarEvent.rrule.isnot(None),
CalendarEvent.rrule != "",
CalendarEvent.dtstart < end_dt,
),
),
)
if calendar:
q = q.filter(
@@ -547,7 +674,15 @@ def setup_calendar_routes() -> APIRouter:
(CalendarCal.name == calendar)
)
events = q.order_by(CalendarEvent.dtstart).all()
return {"events": [_event_to_dict(e) for e in events]}
# Expand recurring events into individual occurrences.
expanded = []
for e in events:
expanded.extend(_expand_rrule(e, start_dt, end_dt))
# Sort by occurrence start time for consistent frontend ordering.
expanded.sort(key=lambda d: d["dtstart"])
return {"events": expanded}
except HTTPException:
raise
except Exception as e:
@@ -617,9 +752,13 @@ def setup_calendar_routes() -> APIRouter:
@router.put("/events/{uid}")
async def update_event(request: Request, uid: str, data: EventUpdate):
owner = _require_user(request)
try:
base_uid = _resolve_base_uid(uid)
except ValueError as e:
raise HTTPException(400, str(e))
db = SessionLocal()
try:
ev = _get_or_404_event(db, uid, owner)
ev = _get_or_404_event(db, base_uid, owner)
if data.summary is not None:
ev.summary = data.summary
if data.description is not None:
@@ -659,9 +798,13 @@ def setup_calendar_routes() -> APIRouter:
@router.delete("/events/{uid}")
async def delete_event(request: Request, uid: str):
owner = _require_user(request)
try:
base_uid = _resolve_base_uid(uid)
except ValueError as e:
raise HTTPException(400, str(e))
db = SessionLocal()
try:
ev = _get_or_404_event(db, uid, owner)
ev = _get_or_404_event(db, base_uid, owner)
db.delete(ev)
db.commit()
return {"ok": True}
@@ -902,8 +1045,8 @@ def setup_calendar_routes() -> APIRouter:
lines.append(f"DTSTART:{ev.dtstart.strftime('%Y%m%dT%H%M%S')}")
lines.append(f"DTEND:{ev.dtend.strftime('%Y%m%dT%H%M%S')}")
if ev.description:
escaped_desc = ev.description.replace(chr(10), "\\n")
lines.append(f"DESCRIPTION:{escaped_desc}")
desc = ev.description.replace(chr(10), '\\n')
lines.append(f"DESCRIPTION:{desc}")
if ev.location:
lines.append(f"LOCATION:{ev.location}")
if ev.rrule:

View File

@@ -1952,7 +1952,7 @@ async def do_manage_calendar(content: str, owner: Optional[str] = None) -> Dict:
"""Handle manage_calendar tool calls: list/create/update/delete calendar events (local SQLite)."""
from datetime import datetime, timedelta
from core.database import SessionLocal, CalendarCal, CalendarEvent, Note
from routes.calendar_routes import _ensure_default_calendar, _parse_dt, _parse_dt_pair, parse_due_for_user
from routes.calendar_routes import _ensure_default_calendar, _parse_dt, _parse_dt_pair, parse_due_for_user, _resolve_base_uid
import uuid as _uuid
try:
@@ -2317,7 +2317,11 @@ async def do_manage_calendar(content: str, owner: Optional[str] = None) -> Dict:
uid = args.get("uid")
if not uid:
return {"error": "uid is required", "exit_code": 1}
ev = _event_query().filter(CalendarEvent.uid == uid).first()
try:
base_uid = _resolve_base_uid(uid)
except ValueError as e:
return {"error": str(e), "exit_code": 1}
ev = _event_query().filter(CalendarEvent.uid == base_uid).first()
if not ev:
return {"error": f"Event {uid} not found", "exit_code": 1}
if args.get("summary") is not None:
@@ -2346,7 +2350,11 @@ async def do_manage_calendar(content: str, owner: Optional[str] = None) -> Dict:
uid = args.get("uid")
if not uid:
return {"error": "uid is required", "exit_code": 1}
ev = _event_query().filter(CalendarEvent.uid == uid).first()
try:
base_uid = _resolve_base_uid(uid)
except ValueError as e:
return {"error": str(e), "exit_code": 1}
ev = _event_query().filter(CalendarEvent.uid == base_uid).first()
if not ev:
return {"error": f"Event {uid} not found", "exit_code": 1}
db.delete(ev)

View File

@@ -265,12 +265,22 @@ async function _updateEvent(uid, data) {
const merged = { ...(_allEvents[uid] || {}), ...data };
const _preMergeBackup = _allEvents[uid];
_allEvents[uid] = _optimisticEvent(merged, uid);
// For recurring events the uid is a compound "{base_uid}::{date}" —
// the backend resolves it to the base series row. After the update,
// other occurrences of the same series are stale. Wipe the cache so
// a re-fetch picks up fresh data (next render + prefetch handles it).
const isRecurring = uid.includes('::');
fetch(`${API_BASE}/api/calendar/events/${encodeURIComponent(uid)}`, {
method: 'PUT', credentials: 'same-origin',
headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(data),
}).then(r => {
if (!r.ok) throw new Error('HTTP ' + r.status);
_saveCache && _saveCache();
if (isRecurring) {
_fetchedRanges = [];
localStorage.removeItem(LS_KEY);
} else {
_saveCache && _saveCache();
}
}).catch((e) => {
if (_preMergeBackup) _allEvents[uid] = _preMergeBackup;
else delete _allEvents[uid];
@@ -283,11 +293,17 @@ async function _updateEvent(uid, data) {
async function _deleteEvent(uid) {
const backup = _allEvents[uid];
delete _allEvents[uid];
const isRecurring = uid.includes('::');
fetch(`${API_BASE}/api/calendar/events/${encodeURIComponent(uid)}`, {
method: 'DELETE', credentials: 'same-origin',
}).then(r => {
if (!r.ok) throw new Error('HTTP ' + r.status);
_saveCache && _saveCache();
if (isRecurring) {
_fetchedRanges = [];
localStorage.removeItem(LS_KEY);
} else {
_saveCache && _saveCache();
}
}).catch((e) => {
if (backup) _allEvents[uid] = backup;
if (window.uiModule) window.uiModule.showError('Failed to delete event: ' + (e?.message || 'unknown'));

View File

@@ -0,0 +1,321 @@
"""Regression tests for calendar recurrence expansion.
Tests _expand_rrule and _resolve_base_uid — imported directly from
routes/calendar_routes using the same stub-friendly import pattern
as test_null_owner_gates.py. No live DB or FastAPI test client needed.
"""
from datetime import datetime, timedelta
from types import SimpleNamespace
import pytest
from tests.test_null_owner_gates import _import_calendar_helpers
# ── _resolve_base_uid ──────────────────────────────────────────────────
def test_resolve_base_uid_plain_passthrough():
cal = _import_calendar_helpers()
assert cal._resolve_base_uid("evt-123") == "evt-123"
def test_resolve_base_uid_compound_strips_suffix_date():
cal = _import_calendar_helpers()
assert cal._resolve_base_uid("evt-123::2026-06-15") == "evt-123"
def test_resolve_base_uid_compound_strips_suffix_datetime():
cal = _import_calendar_helpers()
assert cal._resolve_base_uid("evt-123::2026-06-15T09:00") == "evt-123"
def test_resolve_base_uid_rejects_empty():
cal = _import_calendar_helpers()
with pytest.raises(ValueError, match="empty uid"):
cal._resolve_base_uid("")
def test_resolve_base_uid_rejects_missing_base():
cal = _import_calendar_helpers()
with pytest.raises(ValueError, match="malformed compound UID"):
cal._resolve_base_uid("::2026-06-15")
# ── _expand_rrule ──────────────────────────────────────────────────────
_MOCK_CAL = SimpleNamespace(name="Personal", color="#5b8abf")
def _make_event(**overrides):
"""Build a dict-shaped mock CalendarEvent for _expand_rrule."""
defaults = {
"uid": "evt-test-001",
"summary": "Test Event",
"dtstart": datetime(2026, 6, 1, 9, 0),
"dtend": datetime(2026, 6, 1, 10, 0),
"all_day": False,
"is_utc": False,
"rrule": "",
"calendar": _MOCK_CAL.name,
"calendar_id": "cal-001",
"color": None,
"description": "",
"location": "",
"event_type": None,
"importance": "normal",
}
defaults.update(overrides)
ev = SimpleNamespace(**defaults)
ev.calendar = _MOCK_CAL
return ev
def test_expand_non_recurring_returns_single():
"""Non-recurring events pass through unchanged with series_uid=uid."""
cal = _import_calendar_helpers()
ev = _make_event(rrule="")
results = cal._expand_rrule(ev, datetime(2026, 5, 1), datetime(2026, 7, 1))
assert len(results) == 1
r = results[0]
assert r["uid"] == "evt-test-001"
assert r["series_uid"] == "evt-test-001"
assert r["is_recurrence"] is False
def test_expand_yearly_old_dtstart_later_year_single_occurrence():
"""Create an old DTSTART + FREQ=YEARLY, query a later year, verify
exactly one occurrence is returned.
This is the explicit regression case from PR review feedback.
"""
cal = _import_calendar_helpers()
ev = _make_event(
uid="evt-bday-001",
summary="Annual Review",
dtstart=datetime(2020, 4, 15, 10, 0),
dtend=datetime(2020, 4, 15, 11, 0),
rrule="FREQ=YEARLY",
)
# Query year 2028 — should find the 2028-04-15 occurrence only
results = cal._expand_rrule(ev, datetime(2028, 1, 1), datetime(2029, 1, 1))
assert len(results) == 1, (
f"Expected exactly 1 yearly occurrence in 2028, got {len(results)}: "
f"{[r['uid'] for r in results]}"
)
r = results[0]
assert r["uid"] == "evt-bday-001::2028-04-15T10:00"
assert r["dtstart"] == "2028-04-15T10:00:00"
assert r["series_uid"] == "evt-bday-001"
assert r["is_recurrence"] is True
assert r["summary"] == "Annual Review"
def test_expand_yearly_narrow_window_after_dtstart_returns_one():
"""DTSTART=2020, query just two months in 2029 — should return
exactly one occurrence (the one that falls in that window).
"""
cal = _import_calendar_helpers()
ev = _make_event(
uid="evt-ann",
dtstart=datetime(2020, 3, 1),
dtend=datetime(2020, 3, 2),
all_day=True,
rrule="FREQ=YEARLY",
)
results = cal._expand_rrule(ev, datetime(2029, 1, 1), datetime(2029, 4, 1))
assert len(results) == 1
assert results[0]["uid"] == "evt-ann::2029-03-01"
assert results[0]["all_day"] is True
def test_expand_yearly_strict_before_window_returns_empty():
"""DTSTART=2020, query a window that ends before the yearly
occurrence in that year. Should return zero.
"""
cal = _import_calendar_helpers()
ev = _make_event(
uid="evt-late",
dtstart=datetime(2020, 12, 25),
dtend=datetime(2020, 12, 26),
all_day=True,
rrule="FREQ=YEARLY",
)
results = cal._expand_rrule(ev, datetime(2026, 1, 1), datetime(2026, 6, 1))
assert len(results) == 0
def test_expand_yearly_strict_after_window_returns_empty():
"""DTSTART=2020. Query a window that starts after the occurrence in
that year. Should return zero.
"""
cal = _import_calendar_helpers()
ev = _make_event(
uid="evt-early",
dtstart=datetime(2020, 1, 15),
dtend=datetime(2020, 1, 16),
all_day=True,
rrule="FREQ=YEARLY",
)
results = cal._expand_rrule(ev, datetime(2026, 6, 1), datetime(2026, 12, 31))
assert len(results) == 0
def test_expand_weekly_unique_no_overwrites():
"""Multiple occurrences from the same series must have unique UIDs
so _allEvents[uid] = ev doesn't overwrite earlier ones.
"""
cal = _import_calendar_helpers()
ev = _make_event(
uid="evt-wk",
dtstart=datetime(2026, 6, 1, 9, 0),
dtend=datetime(2026, 6, 1, 10, 0),
rrule="FREQ=WEEKLY;BYDAY=MO,WE,FR",
)
results = cal._expand_rrule(ev, datetime(2026, 6, 1), datetime(2026, 7, 1))
# June 2026 has 4 Mondays, 5 Wednesdays, 4 Fridays = 13 occurrences
assert len(results) >= 10 # sanity lower bound
uids = [r["uid"] for r in results]
assert len(uids) == len(set(uids)), f"Duplicate UIDs found: {uids}"
for r in results:
assert r["series_uid"] == "evt-wk"
assert r["is_recurrence"] is True
def test_expand_monthly_all_day():
cal = _import_calendar_helpers()
ev = _make_event(
uid="evt-rent",
dtstart=datetime(2026, 1, 1),
dtend=datetime(2026, 1, 2),
all_day=True,
rrule="FREQ=MONTHLY",
)
results = cal._expand_rrule(ev, datetime(2026, 1, 1), datetime(2026, 12, 31))
assert len(results) == 12
for r in results:
assert r["uid"].startswith("evt-rent::")
assert r["all_day"] is True
def test_expand_bad_rrule_graceful():
"""Malformed rrule should fall back to returning the base event,
but only when the base event overlaps the requested window."""
cal = _import_calendar_helpers()
ev = _make_event(
uid="evt-broken",
rrule="FREQ=GARBAGE",
)
# Base event (2026-06-01) falls inside the window — should appear
results = cal._expand_rrule(ev, datetime(2026, 1, 1), datetime(2026, 12, 31))
assert len(results) == 1
assert results[0]["uid"] == "evt-broken"
assert results[0]["is_recurrence"] is False
def test_expand_bad_rrule_fallback_rejects_non_overlapping():
"""Malformed rrule with a base event outside the requested window
must return zero results, not leak the event into an unrelated range."""
cal = _import_calendar_helpers()
ev = _make_event(
uid="evt-old-broken",
dtstart=datetime(2020, 1, 1, 9, 0),
dtend=datetime(2020, 1, 1, 10, 0),
rrule="FREQ=GARBAGE",
)
# Query a far-future window that the base event doesn't overlap
results = cal._expand_rrule(ev, datetime(2030, 1, 1), datetime(2030, 2, 1))
assert len(results) == 0, (
f"Malformed rrule base event outside window should return empty, "
f"got {len(results)}: {[r['uid'] for r in results]}"
)
def test_expand_exclusive_end_boundary():
"""An occurrence whose start equals the window end must be excluded.
The contract is [start, end), same as the non-recurring SQL filter."""
cal = _import_calendar_helpers()
ev = _make_event(
uid="evt-daily",
dtstart=datetime(2026, 6, 1, 9, 0),
dtend=datetime(2026, 6, 1, 10, 0),
rrule="FREQ=DAILY",
)
# Query [Jun 1, Jun 5) — occurrences on Jun 1-4 only
results = cal._expand_rrule(ev, datetime(2026, 6, 1), datetime(2026, 6, 5))
uids = [r["uid"] for r in results]
assert len(results) == 4, f"Expected 4 (Jun 1-4), got {len(results)}: {uids}"
assert "evt-daily::2026-06-05T09:00" not in uids, "Jun 5 is at end boundary, must be excluded"
def test_expand_multi_day_crossing_range_start():
"""A multi-day occurrence that starts before the window but ends inside
it must be included (matching non-recurring overlap: dtend > start)."""
cal = _import_calendar_helpers()
ev = _make_event(
uid="evt-weekly-multi",
summary="Weekend Trip",
dtstart=datetime(2026, 5, 29, 18, 0), # Friday evening
dtend=datetime(2026, 6, 1, 12, 0), # Monday noon
rrule="FREQ=WEEKLY",
)
# Query the Monday window — the occurrence starts Fri but ends Mon,
# so it overlaps the query.
results = cal._expand_rrule(ev, datetime(2026, 6, 1), datetime(2026, 6, 2))
# The 2026-06-05 occurrence starts Fri Jun 5 and ends Mon Jun 8 —
# that crosses [Jun 1, Jun 2): occ_start=2026-06-05 >= end=2026-06-02 → excluded.
# The 2026-05-29 occurrence starts Fri May 29 and ends Mon Jun 1 —
# occ_end=2026-06-01T12:00 > start=2026-06-01 → included.
assert len(results) == 1, (
f"Expected 1 occurrence crossing into the window, got {len(results)}: "
f"{[r['uid'] for r in results]}"
)
assert results[0]["uid"] == "evt-weekly-multi::2026-05-29T18:00"
def test_expand_multi_day_fully_before_window():
"""A multi-day occurrence that ends exactly at the window start
must be excluded (occ_end <= start)."""
cal = _import_calendar_helpers()
ev = _make_event(
uid="evt-multi",
dtstart=datetime(2026, 5, 29, 18, 0),
dtend=datetime(2026, 6, 1, 0, 0), # ends at midnight Jun 1
rrule="FREQ=WEEKLY",
)
# Query starting Jun 1 midnight — occ_end <= start, excluded
results = cal._expand_rrule(ev, datetime(2026, 6, 1), datetime(2026, 6, 8))
assert len(results) == 1 # only the next week's occurrence (Jun 5-8)
assert results[0]["uid"] == "evt-multi::2026-06-05T18:00"
def test_expand_metadata_inheritance():
"""Occurrence dicts must carry the base event's metadata
(summary, importance, event_type, color, location)."""
cal = _import_calendar_helpers()
ev = _make_event(
uid="evt-meta",
summary="Board Meeting",
dtstart=datetime(2026, 1, 1, 14, 0),
dtend=datetime(2026, 1, 1, 16, 0),
rrule="FREQ=MONTHLY",
event_type="work",
importance="critical",
location="Room 42",
)
results = cal._expand_rrule(ev, datetime(2026, 1, 1), datetime(2026, 3, 1))
assert len(results) == 2 # Jan + Feb
for r in results:
assert r["summary"] == "Board Meeting"
assert r["importance"] == "critical"
assert r["event_type"] == "work"
assert r["location"] == "Room 42"