#!/usr/bin/env python3 """ Local fixture harness for the calendar plugin requirements. Commands: python3 fixture/server.py init python3 fixture/server.py reset python3 fixture/server.py seed python3 fixture/server.py run --host 127.0.0.1 --port 8080 """ from __future__ import annotations import argparse import base64 import hashlib import hmac import json import os import re import secrets import smtplib import sqlite3 import ssl import textwrap import time import calendar as pycalendar from collections import deque from email.message import EmailMessage from dataclasses import dataclass from datetime import date, datetime, timedelta, timezone from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from urllib.parse import parse_qs, unquote, urlparse ROOT = Path(__file__).resolve().parent DB_PATH = ROOT / "fixture.db" DEFAULT_TIMEZONE = "Europe/London" SHARED_CALENDAR_ID = "public" API_BASE = "/wp-json/calendar/v1" PBKDF2_ITERATIONS = 210_000 VERIFY_TOKEN_TTL_SECONDS = 24 * 60 * 60 RESET_TOKEN_TTL_SECONDS = 30 * 60 _RATE_LIMIT_STATE: Dict[str, List[float]] = {} def fixture_trace_log_path() -> Path: configured = (os.getenv("FIXTURE_HTTP_TRACE_LOG") or "").strip() if configured: p = Path(configured) return p if p.is_absolute() else (ROOT / p) return ROOT / "http_trace.log" def utc_now_iso() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat() def hash_password(value: str) -> str: # Stronger hash format for fixture: pbkdf2_sha256$iterations$salt_hex$digest_hex salt = secrets.token_bytes(16) digest = hashlib.pbkdf2_hmac("sha256", value.encode("utf-8"), salt, PBKDF2_ITERATIONS) return f"pbkdf2_sha256${PBKDF2_ITERATIONS}${salt.hex()}${digest.hex()}" def verify_password(value: str, stored_hash: str) -> bool: if not stored_hash: return False if stored_hash.startswith("pbkdf2_sha256$"): parts = stored_hash.split("$") if len(parts) != 4: return False try: iterations = int(parts[1]) salt = bytes.fromhex(parts[2]) expected = bytes.fromhex(parts[3]) except ValueError: return False actual = hashlib.pbkdf2_hmac("sha256", value.encode("utf-8"), salt, iterations) return hmac.compare_digest(actual, expected) # Backward-compatible legacy support for old unsalted SHA-256 hashes. if re.fullmatch(r"[0-9a-f]{64}", stored_hash): legacy = hashlib.sha256(value.encode("utf-8")).hexdigest() return hmac.compare_digest(legacy, stored_hash) return False def _rate_limited(scope: str, key: str, limit: int, window_seconds: int) -> bool: now = time.monotonic() bucket = f"{scope}:{key}" hits = _RATE_LIMIT_STATE.get(bucket, []) cutoff = now - window_seconds hits = [t for t in hits if t >= cutoff] blocked = len(hits) >= limit hits.append(now) _RATE_LIMIT_STATE[bucket] = hits[-max(limit * 2, 16) :] return blocked def mk_etag(seed: str) -> str: digest = hashlib.sha256(seed.encode("utf-8")).hexdigest()[:16] return f'"{digest}"' def parse_basic_auth(header_value: Optional[str]) -> Optional[Tuple[str, str]]: if not header_value or not header_value.startswith("Basic "): return None encoded = header_value[6:].strip() try: raw = base64.b64decode(encoded).decode("utf-8") except Exception: return None if ":" not in raw: return None username, password = raw.split(":", 1) return username, password def _read_env_file(path: Path) -> Dict[str, str]: out: Dict[str, str] = {} if not path.exists(): return out for raw in path.read_text(encoding="utf-8").splitlines(): line = raw.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) out[key.strip()] = value.strip().strip('"').strip("'") return out def _smtp_settings() -> Dict[str, str]: env = _read_env_file(ROOT.parent / "credentials" / ".env") merged = dict(env) merged.update({k: v for k, v in os.environ.items() if k.startswith("SMTP_")}) return merged def send_fixture_email(to_email: str, subject: str, body_text: str) -> Tuple[bool, str]: cfg = _smtp_settings() host = (cfg.get("SMTP_HOST") or "").strip() port = int((cfg.get("SMTP_PORT") or "587").strip()) username = (cfg.get("SMTP_USERNAME") or "").strip() password = cfg.get("SMTP_PASSWORD") or "" from_email = (cfg.get("SMTP_FROM") or "").strip() use_tls = (cfg.get("SMTP_USE_TLS") or "true").strip().lower() in {"1", "true", "yes", "on"} if not host or not from_email or not to_email: return False, "smtp_not_configured" msg = EmailMessage() msg["From"] = from_email msg["To"] = to_email msg["Subject"] = subject msg.set_content(body_text) try: with smtplib.SMTP(host, port, timeout=10) as server: if use_tls: server.starttls(context=ssl.create_default_context()) if username: server.login(username, password) server.send_message(msg) return True, "sent" except Exception as exc: # pragma: no cover - environment dependent return False, f"smtp_error:{exc.__class__.__name__}:{exc}" def ensure_db() -> sqlite3.Connection: conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def create_schema(conn: sqlite3.Connection) -> None: cur = conn.cursor() cur.executescript( """ CREATE TABLE IF NOT EXISTS events ( id INTEGER PRIMARY KEY AUTOINCREMENT, uid TEXT NOT NULL UNIQUE, title TEXT NOT NULL, description TEXT NOT NULL DEFAULT '', location TEXT NOT NULL DEFAULT '', category TEXT NOT NULL DEFAULT '', all_day_event INTEGER NOT NULL DEFAULT 0, start_datetime TEXT NOT NULL, end_datetime TEXT NOT NULL, repeat_type TEXT NOT NULL DEFAULT 'none', repeat_interval INTEGER NOT NULL DEFAULT 1, repeat_nth_mode TEXT NOT NULL DEFAULT '', repeat_nth_day INTEGER NULL, repeat_nth_pos INTEGER NULL, repeat_nth_weekday INTEGER NULL, repeat_range_mode TEXT NOT NULL DEFAULT 'none', repeat_count INTEGER NULL, repeat_until TEXT NULL, timezone TEXT NOT NULL DEFAULT 'Europe/London', calendar_id TEXT NOT NULL DEFAULT 'public', etag TEXT NOT NULL, sync_version INTEGER NOT NULL DEFAULT 1, last_modified_by_user_id INTEGER NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS recurrence_exceptions ( id INTEGER PRIMARY KEY AUTOINCREMENT, event_id INTEGER NOT NULL, occurrence_key TEXT NOT NULL, exception_type TEXT NOT NULL, override_payload TEXT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, UNIQUE (event_id, occurrence_key) ); CREATE TABLE IF NOT EXISTS caldav_users ( id INTEGER PRIMARY KEY AUTOINCREMENT, email TEXT NOT NULL UNIQUE, password_hash TEXT NOT NULL, email_verified_at TEXT NULL, account_status TEXT NOT NULL, access_level TEXT NOT NULL, request_state TEXT NOT NULL DEFAULT 'none', created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS user_tokens ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, token_type TEXT NOT NULL, token_hash TEXT NOT NULL, expires_at TEXT NOT NULL, used_at TEXT NULL, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS audit_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, actor_type TEXT NOT NULL, actor_id TEXT NOT NULL, action TEXT NOT NULL, target_type TEXT NOT NULL, target_id TEXT NOT NULL, result TEXT NOT NULL, context_json TEXT NOT NULL, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS plugin_settings ( key TEXT PRIMARY KEY, value TEXT NOT NULL, updated_at TEXT NOT NULL ); """ ) cols = {r["name"] for r in conn.execute("PRAGMA table_info(events)").fetchall()} if "caldav_resource" not in cols: conn.execute("ALTER TABLE events ADD COLUMN caldav_resource TEXT NULL") conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_events_caldav_resource ON events(caldav_resource)") conn.execute("UPDATE events SET caldav_resource = id || '.ics' WHERE caldav_resource IS NULL") conn.commit() def drop_all(conn: sqlite3.Connection) -> None: cur = conn.cursor() cur.executescript( """ DROP TABLE IF EXISTS audit_log; DROP TABLE IF EXISTS user_tokens; DROP TABLE IF EXISTS caldav_users; DROP TABLE IF EXISTS recurrence_exceptions; DROP TABLE IF EXISTS events; DROP TABLE IF EXISTS plugin_settings; """ ) conn.commit() def log_audit( conn: sqlite3.Connection, actor_type: str, actor_id: str, action: str, target_type: str, target_id: str, result: str, context: Dict[str, Any], ) -> None: conn.execute( """ INSERT INTO audit_log (actor_type, actor_id, action, target_type, target_id, result, context_json, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( actor_type, actor_id, action, target_type, target_id, result, json.dumps(context, sort_keys=True), utc_now_iso(), ), ) conn.commit() def seed_data(conn: sqlite3.Connection) -> None: now = utc_now_iso() users = [ ("rw_user@example.test", "rwpass123456", now, "active", "write", "approved"), ("adrians@chezstephens.org.uk", "brillig1", now, "active", "write", "approved"), ("pending_user@example.test", "pending123456", now, "pending_approval", "write", "requested"), ] for email, pw, verified, status, level, request_state in users: conn.execute( """ INSERT OR IGNORE INTO caldav_users (email, password_hash, email_verified_at, account_status, access_level, request_state, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, (email, hash_password(pw), verified, status, level, request_state, now, now), ) fixture_events: List[Dict[str, Any]] = [ { "id": 1, "uid": "fixture-ce-001@calendar-wp-plugin", "title": "Board Meeting", "description": "Quarterly board review.", "location": "Room A", "category": "Governance", "all_day_event": 0, "start_datetime": "2026-04-01T10:00:00+01:00", "end_datetime": "2026-04-01T11:30:00+01:00", "repeat_type": "none", "repeat_interval": 1, "repeat_range_mode": "none", "repeat_count": None, "repeat_until": None, }, { "id": 2, "uid": "fixture-ce-002@calendar-wp-plugin", "title": "Office Closed", "description": "Public holiday closure.", "location": "HQ", "category": "Operations", "all_day_event": 1, "start_datetime": "2026-05-04", "end_datetime": "2026-05-05", "repeat_type": "none", "repeat_interval": 1, "repeat_range_mode": "none", "repeat_count": None, "repeat_until": None, }, { "id": 3, "uid": "fixture-ce-003@calendar-wp-plugin", "title": "Daily Standup", "description": "15 minute sync.", "location": "Online", "category": "Team", "all_day_event": 0, "start_datetime": "2026-04-06T09:00:00+01:00", "end_datetime": "2026-04-06T09:15:00+01:00", "repeat_type": "daily", "repeat_interval": 1, "repeat_range_mode": "count", "repeat_count": 10, "repeat_until": None, }, { "id": 4, "uid": "fixture-ce-004@calendar-wp-plugin", "title": "Community Lunch", "description": "Weekly community lunch.", "location": "Cafeteria", "category": "Community", "all_day_event": 0, "start_datetime": "2026-04-08T12:30:00+01:00", "end_datetime": "2026-04-08T13:30:00+01:00", "repeat_type": "weekly", "repeat_interval": 1, "repeat_range_mode": "no_end", "repeat_count": None, "repeat_until": None, }, { "id": 5, "uid": "fixture-ce-005@calendar-wp-plugin", "title": "Finance Close", "description": "Month-end close process.", "location": "Finance Office", "category": "Finance", "all_day_event": 0, "start_datetime": "2026-04-30T17:00:00+01:00", "end_datetime": "2026-04-30T18:00:00+01:00", "repeat_type": "monthly", "repeat_interval": 1, "repeat_range_mode": "until", "repeat_count": None, "repeat_until": "2026-08-31", }, { "id": 6, "uid": "fixture-ce-006@calendar-wp-plugin", "title": "Annual Conference", "description": "Annual community conference.", "location": "Main Hall", "category": "Events", "all_day_event": 0, "start_datetime": "2026-06-15T10:00:00+01:00", "end_datetime": "2026-06-15T17:00:00+01:00", "repeat_type": "yearly", "repeat_interval": 1, "repeat_range_mode": "count", "repeat_count": 3, "repeat_until": None, }, { "id": 7, "uid": "fixture-ce-007@calendar-wp-plugin", "title": "Fortnightly Coaching", "description": "Coaching check-in.", "location": "Online", "category": "Training", "all_day_event": 0, "start_datetime": "2026-04-07T15:00:00+01:00", "end_datetime": "2026-04-07T16:00:00+01:00", "repeat_type": "custom", "repeat_interval": 2, "repeat_range_mode": "until", "repeat_count": None, "repeat_until": "2026-07-31", }, { "id": 8, "uid": "fixture-ce-008@calendar-wp-plugin", "title": "DST Validation Event", "description": "Validates DST transition rendering.", "location": "Lab", "category": "QA", "all_day_event": 0, "start_datetime": "2026-10-25T00:30:00+01:00", "end_datetime": "2026-10-25T02:30:00+00:00", "repeat_type": "none", "repeat_interval": 1, "repeat_range_mode": "none", "repeat_count": None, "repeat_until": None, }, { "id": 9, "uid": "fixture-ce-009@calendar-wp-plugin", "title": "Leap Day Marker", "description": "Leap day recurrence behavior.", "location": "Calendar", "category": "QA", "all_day_event": 0, "start_datetime": "2028-02-29T09:00:00+00:00", "end_datetime": "2028-02-29T10:00:00+00:00", "repeat_type": "yearly", "repeat_interval": 1, "repeat_range_mode": "count", "repeat_count": 3, "repeat_until": None, }, { "id": 10, "uid": "fixture-ce-010@calendar-wp-plugin", "title": "Therapy Session", "description": "Used for single-occurrence delete exception tests.", "location": "Clinic", "category": "Health", "all_day_event": 0, "start_datetime": "2026-04-03T14:00:00+01:00", "end_datetime": "2026-04-03T15:00:00+01:00", "repeat_type": "weekly", "repeat_interval": 1, "repeat_range_mode": "count", "repeat_count": 8, "repeat_until": None, }, ] for event in fixture_events: etag = mk_etag(f"{event['uid']}:{now}:1") conn.execute( """ INSERT OR REPLACE INTO events (id, uid, title, description, location, category, all_day_event, start_datetime, end_datetime, repeat_type, repeat_interval, repeat_nth_mode, repeat_nth_day, repeat_nth_pos, repeat_nth_weekday, repeat_range_mode, repeat_count, repeat_until, timezone, calendar_id, caldav_resource, etag, sync_version, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( event["id"], event["uid"], event["title"], event["description"], event["location"], event["category"], event["all_day_event"], event["start_datetime"], event["end_datetime"], event["repeat_type"], event["repeat_interval"], event.get("repeat_nth_mode", ""), event.get("repeat_nth_day"), event.get("repeat_nth_pos"), event.get("repeat_nth_weekday"), event["repeat_range_mode"], event["repeat_count"], event["repeat_until"], DEFAULT_TIMEZONE, SHARED_CALENDAR_ID, f"{event['id']}.ics", etag, 1, now, now, ), ) conn.execute("DELETE FROM recurrence_exceptions") conn.execute("INSERT OR REPLACE INTO plugin_settings (key, value, updated_at) VALUES (?, ?, ?)", ("presentation_name", "Calendar", now)) conn.execute("INSERT OR REPLACE INTO plugin_settings (key, value, updated_at) VALUES (?, ?, ?)", ("url_slug", "", now)) conn.execute("INSERT OR REPLACE INTO plugin_settings (key, value, updated_at) VALUES (?, ?, ?)", ("ics_access_mode", "public_read", now)) conn.execute("INSERT OR REPLACE INTO plugin_settings (key, value, updated_at) VALUES (?, ?, ?)", ("caldav_calendar_name", "Public Calendar", now)) conn.commit() def event_to_dict(row: sqlite3.Row) -> Dict[str, Any]: return { "id": row["id"], "uid": row["uid"], "title": row["title"], "description": row["description"], "location": row["location"], "category": row["category"], "all_day_event": bool(row["all_day_event"]), "start_datetime": row["start_datetime"], "end_datetime": row["end_datetime"], "repeat_type": row["repeat_type"], "repeat_interval": row["repeat_interval"], "repeat_nth_mode": row["repeat_nth_mode"], "repeat_nth_day": row["repeat_nth_day"], "repeat_nth_pos": row["repeat_nth_pos"], "repeat_nth_weekday": row["repeat_nth_weekday"], "repeat_range_mode": row["repeat_range_mode"], "repeat_count": row["repeat_count"], "repeat_until": row["repeat_until"], "timezone": row["timezone"], "calendar_id": row["calendar_id"], "etag": row["etag"], "sync_version": row["sync_version"], "updated_at": row["updated_at"], } def get_setting(conn: sqlite3.Connection, key: str, default_value: str) -> str: row = conn.execute("SELECT value FROM plugin_settings WHERE key = ?", (key,)).fetchone() if not row: return default_value return row["value"] def set_setting(conn: sqlite3.Connection, key: str, value: str) -> None: conn.execute( "INSERT OR REPLACE INTO plugin_settings (key, value, updated_at) VALUES (?, ?, ?)", (key, value, utc_now_iso()), ) conn.commit() def _parse_dt_maybe_date(value: str) -> datetime: if "T" in value: return datetime.fromisoformat(value.replace("Z", "+00:00")) return datetime.fromisoformat(f"{value}T00:00:00+00:00") def _canonical_occurrence_key(value: str) -> Optional[str]: raw = (value or "").strip() if not raw: return None try: dt = _parse_dt_maybe_date(raw) except ValueError: return None return dt.astimezone(timezone.utc).replace(microsecond=0).isoformat() def _normalize_occurrence_key_for_event(value: str, event_start_iso: str) -> Optional[str]: raw = (value or "").strip() if not raw: return None if "T" in raw: return _canonical_occurrence_key(raw) # Allow date-only delete keys by inheriting time+offset from event start. try: base = _parse_dt_maybe_date(event_start_iso) d = date.fromisoformat(raw) except ValueError: return None candidate = base.replace(year=d.year, month=d.month, day=d.day) return candidate.astimezone(timezone.utc).replace(microsecond=0).isoformat() def _normalize_monthly_anchor( start_iso: str, end_iso: str, repeat_type: str, repeat_nth_mode: str, repeat_nth_day: Optional[int], repeat_nth_pos: Optional[int], repeat_nth_weekday: Optional[int], ) -> Tuple[str, str]: if repeat_type != "monthly": return start_iso, end_iso try: start_dt = datetime.fromisoformat(start_iso.replace("Z", "+00:00")) end_dt = datetime.fromisoformat(end_iso.replace("Z", "+00:00")) except ValueError: return start_iso, end_iso duration = end_dt - start_dt adjusted_start = start_dt if repeat_nth_mode == "day_of_month" and repeat_nth_day: day = max(1, min(int(repeat_nth_day), pycalendar.monthrange(start_dt.year, start_dt.month)[1])) adjusted_start = start_dt.replace(day=day) elif repeat_nth_mode == "weekday_of_month" and repeat_nth_pos and repeat_nth_weekday is not None: nth_day = _nth_weekday_of_month(start_dt.year, start_dt.month, int(repeat_nth_weekday), int(repeat_nth_pos)) if nth_day is not None: adjusted_start = start_dt.replace(day=nth_day) adjusted_end = adjusted_start + duration return adjusted_start.isoformat(), adjusted_end.isoformat() def _add_months(dt: datetime, months: int) -> datetime: year = dt.year + (dt.month - 1 + months) // 12 month = (dt.month - 1 + months) % 12 + 1 day = min(dt.day, pycalendar.monthrange(year, month)[1]) return dt.replace(year=year, month=month, day=day) def _add_years(dt: datetime, years: int) -> datetime: try: return dt.replace(year=dt.year + years) except ValueError: # Leap-day fallback return dt.replace(month=2, day=28, year=dt.year + years) def _nth_weekday_of_month(year: int, month: int, weekday: int, pos: int) -> Optional[int]: # weekday: 0=Sunday..6=Saturday, pos: 1..5 or -1 (last) if not (0 <= weekday <= 6 and (1 <= pos <= 5 or pos == -1)): return None py_weekday = (weekday - 1) % 7 # python: Monday=0..Sunday=6 if pos == -1: days_in_month = pycalendar.monthrange(year, month)[1] for day in range(days_in_month, 0, -1): if date(year, month, day).weekday() == py_weekday: return day return None day = 1 hits = 0 days_in_month = pycalendar.monthrange(year, month)[1] while day <= days_in_month: dt = date(year, month, day) if dt.weekday() == py_weekday: hits += 1 if hits == pos: return day day += 1 return None def _window_for_view(view: str, anchor: date) -> Tuple[datetime, datetime]: if view == "day": start = datetime.fromisoformat(f"{anchor.isoformat()}T00:00:00+00:00") return start, start + timedelta(days=1) if view == "week": monday = anchor - timedelta(days=anchor.weekday()) start = datetime.fromisoformat(f"{monday.isoformat()}T00:00:00+00:00") return start, start + timedelta(days=7) if view == "month": first = anchor.replace(day=1) start = datetime.fromisoformat(f"{first.isoformat()}T00:00:00+00:00") end = _add_months(start, 1) return start, end if view == "year": first = date(anchor.year, 1, 1) start = datetime.fromisoformat(f"{first.isoformat()}T00:00:00+00:00") end = _add_years(start, 1) return start, end start = datetime.fromisoformat(f"{anchor.isoformat()}T00:00:00+00:00") return start - timedelta(days=30), start + timedelta(days=90) def _occurrence_overlap(s1: datetime, e1: datetime, s2: datetime, e2: datetime) -> bool: return s1 < e2 and e1 > s2 def make_uid(seed: str) -> str: digest = hashlib.sha256(seed.encode("utf-8")).hexdigest()[:20] return f"{digest}@calendar-wp-plugin" def create_ics_event(event: Dict[str, Any], exdates: List[str]) -> str: lines = [ "BEGIN:VEVENT", f"UID:{event['uid']}", f"DTSTAMP:{_to_ics_ts(utc_now_iso())}", f"SUMMARY:{_ics_escape(event['title'])}", ] if event.get("description"): lines.append(f"DESCRIPTION:{_ics_escape(event['description'])}") if event.get("location"): lines.append(f"LOCATION:{_ics_escape(event['location'])}") if event.get("category"): lines.append(f"CATEGORIES:{_ics_escape(event['category'])}") if event.get("all_day_event"): lines.append(f"DTSTART;VALUE=DATE:{_to_ics_date(event['start_datetime'])}") lines.append(f"DTEND;VALUE=DATE:{_to_ics_date(event['end_datetime'])}") else: lines.append(f"DTSTART;TZID={DEFAULT_TIMEZONE}:{_to_ics_local(event['start_datetime'])}") lines.append(f"DTEND;TZID={DEFAULT_TIMEZONE}:{_to_ics_local(event['end_datetime'])}") rrule = _build_rrule(event) if rrule: lines.append(f"RRULE:{rrule}") if exdates: start_tz = None try: start_tz = datetime.fromisoformat(event["start_datetime"].replace("Z", "+00:00")).tzinfo except ValueError: start_tz = None normalized_exdates = [] for x in exdates: try: dt = datetime.fromisoformat(x.replace("Z", "+00:00")) if start_tz is not None and dt.tzinfo is not None: dt = dt.astimezone(start_tz) normalized_exdates.append(dt.strftime("%Y%m%dT%H%M%S")) except ValueError: continue joined = ",".join(normalized_exdates) if joined: lines.append(f"EXDATE;TZID={DEFAULT_TIMEZONE}:{joined}") lines.append("END:VEVENT") return "\r\n".join(lines) def _build_rrule(event: Dict[str, Any]) -> Optional[str]: freq_map = { "daily": "DAILY", "weekly": "WEEKLY", "monthly": "MONTHLY", "yearly": "YEARLY", "custom": "WEEKLY", # First-pass simplification for fixture. } repeat_type = event.get("repeat_type") if repeat_type not in freq_map or repeat_type == "none": return None parts = [f"FREQ={freq_map[repeat_type]}"] interval = int(event.get("repeat_interval") or 1) if interval > 1: parts.append(f"INTERVAL={interval}") if repeat_type == "monthly": nth_mode = event.get("repeat_nth_mode") or "" if nth_mode == "day_of_month" and event.get("repeat_nth_day"): parts.append(f"BYMONTHDAY={int(event['repeat_nth_day'])}") elif nth_mode == "weekday_of_month" and event.get("repeat_nth_pos") and event.get("repeat_nth_weekday") is not None: weekday_map = ["SU", "MO", "TU", "WE", "TH", "FR", "SA"] pos = int(event["repeat_nth_pos"]) wd = int(event["repeat_nth_weekday"]) if 0 <= wd <= 6 and (1 <= pos <= 5 or pos == -1): parts.append(f"BYDAY={weekday_map[wd]}") parts.append(f"BYSETPOS={pos}") mode = event.get("repeat_range_mode") if mode == "count" and event.get("repeat_count"): parts.append(f"COUNT={int(event['repeat_count'])}") elif mode == "until" and event.get("repeat_until"): # date-only UNTIL for first pass until = event["repeat_until"].replace("-", "") parts.append(f"UNTIL={until}T235959") return ";".join(parts) def _to_ics_ts(iso_ts: str) -> str: dt = datetime.fromisoformat(iso_ts.replace("Z", "+00:00")).astimezone(timezone.utc) return dt.strftime("%Y%m%dT%H%M%SZ") def _to_ics_local(iso_ts: str) -> str: dt = datetime.fromisoformat(iso_ts.replace("Z", "+00:00")) return dt.strftime("%Y%m%dT%H%M%S") def _to_ics_date(iso_ts: str) -> str: if "T" in iso_ts: dt = datetime.fromisoformat(iso_ts.replace("Z", "+00:00")) return dt.strftime("%Y%m%d") return iso_ts.replace("-", "") def _ics_escape(value: str) -> str: return ( value.replace("\\", "\\\\") .replace(";", r"\;") .replace(",", r"\,") .replace("\n", r"\n") ) @dataclass class AuthContext: actor_type: str actor_id: str role: str class _TraceWriter: def __init__(self, handler: "FixtureHandler", raw): self._handler = handler self._raw = raw def write(self, data): self._handler._trace_capture_response_bytes(data) return self._raw.write(data) def flush(self): return self._raw.flush() def __getattr__(self, name): return getattr(self._raw, name) class FixtureHandler(BaseHTTPRequestHandler): server_version = "CalendarFixture/0.1" _trace_body_cap = 4096 _trace_body_preview_cap = 1200 def setup(self) -> None: super().setup() self.wfile = _TraceWriter(self, self.wfile) self._trace_started = False self._trace_request_body = b"" self._trace_response_status: Optional[int] = None self._trace_response_headers: List[Tuple[str, str]] = [] self._trace_response_body = bytearray() self._cached_request_body: Optional[bytes] = None def _trace_start(self) -> None: self._trace_started = True self._trace_request_body = b"" self._trace_response_status = None self._trace_response_headers = [] self._trace_response_body = bytearray() self._cached_request_body = None def _trace_capture_response_bytes(self, data: Any) -> None: if not self._trace_started: return raw = data.encode("utf-8", errors="replace") if isinstance(data, str) else bytes(data) remaining = self._trace_body_cap - len(self._trace_response_body) if remaining > 0: self._trace_response_body.extend(raw[:remaining]) def send_response(self, code: int, message: Optional[str] = None) -> None: if getattr(self, "_trace_started", False): self._trace_response_status = int(code) super().send_response(code, message) def send_header(self, keyword: str, value: str) -> None: if getattr(self, "_trace_started", False): self._trace_response_headers.append((keyword, value)) super().send_header(keyword, value) def _trace_log_path(self) -> Path: return fixture_trace_log_path() def _redact_headers(self, headers: Dict[str, str]) -> Dict[str, str]: out: Dict[str, str] = {} for k, v in headers.items(): if k.lower() in {"authorization", "cookie", "set-cookie"}: out[k] = "***" else: out[k] = v return out def _sanitize_json_for_log(self, value: Any) -> Any: if isinstance(value, dict): out: Dict[str, Any] = {} for k, v in value.items(): if str(k).lower() in {"password", "new_password", "token"}: out[k] = "***" else: out[k] = self._sanitize_json_for_log(v) return out if isinstance(value, list): return [self._sanitize_json_for_log(v) for v in value] return value def _decode_body_for_log(self, body: bytes, content_type: str) -> str: if not body: return "" ct = (content_type or "").lower() raw = body[: self._trace_body_cap] text = raw.decode("utf-8", errors="replace") if "application/json" in ct: try: compact = json.dumps(self._sanitize_json_for_log(json.loads(text)), ensure_ascii=True) if len(compact) > self._trace_body_preview_cap: return compact[: self._trace_body_preview_cap] + " ...(truncated)" return compact except json.JSONDecodeError: return text[: self._trace_body_preview_cap] if "text/html" in ct: return f"[omitted html payload: {len(body)} bytes]" if "application/xml" in ct or "text/xml" in ct: return text[: self._trace_body_preview_cap] + (" ...(truncated)" if len(text) > self._trace_body_preview_cap else "") if "text/calendar" in ct: return text[: self._trace_body_preview_cap] + (" ...(truncated)" if len(text) > self._trace_body_preview_cap else "") if ct.startswith("text/"): return text[: self._trace_body_preview_cap] + (" ...(truncated)" if len(text) > self._trace_body_preview_cap else "") return f"[omitted non-text payload: {len(body)} bytes, content-type={content_type or 'unknown'}]" def _strip_http_response_preamble(self, raw: bytes) -> bytes: if not raw.startswith(b"HTTP/"): return raw marker = b"\r\n\r\n" idx = raw.find(marker) if idx == -1: return raw return raw[idx + len(marker) :] def _read_trace_entries(self, limit: int = 20) -> List[Dict[str, Any]]: path = self._trace_log_path() if not path.exists(): return [] out: deque[Dict[str, Any]] = deque(maxlen=max(1, min(limit, 200))) with path.open("r", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue try: out.append(json.loads(line)) except json.JSONDecodeError: continue return list(out) def _trace_finish(self) -> None: if not self._trace_started: return req_headers = self._redact_headers({k: v for (k, v) in self.headers.items()}) resp_headers: Dict[str, str] = {} for k, v in self._trace_response_headers: resp_headers[k] = "***" if k.lower() == "set-cookie" else v req_ct = req_headers.get("Content-Type", "") resp_ct = resp_headers.get("Content-Type", "") response_raw = bytes(self._trace_response_body) response_body_raw = self._strip_http_response_preamble(response_raw) entry = { "ts": utc_now_iso(), "client": self.client_address[0] if self.client_address else "", "method": self.command, "path": self.path, "request": { "headers": req_headers, "body": self._decode_body_for_log(self._trace_request_body, req_ct), "body_bytes": len(self._trace_request_body), "body_truncated": len(self._trace_request_body) > self._trace_body_cap, }, "response": { "status": self._trace_response_status, "headers": resp_headers, "body": self._decode_body_for_log(response_body_raw, resp_ct), "body_bytes": len(response_body_raw), "body_truncated": len(self._trace_response_body) >= self._trace_body_cap, }, } path = self._trace_log_path() path.parent.mkdir(parents=True, exist_ok=True) with path.open("a", encoding="utf-8") as f: f.write(json.dumps(entry, ensure_ascii=True) + "\n") self._trace_started = False def _read_request_body_bytes(self) -> bytes: if self._cached_request_body is not None: return self._cached_request_body length = int(self.headers.get("Content-Length", "0") or "0") if length <= 0: data = b"" else: data = self.rfile.read(length) self._cached_request_body = data self._trace_request_body = data[: self._trace_body_cap + 1] return data def _slug_value(self) -> str: conn = ensure_db() try: raw = get_setting(conn, "url_slug", "").strip() finally: conn.close() raw = raw.strip("/") return raw def _slug_prefix(self) -> str: slug = self._slug_value() return f"/{slug}" if slug else "" def _strip_slug(self, path: str) -> Optional[str]: prefix = self._slug_prefix() if not prefix: return path if path == prefix: return "/" if path.startswith(prefix + "/"): return path[len(prefix) :] return None def _with_slug(self, path: str) -> str: prefix = self._slug_prefix() return f"{prefix}{path}" if prefix else path def _request_query(self) -> Dict[str, List[str]]: return parse_qs(urlparse(self.path).query) def _json_body(self) -> Dict[str, Any]: data = self._read_request_body_bytes() if not data: return {} return json.loads(data.decode("utf-8")) def _write_json(self, status: int, payload: Dict[str, Any]) -> None: body = json.dumps(payload).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _write_text(self, status: int, text: str, content_type: str = "text/plain; charset=utf-8") -> None: body = text.encode("utf-8") self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _not_found(self) -> None: self._write_json(404, {"error": {"code": "not_found", "message": "Route not found"}}) def _authenticate_wp_admin(self) -> Optional[AuthContext]: # Fixture-only auth: header X-WP-User or query ?as=admin|editor. wp_user = self.headers.get("X-WP-User", "") if not wp_user: wp_user = (self._request_query().get("as") or [""])[0] if wp_user == "admin": return AuthContext("wp_user", "admin", "wp_admin") if wp_user == "editor": return AuthContext("wp_user", "editor", "wp_editor") return None def _authenticate_caldav_user(self) -> Optional[AuthContext]: creds = parse_basic_auth(self.headers.get("Authorization")) if not creds: return None email, password = creds conn = ensure_db() row = conn.execute( """ SELECT id, email, password_hash, account_status, email_verified_at FROM caldav_users WHERE email = ? """, (email,), ).fetchone() if not row: conn.close() return None if not verify_password(password, row["password_hash"]): conn.close() return None if row["account_status"] != "active" or not row["email_verified_at"]: conn.close() return None # Transparent legacy-hash upgrade on successful login. if re.fullmatch(r"[0-9a-f]{64}", row["password_hash"]): conn.execute( "UPDATE caldav_users SET password_hash = ?, updated_at = ? WHERE id = ?", (hash_password(password), utc_now_iso(), int(row["id"])), ) conn.commit() conn.close() return AuthContext("caldav_user", str(row["id"]), "caldav_write") def _require_api_admin(self) -> Optional[AuthContext]: auth = self._authenticate_wp_admin() if not auth: self._write_json(401, {"error": {"code": "authentication_error", "message": "Admin authentication required"}}) return None return auth def _require_event_read_auth(self) -> Optional[AuthContext]: wp = self._authenticate_wp_admin() if wp: return wp caldav = self._authenticate_caldav_user() if caldav: return caldav self._write_json(401, {"error": {"code": "authentication_error", "message": "Login required"}}) return None def _require_event_write_auth(self) -> Optional[AuthContext]: wp = self._authenticate_wp_admin() if wp and wp.role in {"wp_admin", "wp_editor"}: return wp caldav = self._authenticate_caldav_user() if caldav and caldav.role == "caldav_write": return caldav if wp or caldav: self._write_json(403, {"error": {"code": "authorization_error", "message": "write capability required"}}) return None self._write_json(401, {"error": {"code": "authentication_error", "message": "Login required"}}) return None def _validate_event_payload(self, payload: Dict[str, Any]) -> Optional[Dict[str, Any]]: title = (payload.get("title") or "").strip() start = payload.get("start_datetime") end = payload.get("end_datetime") if not title: self._write_json(422, {"error": {"code": "validation_error", "message": "title is required"}}) return None if not start or not end: self._write_json(422, {"error": {"code": "validation_error", "message": "start_datetime and end_datetime are required"}}) return None try: ds = datetime.fromisoformat(start.replace("Z", "+00:00")) de = datetime.fromisoformat(end.replace("Z", "+00:00")) except ValueError: self._write_json(422, {"error": {"code": "validation_error", "message": "invalid datetime format"}}) return None if de < ds: self._write_json(422, {"error": {"code": "validation_error", "message": "end_datetime must be >= start_datetime"}}) return None normalized = { "title": title, "description": payload.get("description", ""), "location": payload.get("location", ""), "category": payload.get("category", ""), "all_day_event": 1 if payload.get("all_day_event") else 0, "start_datetime": start, "end_datetime": end, "repeat_type": payload.get("repeat_type", "none"), "repeat_interval": int(payload.get("repeat_interval", 1)), "repeat_nth_mode": payload.get("repeat_nth_mode", "") or "", "repeat_nth_day": payload.get("repeat_nth_day"), "repeat_nth_pos": payload.get("repeat_nth_pos"), "repeat_nth_weekday": payload.get("repeat_nth_weekday"), "repeat_range_mode": payload.get("repeat_range_mode", "none"), "repeat_count": payload.get("repeat_count"), "repeat_until": payload.get("repeat_until"), "timezone": payload.get("timezone", DEFAULT_TIMEZONE), } normalized_start, normalized_end = _normalize_monthly_anchor( normalized["start_datetime"], normalized["end_datetime"], str(normalized["repeat_type"] or "none"), str(normalized["repeat_nth_mode"] or ""), int(normalized["repeat_nth_day"]) if normalized["repeat_nth_day"] not in (None, "") else None, int(normalized["repeat_nth_pos"]) if normalized["repeat_nth_pos"] not in (None, "") else None, int(normalized["repeat_nth_weekday"]) if normalized["repeat_nth_weekday"] not in (None, "") else None, ) normalized["start_datetime"] = normalized_start normalized["end_datetime"] = normalized_end return normalized def do_OPTIONS(self) -> None: self._trace_start() try: path = self._strip_slug(urlparse(self.path).path) if path is None: self.send_response(404) self.end_headers() return if path.startswith("/caldav"): self.send_response(200) self.send_header("DAV", "1, 2, calendar-access") self.send_header("Allow", "OPTIONS, PROPFIND, REPORT, GET, PUT, DELETE") self.end_headers() return self.send_response(200) self.send_header("Allow", "GET, POST, PUT, PATCH, DELETE, OPTIONS") self.end_headers() finally: self._trace_finish() def do_GET(self) -> None: self._trace_start() try: parsed = urlparse(self.path) path = self._strip_slug(parsed.path) if path is None: self._not_found() return if path == "/": prefix = self._slug_prefix() self._write_text( 200, "calendar fixture is running\n" f"slug prefix: {prefix or '(none)'}\n" f"admin: {self._with_slug('/admin.php')}?as=admin\n" f"public calendar: {self._with_slug('/calendar')}\n" f"api: {self._with_slug('/wp-json/calendar/v1/events')}\n" f"ics: {self._with_slug('/calendar.ics')}\n" f"caldav: {self._with_slug('/caldav/')}\n", ) return if path in {"/admin.php", "/admin.php/"}: self._handle_admin_page(parsed) return if path.startswith("/wp-admin/admin.php"): self._handle_admin_page(parsed) return if path == "/calendar.ics": self._handle_ics_get() return if path in {"/calendar", "/calendar/"}: self._handle_public_calendar_page() return if path in {"/calendar-sidebar", "/calendar-sidebar/"}: self._handle_sidebar_shortcode_page() return if path.startswith(API_BASE): self._handle_api_get(path, parsed) return if path.startswith("/caldav/"): self._handle_caldav_get(path) return self._not_found() finally: self._trace_finish() def _base_page(self, title: str, body_html: str) -> str: return f""" {_html_escape(title)} {body_html} """ def _handle_public_calendar_page(self) -> None: page_html = """
Calendar Fixture Public Calendar UI
ICS Link CalDAV
""" page_html = page_html.replace("__ICS__", self._with_slug("/calendar.ics")) page_html = page_html.replace("__CALDAV__", self._with_slug("/caldav/")) page_html = page_html.replace("__ROOT__", json.dumps(self._slug_prefix())) html = self._base_page("Calendar", page_html) self._write_text(200, html, "text/html; charset=utf-8") def _handle_sidebar_shortcode_page(self) -> None: page_html = """
Calendar Fixture Sidebar Shortcode Preview
Shortcode: [calendar_sidebar_upcoming]
Window: next 14 days
""" page_html = page_html.replace("__ROOT__", json.dumps(self._slug_prefix())) html = self._base_page("Calendar Sidebar Shortcode", page_html) self._write_text(200, html, "text/html; charset=utf-8") def do_POST(self) -> None: self._trace_start() try: path = self._strip_slug(urlparse(self.path).path) if path is None: self._not_found() return if path.startswith(API_BASE): self._handle_api_post(path) return self._not_found() finally: self._trace_finish() def do_PUT(self) -> None: self._trace_start() try: path = self._strip_slug(urlparse(self.path).path) if path is None: self._not_found() return if path.startswith(API_BASE): self._handle_api_put(path) return if path.startswith("/caldav/"): self._handle_caldav_put(path) return self._not_found() finally: self._trace_finish() def do_PATCH(self) -> None: self._trace_start() try: path = self._strip_slug(urlparse(self.path).path) if path is None: self._not_found() return if path.startswith(API_BASE): self._handle_api_patch(path) return self._not_found() finally: self._trace_finish() def do_DELETE(self) -> None: self._trace_start() try: path = self._strip_slug(urlparse(self.path).path) if path is None: self._not_found() return if path.startswith(API_BASE): self._handle_api_delete(path) return if path.startswith("/caldav/"): self._handle_caldav_delete(path) return self._not_found() finally: self._trace_finish() def do_PROPFIND(self) -> None: self._trace_start() try: self._read_request_body_bytes() path = self._strip_slug(urlparse(self.path).path) if path is None: self._not_found() return if not path.startswith("/caldav/"): self._not_found() return auth = self._authenticate_caldav_user() if not auth: self.send_response(401) self.send_header("WWW-Authenticate", 'Basic realm="calendar-caldav-fixture"') self.end_headers() return prefix = self._slug_prefix() caldav_root = f"{prefix}/caldav/" principals_root = f"{prefix}/caldav/principals/" principal_href = f"{prefix}/caldav/principals/user/" calendars_root = f"{prefix}/caldav/calendars/" public_calendar = f"{prefix}/caldav/calendars/public/" conn = ensure_db() caldav_calendar_name = get_setting(conn, "caldav_calendar_name", "Public Calendar") conn.close() def response_block(href: str, prop_xml: str) -> str: return textwrap.dedent( f"""\ {href} {prop_xml} HTTP/1.1 200 OK """ ).strip() root_props = ( "" f"{principal_href}" ) principal_props = ( "" f"{calendars_root}" ) calendar_home_props = "" public_calendar_props = ( "" f"{_xml_escape(caldav_calendar_name)}" "" "" "" ) responses: List[str] = [] canonical = path if path.endswith("/") else f"{path}/" if canonical in {"/caldav/", "/caldav"}: responses.append(response_block(caldav_root, root_props)) if self.headers.get("Depth", "0") != "0": responses.append(response_block(principal_href, principal_props)) responses.append(response_block(calendars_root, calendar_home_props)) responses.append(response_block(public_calendar, public_calendar_props)) elif canonical == "/caldav/principals/": responses.append(response_block(principals_root, "")) if self.headers.get("Depth", "0") != "0": responses.append(response_block(principal_href, principal_props)) elif canonical == "/caldav/principals/user/": responses.append(response_block(principal_href, principal_props)) elif canonical == "/caldav/calendars/": responses.append(response_block(calendars_root, calendar_home_props)) if self.headers.get("Depth", "0") != "0": responses.append(response_block(public_calendar, public_calendar_props)) elif canonical == "/caldav/calendars/public/": responses.append(response_block(public_calendar, public_calendar_props)) if self.headers.get("Depth", "0") != "0": conn = ensure_db() rows = conn.execute( "SELECT id, etag, caldav_resource AS resource FROM events WHERE calendar_id = ? AND caldav_resource IS NOT NULL ORDER BY id ASC", (SHARED_CALENDAR_ID,), ).fetchall() conn.close() for row in rows: href = f"{prefix}/caldav/calendars/public/{row['resource']}" item_props = ( "" "text/calendar; charset=utf-8" f"{row['etag']}" ) responses.append(response_block(href, item_props)) else: self._not_found() return body = ( '\n' '\n' + "\n".join(responses) + "\n\n" ) self.send_response(207) self.send_header("Content-Type", "application/xml; charset=utf-8") self.send_header("Content-Length", str(len(body.encode("utf-8")))) self.end_headers() self.wfile.write(body.encode("utf-8")) finally: self._trace_finish() def do_REPORT(self) -> None: self._trace_start() try: request_body = self._read_request_body_bytes().decode("utf-8", errors="replace") path = self._strip_slug(urlparse(self.path).path) if path is None: self._not_found() return if not path.startswith("/caldav/"): self._not_found() return auth = self._authenticate_caldav_user() if not auth: self.send_response(401) self.send_header("WWW-Authenticate", 'Basic realm="calendar-caldav-fixture"') self.end_headers() return conn = ensure_db() all_rows = conn.execute( "SELECT * FROM events WHERE calendar_id = ? AND caldav_resource IS NOT NULL ORDER BY id ASC", (SHARED_CALENDAR_ID,) ).fetchall() conn.close() rows = list(all_rows) requested_hrefs = re.findall(r"\s*([^<]+)\s*", request_body, re.IGNORECASE) if "calendar-multiget" in request_body.lower() and requested_hrefs: wanted_resources: set[str] = set() for href in requested_hrefs: href_path = urlparse(href.strip()).path stripped = self._strip_slug(href_path) if stripped is None: stripped = href_path filename = _caldav_filename_from_path(stripped) if filename: wanted_resources.add(filename) if wanted_resources: rows = [r for r in all_rows if str(r["caldav_resource"]) in wanted_resources] else: rows = [] prefix = self._slug_prefix() responses = [] for row in rows: evt = event_to_dict(row) href = f"{prefix}/caldav/calendars/public/{row['caldav_resource']}" ics = self._event_as_ics(evt) responses.append( textwrap.dedent( f"""\ {href} {row["etag"]} {_xml_escape(ics)} HTTP/1.1 200 OK """ ).strip() ) body = ( '\n' '\n' + "\n".join(responses) + "\n\n" ) self.send_response(207) self.send_header("Content-Type", "application/xml; charset=utf-8") self.send_header("Content-Length", str(len(body.encode("utf-8")))) self.end_headers() self.wfile.write(body.encode("utf-8")) finally: self._trace_finish() def _handle_admin_page(self, parsed) -> None: auth = self._authenticate_wp_admin() if not auth: html = self._base_page( "Admin Access", f"""
Calendar Fixture Admin
Admin auth required (use ?as=admin or ?as=editor in fixture).

