diff --git a/ex_app/lib/agent.py b/ex_app/lib/agent.py index 779b108b..05aff9c0 100644 --- a/ex_app/lib/agent.py +++ b/ex_app/lib/agent.py @@ -5,7 +5,7 @@ import random import string from collections.abc import Awaitable, Callable -from datetime import date +from datetime import date, datetime from time import monotonic from typing import Any, cast @@ -14,6 +14,9 @@ from nc_py_api import AsyncNextcloudApp from nc_py_api.ex_app import persistent_storage +import pytz + +from ex_app.lib.all_tools.lib.tz import user_timezone_async from ex_app.lib.all_tools.nextcloud_links import get_absolute_base_url from ex_app.lib.all_tools.skills import list_skills_metadata from ex_app.lib.graph import AgentState, get_graph @@ -112,6 +115,14 @@ async def react( model.bind_nextcloud(nc) model.multimodal = multimodal + tz_name = user_tz = None + try: + tz_name = await user_timezone_async(nc) + if tz_name: + user_tz = pytz.timezone(tz_name) + except Exception: + tz_name = user_tz = None + safe_tools, dangerous_tools = await get_tools(nc) tools = dangerous_tools + safe_tools @@ -130,12 +141,19 @@ async def call_model( state: AgentState, config: RunnableConfig, ): - current_date = date.today().strftime("%Y-%m-%d") + if user_tz is not None: + current_date = datetime.now(user_tz).strftime("%Y-%m-%d") + date_suffix = ( + f" The user's timezone is {tz_name}; use it for the timezone parameter of calendar and task tools." + ) + else: + current_date = date.today().strftime("%Y-%m-%d") + date_suffix = "" system_prompt_text = """ You are a helpful AI assistant with access to tools, please respond to the user's query to the best of your ability, using the provided tools if necessary. If no tool is needed to provide a correct answer, do not use one. If you used a tool, you still need to convey its output to the user. Use the same language for your answers as the user used in their message. -Today is {CURRENT_DATE}. +Today is {CURRENT_DATE}.{DATE_SUFFIX} Intuit the language the user is using (there is no tool for this, you will need to guess). Reply in the language intuited. Do not output the language you intuited. Only use tools if you cannot answer the user without them. If you get a link as a tool output, always add the link to your response. @@ -152,8 +170,10 @@ async def call_model( system_prompt_text += "Use the find_person_in_contacts tool to find a person's email address and location.\n" if tool_enabled("find_person_in_users"): system_prompt_text += "Use the find_person_in_users tool to find a person's userId and user details.\n" - if tool_enabled("find_details_of_current_user"): + if tz_name is None and tool_enabled("find_details_of_current_user"): system_prompt_text += "Use the find_details_of_current_user tool to find the current user's location and timezone.\n" + if tool_enabled("get_current_datetime"): + system_prompt_text += "Use the get_current_datetime tool when the user asks for the current date or time.\n" if tool_enabled("list_mails"): system_prompt_text += "Always check for the mail account id before requesting a folder list.\n" if tool_enabled("web_fetch"): @@ -189,7 +209,7 @@ async def call_model( # this is similar to customizing the create_react_agent with state_modifier, but is a lot more flexible system_prompt = SystemMessage( - system_prompt_text.replace("{CURRENT_DATE}", current_date) + system_prompt_text.replace("{CURRENT_DATE}", current_date).replace("{DATE_SUFFIX}", date_suffix) ) response = await bound_model.ainvoke([system_prompt] + state["messages"], config) diff --git a/ex_app/lib/all_tools/assignments.py b/ex_app/lib/all_tools/assignments.py index c7e36ffd..ee8826ba 100644 --- a/ex_app/lib/all_tools/assignments.py +++ b/ex_app/lib/all_tools/assignments.py @@ -7,6 +7,7 @@ from nc_py_api import AsyncNextcloudApp from ex_app.lib.all_tools.lib.decorator import dangerous_tool, safe_tool +from ex_app.lib.all_tools.lib.tz import get_timezone, starts_at_timestamp, user_timezone_async async def get_tools(nc: AsyncNextcloudApp): @@ -23,17 +24,18 @@ async def create_scheduled_task(title: str, prompt: str, recurrence_rule: str, t :param title: A title for the Scheduled Task, e.g. "Transcribe audio files" -- This is only for the user's reference and has no effect on the execution of the Scheduled Task. :param prompt: The instructions for the agent carrying out the Scheduled Task (Do not tell it to create a scheduled task here, as this is the instruction that runs as the scheduled task already. So instead of "Create a scheduled task to do X", just pass "Do X" here) :param recurrence_rule: An RRule compliant with RFC 5545 that defines the recurrence rule for the Scheduled Task. For example "FREQ=DAILY;INTERVAL=1" to run the Scheduled Task every day, an empty string as the recurrence_rule means the task does not repeat. - :param starts_at: A date time string in ISO 8601 format that defines when the Scheduled Task should start. For example "2025-01-01T09:00:00Z". If not provided, the Scheduled Task will start immediately. Make sure to use the user's timezone for this, obtainable with find_details_of_current_user - :param timezone: Timezone (e.g., 'America/New_York') defaults to the user's current time zone + :param starts_at: A date time string in ISO 8601 format that defines when the Scheduled Task should start. For example "2025-01-01T09:00:00Z". If not provided, the Scheduled Task will start immediately. + :param timezone: Timezone (e.g., 'America/New_York'). Defaults to the user's profile timezone; used to interpret starts_at. Must be a valid IANA timezone name when supplied. :return: """ + tz = get_timezone(timezone or await user_timezone_async(nc)) await nc.ocs('POST', f'/ocs/v2.php/apps/assistant/assignments', json={ 'title': title, 'prompt': prompt, 'recurrence': recurrence_rule, - 'startsAt': int(datetime.datetime.fromisoformat(starts_at.replace('Z', '+00:00')).timestamp()) if starts_at is not None else datetime.datetime.now(datetime.UTC).timestamp(), - 'timezone': pytz.timezone(timezone).zone if timezone is not None else None + 'startsAt': starts_at_timestamp(starts_at, tz) if starts_at is not None else datetime.datetime.now(datetime.UTC).timestamp(), + 'timezone': tz }) return True @@ -56,16 +58,17 @@ async def update_scheduled_task(id: int, prompt: None|str = None, recurrence_rul :param id: The ID of the Scheduled Task to update, you can obtain this from the list_scheduled_tasks tool :param prompt: The instructions for the AI carrying out the Scheduled Task. Pass `None` to leave this unchanged. (Do not tell the other AI to create a scheduled task here, as this is the instruction that runs as the scheduled task already. So instead of "Create a scheduled task to do X", just pass "Do X" here) :param recurrence_rule: An RRule compliant with RFC 5545 that defines the recurrence rule for the Scheduled Task. For example "FREQ=DAILY;INTERVAL=1" to run the Scheduled Task every day. An empty string means, it does not repeat. Pass `None` to leave this unchanged. - :param timezone A timezone for the scheduled task, set to None to leave this as is. + :param timezone: Timezone (e.g., 'America/New_York'). Defaults to the user's profile timezone; used to interpret starts_at. Must be a valid IANA timezone name when supplied. :param starts_at: A date time string in ISO 8601 format that defines when the Scheduled Task should start. For example "2025-01-01T09:00:00Z". If not provided, the Scheduled Task will start immediately. Pass `None` to leave this unchanged. :return: """ + tz = get_timezone(timezone or await user_timezone_async(nc)) return await nc.ocs('PATCH', f'/ocs/v2.php/apps/assistant/assignments/{id}', json={ 'prompt': prompt, 'recurrence': recurrence_rule, - 'startsAt': int(datetime.datetime.fromisoformat(starts_at.replace('Z', '+00:00')).timestamp()) if starts_at is not None else None, - 'timezone': pytz.timezone(timezone).zone if timezone is not None else None + 'startsAt': starts_at_timestamp(starts_at, tz) if starts_at is not None else None, + 'timezone': tz }) @tool diff --git a/ex_app/lib/all_tools/calendar.py b/ex_app/lib/all_tools/calendar.py index 4170131f..ee59bbbc 100644 --- a/ex_app/lib/all_tools/calendar.py +++ b/ex_app/lib/all_tools/calendar.py @@ -2,18 +2,21 @@ # SPDX-License-Identifier: AGPL-3.0-or-later import asyncio import time +import uuid from datetime import datetime, timezone, timedelta from typing import Optional from niquests import ConnectionError, Timeout import pytz -from ics import Calendar, Event, Attendee, Organizer, Todo +from icalendar import Event as ICalendarEvent, vCalAddress, vText +from ics import Calendar, Todo from langchain_core.tools import tool from nc_py_api import AsyncNextcloudApp, NextcloudApp import xml.etree.ElementTree as ET import vobject from ex_app.lib.all_tools.lib.decorator import safe_tool, dangerous_tool +from ex_app.lib.all_tools.lib.tz import get_timezone, user_timezone from ex_app.lib.all_tools.lib.freebusy_finder import find_available_slots, round_to_nearest_half_hour @@ -35,41 +38,48 @@ async def list_calendars(): """ return await asyncio.to_thread(list_calendars_sync) - def schedule_event_sync(calendar_name: str, title: str, description: str, start_date: str, end_date: str, attendees: Optional[list[str]], start_time: Optional[str], end_time: Optional[str], location: Optional[str], timezone: Optional[str]): + def schedule_event_sync(calendar_name: str, title: str, description: str, start_date: str, end_date: str, attendees: Optional[list[str]] = None, start_time: Optional[str] = None, end_time: Optional[str] = None, location: Optional[str] = None, timezone: Optional[str] = None): # Parse date and times start_date = datetime.strptime(start_date, "%Y-%m-%d") end_date = datetime.strptime(end_date, "%Y-%m-%d") + # Set timezone (defaults to the user's profile timezone); validated even for + # all-day events so a bad value errors instead of being silently ignored + timezone = timezone or user_timezone(ncSync) + tz = pytz.timezone(get_timezone(timezone)) + # Combine date and time if start_time and end_time: start_time = datetime.strptime(start_time, "%I:%M %p").time() end_time = datetime.strptime(end_time, "%I:%M %p").time() - start_datetime = datetime.combine(start_date, start_time) - end_datetime = datetime.combine(end_date, end_time) + # icalendar keeps the timezone as a TZID on datetime values + start_datetime = tz.localize(datetime.combine(start_date, start_time)) + end_datetime = tz.localize(datetime.combine(end_date, end_time)) else: - start_datetime = start_date - end_datetime = end_date - - # Set timezone - if timezone != None: - tz = pytz.timezone(timezone) - start_datetime = tz.localize(start_datetime) - end_datetime = tz.localize(end_datetime) + # All-day events are DATE values; DTEND is exclusive, so the day after + start_datetime = start_date.date() + end_datetime = (end_date + timedelta(days=1)).date() description_with_ai_note = f"{description}\n\n---\n\nThis event was scheduled by Nextcloud AI Assistant." - # Create event - c = Calendar() - e = Event() - e.name = title - e.begin = start_datetime - e.end = end_datetime - e.description = description_with_ai_note - e.location = location - if attendees is not None: - for attendee in attendees: - e.add_attendee(Attendee(common_name=attendee, email=attendee, partstat='NEEDS-ACTION', role='REQ-PARTICIPANT', cutype='INDIVIDUAL')) + # Create event (icalendar keeps the event's timezone as a TZID; the ics library would serialize it to UTC) + e = ICalendarEvent() + e.add('summary', title) + e.add('dtstart', start_datetime) + e.add('dtend', end_datetime) + e.add('description', description_with_ai_note) + if location: + e.add('location', location) + e.add('uid', str(uuid.uuid4()) + '@context-agent') + e.add('dtstamp', datetime.now(pytz.UTC)) + for attendee in attendees or []: + a = vCalAddress('mailto:' + attendee) + a.params['CN'] = vText(attendee) + a.params['PARTSTAT'] = vText('NEEDS-ACTION') + a.params['ROLE'] = vText('REQ-PARTICIPANT') + a.params['CUTYPE'] = vText('INDIVIDUAL') + e.add('attendee', a) # let's check who we are... i = 0 @@ -89,19 +99,20 @@ def schedule_event_sync(calendar_name: str, title: str, description: str, start_ raise Exception('Error fetching current user information') # ...and set the organizer - e.organizer = Organizer(common_name=json['displayname'], email=json['email']) + organizer = vCalAddress('mailto:' + (json['email'] or '')) + organizer.params['CN'] = vText(json['displayname']) + e.add('organizer', organizer) # Add event to calendar - c.events.add(e) principal = ncSync.cal.principal() calendars = principal.calendars() calendar = {cal.name: cal for cal in calendars}[calendar_name] - calendar.add_event(str(c)) + calendar.add_event(e.to_ical().decode()) @tool @dangerous_tool - async def schedule_event(calendar_name: str, title: str, description: str, start_date: str, end_date: str, attendees: Optional[list[str]], start_time: Optional[str], end_time: Optional[str], location: Optional[str], timezone: Optional[str]): + async def schedule_event(calendar_name: str, title: str, description: str, start_date: str, end_date: str, attendees: Optional[list[str]] = None, start_time: Optional[str] = None, end_time: Optional[str] = None, location: Optional[str] = None, timezone: Optional[str] = None): """ Crete a new event or meeting in a calendar. Omit start_time and end_time parameters to create an all-day event. :param calendar_name: The name of the calendar to add the event to @@ -113,7 +124,7 @@ async def schedule_event(calendar_name: str, title: str, description: str, start :param start_time: the start time in the following form: HH:MM AM/PM e.g. '3:00 PM' :param end_time: the start time in the following form: HH:MM AM/PM e.g. '4:00 PM' :param location: The location of the event - :param timezone: Timezone (e.g., 'America/New_York'). + :param timezone: Timezone (e.g., 'America/New_York'). Defaults to the user's profile timezone. Must be a valid IANA timezone name. :return: bool """ await asyncio.to_thread(schedule_event_sync, calendar_name, title, description, start_date, end_date, attendees, start_time, end_time, location, timezone) @@ -197,8 +208,8 @@ async def find_free_time_slot_in_calendar(participants: list[str], slot_duration available_slots = await asyncio.to_thread(find_free_time_slot_in_calendar_sync, participants, slot_duration, start_time, end_time) return available_slots - def add_task_sync(calendar_name: str, title: str, description: str, due_date: Optional[str], - due_time: Optional[str], timezone: Optional[str]): + def add_task_sync(calendar_name: str, title: str, description: str, due_date: Optional[str] = None, + due_time: Optional[str] = None, timezone: Optional[str] = None): description_with_ai_note = f"{description}\n\n---\n\nThis task was scheduled by Nextcloud AI Assistant." # Create task @@ -219,10 +230,9 @@ def add_task_sync(calendar_name: str, title: str, description: str, due_date: Op else: due_datetime = due_date - # Set timezone - if timezone != None: - tz = pytz.timezone(timezone) - due_datetime = tz.localize(due_datetime) + # Set timezone (defaults to the user's profile timezone) + timezone = timezone or user_timezone(ncSync) + due_datetime = pytz.timezone(get_timezone(timezone)).localize(due_datetime) t.due = due_datetime @@ -238,7 +248,7 @@ def add_task_sync(calendar_name: str, title: str, description: str, due_date: Op @tool @dangerous_tool - async def add_task(calendar_name: str, title: str, description: str, due_date: Optional[str], due_time: Optional[str], timezone: Optional[str]): + async def add_task(calendar_name: str, title: str, description: str, due_date: Optional[str] = None, due_time: Optional[str] = None, timezone: Optional[str] = None): """ Crete a new task in a calendar. :param calendar_name: The name of the calendar to add the task to @@ -246,10 +256,9 @@ async def add_task(calendar_name: str, title: str, description: str, due_date: O :param description: The description of the task :param due_date: the due date of the event in the following form: YYYY-MM-DD e.g. '2024-12-01' :param due_time: the due time in the following form: HH:MM AM/PM e.g. '3:00 PM' - :param timezone: Timezone (e.g., 'America/New_York'). Is required if there is a specified due date. + :param timezone: Timezone (e.g., 'America/New_York'). Defaults to the user's profile timezone. Must be a valid IANA timezone name. :return: bool """ - await asyncio.to_thread(add_task_sync, calendar_name, title, description, due_date, due_time, timezone) return True @@ -344,7 +353,7 @@ async def complete_task(calendar_name: str, task_uid: str): """ return await asyncio.to_thread(complete_task_sync, calendar_name, task_uid) - def update_task_sync(calendar_name: str, task_uid: str, title: Optional[str] = None, description: Optional[str] = None, due_date: Optional[str] = None, due_time: Optional[str] = None, timezone_str: Optional[str] = None, priority: Optional[int] = None): + def update_task_sync(calendar_name: str, task_uid: str, title: Optional[str] = None, description: Optional[str] = None, due_date: Optional[str] = None, due_time: Optional[str] = None, priority: Optional[int] = None, timezone: Optional[str] = None): principal = ncSync.cal.principal() calendars = principal.calendars() calendar = {cal.name: cal for cal in calendars}[calendar_name] @@ -373,9 +382,8 @@ def update_task_sync(calendar_name: str, task_uid: str, title: Optional[str] = N else: due_datetime = parsed_date - if timezone_str: - tz = pytz.timezone(timezone_str) - due_datetime = tz.localize(due_datetime) + timezone = timezone or user_timezone(ncSync) + due_datetime = pytz.timezone(get_timezone(timezone)).localize(due_datetime) ics_todo.due = due_datetime @@ -390,7 +398,7 @@ def update_task_sync(calendar_name: str, task_uid: str, title: Optional[str] = N @tool @dangerous_tool - async def update_task(calendar_name: str, task_uid: str, title: Optional[str] = None, description: Optional[str] = None, due_date: Optional[str] = None, due_time: Optional[str] = None, timezone: Optional[str] = None, priority: Optional[int] = None): + async def update_task(calendar_name: str, task_uid: str, title: Optional[str] = None, description: Optional[str] = None, due_date: Optional[str] = None, due_time: Optional[str] = None, priority: Optional[int] = None, timezone: Optional[str] = None): """ Update an existing task :param calendar_name: The name of the calendar containing the task (obtainable via list_calendars) @@ -399,11 +407,11 @@ async def update_task(calendar_name: str, task_uid: str, title: Optional[str] = :param description: New description for the task :param due_date: New due date in the form: YYYY-MM-DD e.g. '2024-12-01' :param due_time: New due time in the form: HH:MM AM/PM e.g. '3:00 PM' - :param timezone: Timezone (e.g., 'America/New_York') :param priority: Priority from 0 (undefined) to 9 (lowest), where 1 is highest priority + :param timezone: Timezone (e.g., 'America/New_York'). Defaults to the user's profile timezone. Must be a valid IANA timezone name. :return: bool indicating success """ - return await asyncio.to_thread(update_task_sync, calendar_name, task_uid, title, description, due_date, due_time, timezone, priority) + return await asyncio.to_thread(update_task_sync, calendar_name, task_uid, title, description, due_date, due_time, priority, timezone) def delete_task_sync(calendar_name: str, task_uid: str): principal = ncSync.cal.principal() diff --git a/ex_app/lib/all_tools/datetime_info.py b/ex_app/lib/all_tools/datetime_info.py new file mode 100644 index 00000000..8a072fcc --- /dev/null +++ b/ex_app/lib/all_tools/datetime_info.py @@ -0,0 +1,38 @@ +# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors +# SPDX-License-Identifier: AGPL-3.0-or-later +from datetime import datetime + +import pytz +from langchain_core.tools import tool +from nc_py_api import AsyncNextcloudApp + +from ex_app.lib.all_tools.lib.decorator import safe_tool +from ex_app.lib.all_tools.lib.tz import user_timezone_async + + +async def get_tools(nc: AsyncNextcloudApp): + + @tool + @safe_tool + async def get_current_datetime(): + """ + Get the current date and time in the user's timezone + :return: the current date and time, e.g. '2026-05-11 15:30 (Europe/Berlin)'. + Falls back to UTC if the user's timezone is unknown + """ + tz_name = await user_timezone_async(nc) + try: + tz = pytz.timezone(tz_name) if tz_name else pytz.UTC + except pytz.UnknownTimeZoneError: + tz = pytz.UTC + return f"{datetime.now(tz).strftime('%Y-%m-%d %H:%M')} ({tz.zone})" + + return [ + get_current_datetime, + ] + +def get_category_name(): + return "Date and time" + +async def is_available(nc: AsyncNextcloudApp): + return True diff --git a/ex_app/lib/all_tools/lib/freebusy_finder.py b/ex_app/lib/all_tools/lib/freebusy_finder.py index d65f3f16..a07b3890 100644 --- a/ex_app/lib/all_tools/lib/freebusy_finder.py +++ b/ex_app/lib/all_tools/lib/freebusy_finder.py @@ -1,6 +1,6 @@ # SPDX-FileCopyrightText: 2025 Nextcloud GmbH and Nextcloud contributors # SPDX-License-Identifier: AGPL-3.0-or-later -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone def find_available_slots(start_time, end_time, busy_intervals, slot_duration=timedelta(hours=1), max_slots=3): """ @@ -59,7 +59,7 @@ def round_to_nearest_half_hour(dt=None): :return: A rounded datetime object """ if dt is None: - dt = datetime.utcnow() + dt = datetime.now(timezone.utc) # Get minutes and determine rounding minutes = dt.minute diff --git a/ex_app/lib/all_tools/lib/tz.py b/ex_app/lib/all_tools/lib/tz.py new file mode 100644 index 00000000..38f0f625 --- /dev/null +++ b/ex_app/lib/all_tools/lib/tz.py @@ -0,0 +1,40 @@ +# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors +# SPDX-License-Identifier: AGPL-3.0-or-later +from datetime import datetime + +import pytz + +from ex_app.lib.all_tools.lib.decorator import timed_memoize + + +def get_timezone(timezone: str | None) -> str: + if timezone is None or not timezone.strip(): + raise ValueError("No timezone given and the user's profile has none set; provide a valid IANA timezone name, e.g. 'America/New_York'") + try: + return pytz.timezone(timezone).zone + except pytz.UnknownTimeZoneError: + raise ValueError(f"Invalid timezone '{timezone}'. Must be a valid IANA name, e.g. 'America/New_York'") from None + + +def user_timezone(nc) -> str | None: + try: + user = nc.ocs('GET', '/ocs/v2.php/cloud/user') or {} + return user.get('timezone') or None + except Exception: + return None + + +@timed_memoize(300) +async def user_timezone_async(nc) -> str | None: + try: + user = (await nc.ocs('GET', '/ocs/v2.php/cloud/user')) or {} + return user.get('timezone') or None + except Exception: + return None + + +def starts_at_timestamp(starts_at: str, timezone: str) -> int: + parsed = datetime.fromisoformat(starts_at.replace("Z", "+00:00")) + if parsed.tzinfo is None or parsed.utcoffset() is None: + parsed = pytz.timezone(timezone).localize(parsed) + return int(parsed.timestamp()) diff --git a/tests/test_schedule_event_payload.py b/tests/test_schedule_event_payload.py new file mode 100644 index 00000000..a8a50d0a --- /dev/null +++ b/tests/test_schedule_event_payload.py @@ -0,0 +1,156 @@ +# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors +# SPDX-License-Identifier: AGPL-3.0-or-later +"""Payload-level tests for schedule_event serialization. + +Run standalone (needs the AppAPI env, e.g. `set -a; source test.env; set +a`): + + PYTHONPATH=. .venv/bin/python tests/test_schedule_event_payload.py + +or via pytest when available. The network layer is stubbed: NextcloudApp inside +get_tools is replaced with a fake, and the ICS payload handed to CalDAV +add_event is captured. This is the layer the all-day regression (PR 257 review) +lived at — a dav roundtrip only proves an event got created, not its shape. +""" +import asyncio +import inspect +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +try: + import pytest +except ImportError: # standalone mode, see __main__ block + pytest = None + +from ex_app.lib.all_tools import calendar as cal # noqa: E402 + +CAPTURED = {} + + +class _FakeCalendar: + name = "Personal" + + def add_event(self, ics): + CAPTURED["ics"] = ics + + +class _FakePrincipal: + def calendars(self): + return [_FakeCalendar()] + + +class _FakeCal: + def principal(self): + return _FakePrincipal() + + +class _FakeNc: + cal = _CalNS = None # placeholder replaced below + + +class _FakeNcImpl: + cal = None + + def set_user(self, user): + pass + + @property + def user(self): + async def _u(): + return "payload-test" + + return _u() + + def ocs(self, method, path): + return { + "email": "payload-test@example.com", + "displayname": "Payload Test", + "timezone": "Europe/Berlin", + } + + +_FakeNcImpl.cal = _FakeCal() + + +def _schedule_event_sync(): + cal.NextcloudApp = lambda: _FakeNcImpl() + tools = asyncio.run(cal.get_tools(_FakeNcImpl())) + by_name = {t.name: t for t in tools} + wrapper = getattr(by_name["schedule_event"], "coroutine", None) or by_name["schedule_event"].func + return inspect.getclosurevars(wrapper).nonlocals["schedule_event_sync"] + + +def _schedule(**kwargs): + CAPTURED.clear() + args = { + "calendar_name": "Personal", + "title": "Payload test", + "description": "n/a", + "start_date": "2026-10-17", + "end_date": "2026-10-17", + "attendees": None, + "start_time": None, + "end_time": None, + "location": None, + "timezone": "Europe/Berlin", + } + args.update(kwargs) + _schedule_event_sync()(**args) + return CAPTURED["ics"] + + +def _unfold(ics): + """icalendar line-folds long values; join continuation lines before asserting.""" + lines = ics.splitlines() + out = [] + for line in lines: + if line[:1] in (" ", "\t") and out: + out[-1] += line[1:] + else: + out.append(line) + return "\r\n".join(out) + + +def test_all_day_event_uses_date_values(): + ics = _unfold(_schedule()) + assert "DTSTART;VALUE=DATE:20261017" in ics, ics + assert "DTEND;VALUE=DATE:20261018" in ics, ics # exclusive end: next day + assert "TZID=" not in ics, ics + assert "T000000" not in ics, ics + + +def test_timed_event_keeps_tzid(): + ics = _unfold(_schedule(start_time="12:00 PM", end_time="1:00 PM")) + assert "DTSTART;TZID=Europe/Berlin:20261017T120000" in ics, ics + assert "DTEND;TZID=Europe/Berlin:20261017T130000" in ics, ics + assert "VALUE=DATE" not in ics, ics + + +def test_invalid_timezone_raises(): + try: + _schedule(timezone="Not/AZone") + except ValueError: + return + raise AssertionError("expected ValueError for invalid timezone") + + +def test_omitted_timezone_falls_back_to_profile(): + ics = _unfold(_schedule(timezone=None)) # fake profile carries Europe/Berlin + assert "DTSTART;VALUE=DATE:20261017" in ics, ics + + +if __name__ == "__main__": + if not os.environ.get("NEXTCLOUD_URL"): + print("needs AppAPI env: set -a; source test.env; set +a") + sys.exit(77) + failed = 0 + for name, fn in sorted(globals().items()): + if name.startswith("test_") and callable(fn): + try: + fn() + print(f"PASS {name}") + except AssertionError as exc: + failed += 1 + print(f"FAIL {name}: {exc}") + sys.exit(1 if failed else 0)