From 05f078086e31cc33829a02142e3b5408a8a8fcee Mon Sep 17 00:00:00 2001 From: Tobias Brox Date: Tue, 13 Oct 2026 01:27:19 +0200 Subject: [PATCH] feat: add freebusy-sync command Copies busy time into another calendar as time-only placeholders with derived UIDs, moving and deleting them as the source changes, so it can run from a timer both ways between work and private calendars. It replaces --freebusyhack, now deprecated, which kept the source UID and left recurrence exceptions unstripped. Prompt: [set up regular freebusy sync between work and private calendar] Prompts [some arguments in the review-process, none of it much relevant] Assisted-By: Claude Opus 5.5 Reviewed-by: Tobias Brox --- CHANGELOG.md | 8 ++ USER_GUIDE.md | 14 +++ plann/cli.py | 71 ++++++++++- plann/commands.py | 1 + plann/freebusy.py | 183 +++++++++++++++++++++++++++ tests/test_freebusy.py | 267 ++++++++++++++++++++++++++++++++++++++++ tests/test_plann_cli.py | 96 +++++++++++++++ 7 files changed, 637 insertions(+), 3 deletions(-) create mode 100644 plann/freebusy.py create mode 100644 tests/test_freebusy.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 830fff5..8c14556 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,14 @@ and I do try to adhere to [Semantic Versioning](https://semver.org/spec/v2.0.0.h ## [Unreleased] +### Added + +* `plann freebusy-sync`: keeps a copy of the busy time from one set of calendars in another calendar, holding only the time and a fixed summary. Copies are moved and deleted as the source changes, so it can run from a timer in both directions between a work and a private calendar. See the user guide. + +### Deprecated + +* `select --freebusyhack`, superseded by `freebusy-sync`. It keeps the UID of the original, and it leaves the summary, description and attendees of recurrence exceptions in place. + ### Changed * Shell tab completion scripts (bash, zsh, fish) now ship with the package. They work without shell configuration for system-wide installs (and for bash user/venv installs with bash-completion 2.12+); see the README for other setups. The `make install-completion` targets are gone. diff --git a/USER_GUIDE.md b/USER_GUIDE.md index 6a4c214..3cab278 100644 --- a/USER_GUIDE.md +++ b/USER_GUIDE.md @@ -31,6 +31,7 @@ Those commands are made mostly for making `plann` more convenient to use for the * agenda - list up some of the upcoming events plus some of the upcoming tasks * interactive manage-tasks - go through your tasks and make suggestions +* freebusy-sync - copy busy time between calendars, see below Note that those commands are mostly tested against Xandikos (which plann's functional tests run on) and earlier DAViCal (see the [`docs/CALENDAR_SERVER_RECOMMENDATIONS.md`](docs/CALENDAR_SERVER_RECOMMENDATIONS.md) file). @@ -156,6 +157,19 @@ plann select --todo --category computer-work --start=2022-04-04 --end=2022-05-05 plann select --todo --category computer-work --overdue edit --postpone=5d ``` +## Sharing busy time between calendars + +`freebusy-sync` copies every busy period in the selected calendars (`--config-section`, plus any given with `--caldav-url`/`--calendar-url`) into the calendar of the config section given with `--to-section`, which must resolve to exactly one calendar (set `calendar_url` or `calendar_name` in it). The copies hold only the time, in UTC, and a fixed summary - no description, location, attendees, timezone or UID from the original. Run it regularly, e.g. from a systemd timer: copies are moved or deleted when the original changes, and other events in the target calendar are left alone. Copies are never copied onward, so syncing both ways is safe: + +``` +plann --config-section work freebusy-sync --to-section private --summary 'event from work calendar' +plann --config-section private freebusy-sync --to-section work --summary 'event from the personal calendar' --hours 9-16 --weekdays-only +``` + +`--hours` and `--weekdays-only` limit what is copied to working hours; an all-day event then becomes an event covering those hours. `--dry-run` shows what would change. If any calendar fails to load, the run stops without changing anything. The copies carry an `X-PLANN-FREEBUSY-SOURCE` property naming the sync (`--tag`, defaulting to the `--config-section` value); don't change the tag of a running sync, or the old copies are orphaned. + +Switching from `select --freebusyhack`: delete the copies it made first. They carry neither the tag nor a recognisable UID, so `freebusy-sync` treats them as real events and copies them back the other way. + ## See also [NEXT_LEVEL.md](NEXT_LEVEL.md) describes some of my visions on what a good calendaring system should be capable of, and does an attempt on mapping this down to the icalendar standard. diff --git a/plann/cli.py b/plann/cli.py index 68f8448..b5296d6 100755 --- a/plann/cli.py +++ b/plann/cli.py @@ -17,6 +17,7 @@ ## cli as such. It should be moved out and made available through ## `from plann import ...` +import datetime import os import sys @@ -141,10 +142,14 @@ def cli(ctx, **kwargs): ## never touch the server - notably `plann --help` - don't ## connect to every configured calendar. ctx.obj['calendars'] = _LazyCalendars(lambda: _discover_calendars(kwargs)) + ## the discovery options, for a command that must tighten them before the + ## (lazy) discovery happens + ctx.obj['discovery_options'] = kwargs ## stashed for the `configure` subcommand (and any other command that needs ## to know which config file / section the user pointed at) ctx.obj['config_file'] = kwargs['config_file'] ctx.obj['config_section'] = kwargs['config_section'] + ctx.obj['raise_errors'] = kwargs['raise_errors'] for flag in ('show_native_timezone', 'store_timezone', 'implicit_timezone'): setattr(tz, flag, kwargs[flag]) @@ -156,10 +161,23 @@ def _discover_calendars(kwargs): config = read_config(kwargs['config_file']) if config: for meta_section in kwargs['config_section']: - for section in expand_config_section(config, meta_section): - calendars.extend(find_calendars(config_section(config, section), raise_errors=kwargs['raise_errors'])) + calendars.extend(_config_calendars(config, meta_section, kwargs['raise_errors'])) return calendars +def _config_calendars(config, meta_section, raise_errors): + calendars = [] + for section in expand_config_section(config, meta_section): + calendars.extend(find_calendars(config_section(config, section), raise_errors=raise_errors)) + return calendars + +def _section_calendars(ctx, meta_section): + """Find the calendars of one config-file section, independent of the + sections selected with --config-section. Errors are raised.""" + config = read_config(ctx.obj['config_file']) + if not config: + return [] + return _config_calendars(config, meta_section, raise_errors=True) + @cli.command() @click.pass_context def list_calendars(ctx): @@ -232,6 +250,53 @@ def _set_attr_options_(func, verb, desc=""): def _set_attr_options(verb="", desc=""): return lambda func: _set_attr_options_(func, verb, desc) +@cli.command() +@click.option('--to-section', required=True, help="Config section holding the target calendar (the first calendar found is used)") +@click.option('--summary', required=True, help="Summary of every copied event, e.g. 'event from work calendar'") +@click.option('--tag', help="Name of this sync, stored in the copies. Copies with the tag that no longer match a busy period are deleted. Defaults to the --config-section value(s)") +@click.option('--days', default=60, show_default=True, help="Sync from the start of today and this many days ahead") +@click.option('--hours', help="Only copy busy time within these local hours, e.g. 9-16. All-day events become events covering these hours") +@click.option('--weekdays-only/--all-days', default=False, help="Only copy busy time on Monday to Friday") +@click.option('--dry-run/--no-dry-run', default=False, help="Report what would change without changing anything") +@click.pass_context +def freebusy_sync(ctx, to_section, summary, tag, days, hours, weekdays_only, dry_run): + """ + Copy busy time from the selected calendars into another calendar. + + Each busy period in the calendars selected by --config-section becomes + an event in the --to-section calendar holding only the time and + SUMMARY. Run it regularly: copies are moved and deleted as the + source changes. Example, run from a timer: + + plann --config-section work freebusy-sync --to-section default --summary 'event from work calendar' + """ + from plann.freebusy import freebusy_sync as freebusy_sync_ + if hours: + try: + hours = tuple(int(x) for x in hours.split('-')) + except ValueError: + hours = () + if len(hours) != 2 or not 0 <= hours[0] < hours[1] <= 24: + raise click.BadParameter("expected two hours within 0-24, like 9-16", param_hint='--hours') + ## A source that fails to load must stop the run: read as "no busy time", + ## it would delete every copy. + ctx.obj['discovery_options']['raise_errors'] = True + try: + sources = list_type(ctx.obj['calendars']) + targets = _section_calendars(ctx, to_section) + except Exception as e: + _abort(f"Calendar discovery failed, nothing synced: {e}") + if not sources: + _abort("No source calendars found, nothing synced") + if len(targets) != 1: + _abort(f"Config section {to_section} must resolve to exactly one calendar, found {len(targets)}. Set calendar_url or calendar_name in it.") + start = _now().replace(hour=0, minute=0, second=0) + stats = freebusy_sync_( + sources, targets[0], start=start, end=start + datetime.timedelta(days=days), + summary=summary, tag=tag or ','.join(ctx.obj['config_section']), hours=hours, + weekdays=range(5) if weekdays_only else None, tzinfo=tz.implicit_timezone, dry_run=dry_run) + click.echo(("dry run, nothing changed: " if dry_run else "") + ", ".join(f"{k} {v}" for k, v in stats.items())) + @cli.group() @click.option('--interactive/--no-interactive-select', help="line based interactive filtering") @click.option('--mass-interactive/--no-mass-interactive-select', help="editor based interactive filtering") @@ -258,7 +323,7 @@ def _set_attr_options(verb="", desc=""): @click.option('--skip-children/--include-children', help="Skip children if it's parent is selected. Useful for getting an overview of the big picture if children are subtasks", default=False) @click.option('--limit', help='Number of objects to show', type=int) @click.option('--offset', help='Skip the first objects', type=int) -@click.option('--freebusyhack', help='removes almost everything from the ical and replaces the summary with the provided string. (this option is to be replaced with something better in a future release)') +@click.option('--freebusyhack', help='DEPRECATED, use the freebusy-sync command. Removes almost everything from the ical and replaces the summary with the provided string. Leaks the details of recurrence exceptions.') @click.option('--pinned-tasks/--no-pinned-tasks', default=None, help='select all/no pinned tasks') @click.pass_context def select(*largs, **kwargs): diff --git a/plann/commands.py b/plann/commands.py index e79fe41..5bb9df1 100644 --- a/plann/commands.py +++ b/plann/commands.py @@ -231,6 +231,7 @@ def __select(ctx, extend_objects=False, all=None, uid=[], abort_on_missing_uid=N ## (typically between private and work calendars) without carrying over ## (too much potentially) private/confidential information. if freebusyhack: + logging.warning("--freebusyhack is deprecated, use the freebusy-sync command") for obj in ctx.obj['objs']: comp = obj.icalendar_component attribs = list(comp.keys()) diff --git a/plann/freebusy.py b/plann/freebusy.py new file mode 100644 index 0000000..284e625 --- /dev/null +++ b/plann/freebusy.py @@ -0,0 +1,183 @@ +"""Copy busy time from one set of calendars into another calendar. + +Typical use is between a work calendar and a private calendar: the +target gets a placeholder event for every busy period in the source, +carrying nothing but the time and a fixed summary. + +Every copy has a derived UID and an X-PLANN-FREEBUSY-SOURCE property +holding a tag naming the sync. The tag lets a later run find, move and +delete its own copies without touching anything else in the target +calendar, and lets copies be recognised so they never sync back. +""" + +import datetime +import hashlib +from collections.abc import Iterable +from typing import Any + +import icalendar +from dateutil import tz as dateutil_tz + +SOURCE_PROP = 'X-PLANN-FREEBUSY-SOURCE' +UID_PREFIX = 'plann-freebusy-' + + +def _end_of(comp: icalendar.cal.Component) -> datetime.date | None: + start = comp['DTSTART'].dt + if 'DTEND' in comp: + return comp['DTEND'].dt + if 'DURATION' in comp: + return start + comp['DURATION'].dt + ## RFC 5545 section 3.6.1: an all-day event lasts one day, a timed one no time + if not isinstance(start, datetime.datetime): + return start + datetime.timedelta(days=1) + return None + + +def _clip(start: datetime.date, end: datetime.date, hours: tuple[int, int], weekdays: Iterable[int], + tzinfo: datetime.tzinfo) -> list[tuple[datetime.date, datetime.datetime, datetime.datetime]]: + """Cut a busy period down to the given hours (0-24), one piece per day. + + All-day periods become the full hours on every allowed day. + """ + weekdays = set(weekdays) + if isinstance(start, datetime.datetime): + start = start if start.tzinfo else start.replace(tzinfo=tzinfo) + end = end if end.tzinfo else end.replace(tzinfo=tzinfo) + first_day = start.astimezone(tzinfo).date() + last_day = end.astimezone(tzinfo).date() + else: + first_day = start + last_day = end - datetime.timedelta(days=1) + pieces = [] + day = first_day + while day <= last_day: + if day.weekday() in weekdays: + day_start = datetime.datetime.combine(day, datetime.time(hours[0]), tzinfo) + day_end = datetime.datetime.combine(day + datetime.timedelta(days=hours[1] // 24), + datetime.time(hours[1] % 24), tzinfo) + if isinstance(start, datetime.datetime): + piece_start, piece_end = max(start, day_start), min(end, day_end) + else: + piece_start, piece_end = day_start, day_end + if piece_start < piece_end: + pieces.append((day, piece_start, piece_end)) + day += datetime.timedelta(days=1) + return pieces + + +def _to_utc(value: datetime.date, tzinfo: datetime.tzinfo) -> datetime.date: + """Timestamps go out in UTC: no TZID to resolve, and no zone name leaks.""" + if not isinstance(value, datetime.datetime): + return value + if not value.tzinfo: + value = value.replace(tzinfo=tzinfo) + return value.astimezone(datetime.timezone.utc) + + +def busy_copies(comp: icalendar.cal.Component, summary: str, tag: str, hours: tuple[int, int] | None = None, + weekdays: Iterable[int] | None = None, tzinfo: datetime.tzinfo | None = None, + window: tuple[datetime.datetime, datetime.datetime] | None = None) -> list[icalendar.Event]: + """Return the placeholder events standing in for one (expanded) event. + + The result is empty for events that do not block time (transparent + or cancelled), and for events that are themselves copies. With + ``hours`` (a range within 0-24) or ``weekdays`` (0 is Monday) set, + the event is clipped to them in ``tzinfo``, which may split it into + one copy per day; ``window`` then drops the days outside it. + """ + if (str(comp.get('TRANSP', '')).upper() == 'TRANSPARENT' + or str(comp.get('STATUS', '')).upper() == 'CANCELLED' + or SOURCE_PROP in comp + or str(comp.get('UID', '')).startswith(UID_PREFIX) + or 'DTSTART' not in comp): + return [] + start = comp['DTSTART'].dt + end = _end_of(comp) + if end is None: + return [] + tzinfo = tzinfo or dateutil_tz.tzlocal() + recurrence_id = comp['RECURRENCE-ID'].to_ical().decode() if 'RECURRENCE-ID' in comp else '' + key = f"{tag}\0{comp['UID']}\0{recurrence_id}" + if hours or weekdays is not None: + pieces = _clip(start, end, hours or (0, 24), weekdays if weekdays is not None else range(7), tzinfo) + if window: + pieces = [p for p in pieces if p[1] < window[1] and p[2] > window[0]] + pieces = [(f"{key}\0{day.isoformat()}", s, e) for day, s, e in pieces] + else: + pieces = [(key, start, end)] + copies = [] + for piece_key, piece_start, piece_end in pieces: + copy = icalendar.Event() + copy.add('UID', UID_PREFIX + hashlib.sha1(piece_key.encode()).hexdigest()) + copy.add('DTSTAMP', datetime.datetime.now(datetime.timezone.utc)) + copy.add('DTSTART', _to_utc(piece_start, tzinfo)) + copy.add('DTEND', _to_utc(piece_end, tzinfo)) + copy.add('SUMMARY', summary) + copy.add('TRANSP', 'OPAQUE') + copy.add(SOURCE_PROP, tag) + copies.append(copy) + return copies + + +def _vcal(comp: icalendar.cal.Component) -> str: + cal = icalendar.Calendar() + cal.add('PRODID', '-//plann//freebusy-sync//EN') + cal.add('VERSION', '2.0') + cal.add_component(comp) + return cal.to_ical().decode() + + +def _same(a: icalendar.cal.Component, b: icalendar.cal.Component) -> bool: + return all(a.get(p) is not None and b.get(p) is not None and a[p].dt == b[p].dt + for p in ('DTSTART', 'DTEND')) and str(a.get('SUMMARY')) == str(b.get('SUMMARY')) + + +def freebusy_sync(source_calendars: Iterable[Any], target: Any, start: datetime.datetime, end: datetime.datetime, + summary: str, tag: str, hours: tuple[int, int] | None = None, weekdays: Iterable[int] | None = None, + tzinfo: datetime.tzinfo | None = None, dry_run: bool = False) -> dict[str, int]: + """Make the copies tagged ``tag`` in ``target`` mirror the busy time in + ``source_calendars`` between ``start`` and ``end``. + + Copies are added, moved or deleted as needed; other objects in the + target are never touched. Returns how many copies were added, + updated, deleted and left unchanged. + + Raises ValueError when there are no source calendars, since an + unreachable source would otherwise read as no busy time at all and + delete every copy. + """ + source_calendars = list(source_calendars) + if not source_calendars: + raise ValueError("no source calendars; refusing to sync, as that would delete every copy") + wanted = {} + for calendar in source_calendars: + for obj in calendar.search(event=True, start=start, end=end, expand=True): + for copy in busy_copies(obj.icalendar_component, summary, tag, hours, weekdays, tzinfo, (start, end)): + wanted[str(copy['UID'])] = copy + + existing = {} + for obj in target.search(event=True, start=start, end=end, expand=False): + comp = obj.icalendar_component + if str(comp.get(SOURCE_PROP, '')) == tag: + existing[str(comp['UID'])] = obj + + stats = {'added': 0, 'updated': 0, 'deleted': 0, 'unchanged': 0} + for uid, copy in wanted.items(): + obj = existing.pop(uid, None) + if obj is None: + stats['added'] += 1 + if not dry_run: + target.add_event(_vcal(copy)) + elif _same(obj.icalendar_component, copy): + stats['unchanged'] += 1 + else: + stats['updated'] += 1 + if not dry_run: + obj.data = _vcal(copy) + obj.save() + for obj in existing.values(): + stats['deleted'] += 1 + if not dry_run: + obj.delete() + return stats diff --git a/tests/test_freebusy.py b/tests/test_freebusy.py new file mode 100644 index 0000000..f31a106 --- /dev/null +++ b/tests/test_freebusy.py @@ -0,0 +1,267 @@ +from datetime import date, datetime, timedelta +from zoneinfo import ZoneInfo + +import icalendar +import pytest + +from plann.freebusy import SOURCE_PROP, _end_of, busy_copies, freebusy_sync + +oslo = ZoneInfo('Europe/Oslo') + + +def _event(uid='ev1', summary='Secret meeting with ACME', start=datetime(2026, 10, 14, 10, 0, tzinfo=oslo), + end=datetime(2026, 10, 14, 11, 0, tzinfo=oslo), **props): + ev = icalendar.Event() + ev.add('UID', uid) + ev.add('DTSTAMP', datetime(2026, 10, 1, tzinfo=oslo)) + ev.add('SUMMARY', summary) + ev.add('DESCRIPTION', 'confidential details') + ev.add('LOCATION', 'ACME HQ') + ev.add('DTSTART', start) + if end is not None: + ev.add('DTEND', end) + for key, value in props.items(): + ev.add(key.replace('_', '-'), value) + return ev + + +class FakeObject: + def __init__(self, calendar, ical): + self.calendar = calendar + self.data = ical + + @property + def icalendar_component(self): + return [c for c in icalendar.Calendar.from_ical(self.data).walk('VEVENT')][0] + + def save(self): + self.calendar.saves += 1 + + def delete(self): + self.calendar.objects.remove(self) + + +def _as_dt(value): + if isinstance(value, datetime): + return value + return datetime.combine(value, datetime.min.time(), oslo) + + +class FakeCalendar: + """Just enough of caldav.Calendar for freebusy_sync. + + search() ignores the time range; the tests keep everything inside it. + """ + def __init__(self, components=()): + self.objects = [] + self.saves = 0 + for comp in components: + self.add_event(_vcal(comp)) + + def search(self, event=None, start=None, end=None, expand=None): + ## time-range overlap only; no recurrence expansion + def overlaps(obj): + comp = obj.icalendar_component + s, e = _as_dt(comp['DTSTART'].dt), _as_dt(_end_of(comp) or comp['DTSTART'].dt) + return (end is None or s < end) and (start is None or e > start) + return [o for o in self.objects if overlaps(o)] + + def add_event(self, ical): + obj = FakeObject(self, ical if isinstance(ical, str) else ical.decode()) + self.objects.append(obj) + return obj + + def components(self): + return [o.icalendar_component for o in self.objects] + + +def _vcal(comp): + cal = icalendar.Calendar() + cal.add('PRODID', '-//test//') + cal.add('VERSION', '2.0') + cal.add_component(comp) + return cal.to_ical().decode() + + +class TestBusyCopies: + def test_strips_everything_but_the_time(self): + [copy] = busy_copies(_event(), summary='busy', tag='work') + assert str(copy['SUMMARY']) == 'busy' + for prop in ('DESCRIPTION', 'LOCATION'): + assert prop not in copy + assert copy['DTSTART'].dt == datetime(2026, 10, 14, 10, 0, tzinfo=oslo) + assert copy['DTEND'].dt == datetime(2026, 10, 14, 11, 0, tzinfo=oslo) + assert str(copy[SOURCE_PROP]) == 'work' + + def test_times_are_stored_in_utc(self): + ## no TZID to resolve, and the source's zone name does not leak + ev = _event() + ev['DTSTART'].params['TZID'] = 'W. Europe Standard Time' + [copy] = busy_copies(ev, summary='busy', tag='work') + assert copy['DTSTART'].dt.utcoffset() == timedelta(0) + assert b'TZID' not in copy.to_ical() + [clipped] = busy_copies(_event(start=date(2026, 10, 14), end=date(2026, 10, 15)), summary='busy', tag='home', + hours=(9, 16), tzinfo=oslo) + assert clipped['DTSTART'].dt == datetime(2026, 10, 14, 7, 0, tzinfo=ZoneInfo('UTC')) + assert b'TZID' not in clipped.to_ical() + + def test_skips_copies_whose_tag_property_was_dropped(self): + [copy] = busy_copies(_event(), summary='busy', tag='work') + del copy[SOURCE_PROP] + assert busy_copies(copy, summary='busy', tag='home') == [] + + def test_uid_is_derived_stable_and_hides_the_source_uid(self): + [a] = busy_copies(_event(uid='secret-uid@acme.example.com'), summary='busy', tag='work') + [b] = busy_copies(_event(uid='secret-uid@acme.example.com'), summary='busy', tag='work') + [c] = busy_copies(_event(uid='secret-uid@acme.example.com'), summary='busy', tag='home') + assert a['UID'] == b['UID'] + assert a['UID'] != c['UID'] + assert 'acme' not in str(a['UID']) + + def test_recurrence_instances_get_distinct_uids(self): + first = _event(recurrence_id=datetime(2026, 10, 14, 10, 0, tzinfo=oslo)) + second = _event(start=datetime(2026, 10, 21, 10, 0, tzinfo=oslo), end=datetime(2026, 10, 21, 11, 0, tzinfo=oslo), + recurrence_id=datetime(2026, 10, 21, 10, 0, tzinfo=oslo)) + [a] = busy_copies(first, summary='busy', tag='work') + [b] = busy_copies(second, summary='busy', tag='work') + assert a['UID'] != b['UID'] + assert 'RECURRENCE-ID' not in a + + def test_skips_free_cancelled_and_copies(self): + assert busy_copies(_event(transp='TRANSPARENT'), summary='busy', tag='work') == [] + assert busy_copies(_event(status='CANCELLED'), summary='busy', tag='work') == [] + [copy] = busy_copies(_event(), summary='busy', tag='work') + assert busy_copies(copy, summary='busy', tag='home') == [] + + def test_duration_and_missing_end(self): + [copy] = busy_copies(_event(end=None, duration=timedelta(minutes=30)), summary='busy', tag='work') + assert copy['DTEND'].dt == datetime(2026, 10, 14, 10, 30, tzinfo=oslo) + ## RFC 5545: a timed event without DTEND or DURATION takes no time + assert busy_copies(_event(end=None), summary='busy', tag='work') == [] + + def test_all_day_event_kept_as_all_day_without_hours(self): + [copy] = busy_copies(_event(start=date(2026, 10, 14), end=date(2026, 10, 15)), summary='busy', tag='work') + assert copy['DTSTART'].dt == date(2026, 10, 14) + assert copy['DTEND'].dt == date(2026, 10, 15) + + def test_hours_clip_a_timed_event(self): + ev = _event(start=datetime(2026, 10, 14, 7, 0, tzinfo=oslo), end=datetime(2026, 10, 14, 10, 0, tzinfo=oslo)) + [copy] = busy_copies(ev, summary='busy', tag='home', hours=(9, 16), weekdays=range(5), tzinfo=oslo) + assert copy['DTSTART'].dt == datetime(2026, 10, 14, 9, 0, tzinfo=oslo) + assert copy['DTEND'].dt == datetime(2026, 10, 14, 10, 0, tzinfo=oslo) + + def test_hours_drop_events_outside_and_weekends(self): + evening = _event(start=datetime(2026, 10, 14, 18, 0, tzinfo=oslo), end=datetime(2026, 10, 14, 20, 0, tzinfo=oslo)) + saturday = _event(start=datetime(2026, 10, 17, 10, 0, tzinfo=oslo), end=datetime(2026, 10, 17, 11, 0, tzinfo=oslo)) + for ev in (evening, saturday): + assert busy_copies(ev, summary='busy', tag='home', hours=(9, 16), weekdays=range(5), tzinfo=oslo) == [] + + def test_hours_turn_all_day_events_into_working_hours_per_weekday(self): + ## Friday to Monday (exclusive end Tuesday): Friday and Monday remain + ev = _event(start=date(2026, 10, 16), end=date(2026, 10, 20)) + copies = busy_copies(ev, summary='busy', tag='home', hours=(9, 16), weekdays=range(5), tzinfo=oslo) + assert [(c['DTSTART'].dt, c['DTEND'].dt) for c in copies] == [ + (datetime(2026, 10, 16, 9, 0, tzinfo=oslo), datetime(2026, 10, 16, 16, 0, tzinfo=oslo)), + (datetime(2026, 10, 19, 9, 0, tzinfo=oslo), datetime(2026, 10, 19, 16, 0, tzinfo=oslo)), + ] + assert len({c['UID'] for c in copies}) == 2 + + + def test_hours_until_midnight(self): + ev = _event(start=datetime(2026, 10, 14, 22, 0, tzinfo=oslo), end=datetime(2026, 10, 15, 1, 0, tzinfo=oslo)) + copies = busy_copies(ev, summary='busy', tag='home', hours=(20, 24), tzinfo=oslo) + assert [(c['DTSTART'].dt, c['DTEND'].dt) for c in copies] == [ + (datetime(2026, 10, 14, 22, 0, tzinfo=oslo), datetime(2026, 10, 15, 0, 0, tzinfo=oslo))] + + def test_weekdays_without_hours(self): + saturday = _event(start=datetime(2026, 10, 17, 10, 0, tzinfo=oslo), end=datetime(2026, 10, 17, 11, 0, tzinfo=oslo)) + assert busy_copies(saturday, summary='busy', tag='home', weekdays=range(5), tzinfo=oslo) == [] + [copy] = busy_copies(_event(), summary='busy', tag='home', weekdays=range(5), tzinfo=oslo) + assert copy['DTSTART'].dt == datetime(2026, 10, 14, 10, 0, tzinfo=oslo) + + def test_pieces_outside_the_window_are_dropped(self): + holiday = _event(start=date(2026, 10, 12), end=date(2026, 10, 24)) + copies = busy_copies(holiday, summary='busy', tag='home', hours=(9, 16), weekdays=range(5), tzinfo=oslo, + window=(datetime(2026, 10, 14, tzinfo=oslo), datetime(2026, 10, 16, tzinfo=oslo))) + assert [c['DTSTART'].dt.astimezone(oslo).date() for c in copies] == [date(2026, 10, 14), date(2026, 10, 15)] + + +class TestFreebusySync: + window = dict(start=datetime(2026, 10, 13, tzinfo=oslo), end=datetime(2026, 12, 12, tzinfo=oslo)) + + def _sync(self, source, target, **kwargs): + return freebusy_sync([source], target, summary='event from work calendar', tag='work', **self.window, **kwargs) + + def test_adds_copies(self): + source = FakeCalendar([_event()]) + target = FakeCalendar() + assert self._sync(source, target) == {'added': 1, 'updated': 0, 'deleted': 0, 'unchanged': 0} + [copy] = target.components() + assert str(copy['SUMMARY']) == 'event from work calendar' + assert 'DESCRIPTION' not in copy + + def test_second_run_is_a_noop(self): + source = FakeCalendar([_event()]) + target = FakeCalendar() + self._sync(source, target) + assert self._sync(source, target) == {'added': 0, 'updated': 0, 'deleted': 0, 'unchanged': 1} + assert len(target.objects) == 1 + + def test_moved_source_updates_the_copy(self): + source = FakeCalendar([_event()]) + target = FakeCalendar() + self._sync(source, target) + source = FakeCalendar([_event(start=datetime(2026, 10, 14, 12, 0, tzinfo=oslo), end=datetime(2026, 10, 14, 13, 0, tzinfo=oslo))]) + assert self._sync(source, target)['updated'] == 1 + [copy] = target.components() + assert copy['DTSTART'].dt == datetime(2026, 10, 14, 12, 0, tzinfo=oslo) + assert target.saves == 1 + + def test_removed_source_deletes_the_copy_but_nothing_else(self): + own = _event(uid='my-own-event', summary='dentist') + target = FakeCalendar([own]) + self._sync(FakeCalendar([_event()]), target) + assert len(target.objects) == 2 + assert self._sync(FakeCalendar(), target)['deleted'] == 1 + assert [str(c['UID']) for c in target.components()] == ['my-own-event'] + + def test_copies_from_another_tag_are_left_alone(self): + target = FakeCalendar() + freebusy_sync([FakeCalendar([_event()])], target, summary='x', tag='other', **self.window) + assert self._sync(FakeCalendar(), target)['deleted'] == 0 + assert len(target.objects) == 1 + + def test_copies_in_the_source_are_not_synced_back(self): + ## private -> work, then work -> private must not bounce the copy back + private = FakeCalendar([_event(uid='p1', summary='dentist')]) + work = FakeCalendar() + freebusy_sync([private], work, summary='private', tag='private', **self.window) + self._sync(work, private) + assert len(private.objects) == 1 + + def test_dry_run_changes_nothing(self): + target = FakeCalendar() + assert self._sync(FakeCalendar([_event()]), target, dry_run=True)['added'] == 1 + assert target.objects == [] + + def test_window_edge_settles(self): + ## a holiday starting before the window must not be re-added every run + source = FakeCalendar([_event(start=date(2026, 10, 5), end=date(2026, 10, 24))]) + target = FakeCalendar() + self._sync(source, target, hours=(9, 16), weekdays=range(5), tzinfo=oslo) + assert self._sync(source, target, hours=(9, 16), weekdays=range(5), tzinfo=oslo)['added'] == 0 + + def test_move_to_another_zone_at_the_same_wall_time(self): + target = FakeCalendar() + self._sync(FakeCalendar([_event()]), target) + ny = ZoneInfo('America/New_York') + moved = _event(start=datetime(2026, 10, 14, 10, 0, tzinfo=ny), end=datetime(2026, 10, 14, 11, 0, tzinfo=ny)) + assert self._sync(FakeCalendar([moved]), target)['updated'] == 1 + + def test_refuses_without_source_calendars(self): + ## an unreachable source must not read as "no busy time": that would delete every copy + target = FakeCalendar() + self._sync(FakeCalendar([_event()]), target) + with pytest.raises(ValueError): + freebusy_sync([], target, summary='x', tag='work', **self.window) + assert len(target.objects) == 1 diff --git a/tests/test_plann_cli.py b/tests/test_plann_cli.py index cd9de7d..2db67a7 100644 --- a/tests/test_plann_cli.py +++ b/tests/test_plann_cli.py @@ -1,5 +1,6 @@ ## Check https://click.palletsprojects.com/en/8.1.x/testing/ +from datetime import datetime, timedelta from unittest.mock import patch import caldav @@ -10,6 +11,7 @@ import plann.commands as commands_mod from plann.cli import _LazyCalendars, cli, delete, edit from plann.commands import _process_add_args, _set_task_attribs, _sort_key_function, _todos_missing +from tests.test_freebusy import FakeCalendar, _event, oslo def _make_todo(uid, *, category=False, due=False, priority=False, dtstart=False): @@ -350,3 +352,97 @@ def test_delete_reports_empty_selection(): res = CliRunner().invoke(delete, obj={'objs': []}) assert res.exit_code == 0, res.output assert res.output.strip(), "expected some feedback on an empty selection" + + +def test_freebusyhack_marked_deprecated(): + """--freebusyhack is superseded by the freebusy-sync command.""" + opt = _find_option(cli.commands['select'], '--freebusyhack') + assert 'deprecat' in (opt.help or '').lower() + + +def test_freebusy_sync_command(): + """freebusy-sync copies from the selected calendars into the first + calendar of --to-section, clipping with --hours and --weekdays-only.""" + day = (datetime.now(oslo) + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0) + while day.weekday() > 4: + day += timedelta(days=1) + source = FakeCalendar([_event(start=day.replace(hour=7), end=day.replace(hour=10))]) + target = FakeCalendar() + sections = [] + + def section_calendars(ctx, section): + sections.append(section) + return [target] + + ## --implicit-timezone sets the process-wide tz; restore it for later tests + with patch.object(cli_mod, '_discover_calendars', lambda kwargs: [source]), \ + patch.object(cli_mod, '_section_calendars', section_calendars), \ + patch.object(cli_mod.tz, '_implicit_timezone', cli_mod.tz.implicit_timezone), \ + patch.object(cli_mod.tz, '_store_timezone', cli_mod.tz.store_timezone): + res = CliRunner().invoke(cli, [ + '--implicit-timezone', 'Europe/Oslo', 'freebusy-sync', '--to-section', 'work', + '--summary', 'event from the personal calendar', '--hours', '9-16', '--weekdays-only']) + assert res.exit_code == 0, res.output + assert sections == ['work'] + assert 'added 1' in res.output + assert 'dry run' not in res.output + [copy] = target.components() + assert copy['DTSTART'].dt == day.replace(hour=9) + assert copy['DTEND'].dt == day.replace(hour=10) + assert str(copy['X-PLANN-FREEBUSY-SOURCE']) == 'default' + + +def _run_freebusy_sync(source_calendars, target_calendars, *args, config=None): + config = config or {'work': {}} + with patch.object(cli_mod, '_discover_calendars', lambda kwargs: list(source_calendars)), \ + patch.object(cli_mod, 'read_config', lambda *a, **k: config), \ + patch.object(cli_mod, 'find_calendars', lambda *a, **k: list(target_calendars)), \ + patch.object(cli_mod.tz, '_implicit_timezone', cli_mod.tz.implicit_timezone), \ + patch.object(cli_mod.tz, '_store_timezone', cli_mod.tz.store_timezone): + return CliRunner().invoke(cli, ['freebusy-sync', '--to-section', 'work', '--summary', 'x', *args]) + + +def test_freebusy_sync_needs_exactly_one_target(): + res = _run_freebusy_sync([FakeCalendar()], [FakeCalendar(), FakeCalendar()]) + assert res.exit_code != 0 + assert 'exactly one' in res.output + + +def test_freebusy_sync_aborts_without_source_calendars(): + target = FakeCalendar() + res = _run_freebusy_sync([], [target]) + assert res.exit_code != 0 + assert target.objects == [] + + +def test_freebusy_sync_rejects_bad_hours(): + res = _run_freebusy_sync([FakeCalendar()], [FakeCalendar()], '--hours', '9-25') + assert res.exit_code != 0 + assert '--hours' in res.output + + +def test_freebusy_sync_dry_run_wording(): + res = _run_freebusy_sync([FakeCalendar()], [FakeCalendar()], '--dry-run') + assert res.exit_code == 0, res.output + assert res.output.startswith('dry run, nothing changed: added 0') + + +def test_freebusy_sync_discovers_with_raise_errors(): + """A source that fails to load must abort the run, not count as empty.""" + seen = {} + + def discover(kwargs): + seen['raise_errors'] = kwargs['raise_errors'] + return [FakeCalendar()] + + with patch.object(cli_mod, '_discover_calendars', discover), \ + patch.object(cli_mod, '_section_calendars', lambda ctx, section: [FakeCalendar()]): + res = CliRunner().invoke(cli, ['freebusy-sync', '--to-section', 'work', '--summary', 'x']) + assert res.exit_code == 0, res.output + assert seen['raise_errors'] is True + + +def test_freebusyhack_warns(caplog): + ctx = _FakeCtx() + commands_mod._select(ctx, freebusyhack='busy') + assert 'deprecated' in caplog.text