Useful Links

""", ) self._write_text(401, html, "text/html; charset=utf-8") return query = parse_qs(parsed.query) page_values = query.get("page") or [] page = page_values[0] if page_values else "" as_user = (query.get("as") or ["admin"])[0] allowed = {"calendar-users", "calendar-setup", "calendar-diagnostics"} if page and page not in allowed: self._write_text(404, "unknown admin page") return if page in {"calendar-users", "calendar-setup", "calendar-diagnostics"} and auth.role != "wp_admin": self._write_text(403, "forbidden") return nav = f"""
Calendar Fixture Admin   Users Setup Diagnostics Public Calendar
Authenticated as: {auth.actor_id} ({auth.role})
""" if not page: body = nav + f"""

Admin Pages

""" html = self._base_page("Admin Index", body) if page == "calendar-users": body = nav + self._admin_users_body(as_user) + "" html = self._base_page("Users", body) elif page == "calendar-setup": body = nav + self._admin_setup_body(as_user) + "" html = self._base_page("Setup", body) elif page == "calendar-diagnostics": body = nav + self._admin_diagnostics_body(as_user) + "" html = self._base_page("Diagnostics", body) self._write_text(200, html, "text/html; charset=utf-8") def _admin_edit_body(self, as_user: str) -> str: return f"""

