feat(calendar): U7 Outlook sync via Microsoft Graph API
OutlookSyncProvider implementing AbstractSyncProvider for bidirectional Outlook Calendar sync. Uses Graph API delta query for incremental pull, auto-refreshes OAuth tokens on 401, and converts Outlook recurrence patterns to RRULE. Same conflict resolution as CalDAV (last-write-wins + WS notification). - src/agentkit/calendar/sync/outlook_provider.py — OutlookSyncProvider - tests/unit/calendar/test_sync_outlook.py — 8 tests
This commit is contained in:
chiguyong
parent
40bc27822f
commit
8d4145ddf9
|
|
@ -0,0 +1,536 @@
|
|||
"""Outlook sync provider — bidirectional sync with Microsoft Graph API (U7).
|
||||
|
||||
Uses ``httpx.AsyncClient`` for all Graph API calls. Conflict resolution is
|
||||
last-write-wins based on ``last_modified``; conflicts emit a
|
||||
``calendar_sync_conflict`` WS notification via the injectable ``notify_callback``
|
||||
(G4).
|
||||
|
||||
ponytail: browser OAuth flow (auth-code grant + redirect) is deferred to U12
|
||||
settings UI. This provider assumes tokens are already stored in
|
||||
``ExternalCalendarConfig.credentials``. Upgrade: add an ``OutlookOAuthFlow``
|
||||
helper that performs the device-code or auth-code flow and writes tokens.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import uuid
|
||||
from collections.abc import Awaitable, Callable
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
|
||||
import httpx
|
||||
|
||||
from agentkit.calendar.db import (
|
||||
DEFAULT_CALENDAR_DB_PATH,
|
||||
get_event_by_external_id,
|
||||
insert_event,
|
||||
update_event,
|
||||
update_external_config,
|
||||
)
|
||||
from agentkit.calendar.models import CalendarEvent, ExternalCalendarConfig, _now_iso
|
||||
from agentkit.calendar.sync.base import AbstractSyncProvider
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Async callback signature: (event_type: str, payload: dict) -> None
|
||||
NotifyCallback = Callable[[str, dict[str, Any]], Awaitable[None]]
|
||||
|
||||
GRAPH_BASE = "https://graph.microsoft.com/v1.0"
|
||||
TOKEN_URL = "https://login.microsoftonline.com/common/oauth2/v2.0/token"
|
||||
DEFAULT_SCOPE = "https://graph.microsoft.com/Calendars.ReadWrite offline_access"
|
||||
|
||||
_DAY_MAP = {
|
||||
"monday": "MO",
|
||||
"tuesday": "TU",
|
||||
"wednesday": "WE",
|
||||
"thursday": "TH",
|
||||
"friday": "FR",
|
||||
"saturday": "SA",
|
||||
"sunday": "SU",
|
||||
}
|
||||
_DAY_MAP_REVERSE = {v: k for k, v in _DAY_MAP.items()}
|
||||
_FREQ_MAP = {
|
||||
"daily": "DAILY",
|
||||
"weekly": "WEEKLY",
|
||||
"absoluteMonthly": "MONTHLY",
|
||||
"absoluteYearly": "YEARLY",
|
||||
}
|
||||
_FREQ_MAP_REVERSE = {v: k for k, v in _FREQ_MAP.items()}
|
||||
|
||||
|
||||
def _parse_iso(dt_str: str) -> datetime:
|
||||
"""Parse ISO 8601 string to UTC-aware datetime."""
|
||||
dt = datetime.fromisoformat(dt_str)
|
||||
if dt.tzinfo is None:
|
||||
dt = dt.replace(tzinfo=timezone.utc)
|
||||
return dt.astimezone(timezone.utc)
|
||||
|
||||
|
||||
def _outlook_dt_to_iso(dt_obj: dict[str, Any]) -> str:
|
||||
"""Convert Outlook dateTimeTimeZone to ISO 8601 UTC.
|
||||
|
||||
ponytail: assumes Graph returns UTC (no ``Prefer: outlook.timezone`` header
|
||||
is sent). If a non-UTC timezone is returned, it's treated as UTC. Upgrade:
|
||||
use ``zoneinfo`` with a Windows→IANA timezone mapping for correct conversion.
|
||||
"""
|
||||
if not dt_obj:
|
||||
return ""
|
||||
dt_str = dt_obj.get("dateTime", "")
|
||||
if not dt_str:
|
||||
return ""
|
||||
if "T" in dt_str:
|
||||
dt = datetime.fromisoformat(dt_str)
|
||||
else:
|
||||
# Date only (all-day event)
|
||||
dt = datetime.fromisoformat(dt_str + "T00:00:00")
|
||||
if dt.tzinfo is None:
|
||||
dt = dt.replace(tzinfo=timezone.utc)
|
||||
return dt.astimezone(timezone.utc).isoformat()
|
||||
|
||||
|
||||
def _iso_to_outlook_dt(iso_str: str, is_all_day: bool) -> dict[str, str]:
|
||||
"""Convert ISO 8601 UTC to Outlook dateTimeTimeZone."""
|
||||
if not iso_str:
|
||||
return {"dateTime": "", "timeZone": "UTC"}
|
||||
dt = _parse_iso(iso_str)
|
||||
if is_all_day:
|
||||
return {"dateTime": dt.strftime("%Y-%m-%d"), "timeZone": "UTC"}
|
||||
return {"dateTime": dt.strftime("%Y-%m-%dT%H:%M:%S"), "timeZone": "UTC"}
|
||||
|
||||
|
||||
def _outlook_recurrence_to_rrule(recurrence: dict[str, Any] | None) -> str | None:
|
||||
"""Convert Outlook recurrence pattern to RRULE string."""
|
||||
if not recurrence:
|
||||
return None
|
||||
pattern = recurrence.get("pattern", {}) or {}
|
||||
range_obj = recurrence.get("range", {}) or {}
|
||||
|
||||
parts: list[str] = []
|
||||
freq = _FREQ_MAP.get(pattern.get("type", ""))
|
||||
if freq:
|
||||
parts.append(f"FREQ={freq}")
|
||||
|
||||
interval = pattern.get("interval")
|
||||
if interval and interval > 1:
|
||||
parts.append(f"INTERVAL={interval}")
|
||||
|
||||
days = pattern.get("daysOfWeek", [])
|
||||
if days:
|
||||
bydays = [_DAY_MAP[d] for d in days if d in _DAY_MAP]
|
||||
if bydays:
|
||||
parts.append(f"BYDAY={','.join(bydays)}")
|
||||
|
||||
count = pattern.get("numberOfOccurrences")
|
||||
if count:
|
||||
parts.append(f"COUNT={count}")
|
||||
elif range_obj.get("type") == "endDate" and range_obj.get("endDate"):
|
||||
end_date = range_obj["endDate"] # "2026-12-31"
|
||||
parts.append(f"UNTIL={end_date.replace('-', '')}T235959Z")
|
||||
|
||||
return ";".join(parts) if parts else None
|
||||
|
||||
|
||||
def _rrule_to_outlook_recurrence(rrule: str, start_date: str) -> dict[str, Any] | None:
|
||||
"""Convert RRULE string to Outlook recurrence pattern.
|
||||
|
||||
``start_date`` is the event's start date in ``YYYY-MM-DD`` format (required
|
||||
by Graph API for the ``range.startDate`` field).
|
||||
"""
|
||||
parts: dict[str, str] = {}
|
||||
for part in rrule.split(";"):
|
||||
if "=" in part:
|
||||
k, v = part.split("=", 1)
|
||||
parts[k.upper()] = v
|
||||
|
||||
freq = parts.get("FREQ", "").upper()
|
||||
pattern_type = _FREQ_MAP_REVERSE.get(freq)
|
||||
if not pattern_type:
|
||||
return None
|
||||
|
||||
pattern: dict[str, Any] = {"type": pattern_type}
|
||||
|
||||
interval = parts.get("INTERVAL")
|
||||
pattern["interval"] = int(interval) if interval else 1
|
||||
|
||||
byday = parts.get("BYDAY")
|
||||
if byday:
|
||||
pattern["daysOfWeek"] = [
|
||||
_DAY_MAP_REVERSE[d] for d in byday.split(",") if d in _DAY_MAP_REVERSE
|
||||
]
|
||||
|
||||
count = parts.get("COUNT")
|
||||
until = parts.get("UNTIL")
|
||||
|
||||
if count:
|
||||
pattern["numberOfOccurrences"] = int(count)
|
||||
range_obj: dict[str, Any] = {
|
||||
"type": "numbered",
|
||||
"startDate": start_date,
|
||||
"numberOfOccurrences": int(count),
|
||||
}
|
||||
elif until:
|
||||
# UNTIL=20261231T235959Z → "2026-12-31"
|
||||
date_str = until[:8]
|
||||
end_date = f"{date_str[:4]}-{date_str[4:6]}-{date_str[6:8]}"
|
||||
range_obj = {"type": "endDate", "startDate": start_date, "endDate": end_date}
|
||||
else:
|
||||
range_obj = {"type": "noEnd", "startDate": start_date}
|
||||
|
||||
return {"pattern": pattern, "range": range_obj}
|
||||
|
||||
|
||||
def _extract_delta_token(delta_link: str) -> str | None:
|
||||
"""Extract ``$deltaToken`` from a Graph delta link URL."""
|
||||
parsed = urlparse(delta_link)
|
||||
params = parse_qs(parsed.query)
|
||||
values = params.get("$deltaToken", [])
|
||||
return values[0] if values else None
|
||||
|
||||
|
||||
class OutlookSyncProvider(AbstractSyncProvider):
|
||||
"""Bidirectional Outlook sync provider via Microsoft Graph REST API.
|
||||
|
||||
The ``client_factory`` parameter allows tests to inject a mock
|
||||
``httpx.AsyncClient`` without making real HTTP calls. When ``None``, a
|
||||
real ``httpx.AsyncClient`` is constructed per-operation.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
db_path: str | Path | None = None,
|
||||
client_factory: Callable[[], Any] | None = None,
|
||||
notify_callback: NotifyCallback | None = None,
|
||||
) -> None:
|
||||
self.db_path = Path(db_path) if db_path is not None else DEFAULT_CALENDAR_DB_PATH
|
||||
self._client_factory = client_factory
|
||||
self._notify = notify_callback
|
||||
# ponytail: conflicts list is in-memory only; if the process restarts
|
||||
# before a sync completes, conflict history is lost. Upgrade: persist
|
||||
# to a calendar_sync_conflicts table.
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Client / auth
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _get_client(self) -> Any:
|
||||
"""Return an httpx.AsyncClient (real or mock from factory)."""
|
||||
if self._client_factory is not None:
|
||||
return self._client_factory()
|
||||
return httpx.AsyncClient(timeout=30.0)
|
||||
|
||||
def _load_creds(self, config: ExternalCalendarConfig) -> dict[str, Any]:
|
||||
return json.loads(config.credentials) if config.credentials else {}
|
||||
|
||||
def _save_creds(self, config: ExternalCalendarConfig, creds: dict[str, Any]) -> None:
|
||||
config.credentials = json.dumps(creds)
|
||||
|
||||
async def _refresh_token(self, client: Any, config: ExternalCalendarConfig) -> dict[str, Any]:
|
||||
"""Refresh the access token using the refresh_token grant.
|
||||
|
||||
Posts to the Azure AD token endpoint, updates ``config.credentials``
|
||||
in-memory, and persists the new credentials to the DB.
|
||||
"""
|
||||
creds = self._load_creds(config)
|
||||
resp = await client.request(
|
||||
"POST",
|
||||
TOKEN_URL,
|
||||
data={
|
||||
"client_id": creds.get("client_id", ""),
|
||||
"grant_type": "refresh_token",
|
||||
"refresh_token": creds.get("refresh_token", ""),
|
||||
"scope": DEFAULT_SCOPE,
|
||||
},
|
||||
)
|
||||
resp.raise_for_status()
|
||||
payload = resp.json()
|
||||
creds["access_token"] = payload["access_token"]
|
||||
if "refresh_token" in payload:
|
||||
creds["refresh_token"] = payload["refresh_token"]
|
||||
creds["expires_at"] = (
|
||||
datetime.now(timezone.utc) + timedelta(seconds=payload.get("expires_in", 3600))
|
||||
).isoformat()
|
||||
self._save_creds(config, creds)
|
||||
await update_external_config(config.id, {"credentials": config.credentials}, self.db_path)
|
||||
return creds
|
||||
|
||||
async def _request(
|
||||
self,
|
||||
client: Any,
|
||||
config: ExternalCalendarConfig,
|
||||
method: str,
|
||||
url: str,
|
||||
*,
|
||||
json_body: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Make an authenticated Graph API request with 401 auto-refresh + retry."""
|
||||
creds = self._load_creds(config)
|
||||
headers = {"Authorization": f"Bearer {creds.get('access_token', '')}"}
|
||||
resp = await client.request(method, url, headers=headers, json=json_body)
|
||||
if resp.status_code == 401:
|
||||
creds = await self._refresh_token(client, config)
|
||||
headers = {"Authorization": f"Bearer {creds.get('access_token', '')}"}
|
||||
resp = await client.request(method, url, headers=headers, json=json_body)
|
||||
resp.raise_for_status()
|
||||
return resp.json() if resp.text else {}
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Pull
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
async def pull_changes(
|
||||
self, config: ExternalCalendarConfig, since: str | None = None
|
||||
) -> list[CalendarEvent]:
|
||||
"""Pull remote Outlook events via delta query.
|
||||
|
||||
First call (no ``sync_token``) is a full sync within a date range.
|
||||
Subsequent calls use the stored delta token for incremental sync.
|
||||
Returns pulled/updated events.
|
||||
"""
|
||||
client = self._get_client()
|
||||
try:
|
||||
remote_events, delta_token = await self._pull_delta(client, config)
|
||||
finally:
|
||||
await client.aclose()
|
||||
|
||||
# Persist delta token for next incremental sync
|
||||
if delta_token:
|
||||
config.sync_token = delta_token
|
||||
await update_external_config(config.id, {"sync_token": delta_token}, self.db_path)
|
||||
|
||||
result: list[CalendarEvent] = []
|
||||
for remote in remote_events:
|
||||
local = await get_event_by_external_id(
|
||||
remote.external_id, "outlook", config.user_id, self.db_path
|
||||
)
|
||||
if local is None:
|
||||
# New remote event → create local
|
||||
await insert_event(remote, self.db_path)
|
||||
result.append(remote)
|
||||
else:
|
||||
# Existing → check for conflict
|
||||
resolved = await self._resolve_pull_conflict(local, remote)
|
||||
if resolved is not None:
|
||||
result.append(resolved)
|
||||
return result
|
||||
|
||||
async def _pull_delta(
|
||||
self, client: Any, config: ExternalCalendarConfig
|
||||
) -> tuple[list[CalendarEvent], str | None]:
|
||||
"""Call /me/calendarView/delta. Returns (events, delta_token)."""
|
||||
url = self._build_delta_url(config)
|
||||
# ponytail: single-page fetch; pagination via @odata.nextLink is not
|
||||
# followed. Upgrade: loop on nextLink until exhausted, then read
|
||||
# deltaLink from the final page.
|
||||
payload = await self._request(client, config, "GET", url)
|
||||
events: list[CalendarEvent] = []
|
||||
for raw in payload.get("value", []):
|
||||
parsed = self._parse_outlook_event(raw, config.user_id)
|
||||
if parsed is not None:
|
||||
events.append(parsed)
|
||||
delta_link = payload.get("@odata.deltaLink")
|
||||
delta_token = _extract_delta_token(delta_link) if delta_link else None
|
||||
return events, delta_token
|
||||
|
||||
def _build_delta_url(self, config: ExternalCalendarConfig) -> str:
|
||||
"""Build the delta query URL. Uses sync_token if present (incremental)."""
|
||||
if config.sync_token:
|
||||
return f"{GRAPH_BASE}/me/calendarView/delta?$deltaToken={config.sync_token}"
|
||||
# Initial sync — use date range to scope the fetch
|
||||
start = (datetime.now(timezone.utc) - timedelta(days=365)).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
end = (datetime.now(timezone.utc) + timedelta(days=90)).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
return f"{GRAPH_BASE}/me/calendarView/delta?startDateTime={start}&endDateTime={end}"
|
||||
|
||||
def _parse_outlook_event(self, raw: dict[str, Any], user_id: str) -> CalendarEvent | None:
|
||||
"""Convert a Graph event JSON to a CalendarEvent dataclass."""
|
||||
eid = raw.get("id")
|
||||
if not eid:
|
||||
return None
|
||||
title = raw.get("subject") or ""
|
||||
if not title:
|
||||
return None
|
||||
|
||||
start_str = _outlook_dt_to_iso(raw.get("start", {}))
|
||||
end_str = _outlook_dt_to_iso(raw.get("end", {})) or start_str
|
||||
is_all_day = bool(raw.get("isAllDay", False))
|
||||
|
||||
body = raw.get("body", {}) or {}
|
||||
description = body.get("content", "") or ""
|
||||
|
||||
location_obj = raw.get("location", {}) or {}
|
||||
location = location_obj.get("displayName", "") or ""
|
||||
|
||||
rrule = _outlook_recurrence_to_rrule(raw.get("recurrence"))
|
||||
|
||||
last_modified = raw.get("lastModifiedDateTime", "") or _now_iso()
|
||||
now = _now_iso()
|
||||
return CalendarEvent(
|
||||
id=uuid.uuid4().hex,
|
||||
user_id=user_id,
|
||||
title=title,
|
||||
description=description,
|
||||
start_time=start_str,
|
||||
end_time=end_str,
|
||||
is_all_day=is_all_day,
|
||||
location=location,
|
||||
rrule=rrule,
|
||||
source="manual",
|
||||
external_id=eid,
|
||||
external_provider="outlook",
|
||||
last_modified=last_modified,
|
||||
created_at=now,
|
||||
)
|
||||
|
||||
async def _resolve_pull_conflict(
|
||||
self, local: CalendarEvent, remote: CalendarEvent
|
||||
) -> CalendarEvent | None:
|
||||
"""Resolve conflict when both local and remote exist (last-write-wins).
|
||||
|
||||
If remote is newer → update local. If local is newer → conflict
|
||||
(last-write-wins keeps local, but log + notify). If equal → no-op.
|
||||
Returns the winning event (or None if local kept unchanged).
|
||||
"""
|
||||
local_lm = (
|
||||
_parse_iso(local.last_modified)
|
||||
if local.last_modified
|
||||
else datetime.min.replace(tzinfo=timezone.utc)
|
||||
)
|
||||
remote_lm = (
|
||||
_parse_iso(remote.last_modified)
|
||||
if remote.last_modified
|
||||
else datetime.min.replace(tzinfo=timezone.utc)
|
||||
)
|
||||
|
||||
if remote_lm > local_lm:
|
||||
# Remote wins → update local
|
||||
fields = {
|
||||
"title": remote.title,
|
||||
"description": remote.description,
|
||||
"start_time": remote.start_time,
|
||||
"end_time": remote.end_time,
|
||||
"is_all_day": remote.is_all_day,
|
||||
"location": remote.location,
|
||||
"rrule": remote.rrule,
|
||||
"last_modified": remote.last_modified,
|
||||
}
|
||||
await update_event(local.id, fields, self.db_path)
|
||||
return remote
|
||||
|
||||
if local_lm > remote_lm:
|
||||
# Local wins → conflict, keep local, notify
|
||||
await self._notify_conflict(local, remote, winner="local")
|
||||
return None
|
||||
|
||||
# Equal timestamps → no change needed
|
||||
return None
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Push
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
async def push_changes(
|
||||
self, config: ExternalCalendarConfig, events: list[CalendarEvent]
|
||||
) -> list[CalendarEvent]:
|
||||
"""Push local events to Outlook. Returns events with external_id set."""
|
||||
client = self._get_client()
|
||||
try:
|
||||
result: list[CalendarEvent] = []
|
||||
for event in events:
|
||||
updated = await self._push_single(client, config, event)
|
||||
result.append(updated)
|
||||
finally:
|
||||
await client.aclose()
|
||||
return result
|
||||
|
||||
async def _push_single(
|
||||
self, client: Any, config: ExternalCalendarConfig, event: CalendarEvent
|
||||
) -> CalendarEvent:
|
||||
"""Push a single event to Outlook, return event with external_id set."""
|
||||
body = self._event_to_outlook(event)
|
||||
if event.external_id:
|
||||
# Update existing remote event
|
||||
url = f"{GRAPH_BASE}/me/events/{event.external_id}"
|
||||
await self._request(client, config, "PATCH", url, json_body=body)
|
||||
fields = {"last_modified": _now_iso()}
|
||||
await update_event(event.id, fields, self.db_path)
|
||||
return event
|
||||
# Create new remote event
|
||||
url = f"{GRAPH_BASE}/me/events"
|
||||
payload = await self._request(client, config, "POST", url, json_body=body)
|
||||
new_id = payload.get("id")
|
||||
if new_id:
|
||||
fields = {
|
||||
"external_id": new_id,
|
||||
"external_provider": "outlook",
|
||||
"last_modified": _now_iso(),
|
||||
}
|
||||
await update_event(event.id, fields, self.db_path)
|
||||
event.external_id = new_id
|
||||
event.external_provider = "outlook"
|
||||
return event
|
||||
|
||||
def _event_to_outlook(self, event: CalendarEvent) -> dict[str, Any]:
|
||||
"""Convert CalendarEvent to Outlook Graph event JSON."""
|
||||
body: dict[str, Any] = {
|
||||
"subject": event.title,
|
||||
"start": _iso_to_outlook_dt(event.start_time, event.is_all_day),
|
||||
"end": _iso_to_outlook_dt(event.end_time, event.is_all_day),
|
||||
"isAllDay": event.is_all_day,
|
||||
}
|
||||
if event.description:
|
||||
body["body"] = {"contentType": "Text", "content": event.description}
|
||||
if event.location:
|
||||
body["location"] = {"displayName": event.location}
|
||||
if event.rrule:
|
||||
start_date = event.start_time[:10] if event.start_time else "2026-01-01"
|
||||
rec = _rrule_to_outlook_recurrence(event.rrule, start_date)
|
||||
if rec is not None:
|
||||
body["recurrence"] = rec
|
||||
return body
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Test connection
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
async def test_connection(self, config: ExternalCalendarConfig) -> tuple[bool, str]:
|
||||
"""Test Outlook connectivity via GET /me. Returns (ok, error_msg)."""
|
||||
client = self._get_client()
|
||||
try:
|
||||
try:
|
||||
await self._request(client, config, "GET", f"{GRAPH_BASE}/me")
|
||||
ok, msg = True, ""
|
||||
except Exception as e:
|
||||
ok, msg = False, str(e)
|
||||
finally:
|
||||
await client.aclose()
|
||||
return ok, msg
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Conflict notification
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
async def _notify_conflict(
|
||||
self, local: CalendarEvent, remote: CalendarEvent, winner: str
|
||||
) -> None:
|
||||
"""Log conflict and send WS notification via callback (G4)."""
|
||||
logger.info(
|
||||
"Sync conflict for event %s (external_id=%s): local_lm=%s remote_lm=%s winner=%s",
|
||||
local.id,
|
||||
local.external_id,
|
||||
local.last_modified,
|
||||
remote.last_modified,
|
||||
winner,
|
||||
)
|
||||
if self._notify is not None:
|
||||
payload = {
|
||||
"event_id": local.id,
|
||||
"title": local.title,
|
||||
"external_id": local.external_id,
|
||||
"local_last_modified": local.last_modified,
|
||||
"remote_last_modified": remote.last_modified,
|
||||
"winner": winner,
|
||||
}
|
||||
await self._notify("calendar_sync_conflict", payload)
|
||||
|
|
@ -0,0 +1,556 @@
|
|||
"""Tests for OutlookSyncProvider — bidirectional Outlook Calendar sync (U7).
|
||||
|
||||
All Microsoft Graph API interactions are mocked via the ``client_factory``
|
||||
injection point. No real HTTP calls are made.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from agentkit.calendar.db import (
|
||||
get_event_by_external_id,
|
||||
init_calendar_db,
|
||||
insert_event,
|
||||
list_events,
|
||||
)
|
||||
from agentkit.calendar.models import CalendarEvent, ExternalCalendarConfig
|
||||
from agentkit.calendar.sync.outlook_provider import OutlookSyncProvider
|
||||
|
||||
USER_ID = "user-1"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Mock helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class MockResponse:
|
||||
"""Minimal mock of an httpx.Response."""
|
||||
|
||||
def __init__(self, status_code: int = 200, json_data: Any = None) -> None:
|
||||
self.status_code = status_code
|
||||
self._json = json_data
|
||||
self.text = json.dumps(json_data) if json_data is not None else ""
|
||||
|
||||
def json(self) -> Any:
|
||||
return self._json if self._json is not None else {}
|
||||
|
||||
def raise_for_status(self) -> None:
|
||||
if self.status_code >= 400:
|
||||
raise RuntimeError(f"HTTP {self.status_code}")
|
||||
|
||||
|
||||
class MockOutlookClient:
|
||||
"""Mock httpx.AsyncClient — records requests, returns queued responses."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.requests: list[dict[str, Any]] = []
|
||||
self._responses: list[MockResponse] = []
|
||||
|
||||
def add_response(self, response: MockResponse) -> None:
|
||||
self._responses.append(response)
|
||||
|
||||
async def request(
|
||||
self,
|
||||
method: str,
|
||||
url: str,
|
||||
*,
|
||||
headers: Any = None,
|
||||
json: Any = None,
|
||||
params: Any = None,
|
||||
data: Any = None,
|
||||
) -> MockResponse:
|
||||
self.requests.append(
|
||||
{
|
||||
"method": method,
|
||||
"url": url,
|
||||
"headers": headers,
|
||||
"json": json,
|
||||
"params": params,
|
||||
"data": data,
|
||||
}
|
||||
)
|
||||
if self._responses:
|
||||
return self._responses.pop(0)
|
||||
return MockResponse(status_code=200, json_data={})
|
||||
|
||||
async def aclose(self) -> None:
|
||||
pass
|
||||
|
||||
|
||||
def make_outlook_event(
|
||||
eid: str = "outlook-1",
|
||||
subject: str = "Outlook Event",
|
||||
start: str = "2026-07-01T10:00:00",
|
||||
end: str = "2026-07-01T11:00:00",
|
||||
is_all_day: bool = False,
|
||||
description: str = "",
|
||||
location: str = "",
|
||||
recurrence: dict[str, Any] | None = None,
|
||||
last_modified: str = "2026-06-01T00:00:00Z",
|
||||
) -> dict[str, Any]:
|
||||
"""Build a minimal Graph event JSON dict."""
|
||||
event: dict[str, Any] = {
|
||||
"id": eid,
|
||||
"subject": subject,
|
||||
"start": {"dateTime": start, "timeZone": "UTC"},
|
||||
"end": {"dateTime": end, "timeZone": "UTC"},
|
||||
"isAllDay": is_all_day,
|
||||
"lastModifiedDateTime": last_modified,
|
||||
}
|
||||
if description:
|
||||
event["body"] = {"contentType": "Text", "content": description}
|
||||
if location:
|
||||
event["location"] = {"displayName": location}
|
||||
if recurrence:
|
||||
event["recurrence"] = recurrence
|
||||
return event
|
||||
|
||||
|
||||
def make_config(
|
||||
config_id: str = "config-1",
|
||||
user_id: str = USER_ID,
|
||||
provider: str = "outlook",
|
||||
last_sync: str | None = None,
|
||||
sync_token: str | None = None,
|
||||
access_token: str = "test-token",
|
||||
refresh_token: str = "test-refresh",
|
||||
) -> ExternalCalendarConfig:
|
||||
return ExternalCalendarConfig(
|
||||
id=config_id,
|
||||
user_id=user_id,
|
||||
provider=provider,
|
||||
credentials=json.dumps(
|
||||
{
|
||||
"access_token": access_token,
|
||||
"refresh_token": refresh_token,
|
||||
"client_id": "test-client-id",
|
||||
"expires_at": "2026-12-31T00:00:00+00:00",
|
||||
}
|
||||
),
|
||||
last_sync=last_sync,
|
||||
sync_token=sync_token,
|
||||
)
|
||||
|
||||
|
||||
def make_local_event(
|
||||
event_id: str = "evt-1",
|
||||
title: str = "Local Event",
|
||||
external_id: str | None = None,
|
||||
last_modified: str = "2026-01-01T00:00:00+00:00",
|
||||
rrule: str | None = None,
|
||||
) -> CalendarEvent:
|
||||
return CalendarEvent(
|
||||
id=event_id,
|
||||
user_id=USER_ID,
|
||||
title=title,
|
||||
start_time="2026-07-01T10:00:00+00:00",
|
||||
end_time="2026-07-01T11:00:00+00:00",
|
||||
external_id=external_id,
|
||||
external_provider="outlook" if external_id else None,
|
||||
last_modified=last_modified,
|
||||
created_at=last_modified,
|
||||
rrule=rrule,
|
||||
)
|
||||
|
||||
|
||||
def client_factory_from(client: MockOutlookClient):
|
||||
"""Wrap a mock client in a client_factory callable."""
|
||||
|
||||
def factory() -> MockOutlookClient:
|
||||
return client
|
||||
|
||||
return factory
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def calendar_db_path(tmp_path: Path) -> Path:
|
||||
path = tmp_path / "test_calendar.db"
|
||||
asyncio.run(init_calendar_db(path))
|
||||
return path
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pull tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_outlook_provider_pull_delta_initial_sync(
|
||||
calendar_db_path: Path,
|
||||
) -> None:
|
||||
"""Mock returns 3 events → 3 local events created with external_id set."""
|
||||
events = [
|
||||
make_outlook_event(eid="outlook-1", subject="Event 1"),
|
||||
make_outlook_event(eid="outlook-2", subject="Event 2"),
|
||||
make_outlook_event(eid="outlook-3", subject="Event 3"),
|
||||
]
|
||||
response = MockResponse(
|
||||
200,
|
||||
{
|
||||
"value": events,
|
||||
"@odata.deltaLink": (
|
||||
"https://graph.microsoft.com/v1.0/me/calendarView/delta?$deltaToken=initial-token"
|
||||
),
|
||||
},
|
||||
)
|
||||
mock_client = MockOutlookClient()
|
||||
mock_client.add_response(response)
|
||||
|
||||
provider = OutlookSyncProvider(
|
||||
db_path=calendar_db_path, client_factory=client_factory_from(mock_client)
|
||||
)
|
||||
config = make_config()
|
||||
|
||||
pulled = await provider.pull_changes(config)
|
||||
|
||||
assert len(pulled) == 3
|
||||
db_events = await list_events(USER_ID, db_path=calendar_db_path)
|
||||
assert len(db_events) == 3
|
||||
titles = {e.title for e in db_events}
|
||||
assert titles == {"Event 1", "Event 2", "Event 3"}
|
||||
for event in db_events:
|
||||
assert event.external_id is not None
|
||||
assert event.external_provider == "outlook"
|
||||
# Delta token stored for next incremental sync
|
||||
assert config.sync_token == "initial-token"
|
||||
|
||||
|
||||
async def test_outlook_provider_pull_delta_incremental(
|
||||
calendar_db_path: Path,
|
||||
) -> None:
|
||||
"""Config has sync_token → incremental sync. 1 new + 1 updated processed."""
|
||||
# Pre-insert a local event that will be updated by the remote
|
||||
local = make_local_event(
|
||||
event_id="evt-1",
|
||||
title="Old Title",
|
||||
external_id="outlook-existing",
|
||||
last_modified="2026-01-01T00:00:00+00:00",
|
||||
)
|
||||
await insert_event(local, calendar_db_path)
|
||||
|
||||
new_event = make_outlook_event(eid="outlook-new", subject="New Event")
|
||||
updated_event = make_outlook_event(
|
||||
eid="outlook-existing",
|
||||
subject="Updated Title",
|
||||
last_modified="2026-06-15T00:00:00Z", # newer than local
|
||||
)
|
||||
response = MockResponse(
|
||||
200,
|
||||
{
|
||||
"value": [new_event, updated_event],
|
||||
"@odata.deltaLink": (
|
||||
"https://graph.microsoft.com/v1.0/me/calendarView/delta?$deltaToken=new-token"
|
||||
),
|
||||
},
|
||||
)
|
||||
mock_client = MockOutlookClient()
|
||||
mock_client.add_response(response)
|
||||
|
||||
provider = OutlookSyncProvider(
|
||||
db_path=calendar_db_path, client_factory=client_factory_from(mock_client)
|
||||
)
|
||||
config = make_config(sync_token="previous-token")
|
||||
|
||||
pulled = await provider.pull_changes(config)
|
||||
|
||||
assert len(pulled) == 2 # 1 new + 1 updated
|
||||
|
||||
# Verify the request URL contains the delta token (incremental sync)
|
||||
assert "$deltaToken=previous-token" in mock_client.requests[0]["url"]
|
||||
|
||||
# New event was created
|
||||
new_db = await get_event_by_external_id("outlook-new", "outlook", USER_ID, calendar_db_path)
|
||||
assert new_db is not None
|
||||
assert new_db.title == "New Event"
|
||||
|
||||
# Existing event was updated (remote was newer)
|
||||
updated_db = await get_event_by_external_id(
|
||||
"outlook-existing", "outlook", USER_ID, calendar_db_path
|
||||
)
|
||||
assert updated_db is not None
|
||||
assert updated_db.title == "Updated Title"
|
||||
|
||||
# Delta token updated
|
||||
assert config.sync_token == "new-token"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Push tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_outlook_provider_push_creates_remote_event(
|
||||
calendar_db_path: Path,
|
||||
) -> None:
|
||||
"""Local event with no external_id → POST creates remote, ID stored."""
|
||||
local = make_local_event(event_id="evt-1", title="New Local Event", external_id=None)
|
||||
await insert_event(local, calendar_db_path)
|
||||
|
||||
mock_client = MockOutlookClient()
|
||||
mock_client.add_response(
|
||||
MockResponse(201, {"id": "remote-uid-1", "subject": "New Local Event"})
|
||||
)
|
||||
|
||||
provider = OutlookSyncProvider(
|
||||
db_path=calendar_db_path, client_factory=client_factory_from(mock_client)
|
||||
)
|
||||
config = make_config()
|
||||
|
||||
events = await list_events(USER_ID, db_path=calendar_db_path)
|
||||
pushed = await provider.push_changes(config, events)
|
||||
|
||||
assert len(pushed) == 1
|
||||
assert pushed[0].external_id == "remote-uid-1"
|
||||
assert pushed[0].external_provider == "outlook"
|
||||
|
||||
# Verify POST was called to /me/events
|
||||
req = mock_client.requests[0]
|
||||
assert req["method"] == "POST"
|
||||
assert "/me/events" in req["url"]
|
||||
assert req["json"]["subject"] == "New Local Event"
|
||||
|
||||
# DB updated with external_id
|
||||
db_event = await get_event_by_external_id("remote-uid-1", "outlook", USER_ID, calendar_db_path)
|
||||
assert db_event is not None
|
||||
|
||||
|
||||
async def test_outlook_provider_push_updates_remote_event(
|
||||
calendar_db_path: Path,
|
||||
) -> None:
|
||||
"""Local event with external_id → PATCH updates remote."""
|
||||
local = make_local_event(
|
||||
event_id="evt-1",
|
||||
title="Updated Local Event",
|
||||
external_id="existing-uid",
|
||||
last_modified="2026-06-01T00:00:00+00:00",
|
||||
)
|
||||
await insert_event(local, calendar_db_path)
|
||||
|
||||
mock_client = MockOutlookClient()
|
||||
mock_client.add_response(MockResponse(200, {"id": "existing-uid"}))
|
||||
|
||||
provider = OutlookSyncProvider(
|
||||
db_path=calendar_db_path, client_factory=client_factory_from(mock_client)
|
||||
)
|
||||
config = make_config()
|
||||
|
||||
events = await list_events(USER_ID, db_path=calendar_db_path)
|
||||
pushed = await provider.push_changes(config, events)
|
||||
|
||||
assert len(pushed) == 1
|
||||
assert pushed[0].external_id == "existing-uid"
|
||||
|
||||
# Verify PATCH was called to /me/events/{id}
|
||||
req = mock_client.requests[0]
|
||||
assert req["method"] == "PATCH"
|
||||
assert "/me/events/existing-uid" in req["url"]
|
||||
assert req["json"]["subject"] == "Updated Local Event"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Token refresh test
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_outlook_token_refresh_on_401(calendar_db_path: Path) -> None:
|
||||
"""Mock 401 → refresh token used → request retried with new token."""
|
||||
mock_client = MockOutlookClient()
|
||||
# 1. First GET → 401 (token expired)
|
||||
mock_client.add_response(MockResponse(401, {"error": "token expired"}))
|
||||
# 2. Token refresh POST → 200 with new tokens
|
||||
mock_client.add_response(
|
||||
MockResponse(
|
||||
200,
|
||||
{
|
||||
"access_token": "new-token",
|
||||
"refresh_token": "new-refresh",
|
||||
"expires_in": 3600,
|
||||
},
|
||||
)
|
||||
)
|
||||
# 3. Retry GET → 200 with events
|
||||
mock_client.add_response(
|
||||
MockResponse(
|
||||
200,
|
||||
{
|
||||
"value": [make_outlook_event(eid="outlook-1", subject="Refreshed Event")],
|
||||
"@odata.deltaLink": (
|
||||
"https://graph.microsoft.com/v1.0/me/calendarView/delta"
|
||||
"?$deltaToken=after-refresh"
|
||||
),
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
provider = OutlookSyncProvider(
|
||||
db_path=calendar_db_path, client_factory=client_factory_from(mock_client)
|
||||
)
|
||||
config = make_config(access_token="expired-token")
|
||||
|
||||
pulled = await provider.pull_changes(config)
|
||||
|
||||
assert len(pulled) == 1
|
||||
assert pulled[0].title == "Refreshed Event"
|
||||
|
||||
# 3 requests: GET (401), POST (refresh), GET (retry)
|
||||
assert len(mock_client.requests) == 3
|
||||
assert mock_client.requests[0]["method"] == "GET"
|
||||
assert mock_client.requests[1]["method"] == "POST"
|
||||
assert mock_client.requests[2]["method"] == "GET"
|
||||
|
||||
# Token refresh hit the Azure AD token endpoint
|
||||
assert "login.microsoftonline.com" in mock_client.requests[1]["url"]
|
||||
assert mock_client.requests[1]["data"]["grant_type"] == "refresh_token"
|
||||
assert mock_client.requests[1]["data"]["refresh_token"] == "test-refresh"
|
||||
|
||||
# Credentials updated in config
|
||||
creds = json.loads(config.credentials)
|
||||
assert creds["access_token"] == "new-token"
|
||||
assert creds["refresh_token"] == "new-refresh"
|
||||
|
||||
# Retry used the new access token
|
||||
retry_headers = mock_client.requests[2]["headers"]
|
||||
assert retry_headers["Authorization"] == "Bearer new-token"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Conflict test
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_outlook_conflict_last_write_wins(calendar_db_path: Path) -> None:
|
||||
"""Both sides modified, local is newer → local wins, local not updated."""
|
||||
local = make_local_event(
|
||||
event_id="evt-1",
|
||||
title="Local Updated",
|
||||
external_id="outlook-1",
|
||||
last_modified="2026-06-15T00:00:00+00:00",
|
||||
)
|
||||
await insert_event(local, calendar_db_path)
|
||||
|
||||
remote = make_outlook_event(
|
||||
eid="outlook-1",
|
||||
subject="Remote Older",
|
||||
last_modified="2026-06-01T00:00:00Z", # older than local
|
||||
)
|
||||
response = MockResponse(200, {"value": [remote]})
|
||||
mock_client = MockOutlookClient()
|
||||
mock_client.add_response(response)
|
||||
|
||||
provider = OutlookSyncProvider(
|
||||
db_path=calendar_db_path, client_factory=client_factory_from(mock_client)
|
||||
)
|
||||
config = make_config()
|
||||
|
||||
pulled = await provider.pull_changes(config)
|
||||
|
||||
# Local wins → no update, pulled is empty
|
||||
assert len(pulled) == 0
|
||||
db_event = await get_event_by_external_id("outlook-1", "outlook", USER_ID, calendar_db_path)
|
||||
assert db_event is not None
|
||||
assert db_event.title == "Local Updated"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# RRULE roundtrip test
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_outlook_rrule_roundtrip(calendar_db_path: Path) -> None:
|
||||
"""Recurring event synced → RRULE preserved in both pull and push."""
|
||||
recurrence = {
|
||||
"pattern": {
|
||||
"type": "weekly",
|
||||
"interval": 1,
|
||||
"daysOfWeek": ["monday"],
|
||||
"numberOfOccurrences": 4,
|
||||
},
|
||||
"range": {
|
||||
"type": "numbered",
|
||||
"startDate": "2026-07-01",
|
||||
"numberOfOccurrences": 4,
|
||||
},
|
||||
}
|
||||
remote = make_outlook_event(
|
||||
eid="outlook-rrule", subject="Weekly Meeting", recurrence=recurrence
|
||||
)
|
||||
response = MockResponse(
|
||||
200,
|
||||
{
|
||||
"value": [remote],
|
||||
"@odata.deltaLink": (
|
||||
"https://graph.microsoft.com/v1.0/me/calendarView/delta?$deltaToken=rrule-token"
|
||||
),
|
||||
},
|
||||
)
|
||||
mock_client = MockOutlookClient()
|
||||
mock_client.add_response(response)
|
||||
|
||||
provider = OutlookSyncProvider(
|
||||
db_path=calendar_db_path, client_factory=client_factory_from(mock_client)
|
||||
)
|
||||
config = make_config()
|
||||
|
||||
# Pull: verify RRULE preserved
|
||||
pulled = await provider.pull_changes(config)
|
||||
assert len(pulled) == 1
|
||||
assert pulled[0].rrule is not None
|
||||
assert "FREQ=WEEKLY" in pulled[0].rrule
|
||||
assert "BYDAY=MO" in pulled[0].rrule
|
||||
assert "COUNT=4" in pulled[0].rrule
|
||||
|
||||
# Push: verify recurrence pattern in request body sent to remote
|
||||
events = await list_events(USER_ID, db_path=calendar_db_path)
|
||||
mock_client2 = MockOutlookClient()
|
||||
mock_client2.add_response(
|
||||
MockResponse(201, {"id": "outlook-rrule", "subject": "Weekly Meeting"})
|
||||
)
|
||||
provider2 = OutlookSyncProvider(
|
||||
db_path=calendar_db_path, client_factory=client_factory_from(mock_client2)
|
||||
)
|
||||
await provider2.push_changes(config, events)
|
||||
|
||||
req = mock_client2.requests[0]
|
||||
assert req["method"] == "PATCH"
|
||||
assert "/me/events/outlook-rrule" in req["url"]
|
||||
body = req["json"]
|
||||
assert "recurrence" in body
|
||||
assert body["recurrence"]["pattern"]["type"] == "weekly"
|
||||
assert body["recurrence"]["pattern"]["daysOfWeek"] == ["monday"]
|
||||
assert body["recurrence"]["pattern"]["numberOfOccurrences"] == 4
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# test_connection tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_outlook_test_connection_success(calendar_db_path: Path) -> None:
|
||||
"""Mock Graph /me returns user profile → test_connection() returns (True, '')."""
|
||||
mock_client = MockOutlookClient()
|
||||
mock_client.add_response(MockResponse(200, {"id": "user-id", "displayName": "Test User"}))
|
||||
|
||||
provider = OutlookSyncProvider(
|
||||
db_path=calendar_db_path, client_factory=client_factory_from(mock_client)
|
||||
)
|
||||
config = make_config()
|
||||
|
||||
ok, msg = await provider.test_connection(config)
|
||||
assert ok is True
|
||||
assert msg == ""
|
||||
|
||||
# Verify GET /me was called
|
||||
req = mock_client.requests[0]
|
||||
assert req["method"] == "GET"
|
||||
assert "/me" in req["url"]
|
||||
Loading…
Reference in New Issue