diff options
Diffstat (limited to 'desktop/island/calendar')
| -rwxr-xr-x | desktop/island/calendar/calctl | 158 | ||||
| -rw-r--r-- | desktop/island/calendar/test_calctl.py | 80 |
2 files changed, 237 insertions, 1 deletions
diff --git a/desktop/island/calendar/calctl b/desktop/island/calendar/calctl index 7988f0b..d79d8b9 100755 --- a/desktop/island/calendar/calctl +++ b/desktop/island/calendar/calctl @@ -280,6 +280,158 @@ def cmd_get(uid): return to_json(cal, ev, start, end, all_day, overrides()) +def parse_payload(text, cals): + """The form's JSON, checked. Dates become date/datetime; an all-day end + becomes exclusive.""" + try: + p = json.loads(text) + except json.JSONDecodeError as e: + raise Fail(f"bad event json: {e}") + if not isinstance(p, dict): + raise Fail("bad event json: not an object") + by_name = {c["name"]: c for c in cals} + if p.get("calendar") not in by_name: + raise Fail(f"unknown calendar {p.get('calendar')!r}") + title = str(p.get("title", "")).strip() + if not title: + raise Fail("title is empty") + all_day = bool(p.get("allDay")) + try: + if all_day: + start = date.fromisoformat(p["start"]) + end = date.fromisoformat(p["end"]) + timedelta(days=1) + else: + start = datetime.fromisoformat(p["start"]) + end = datetime.fromisoformat(p["end"]) + until = date.fromisoformat(p["until"]) if p.get("until") else None + except (KeyError, TypeError, ValueError) as e: + raise Fail(f"bad date: {e}") + if end < start: + raise Fail("end is before start") + repeat = p.get("repeat", "none") + if repeat not in PRESET_REPEAT and repeat not in ("none", "custom"): + raise Fail(f"unknown repeat {repeat!r}") + alarm = p.get("alarm", "none") + if alarm not in PRESET_ALARM and alarm not in ("none", "custom"): + raise Fail(f"unknown alarm {alarm!r}") + return {"cal": by_name[p["calendar"]], "title": title, "allDay": all_day, + "start": start, "end": end, "until": until, "repeat": repeat, "alarm": alarm, + "location": str(p.get("location", "")).strip(), "notes": str(p.get("notes", ""))} + + +def apply(ev, p): + """Write the form's fields into a VEVENT. Everything else on it is left + alone; a `custom` repeat or alarm keeps what is there.""" + for key in ("DTSTART", "DTEND", "DURATION", "SUMMARY", "LOCATION", "DESCRIPTION", + "DTSTAMP", "LAST-MODIFIED"): + ev.pop(key, None) + if p["allDay"]: + ev.add("DTSTART", p["start"]) + ev.add("DTEND", p["end"]) + else: + ev.add("DTSTART", p["start"].replace(tzinfo=LOCAL)) + ev.add("DTEND", p["end"].replace(tzinfo=LOCAL)) + ev.add("SUMMARY", p["title"]) + if p["location"]: + ev.add("LOCATION", p["location"]) + if p["notes"]: + ev.add("DESCRIPTION", p["notes"]) + if p["repeat"] != "custom": + ev.pop("RRULE", None) + if p["repeat"] != "none": + rule = {"FREQ": PRESET_REPEAT[p["repeat"]]} + if p["until"]: + # RFC 5545: UNTIL is a DATE for a DATE start, UTC otherwise. + rule["UNTIL"] = (p["until"] if p["allDay"] else + datetime.combine(p["until"], time(23, 59, 59), LOCAL) + .astimezone(timezone.utc)) + ev.add("RRULE", rule) + if p["alarm"] != "custom": + ev.subcomponents = [c for c in ev.subcomponents if c.name != "VALARM"] + if p["alarm"] != "none": + alarm = icalendar.Alarm() + alarm.add("ACTION", "DISPLAY") + alarm.add("DESCRIPTION", p["title"]) + alarm.add("TRIGGER", -PRESET_ALARM[p["alarm"]]) + ev.add_component(alarm) + now = datetime.now(timezone.utc).replace(microsecond=0) + ev.add("DTSTAMP", now) + ev.add("LAST-MODIFIED", now) + + +def write(path, ics): + """Temporary file in the same directory, then rename: a reader (khal, + cal-notif, vdirsyncer, the panel's watch) never sees half a file. The + temporary name does not end in .ics, so nothing lists it.""" + fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=".calctl-", suffix=".tmp") + try: + with os.fdopen(fd, "wb") as f: + f.write(ics.to_ical()) + os.replace(tmp, path) + except BaseException: + if os.path.exists(tmp): + os.unlink(tmp) + raise + + +def sync(): + """Push the change now rather than at the next cron run. Detached and + not waited for; a failure only means the cron run carries it.""" + if os.environ.get("CALCTL_NOSYNC"): + return + try: + SYNC_LOG.parent.mkdir(parents=True, exist_ok=True) + with open(SYNC_LOG, "ab") as log: + subprocess.Popen(["vdirsyncer", "-v", "ERROR", "sync"], stdin=subprocess.DEVNULL, + stdout=log, stderr=subprocess.STDOUT, start_new_session=True) + except OSError: + pass + + +def cmd_new(text): + cals, _ = calendars() + p = parse_payload(text, cals) + uid = str(uuid.uuid4()) + ics = icalendar.Calendar() + ics.add("PRODID", "-//quickshell//calctl//EN") + ics.add("VERSION", "2.0") + ev = icalendar.Event() + ev.add("UID", uid) + apply(ev, p) + ics.add_component(ev) + ics.add_missing_timezones() + write(p["cal"]["path"] / f"{uid}.ics", ics) + sync() + return {"uid": uid} + + +def cmd_edit(uid, text): + cals, _ = calendars() + p = parse_payload(text, cals) + _, path = find(uid) + ics = icalendar.Calendar.from_ical(path.read_bytes()) + ev = master(ics) + apply(ev, p) + sequence = int(ev.pop("SEQUENCE", 0)) + ev.add("SEQUENCE", sequence + 1) + ics.add_missing_timezones() + target = p["cal"]["path"] / path.name + if target != path and target.exists(): + raise Fail(f"{target.name} already exists in {p['cal']['name']}") + write(target, ics) + if target != path: + path.unlink() + sync() + return {"uid": uid} + + +def cmd_delete(uid): + _, path = find(uid) + path.unlink() + sync() + return {} + + def main(argv): cmd, args = (argv[1], argv[2:]) if len(argv) > 1 else ("", []) try: @@ -292,6 +444,12 @@ def main(argv): data = cmd_list(*args) elif cmd == "get" and len(args) == 1: data = cmd_get(*args) + elif cmd == "new" and len(args) == 1: + data = cmd_new(*args) + elif cmd == "edit" and len(args) == 2: + data = cmd_edit(*args) + elif cmd == "delete" and len(args) == 1: + data = cmd_delete(*args) else: raise Fail(USAGE) except Exception as e: diff --git a/desktop/island/calendar/test_calctl.py b/desktop/island/calendar/test_calctl.py index e6acd10..98973ee 100644 --- a/desktop/island/calendar/test_calctl.py +++ b/desktop/island/calendar/test_calctl.py @@ -131,7 +131,85 @@ assert run("list", "2026-10-01", "2026-11-01") assert "no event" in run("get", "nope", ok=False) assert "usage" in run("frobnicate", ok=False) -# --- writes (Task 3) go here --- +def ev(**kw): + e = {"calendar": "home", "title": "Dentist", "allDay": False, + "start": "2026-10-15T14:30", "end": "2026-10-15T15:30", + "repeat": "none", "until": "", "alarm": "1h", "location": "", "notes": ""} + e.update(kw) + return json.dumps(e) + + +def no_temp(): + assert not list(tmp.rglob(".calctl-*")), "temp file left behind" + + +# new, timed: the local zone with its VTIMEZONE, the alarm; read back in +# local time. +uid = run("new", ev())["uid"] +f = tmp / "a" / f"{uid}.ics" +t = f.read_text() +assert "DTSTART;TZID=Europe/Rome:20261015T143000" in t, t +assert "BEGIN:VTIMEZONE" in t and "TRIGGER:-PT1H" in t, t +no_temp() +mine = [e for e in run("list", "2026-10-01", "2026-11-01") if e["uid"] == uid] +assert len(mine) == 1, mine +assert mine[0]["start"] == "2026-10-15T14:30" and mine[0]["end"] == "2026-10-15T15:30" +assert mine[0]["alarm"] == "1h" and mine[0]["repeat"] == "none" and mine[0]["color"] == "green" +assert mine[0]["override"] is False +assert run("get", uid)["title"] == "Dentist" + +# all-day, multi-day: exclusive DTEND on disk, inclusive end in JSON. +trip = run("new", ev(title="Trip", allDay=True, start="2026-10-30", end="2026-11-02", alarm="none"))["uid"] +t = (tmp / "a" / f"{trip}.ics").read_text() +assert "DTSTART;VALUE=DATE:20261030" in t and "DTEND;VALUE=DATE:20261103" in t, t +assert "VALARM" not in t +nov = [e for e in run("list", "2026-11-01", "2026-12-01") if e["uid"] == trip] +assert len(nov) == 1 and nov[0]["start"] == "2026-10-30" and nov[0]["end"] == "2026-11-02", nov +assert not [e for e in run("list", "2026-10-01", "2026-10-30") if e["uid"] == trip] + +# weekly with an until-date, across the DST change. +gym = run("new", ev(title="Gym", repeat="weekly", until="2026-10-29"))["uid"] +occ = [e["start"] for e in run("list", "2026-10-01", "2026-12-01") if e["uid"] == gym] +assert occ == ["2026-10-15T14:30", "2026-10-22T14:30", "2026-10-29T14:30"], occ +assert run("get", gym)["until"] == "2026-10-29" + +# edit keeping custom: the rich rule, both alarms and an unknown property +# survive; SEQUENCE goes up. +run("edit", "weird-1", ev(title="Standup v2", start="2026-10-05T09:30", + end="2026-10-05T10:00", repeat="custom", alarm="custom")) +t = (tmp / "a" / "weird.ics").read_text() +assert "BYDAY=MO,WE" in t and t.count("BEGIN:VALARM") == 2 and "X-FOO:bar" in t, t +assert "SUMMARY:Standup v2" in t and "SEQUENCE:1" in t, t +no_temp() + +# edit to presets: the rule goes, the alarms become one. +run("edit", "weird-1", ev(title="Standup v2", start="2026-10-05T09:30", + end="2026-10-05T10:00", repeat="none", alarm="10m")) +t = (tmp / "a" / "weird.ics").read_text() +assert "RRULE" not in t and t.count("BEGIN:VALARM") == 1 and "TRIGGER:-PT10M" in t, t + +# a calendar change moves the file. +run("edit", uid, ev(calendar="birthdays")) +assert not f.exists() and (tmp / "b" / f.name).exists() +assert run("get", uid)["calendar"] == "birthdays" + +# cal-notif's per-event override is reported. +(tmp / "overrides.toml").write_text(f'"{uid}" = ["10m"]\n') +assert run("get", uid)["override"] is True + +# delete. +assert run("delete", uid) == {} +assert not (tmp / "b" / f.name).exists() +assert "no event" in run("get", uid, ok=False) + +# validation: the error envelope, exit 1, nothing written. +before = sorted(p.name for p in tmp.rglob("*.ics")) +assert "title" in run("new", ev(title=" "), ok=False) +assert "end" in run("new", ev(end="2026-10-15T14:00"), ok=False) +assert "calendar" in run("new", ev(calendar="nope"), ok=False) +assert "json" in run("new", "{not json", ok=False) +assert sorted(p.name for p in tmp.rglob("*.ics")) == before +no_temp() shutil.rmtree(tmp) print("test_calctl: ok") |