Events

Events

TitleStartRepeatActions

  
"""

    def _admin_users_body(self, as_user: str) -> str:
        return f"""
  

    
Users

    
    

      
IDEmailEmail VerifiedStatusAction

      
    

  

  

  
"""

    def _admin_setup_body(self, as_user: str) -> str:
        return f"""
  

    
Setup

    

    

    

    

    
    
  

  

    
Shortcode Semantics

    
[calendar] renders the full interactive calendar page (views, filters, login, and event edit for approved users).

    
[calendar_sidebar_upcoming] renders a compact upcoming-events list for sidebar/widget use, showing occurrences for the next 14 days.

    
If URL Slug is set, all plugin routes are prefixed with that slug (for example: /my-slug/calendar, /my-slug/calendar-sidebar, /my-slug/calendar.ics, /my-slug/caldav/).

  

  

  
"""

    def _admin_diagnostics_body(self, as_user: str) -> str:
        return f"""
  

    
Diagnostics

    
Shows the most recent HTTP trace events (requests/responses) from the fixture log.

    
    

      
TimestampMethodPathStatusRequestResponse

      
    

  

  

  
"""

    def _handle_api_get(self, path: str, parsed) -> None:
        if path == f"{API_BASE}/public/events":
            query = parse_qs(parsed.query)
            view = (query.get("view") or ["month"])[0]
            date_str = (query.get("date") or [date.today().isoformat()])[0]
            try:
                anchor = date.fromisoformat(date_str)
            except ValueError:
                self._write_json(422, {"error": {"code": "validation_error", "message": "invalid date"}})
                return
            window_start, window_end = _window_for_view(view, anchor)
            conn = ensure_db()
            rows = conn.execute("SELECT * FROM events WHERE calendar_id = ? ORDER BY id ASC", (SHARED_CALENDAR_ID,)).fetchall()
            ex_rows = conn.execute(
                "SELECT event_id, occurrence_key FROM recurrence_exceptions WHERE exception_type = 'deleted_occurrence'"
            ).fetchall()
            conn.close()
            ex_map: Dict[int, set] = {}
            for ex in ex_rows:
                ex_map.setdefault(ex["event_id"], set()).add(ex["occurrence_key"])
            occurrences: List[Dict[str, Any]] = []
            for row in rows:
                evt = event_to_dict(row)
                occurrences.extend(self._expand_event(evt, window_start, window_end, ex_map.get(evt["id"], set())))
            occurrences.sort(key=lambda x: x["occurrence_start"])
            self._write_json(200, {"data": occurrences, "meta": {"count": len(occurrences), "view": view}})
            return

        if path == f"{API_BASE}/public/sidebar-events":
            anchor = date.today()
            window_start = datetime.fromisoformat(f"{anchor.isoformat()}T00:00:00+00:00")
            window_end = window_start + timedelta(days=14)
            conn = ensure_db()
            rows = conn.execute("SELECT * FROM events WHERE calendar_id = ? ORDER BY id ASC", (SHARED_CALENDAR_ID,)).fetchall()
            ex_rows = conn.execute(
                "SELECT event_id, occurrence_key FROM recurrence_exceptions WHERE exception_type = 'deleted_occurrence'"
            ).fetchall()
            conn.close()
            ex_map: Dict[int, set] = {}
            for ex in ex_rows:
                ex_map.setdefault(ex["event_id"], set()).add(ex["occurrence_key"])
            occurrences: List[Dict[str, Any]] = []
            for row in rows:
                evt = event_to_dict(row)
                occurrences.extend(self._expand_event(evt, window_start, window_end, ex_map.get(evt["id"], set())))
            occurrences.sort(key=lambda x: x["occurrence_start"])
            self._write_json(200, {"data": occurrences, "meta": {"count": len(occurrences), "window_days": 14}})
            return

        if path == f"{API_BASE}/events":
            auth = self._require_event_read_auth()
            if not auth:
                return
            conn = ensure_db()
            rows = conn.execute(
                "SELECT * FROM events WHERE calendar_id = ? ORDER BY id ASC", (SHARED_CALENDAR_ID,)
            ).fetchall()
            conn.close()
            items = [event_to_dict(r) for r in rows]
            self._write_json(200, {"data": items, "meta": {"count": len(items)}})
            return

        m = re.fullmatch(rf"{re.escape(API_BASE)}/events/(\d+)", path)
        if m:
            auth = self._require_event_read_auth()
            if not auth:
                return
            event_id = int(m.group(1))
            conn = ensure_db()
            row = conn.execute(
                "SELECT * FROM events WHERE id = ? AND calendar_id = ?", (event_id, SHARED_CALENDAR_ID)
            ).fetchone()
            conn.close()
            if not row:
                self._write_json(404, {"error": {"code": "not_found", "message": "event not found"}})
                return
            self._write_json(200, {"data": event_to_dict(row)})
            return

        if path == f"{API_BASE}/admin/users":
            auth = self._require_api_admin()
            if not auth:
                return
            if auth.role != "wp_admin":
                self._write_json(403, {"error": {"code": "authorization_error", "message": "admin role required"}})
                return
            conn = ensure_db()
            rows = conn.execute(
                "SELECT id, email, email_verified_at, account_status, updated_at FROM caldav_users ORDER BY id ASC"
            ).fetchall()
            conn.close()
            self._write_json(200, {"data": [dict(r) for r in rows]})
            return

        m = re.fullmatch(rf"{re.escape(API_BASE)}/events/(\d+)/occurrences", path)
        if m:
            auth = self._require_event_read_auth()
            if not auth:
                return
            event_id = int(m.group(1))
            query = parse_qs(parsed.query)
            from_s = (query.get("from") or [date.today().isoformat()])[0]
            months = int((query.get("months") or ["3"])[0])
            try:
                from_d = date.fromisoformat(from_s)
            except ValueError:
                self._write_json(422, {"error": {"code": "validation_error", "message": "invalid from date"}})
                return
            start = datetime.fromisoformat(f"{from_d.isoformat()}T00:00:00+00:00")
            end = _add_months(start, max(1, min(months, 12)))
            conn = ensure_db()
            row = conn.execute("SELECT * FROM events WHERE id = ? AND calendar_id = ?", (event_id, SHARED_CALENDAR_ID)).fetchone()
            if not row:
                conn.close()
                self._write_json(404, {"error": {"code": "not_found", "message": "event not found"}})
                return
            ex_rows = conn.execute(
                "SELECT occurrence_key FROM recurrence_exceptions WHERE event_id = ? AND exception_type = 'deleted_occurrence'",
                (event_id,),
            ).fetchall()
            conn.close()
            deleted = {r["occurrence_key"] for r in ex_rows}
            occ = self._expand_event(event_to_dict(row), start, end, deleted)
            self._write_json(200, {"data": occ})
            return

        if path == f"{API_BASE}/admin/setup":
            auth = self._require_api_admin()
            if not auth:
                return
            if auth.role != "wp_admin":
                self._write_json(403, {"error": {"code": "authorization_error", "message": "admin role required"}})
                return
            conn = ensure_db()
            payload = {
                "presentation_name": get_setting(conn, "presentation_name", "Calendar"),
                "caldav_calendar_name": get_setting(conn, "caldav_calendar_name", "Public Calendar"),
                "url_slug": get_setting(conn, "url_slug", ""),
                "ics_access_mode": get_setting(conn, "ics_access_mode", "public_read"),
            }
            conn.close()
            self._write_json(200, {"data": payload})
            return

        if path == f"{API_BASE}/admin/diagnostics":
            auth = self._require_api_admin()
            if not auth:
                return
            if auth.role != "wp_admin":
                self._write_json(403, {"error": {"code": "authorization_error", "message": "admin role required"}})
                return
            query = parse_qs(parsed.query)
            try:
                limit = int((query.get("limit") or ["20"])[0])
            except ValueError:
                self._write_json(422, {"error": {"code": "validation_error", "message": "invalid limit"}})
                return
            entries = self._read_trace_entries(limit=max(1, min(limit, 200)))
            self._write_json(200, {"data": entries, "meta": {"count": len(entries)}})
            return

        if path == f"{API_BASE}/users/me":
            creds = parse_basic_auth(self.headers.get("Authorization"))
            email_hint = (creds[0].strip().lower() if creds else "unknown")
            if _rate_limited("user_login", f"{self.client_address[0]}:{email_hint}", limit=20, window_seconds=300):
                self._write_json(429, {"error": {"code": "rate_limited", "message": "Too many login attempts. Try again later."}})
                return
            auth = self._authenticate_caldav_user()
            if not auth:
                self._write_json(401, {"error": {"code": "authentication_error", "message": "Login failure"}})
                return
            conn = ensure_db()
            row = conn.execute(
                "SELECT id, email, account_status, email_verified_at FROM caldav_users WHERE id = ?",
                (int(auth.actor_id),),
            ).fetchone()
            conn.close()
            self._write_json(200, {"data": dict(row)})
            return

        self._not_found()

    def _handle_api_post(self, path: str) -> None:
        if path == f"{API_BASE}/events":
            auth = self._require_event_write_auth()
            if not auth:
                return
            payload = self._json_body()
            normalized = self._validate_event_payload(payload)
            if normalized is None:
                return
            now = utc_now_iso()
            uid = payload.get("uid") or make_uid(f"{normalized['title']}:{normalized['start_datetime']}:{now}")
            etag = mk_etag(f"{uid}:{now}:1")
            conn = ensure_db()
            cur = conn.execute(
                """
                INSERT INTO events
                  (uid, title, description, location, category, all_day_event, start_datetime, end_datetime,
                   repeat_type, repeat_interval, repeat_nth_mode, repeat_nth_day, repeat_nth_pos, repeat_nth_weekday,
                   repeat_range_mode, repeat_count, repeat_until, timezone,
                   calendar_id, caldav_resource, etag, sync_version, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    uid,
                    normalized["title"],
                    normalized["description"],
                    normalized["location"],
                    normalized["category"],
                    normalized["all_day_event"],
                    normalized["start_datetime"],
                    normalized["end_datetime"],
                    normalized["repeat_type"],
                    normalized["repeat_interval"],
                    normalized["repeat_nth_mode"],
                    normalized["repeat_nth_day"],
                    normalized["repeat_nth_pos"],
                    normalized["repeat_nth_weekday"],
                    normalized["repeat_range_mode"],
                    normalized["repeat_count"],
                    normalized["repeat_until"],
                    normalized["timezone"],
                    SHARED_CALENDAR_ID,
                    None,
                    etag,
                    1,
                    now,
                    now,
                ),
            )
            event_id = cur.lastrowid
            conn.execute("UPDATE events SET caldav_resource = ? WHERE id = ?", (f"{event_id}.ics", event_id))
            conn.commit()
            row = conn.execute("SELECT * FROM events WHERE id = ?", (event_id,)).fetchone()
            log_audit(conn, auth.actor_type, auth.actor_id, "event.create", "event", str(event_id), "success", {})
            conn.close()
            self._write_json(201, {"data": event_to_dict(row)})
            return

        if path == f"{API_BASE}/users/register":
            if _rate_limited("user_register", self.client_address[0], limit=8, window_seconds=300):
                self._write_json(429, {"error": {"code": "rate_limited", "message": "Too many registration attempts. Try again later."}})
                return
            payload = self._json_body()
            email = (payload.get("email") or "").strip().lower()
            password = payload.get("password") or ""
            if not email or len(password) < 8:
                self._write_json(422, {"error": {"code": "validation_error", "message": "email and strong password required"}})
                return
            now = utc_now_iso()
            conn = ensure_db()
            try:
                cur = conn.execute(
                    """
                    INSERT INTO caldav_users
                      (email, password_hash, email_verified_at, account_status, access_level, request_state, created_at, updated_at)
                    VALUES (?, ?, NULL, 'pending_approval', 'write', 'requested', ?, ?)
                    """,
                    (email, hash_password(password), now, now),
                )
            except sqlite3.IntegrityError:
                conn.close()
                self._write_json(409, {"error": {"code": "conflict_error", "message": "user already exists"}})
                return
            user_id = cur.lastrowid
            token_plain = secrets.token_urlsafe(24)
            conn.execute(
                """
                INSERT INTO user_tokens (user_id, token_type, token_hash, expires_at, used_at, created_at)
                VALUES (?, 'verify_email', ?, ?, NULL, ?)
                """,
                (user_id, hash_password(token_plain), (datetime.now(timezone.utc) + timedelta(seconds=VERIFY_TOKEN_TTL_SECONDS)).replace(microsecond=0).isoformat(), now),
            )
            conn.commit()
            conn.close()
            sent, send_info = send_fixture_email(
                email,
                "Calendar account verification",
                f"Your verification token is: {token_plain}\nUse this token in the calendar login dialog to verify your email.",
            )
            self._write_json(
                201,
                {
                    "data": {
                        "user_id": user_id,
                        "status": "pending_approval",
                        "email_status": send_info,
                        "email_sent": sent,
                    }
                },
            )
            return

        if path == f"{API_BASE}/users/verify":
            if _rate_limited("user_verify", self.client_address[0], limit=20, window_seconds=300):
                self._write_json(429, {"error": {"code": "rate_limited", "message": "Too many verification attempts. Try again later."}})
                return
            payload = self._json_body()
            token = payload.get("token") or ""
            conn = ensure_db()
            now = utc_now_iso()
            candidates = conn.execute(
                """
                SELECT id, user_id, token_hash, expires_at FROM user_tokens
                WHERE token_type = 'verify_email' AND used_at IS NULL
                ORDER BY id DESC
                """
            ).fetchall()
            row = None
            for cand in candidates:
                if cand["expires_at"] < now:
                    continue
                if verify_password(token, cand["token_hash"]):
                    row = cand
                    break
            if not row:
                conn.close()
                self._write_json(422, {"error": {"code": "validation_error", "message": "invalid token"}})
                return
            conn.execute("UPDATE user_tokens SET used_at = ? WHERE id = ?", (now, row["id"]))
            conn.execute(
                "UPDATE caldav_users SET email_verified_at = ?, account_status = 'pending_approval', request_state = 'requested', updated_at = ? WHERE id = ?",
                (now, now, row["user_id"]),
            )
            conn.commit()
            requester = conn.execute("SELECT email FROM caldav_users WHERE id = ?", (row["user_id"],)).fetchone()
            conn.close()
            cfg = _smtp_settings()
            admin_to = (cfg.get("SMTP_ADMIN_TO") or "").strip()
            sent = False
            send_info = "admin_email_not_configured"
            if admin_to:
                req_email = requester["email"] if requester else f"user_id={row['user_id']}"
                host = (self.headers.get("Host") or "127.0.0.1:8080").strip()
                proto = (self.headers.get("X-Forwarded-Proto") or "http").strip().lower()
                scheme = "https" if proto == "https" else "http"
                approval_path = self._with_slug("/wp-admin/admin.php?page=calendar-users") + "&as=admin"
                approval_url = f"{scheme}://{host}{approval_path}"
                sent, send_info = send_fixture_email(
                    admin_to,
                    "Calendar registration pending approval",
                    f"User {req_email} (id {row['user_id']}) verified email and is awaiting approval.\nApprove here: {approval_url}",
                )
            self._write_json(200, {"data": {"status": "pending_approval", "email_status": send_info, "email_sent": sent}})
            return

        if path == f"{API_BASE}/users/forgot-password":
            if _rate_limited("user_forgot", self.client_address[0], limit=12, window_seconds=300):
                self._write_json(429, {"error": {"code": "rate_limited", "message": "Too many password reset requests. Try again later."}})
                return
            payload = self._json_body()
            email = (payload.get("email") or "").strip().lower()
            conn = ensure_db()
            row = conn.execute("SELECT id FROM caldav_users WHERE email = ?", (email,)).fetchone()
            token_plain = None
            if row:
                now = utc_now_iso()
                token_plain = secrets.token_urlsafe(24)
                conn.execute(
                    """
                    INSERT INTO user_tokens (user_id, token_type, token_hash, expires_at, used_at, created_at)
                    VALUES (?, 'reset_password', ?, ?, NULL, ?)
                    """,
                    (row["id"], hash_password(token_plain), (datetime.now(timezone.utc) + timedelta(seconds=RESET_TOKEN_TTL_SECONDS)).replace(microsecond=0).isoformat(), now),
                )
                conn.commit()
            conn.close()
            sent = False
            send_info = "no_matching_user"
            if token_plain:
                sent, send_info = send_fixture_email(
                    email,
                    "Calendar password reset",
                    f"Your password reset token is: {token_plain}\nUse this token in the calendar login dialog to set a new password.",
                )
            self._write_json(200, {"data": {"status": "ok", "email_status": send_info, "email_sent": sent}})
            return

        if path == f"{API_BASE}/users/reset-password":
            if _rate_limited("user_reset", self.client_address[0], limit=20, window_seconds=300):
                self._write_json(429, {"error": {"code": "rate_limited", "message": "Too many reset attempts. Try again later."}})
                return
            payload = self._json_body()
            token = payload.get("token") or ""
            new_password = payload.get("new_password") or ""
            if len(new_password) < 8:
                self._write_json(422, {"error": {"code": "validation_error", "message": "weak password"}})
                return
            conn = ensure_db()
            now = utc_now_iso()
            candidates = conn.execute(
                """
                SELECT id, user_id, token_hash, expires_at FROM user_tokens
                WHERE token_type = 'reset_password' AND used_at IS NULL
                ORDER BY id DESC
                """
            ).fetchall()
            row = None
            for cand in candidates:
                if cand["expires_at"] < now:
                    continue
                if verify_password(token, cand["token_hash"]):
                    row = cand
                    break
            if not row:
                conn.close()
                self._write_json(422, {"error": {"code": "validation_error", "message": "invalid token"}})
                return
            conn.execute("UPDATE user_tokens SET used_at = ? WHERE id = ?", (now, row["id"]))
            conn.execute("UPDATE caldav_users SET password_hash = ?, updated_at = ? WHERE id = ?", (hash_password(new_password), now, row["user_id"]))
            conn.commit()
            conn.close()
            self._write_json(200, {"data": {"status": "password_reset"}})
            return

        self._not_found()

    def _handle_api_put(self, path: str) -> None:
        self._handle_api_patch(path)

    def _handle_api_patch(self, path: str) -> None:
        m = re.fullmatch(rf"{re.escape(API_BASE)}/events/(\d+)", path)
        if m:
            auth = self._require_event_write_auth()
            if not auth:
                return
            event_id = int(m.group(1))
            payload = self._json_body()
            normalized = self._validate_event_payload(payload)
            if normalized is None:
                return
            conn = ensure_db()
            row = conn.execute("SELECT * FROM events WHERE id = ? AND calendar_id = ?", (event_id, SHARED_CALENDAR_ID)).fetchone()
            if not row:
                conn.close()
                self._write_json(404, {"error": {"code": "not_found", "message": "event not found"}})
                return
            if_match = self.headers.get("If-Match")
            if if_match and if_match != row["etag"]:
                conn.close()
                self._write_json(412, {"error": {"code": "precondition_failed", "message": "etag mismatch"}})
                return
            now = utc_now_iso()
            sync_version = int(row["sync_version"]) + 1
            etag = mk_etag(f"{row['uid']}:{now}:{sync_version}")
            conn.execute(
                """
                UPDATE events
                SET title = ?, description = ?, location = ?, category = ?, all_day_event = ?,
                    start_datetime = ?, end_datetime = ?, repeat_type = ?, repeat_interval = ?,
                    repeat_nth_mode = ?, repeat_nth_day = ?, repeat_nth_pos = ?, repeat_nth_weekday = ?,
                    repeat_range_mode = ?, repeat_count = ?, repeat_until = ?, timezone = ?,
                    etag = ?, sync_version = ?, updated_at = ?
                WHERE id = ?
                """,
                (
                    normalized["title"],
                    normalized["description"],
                    normalized["location"],
                    normalized["category"],
                    normalized["all_day_event"],
                    normalized["start_datetime"],
                    normalized["end_datetime"],
                    normalized["repeat_type"],
                    normalized["repeat_interval"],
                    normalized["repeat_nth_mode"],
                    normalized["repeat_nth_day"],
                    normalized["repeat_nth_pos"],
                    normalized["repeat_nth_weekday"],
                    normalized["repeat_range_mode"],
                    normalized["repeat_count"],
                    normalized["repeat_until"],
                    normalized["timezone"],
                    etag,
                    sync_version,
                    now,
                    event_id,
                ),
            )
            conn.commit()
            updated = conn.execute("SELECT * FROM events WHERE id = ?", (event_id,)).fetchone()
            log_audit(conn, auth.actor_type, auth.actor_id, "event.update", "event", str(event_id), "success", {})
            conn.close()
            self._write_json(200, {"data": event_to_dict(updated)})
            return

        m = re.fullmatch(rf"{re.escape(API_BASE)}/admin/users/(\d+)", path)
        if m:
            auth = self._require_api_admin()
            if not auth:
                return
            if auth.role != "wp_admin":
                self._write_json(403, {"error": {"code": "authorization_error", "message": "admin role required"}})
                return
            user_id = int(m.group(1))
            payload = self._json_body()
            account_status = payload.get("account_status")
            conn = ensure_db()
            row = conn.execute("SELECT * FROM caldav_users WHERE id = ?", (user_id,)).fetchone()
            if not row:
                conn.close()
                self._write_json(404, {"error": {"code": "not_found", "message": "user not found"}})
                return
            updates = []
            params: List[Any] = []
            if account_status in {"pending_approval", "active"}:
                if account_status == "active" and not row["email_verified_at"]:
                    conn.close()
                    self._write_json(422, {"error": {"code": "validation_error", "message": "email must be verified before approval"}})
                    return
                updates.append("account_status = ?")
                params.append(account_status)
                updates.append("request_state = ?")
                params.append("approved" if account_status == "active" else "requested")
            if not updates:
                conn.close()
                self._write_json(422, {"error": {"code": "validation_error", "message": "valid account_status required"}})
                return
            if account_status == "active":
                updates.append("access_level = ?")
                params.append("write")
            updates.append("updated_at = ?")
            params.append(utc_now_iso())
            params.append(user_id)
            conn.execute(f"UPDATE caldav_users SET {', '.join(updates)} WHERE id = ?", params)
            conn.commit()
            updated = conn.execute(
                "SELECT id, email, email_verified_at, account_status, updated_at FROM caldav_users WHERE id = ?",
                (user_id,),
            ).fetchone()
            log_audit(conn, auth.actor_type, auth.actor_id, "user.update", "caldav_user", str(user_id), "success", payload)
            conn.close()
            self._write_json(200, {"data": dict(updated)})
            return

        if path == f"{API_BASE}/admin/setup":
            auth = self._require_api_admin()
            if not auth:
                return
            if auth.role != "wp_admin":
                self._write_json(403, {"error": {"code": "authorization_error", "message": "admin role required"}})
                return
            payload = self._json_body()
            presentation_name = (payload.get("presentation_name") or "").strip()
            caldav_calendar_name = (payload.get("caldav_calendar_name") or "").strip()
            raw_slug = (payload.get("url_slug") or "").strip().strip("/")
            ics_access_mode = payload.get("ics_access_mode")
            if not presentation_name:
                self._write_json(422, {"error": {"code": "validation_error", "message": "presentation_name required"}})
                return
            if not caldav_calendar_name:
                self._write_json(422, {"error": {"code": "validation_error", "message": "caldav_calendar_name required"}})
                return
            if raw_slug and not re.fullmatch(r"[a-z0-9-]+", raw_slug):
                self._write_json(422, {"error": {"code": "validation_error", "message": "url_slug must match [a-z0-9-]+"}})
                return
            if ics_access_mode not in {"public_read", "authenticated_read"}:
                self._write_json(422, {"error": {"code": "validation_error", "message": "invalid ics_access_mode"}})
                return
            conn = ensure_db()
            set_setting(conn, "presentation_name", presentation_name)
            set_setting(conn, "caldav_calendar_name", caldav_calendar_name)
            set_setting(conn, "url_slug", raw_slug)
            set_setting(conn, "ics_access_mode", ics_access_mode)
            log_audit(conn, auth.actor_type, auth.actor_id, "setup.update", "plugin_settings", "global", "success", payload)
            conn.close()
            self._write_json(200, {"data": {"presentation_name": presentation_name, "caldav_calendar_name": caldav_calendar_name, "url_slug": raw_slug, "ics_access_mode": ics_access_mode}})
            return

        self._not_found()

    def _handle_api_delete(self, path: str) -> None:
        m = re.fullmatch(rf"{re.escape(API_BASE)}/admin/users/(\d+)$", path)
        if m:
            auth = self._require_api_admin()
            if not auth:
                return
            if auth.role != "wp_admin":
                self._write_json(403, {"error": {"code": "authorization_error", "message": "admin role required"}})
                return
            user_id = int(m.group(1))
            conn = ensure_db()
            row = conn.execute("SELECT id FROM caldav_users WHERE id = ?", (user_id,)).fetchone()
            if not row:
                conn.close()
                self._write_json(404, {"error": {"code": "not_found", "message": "user not found"}})
                return
            conn.execute("DELETE FROM user_tokens WHERE user_id = ?", (user_id,))
            conn.execute("DELETE FROM caldav_users WHERE id = ?", (user_id,))
            conn.commit()
            log_audit(conn, auth.actor_type, auth.actor_id, "user.delete", "caldav_user", str(user_id), "success", {})
            conn.close()
            self.send_response(204)
            self.end_headers()
            return

        m = re.fullmatch(rf"{re.escape(API_BASE)}/events/(\d+)$", path)
        if m:
            auth = self._require_event_write_auth()
            if not auth:
                return
            event_id = int(m.group(1))
            conn = ensure_db()
            row = conn.execute("SELECT * FROM events WHERE id = ? AND calendar_id = ?", (event_id, SHARED_CALENDAR_ID)).fetchone()
            if not row:
                conn.close()
                self._write_json(404, {"error": {"code": "not_found", "message": "event not found"}})
                return
            conn.execute("DELETE FROM recurrence_exceptions WHERE event_id = ?", (event_id,))
            conn.execute("DELETE FROM events WHERE id = ?", (event_id,))
            conn.commit()
            log_audit(conn, auth.actor_type, auth.actor_id, "event.delete", "event", str(event_id), "success", {})
            conn.close()
            self.send_response(204)
            self.end_headers()
            return

        m = re.fullmatch(rf"{re.escape(API_BASE)}/events/(\d+)/occurrences/(.+)$", path)
        if m:
            auth = self._require_event_write_auth()
            if not auth:
                return
            event_id = int(m.group(1))
            occurrence_key = unquote(m.group(2))
            conn = ensure_db()
            row = conn.execute("SELECT * FROM events WHERE id = ? AND calendar_id = ?", (event_id, SHARED_CALENDAR_ID)).fetchone()
            if not row:
                conn.close()
                self._write_json(404, {"error": {"code": "not_found", "message": "event not found"}})
                return
            if row["repeat_type"] == "none":
                conn.close()
                self._write_json(422, {"error": {"code": "validation_error", "message": "event is not recurring"}})
                return
            normalized_occurrence_key = _normalize_occurrence_key_for_event(occurrence_key, row["start_datetime"])
            if not normalized_occurrence_key:
                conn.close()
                self._write_json(422, {"error": {"code": "validation_error", "message": "invalid occurrence key"}})
                return
            now = utc_now_iso()
            try:
                conn.execute(
                    """
                    INSERT INTO recurrence_exceptions (event_id, occurrence_key, exception_type, override_payload, created_at, updated_at)
                    VALUES (?, ?, 'deleted_occurrence', NULL, ?, ?)
                    """,
                    (event_id, normalized_occurrence_key, now, now),
                )
                sync_version = int(row["sync_version"]) + 1
                etag = mk_etag(f"{row['uid']}:{now}:{sync_version}")
                conn.execute("UPDATE events SET etag = ?, sync_version = ?, updated_at = ? WHERE id = ?", (etag, sync_version, now, event_id))
                conn.commit()
                log_audit(
                    conn,
                    auth.actor_type,
                    auth.actor_id,
                    "event.occurrence.delete",
                    "event",
                    str(event_id),
                    "success",
                    {"occurrence_key": normalized_occurrence_key},
                )
            except sqlite3.IntegrityError:
                conn.close()
                # Idempotent behavior: deleting an already-excepted occurrence is still success.
                self.send_response(204)
                self.end_headers()
                return
            conn.close()
            self.send_response(204)
            self.end_headers()
            return

        self._not_found()

    def _expand_event(
        self,
        evt: Dict[str, Any],
        window_start: datetime,
        window_end: datetime,
        deleted_occurrence_keys: set,
    ) -> List[Dict[str, Any]]:
        deleted_canonical = set()
        for key in deleted_occurrence_keys:
            deleted_canonical.add(str(key))
            canonical = _canonical_occurrence_key(str(key))
            if canonical:
                deleted_canonical.add(canonical)

        def is_deleted(occ_start: datetime) -> bool:
            raw = occ_start.isoformat()
            if raw in deleted_canonical:
                return True
            canonical = _canonical_occurrence_key(raw)
            return canonical in deleted_canonical if canonical else False

        start = _parse_dt_maybe_date(evt["start_datetime"])
        end = _parse_dt_maybe_date(evt["end_datetime"])
        duration = end - start
        repeat_type = evt.get("repeat_type", "none")
        interval = max(1, int(evt.get("repeat_interval") or 1))
        repeat_mode = evt.get("repeat_range_mode") or "none"
        max_count: Optional[int] = None
        until_dt: Optional[datetime] = None
        if repeat_mode == "count" and evt.get("repeat_count"):
            max_count = int(evt["repeat_count"])
        if repeat_mode == "until" and evt.get("repeat_until"):
            try:
                until_dt = _parse_dt_maybe_date(f"{evt['repeat_until']}T23:59:59+00:00")
            except ValueError:
                until_dt = None

        def make_occurrence(occ_start: datetime) -> Dict[str, Any]:
            occ_end = occ_start + duration
            return {
                "event_id": evt["id"],
                "uid": evt["uid"],
                "title": evt["title"],
                "description": evt["description"],
                "location": evt["location"],
                "category": evt["category"],
                "all_day_event": bool(evt.get("all_day_event")),
                "occurrence_start": occ_start.isoformat(),
                "occurrence_end": occ_end.isoformat(),
                "repeat_type": repeat_type,
            }

        occurrences: List[Dict[str, Any]] = []
        if repeat_type == "none":
            if _occurrence_overlap(start, end, window_start, window_end):
                if not is_deleted(start):
                    occurrences.append(make_occurrence(start))
            return occurrences

        current = start
        produced = 0
        for _ in range(0, 512):
            if max_count is not None and produced >= max_count:
                break
            if until_dt is not None and current > until_dt:
                break
            occ_end = current + duration
            if _occurrence_overlap(current, occ_end, window_start, window_end):
                if not is_deleted(current):
                    occurrences.append(make_occurrence(current))
            if current > window_end + timedelta(days=400):
                break
            produced += 1
            if repeat_type == "daily":
                current = current + timedelta(days=interval)
            elif repeat_type in {"weekly", "custom"}:
                current = current + timedelta(weeks=interval)
            elif repeat_type == "monthly":
                nth_mode = evt.get("repeat_nth_mode") or ""
                if nth_mode == "day_of_month" and evt.get("repeat_nth_day"):
                    next_month = _add_months(current, interval)
                    day = int(evt["repeat_nth_day"])
                    day = max(1, min(day, pycalendar.monthrange(next_month.year, next_month.month)[1]))
                    current = next_month.replace(day=day)
                elif nth_mode == "weekday_of_month" and evt.get("repeat_nth_pos") and evt.get("repeat_nth_weekday") is not None:
                    next_month = _add_months(current, interval)
                    pos = int(evt["repeat_nth_pos"])
                    wd = int(evt["repeat_nth_weekday"])
                    nth_day = _nth_weekday_of_month(next_month.year, next_month.month, wd, pos)
                    if nth_day is None:
                        current = next_month
                    else:
                        current = next_month.replace(day=nth_day)
                else:
                    current = _add_months(current, interval)
            elif repeat_type == "yearly":
                current = _add_years(current, interval)
            else:
                break
        return occurrences

    def _handle_ics_get(self) -> None:
        conn = ensure_db()
        ics_mode = get_setting(conn, "ics_access_mode", "public_read")
        if ics_mode == "authenticated_read":
            auth = self._authenticate_caldav_user() or self._authenticate_wp_admin()
            if not auth:
                conn.close()
                self.send_response(401)
                self.send_header("WWW-Authenticate", 'Basic realm="calendar-ics-fixture"')
                self.end_headers()
                return
        rows = conn.execute(
            "SELECT * FROM events WHERE calendar_id = ? ORDER BY id ASC", (SHARED_CALENDAR_ID,)
        ).fetchall()
        payloads = []
        for row in rows:
            evt = event_to_dict(row)
            ex_rows = conn.execute(
                "SELECT occurrence_key FROM recurrence_exceptions WHERE event_id = ? AND exception_type = 'deleted_occurrence' ORDER BY occurrence_key",
                (row["id"],),
            ).fetchall()
            exdates = [r["occurrence_key"] for r in ex_rows]
            payloads.append(create_ics_event(evt, exdates))
        conn.close()
        body = (
            "BEGIN:VCALENDAR\r\n"
            "VERSION:2.0\r\n"
            "PRODID:-//Calendar WP Plugin Fixture//EN\r\n"
            "CALSCALE:GREGORIAN\r\n"
            f"X-WR-TIMEZONE:{DEFAULT_TIMEZONE}\r\n"
            + ("\r\n".join(payloads) + "\r\n" if payloads else "")
            + "END:VCALENDAR\r\n"
        )
        encoded = body.encode("utf-8")
        etag = mk_etag(body)
        self.send_response(200)
        self.send_header("Content-Type", "text/calendar; charset=utf-8")
        self.send_header("ETag", etag)
        self.send_header("Content-Length", str(len(encoded)))
        self.end_headers()
        self.wfile.write(encoded)

    def _event_as_ics(self, evt: Dict[str, Any]) -> str:
        conn = ensure_db()
        ex_rows = conn.execute(
            "SELECT occurrence_key FROM recurrence_exceptions WHERE event_id = ? AND exception_type = 'deleted_occurrence' ORDER BY occurrence_key",
            (evt["id"],),
        ).fetchall()
        conn.close()
        exdates = [r["occurrence_key"] for r in ex_rows]
        event_ics = create_ics_event(evt, exdates)
        return "BEGIN:VCALENDAR\r\nVERSION:2.0\r\nPRODID:-//Calendar WP Plugin Fixture//EN\r\n" + event_ics + "\r\nEND:VCALENDAR\r\n"

    def _require_caldav_auth(self, write: bool = False) -> Optional[AuthContext]:
        auth = self._authenticate_caldav_user()
        if not auth:
            self.send_response(401)
            self.send_header("WWW-Authenticate", 'Basic realm="calendar-caldav-fixture"')
            self.end_headers()
            return None
        if write and auth.role != "caldav_write":
            self.send_response(403)
            self.end_headers()
            return None
        return auth

    def _handle_caldav_get(self, path: str) -> None:
        auth = self._require_caldav_auth(write=False)
        if not auth:
            return

        if path in {"/caldav/", "/caldav/calendars/", "/caldav/calendars/public/"}:
            self._write_text(200, "caldav shared public calendar\n")
            return

        filename = _caldav_filename_from_path(path)
        if not filename:
            self._not_found()
            return
        conn = ensure_db()
        row = _caldav_lookup_event(conn, filename)
        conn.close()
        if not row:
            self._not_found()
            return
        evt = event_to_dict(row)
        body = self._event_as_ics(evt).encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "text/calendar; charset=utf-8")
        self.send_header("ETag", row["etag"])
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _handle_caldav_put(self, path: str) -> None:
        auth = self._require_caldav_auth(write=True)
        if not auth:
            return
        filename = _caldav_filename_from_path(path)
        if not filename:
            self._not_found()
            return
        raw = self._read_request_body_bytes().decode("utf-8", errors="replace")
        parsed = self._parse_ics_payload(raw)
        if parsed is None:
            self.send_response(415)
            self.end_headers()
            return
        conn = ensure_db()
        row = _caldav_lookup_event(conn, filename)
        now = utc_now_iso()
        status = 200
        existing_by_uid = None
        if row is None and parsed.get("uid"):
            existing_by_uid = conn.execute(
                "SELECT * FROM events WHERE calendar_id = ? AND uid = ?",
                (SHARED_CALENDAR_ID, parsed["uid"]),
            ).fetchone()
        if row is None:
            if existing_by_uid is not None:
                row = existing_by_uid
            else:
                uid = parsed["uid"] or make_uid(f"caldav:{filename}:{now}")
                etag = mk_etag(f"{uid}:{now}:1")
                conn.execute(
                    """
                    INSERT INTO events
                      (uid, title, description, location, category, all_day_event, start_datetime, end_datetime,
                       repeat_type, repeat_interval, repeat_nth_mode, repeat_nth_day, repeat_nth_pos, repeat_nth_weekday,
                       repeat_range_mode, repeat_count, repeat_until, timezone,
                       calendar_id, caldav_resource, etag, sync_version, last_modified_by_user_id, created_at, updated_at)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    (
                        uid,
                        parsed["summary"],
                        parsed["description"],
                        parsed["location"],
                        "",
                        0,
                        parsed["start_datetime"],
                        parsed["end_datetime"],
                        parsed["repeat_type"],
                        parsed["repeat_interval"],
                        parsed["repeat_nth_mode"],
                        parsed["repeat_nth_day"],
                        parsed["repeat_nth_pos"],
                        parsed["repeat_nth_weekday"],
                        parsed["repeat_range_mode"],
                        parsed["repeat_count"],
                        parsed["repeat_until"],
                        DEFAULT_TIMEZONE,
                        SHARED_CALENDAR_ID,
                        filename,
                        etag,
                        1,
                        int(auth.actor_id),
                        now,
                        now,
                    ),
                )
                status = 201
                conn.commit()
                row = conn.execute("SELECT * FROM events WHERE id = last_insert_rowid()").fetchone()
                if row is not None and parsed.get("exdates_specified"):
                    _sync_deleted_occurrences(
                        conn,
                        int(row["id"]),
                        list(parsed.get("exdates") or []),
                        True,
                        now,
                    )
                    conn.commit()
        if row is not None and status != 201:
            if_match = self.headers.get("If-Match")
            if if_match and if_match != row["etag"]:
                conn.close()
                self.send_response(412)
                self.end_headers()
                return
            sync_version = int(row["sync_version"]) + 1
            etag = mk_etag(f"{row['uid']}:{now}:{sync_version}")
            conn.execute(
                """
                UPDATE events SET title = ?, description = ?, location = ?, start_datetime = ?, end_datetime = ?,
                  repeat_type = ?, repeat_interval = ?, repeat_nth_mode = ?, repeat_nth_day = ?, repeat_nth_pos = ?, repeat_nth_weekday = ?,
                  repeat_range_mode = ?, repeat_count = ?, repeat_until = ?,
                  caldav_resource = ?, etag = ?, sync_version = ?, last_modified_by_user_id = ?, updated_at = ?
                WHERE id = ?
                """,
                (
                    parsed["summary"],
                    parsed["description"],
                    parsed["location"],
                    parsed["start_datetime"],
                    parsed["end_datetime"],
                    parsed["repeat_type"],
                    parsed["repeat_interval"],
                    parsed["repeat_nth_mode"],
                    parsed["repeat_nth_day"],
                    parsed["repeat_nth_pos"],
                    parsed["repeat_nth_weekday"],
                    parsed["repeat_range_mode"],
                    parsed["repeat_count"],
                    parsed["repeat_until"],
                    filename,
                    etag,
                    sync_version,
                    int(auth.actor_id),
                    now,
                    int(row["id"]),
                ),
            )
            if parsed.get("exdates_specified"):
                _sync_deleted_occurrences(
                    conn,
                    int(row["id"]),
                    list(parsed.get("exdates") or []),
                    True,
                    now,
                )
            conn.commit()
        updated = conn.execute("SELECT id, etag FROM events WHERE id = ?", (int(row["id"]),)).fetchone()
        log_audit(conn, auth.actor_type, auth.actor_id, "caldav.put", "event", str(updated["id"]), "success", {"resource": filename})
        conn.close()
        self.send_response(status)
        self.send_header("ETag", updated["etag"])
        self.end_headers()
        return

    def _handle_caldav_delete(self, path: str) -> None:
        auth = self._require_caldav_auth(write=True)
        if not auth:
            return
        filename = _caldav_filename_from_path(path)
        if not filename:
            self._not_found()
            return
        conn = ensure_db()
        row = _caldav_lookup_event(conn, filename)
        if not row:
            conn.close()
            self._not_found()
            return
        event_id = int(row["id"])
        conn.execute("DELETE FROM recurrence_exceptions WHERE event_id = ?", (event_id,))
        conn.execute("DELETE FROM events WHERE id = ?", (event_id,))
        conn.commit()
        log_audit(conn, auth.actor_type, auth.actor_id, "caldav.delete", "event", str(event_id), "success", {"resource": filename})
        conn.close()
        self.send_response(204)
        self.end_headers()

    def _parse_ics_payload(self, raw: str) -> Optional[Dict[str, Any]]:
        vevents = _extract_vevents(raw)
        vevent = _select_master_vevent(vevents)
        if not vevent:
            return None
        summary = _extract_line(vevent, "SUMMARY:") or "Untitled"
        description = _extract_line(vevent, "DESCRIPTION:") or ""
        location = _extract_line(vevent, "LOCATION:") or ""
        uid = _extract_line(vevent, "UID:")
        dtstart = _extract_dt(vevent, "DTSTART")
        dtend = _extract_dt(vevent, "DTEND")
        if not dtstart or not dtend:
            return None
        rrule = _extract_line(vevent, "RRULE:")
        exdates = _extract_exdates(vevent)
        cancelled_occurrence_exdates = _extract_cancelled_recurrence_ids(vevents)
        if cancelled_occurrence_exdates:
            dedup = set(exdates)
            for ex in cancelled_occurrence_exdates:
                if ex not in dedup:
                    exdates.append(ex)
                    dedup.add(ex)
        repeat_type = "none"
        repeat_interval = 1
        repeat_nth_mode = ""
        repeat_nth_day = None
        repeat_nth_pos = None
        repeat_nth_weekday = None
        repeat_range_mode = "none"
        repeat_count = None
        repeat_until = None
        if rrule:
            parts = {p.split("=", 1)[0]: p.split("=", 1)[1] for p in rrule.split(";") if "=" in p}
            freq = parts.get("FREQ", "")
            freq_map = {"DAILY": "daily", "WEEKLY": "weekly", "MONTHLY": "monthly", "YEARLY": "yearly"}
            repeat_type = freq_map.get(freq, "custom")
            if "INTERVAL" in parts:
                try:
                    repeat_interval = max(1, int(parts["INTERVAL"]))
                except ValueError:
                    repeat_interval = 1
            if repeat_type == "monthly":
                bymonthday = parts.get("BYMONTHDAY")
                byday = parts.get("BYDAY", "")
                bysetpos = parts.get("BYSETPOS")
                if bymonthday:
                    raw_day = bymonthday.split(",")[0].strip()
                    try:
                        parsed_day = int(raw_day)
                    except ValueError:
                        parsed_day = None
                    if parsed_day is not None and 1 <= parsed_day <= 31:
                        repeat_nth_mode = "day_of_month"
                        repeat_nth_day = parsed_day
                elif byday:
                    day_map = {"SU": 0, "MO": 1, "TU": 2, "WE": 3, "TH": 4, "FR": 5, "SA": 6}
                    token = byday.split(",")[0].strip().upper()
                    m = re.fullmatch(r"([+-]?\d+)?(SU|MO|TU|WE|TH|FR|SA)", token)
                    if m:
                        wd = day_map[m.group(2)]
                        pos_raw = m.group(1)
                        if not pos_raw and bysetpos:
                            pos_raw = bysetpos.split(",")[0].strip()
                        try:
                            pos = int(pos_raw) if pos_raw else None
                        except (TypeError, ValueError):
                            pos = None
                        if pos is not None and (1 <= pos <= 5 or pos == -1):
                            repeat_nth_mode = "weekday_of_month"
                            repeat_nth_pos = pos
                            repeat_nth_weekday = wd
            if "COUNT" in parts:
                repeat_range_mode = "count"
                repeat_count = int(parts["COUNT"])
            elif "UNTIL" in parts:
                repeat_range_mode = "until"
                until_raw = parts["UNTIL"].strip()
                if len(until_raw) >= 8:
                    repeat_until = f"{until_raw[0:4]}-{until_raw[4:6]}-{until_raw[6:8]}"
            else:
                repeat_range_mode = "no_end"
        dtstart, dtend = _normalize_monthly_anchor(
            dtstart,
            dtend,
            repeat_type,
            repeat_nth_mode,
            repeat_nth_day,
            repeat_nth_pos,
            repeat_nth_weekday,
        )
        return {
            "uid": uid,
            "summary": summary,
            "description": description,
            "location": location,
            "start_datetime": dtstart,
            "end_datetime": dtend,
            "repeat_type": repeat_type,
            "repeat_interval": repeat_interval,
            "repeat_nth_mode": repeat_nth_mode,
            "repeat_nth_day": repeat_nth_day,
            "repeat_nth_pos": repeat_nth_pos,
            "repeat_nth_weekday": repeat_nth_weekday,
            "repeat_range_mode": repeat_range_mode,
            "repeat_count": repeat_count,
            "repeat_until": repeat_until,
            "exdates": exdates,
            "exdates_specified": bool(_extract_prop_values(vevent, "EXDATE") or cancelled_occurrence_exdates),
        }

    def log_message(self, fmt: str, *args: Any) -> None:
        # Keep output concise for fixture runs.
        print(f"[fixture] {self.address_string()} - {fmt % args}")


def _extract_line(raw: str, prefix: str) -> Optional[str]:
    for line in raw.splitlines():
        if line.startswith(prefix):
            return line[len(prefix) :].strip()
    return None


def _extract_prop_values(raw: str, name: str) -> List[str]:
    out: List[str] = []
    pattern = re.compile(rf"^{re.escape(name)}(?:;[^:]+)?:([^\r\n]+)$", re.IGNORECASE)
    for line in raw.splitlines():
        m = pattern.match(line.strip())
        if m:
            out.append(m.group(1).strip())
    return out


def _extract_first_vevent(raw: str) -> Optional[str]:
    vevents = _extract_vevents(raw)
    return vevents[0] if vevents else None


def _extract_vevents(raw: str) -> List[str]:
    out: List[str] = []
    for m in re.finditer(r"BEGIN:VEVENT\r?\n(.*?)\r?\nEND:VEVENT", raw, re.DOTALL):
        out.append("BEGIN:VEVENT\n" + m.group(1) + "\nEND:VEVENT")
    return out


def _is_cancelled_vevent(vevent: str) -> bool:
    status = (_extract_line(vevent, "STATUS:") or "").strip().upper()
    return status == "CANCELLED"


def _select_master_vevent(vevents: List[str]) -> str:
    if not vevents:
        return ""
    # Prefer the true series master, not detached/cancelled overrides.
    for v in vevents:
        if not _extract_prop_values(v, "RECURRENCE-ID") and not _is_cancelled_vevent(v):
            return v
    for v in vevents:
        if not _extract_prop_values(v, "RECURRENCE-ID"):
            return v
    for v in vevents:
        if _extract_line(v, "RRULE:") and not _is_cancelled_vevent(v):
            return v
    for v in vevents:
        if not _is_cancelled_vevent(v):
            return v
    return vevents[0]


def _parse_ics_token_to_iso(value: str) -> Optional[str]:
    token = (value or "").strip()
    if not token:
        return None
    # DATE value
    if "T" not in token:
        if re.fullmatch(r"\d{8}", token):
            return f"{token[0:4]}-{token[4:6]}-{token[6:8]}T00:00:00+00:00"
        return None
    # DATE-TIME value
    if token.endswith("Z"):
        core = token[:-1]
        try:
            dt = datetime.strptime(core, "%Y%m%dT%H%M%S").replace(tzinfo=timezone.utc)
            return dt.isoformat()
        except ValueError:
            return None
    try:
        dt = datetime.strptime(token[:15], "%Y%m%dT%H%M%S")
        return dt.strftime("%Y-%m-%dT%H:%M:%S+01:00")
    except ValueError:
        return None


def _extract_exdates(raw: str) -> List[str]:
    values = _extract_prop_values(raw, "EXDATE")
    out: List[str] = []
    for value in values:
        for token in value.split(","):
            iso = _parse_ics_token_to_iso(token)
            if iso:
                out.append(iso)
    return out


def _extract_cancelled_recurrence_ids(vevents: List[str]) -> List[str]:
    out: List[str] = []
    for vevent in vevents:
        rec_values = _extract_prop_values(vevent, "RECURRENCE-ID")
        if not rec_values:
            continue
        if not _is_cancelled_vevent(vevent):
            continue
        for value in rec_values:
            for token in value.split(","):
                iso = _parse_ics_token_to_iso(token)
                if iso:
                    out.append(iso)
    return out


def _extract_dt(raw: str, field: str) -> Optional[str]:
    pattern = re.compile(rf"^{field}(?:;[^:]+)?:([0-9TzZ]+)$")
    for line in raw.splitlines():
        m = pattern.match(line.strip())
        if m:
            value = m.group(1)
            if "T" in value:
                # Interpret as local naive fixture time and inject +01:00 offset for test determinism.
                try:
                    dt = datetime.strptime(value[:15], "%Y%m%dT%H%M%S")
                    return dt.strftime("%Y-%m-%dT%H:%M:%S+01:00")
                except ValueError:
                    return None
    return None


def _sync_deleted_occurrences(
    conn: sqlite3.Connection,
    event_id: int,
    exdates: List[str],
    replace: bool,
    now: str,
) -> None:
    if replace:
        conn.execute(
            "DELETE FROM recurrence_exceptions WHERE event_id = ? AND exception_type = 'deleted_occurrence'",
            (event_id,),
        )
    if not exdates:
        return
    for ex in exdates:
        canonical = _canonical_occurrence_key(ex)
        if not canonical:
            continue
        conn.execute(
            """
            INSERT OR IGNORE INTO recurrence_exceptions
              (event_id, occurrence_key, exception_type, override_payload, created_at, updated_at)
            VALUES (?, ?, 'deleted_occurrence', NULL, ?, ?)
            """,
            (event_id, canonical, now, now),
        )


def _xml_escape(value: str) -> str:
    return (
        value.replace("&", "&")
        .replace("<", "<")
        .replace(">", ">")
        .replace('"', """)
        .replace("'", "'")
    )


def _html_escape(value: str) -> str:
    return _xml_escape(value)


def _caldav_filename_from_path(path: str) -> Optional[str]:
    m = re.fullmatch(r"/caldav/calendars/public/([^/]+\.ics)", path)
    if not m:
        return None
    return unquote(m.group(1))


def _caldav_lookup_event(conn: sqlite3.Connection, filename: str) -> Optional[sqlite3.Row]:
    return conn.execute(
        "SELECT * FROM events WHERE calendar_id = ? AND caldav_resource = ?",
        (SHARED_CALENDAR_ID, filename),
    ).fetchone()


def cmd_init(_: argparse.Namespace) -> int:
    conn = ensure_db()
    create_schema(conn)
    conn.close()
    print(f"initialized fixture db at {DB_PATH}")
    return 0


def cmd_reset(_: argparse.Namespace) -> int:
    conn = ensure_db()
    drop_all(conn)
    create_schema(conn)
    conn.close()
    print(f"reset fixture db at {DB_PATH}")
    return 0


def cmd_seed(_: argparse.Namespace) -> int:
    conn = ensure_db()
    create_schema(conn)
    seed_data(conn)
    conn.close()
    print("seeded fixture data")
    return 0


def cmd_run(args: argparse.Namespace) -> int:
    conn = ensure_db()
    create_schema(conn)
    conn.close()
    server = ThreadingHTTPServer((args.host, args.port), FixtureHandler)
    trace_log = fixture_trace_log_path()
    trace_log.parent.mkdir(parents=True, exist_ok=True)
    print(f"fixture server listening at http://{args.host}:{args.port}")
    print(f"http trace log: {trace_log}")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()
    return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Calendar plugin local fixture server")
    sub = parser.add_subparsers(dest="command", required=True)

    p_init = sub.add_parser("init", help="initialize fixture db schema")
    p_init.set_defaults(func=cmd_init)

    p_reset = sub.add_parser("reset", help="reset fixture db schema")
    p_reset.set_defaults(func=cmd_reset)

    p_seed = sub.add_parser("seed", help="seed fixture db data")
    p_seed.set_defaults(func=cmd_seed)

    p_run = sub.add_parser("run", help="run fixture server")
    p_run.add_argument("--host", default="127.0.0.1")
    p_run.add_argument("--port", type=int, default=8080)
    p_run.set_defaults(func=cmd_run)

    return parser


def main(argv: Optional[List[str]] = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    return args.func(args)


if __name__ == "__main__":
    raise SystemExit(main())