237 lines
7.8 KiB
Python
237 lines
7.8 KiB
Python
import socket
|
|
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from backend.calendar import (
|
|
FetchResult,
|
|
parse_ics_events,
|
|
validate_calendar_url,
|
|
)
|
|
|
|
ICS = b"""BEGIN:VCALENDAR\r
|
|
VERSION:2.0\r
|
|
BEGIN:VEVENT\r
|
|
UID:one\r
|
|
DTSTART:20260920T090000Z\r
|
|
DTEND:20260920T100000Z\r
|
|
SUMMARY:Meeting\r
|
|
DESCRIPTION:Body line one\\nBody line two\r
|
|
LOCATION:Meeting room\r
|
|
END:VEVENT\r
|
|
END:VCALENDAR\r
|
|
"""
|
|
RECURRING_ICS = b"""BEGIN:VCALENDAR\r
|
|
VERSION:2.0\r
|
|
BEGIN:VEVENT\r
|
|
UID:daily\r
|
|
DTSTART;TZID=Asia/Shanghai:20260920T090000\r
|
|
DTEND;TZID=Asia/Shanghai:20260920T100000\r
|
|
RRULE:FREQ=DAILY;COUNT=3\r
|
|
EXDATE;TZID=Asia/Shanghai:20260921T090000\r
|
|
SUMMARY:Daily\r
|
|
END:VEVENT\r
|
|
BEGIN:VEVENT\r
|
|
UID:daily\r
|
|
RECURRENCE-ID;TZID=Asia/Shanghai:20260922T090000\r
|
|
DTSTART;TZID=Asia/Shanghai:20260922T110000\r
|
|
DTEND;TZID=Asia/Shanghai:20260922T120000\r
|
|
SUMMARY:Moved\r
|
|
END:VEVENT\r
|
|
END:VCALENDAR\r
|
|
"""
|
|
INHERITED_DURATION_ICS = RECURRING_ICS.replace(
|
|
b"DTEND;TZID=Asia/Shanghai:20260922T120000\r\n", b""
|
|
).replace(
|
|
b"DTEND;TZID=Asia/Shanghai:20260920T100000\r\n",
|
|
b"DTEND;TZID=Asia/Shanghai:20260920T103000\r\n",
|
|
)
|
|
|
|
|
|
def initialized(client, username="owner"):
|
|
response = client.post(
|
|
"/api/v1/setup/initialize",
|
|
json={"username": username, "password": "correct horse battery staple"},
|
|
)
|
|
assert response.status_code == 201
|
|
return client
|
|
|
|
|
|
def test_parser_restored_with_recurrence_exdates_overrides_and_timezone():
|
|
events = parse_ics_events(
|
|
RECURRING_ICS,
|
|
"Work",
|
|
"#123456",
|
|
datetime(2026, 9, 19, tzinfo=UTC),
|
|
datetime(2026, 9, 24, tzinfo=UTC),
|
|
"Asia/Shanghai",
|
|
)
|
|
assert [(event["title"], event["starts_at"].isoformat()) for event in events] == [
|
|
("Daily", "2026-09-20T01:00:00+00:00"),
|
|
("Moved", "2026-09-22T03:00:00+00:00"),
|
|
]
|
|
|
|
|
|
def test_recurrence_override_inherits_master_duration_and_source_id():
|
|
events = parse_ics_events(
|
|
INHERITED_DURATION_ICS,
|
|
"Work",
|
|
"#123456",
|
|
datetime(2026, 9, 19, tzinfo=UTC),
|
|
datetime(2026, 9, 24, tzinfo=UTC),
|
|
"Asia/Shanghai",
|
|
source_id="source-1",
|
|
)
|
|
moved = next(event for event in events if event["title"] == "Moved")
|
|
assert (moved["ends_at"] - moved["starts_at"]).total_seconds() == 90 * 60
|
|
assert moved["source_id"] == "source-1"
|
|
|
|
|
|
def test_parser_limits_recurrence_expansion():
|
|
endless = ICS.replace(b"UID:one", b"UID:one\r\nRRULE:FREQ=SECONDLY")
|
|
with pytest.raises(ValueError, match="recurrence limit"):
|
|
parse_ics_events(
|
|
endless,
|
|
"x",
|
|
"#000000",
|
|
datetime(2026, 9, 20, tzinfo=UTC),
|
|
datetime(2026, 9, 21, tzinfo=UTC),
|
|
"UTC",
|
|
recurrence_limit=10,
|
|
)
|
|
|
|
|
|
def test_url_validation_rejects_fragments_userinfo_and_mixed_dns(monkeypatch):
|
|
with pytest.raises(HTTPException):
|
|
validate_calendar_url("https://example.com/a.ics#secret")
|
|
with pytest.raises(HTTPException):
|
|
validate_calendar_url("https://[email protected]/a.ics")
|
|
monkeypatch.setattr(socket, "getaddrinfo", lambda *args, **kwargs: [
|
|
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443)),
|
|
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 443)),
|
|
])
|
|
with pytest.raises(HTTPException, match="public"):
|
|
validate_calendar_url("https://example.com/a.ics")
|
|
|
|
|
|
def test_subscription_crud_refresh_events_and_stale_cache(client, monkeypatch):
|
|
client = initialized(client)
|
|
monkeypatch.setattr(
|
|
"backend.calendar.validate_calendar_url",
|
|
lambda url: (url, "93.184.216.34", 443),
|
|
)
|
|
calls = []
|
|
|
|
def fetch(url, *, etag=None, last_modified=None):
|
|
calls.append((url, etag, last_modified))
|
|
if len(calls) == 1:
|
|
return FetchResult(ICS, '"v1"', "Sun, 20 Sep 2026 00:00:00 GMT", False)
|
|
raise HTTPException(502, "upstream down")
|
|
|
|
monkeypatch.setattr("backend.calendar.fetch_calendar", fetch)
|
|
created = client.post(
|
|
"/api/v1/calendar-subscriptions",
|
|
json={"name": " Work ", "url": "https://example.com/work.ics", "color": "#123abc"},
|
|
)
|
|
assert created.status_code == 201
|
|
body = created.json()
|
|
assert body["name"] == "Work"
|
|
assert body["enabled"] is True
|
|
assert body["stale"] is False
|
|
assert calls == [("https://example.com/work.ics", None, None)]
|
|
|
|
listed = client.get("/api/v1/calendar-subscriptions")
|
|
assert [item["id"] for item in listed.json()] == [body["id"]]
|
|
|
|
events = client.get(
|
|
"/api/v1/calendar-events",
|
|
params={"start": "2026-09-20T00:00:00Z", "end": "2026-09-21T00:00:00Z"},
|
|
)
|
|
assert events.status_code == 200
|
|
assert events.json()["events"][0]["title"] == "Meeting"
|
|
assert events.json()["events"][0]["description"] == "Body line one\nBody line two"
|
|
assert events.json()["events"][0]["location"] == "Meeting room"
|
|
assert events.json()["sources"][0]["stale"] is False
|
|
|
|
refreshed = client.post(f"/api/v1/calendar-subscriptions/{body['id']}/refresh")
|
|
assert refreshed.status_code == 200
|
|
assert refreshed.json()["stale"] is True
|
|
assert refreshed.json()["last_error"] == "upstream down"
|
|
assert calls[1][1:] == ('"v1"', "Sun, 20 Sep 2026 00:00:00 GMT")
|
|
|
|
patched = client.patch(
|
|
f"/api/v1/calendar-subscriptions/{body['id']}",
|
|
json={"name": "Personal", "enabled": False, "color": "#abcdef"},
|
|
)
|
|
assert patched.status_code == 200
|
|
assert patched.json()["name"] == "Personal"
|
|
assert patched.json()["enabled"] is False
|
|
assert client.delete(f"/api/v1/calendar-subscriptions/{body['id']}").status_code == 204
|
|
|
|
|
|
def test_parser_bounds_large_event_text_fields():
|
|
oversized = ICS.replace(
|
|
b"DESCRIPTION:Body line one\\nBody line two",
|
|
b"DESCRIPTION:" + b"x" * 3_000,
|
|
).replace(
|
|
b"LOCATION:Meeting room",
|
|
b"LOCATION:" + b"y" * 1_000,
|
|
)
|
|
event = parse_ics_events(
|
|
oversized,
|
|
"Work",
|
|
"#123456",
|
|
datetime(2026, 9, 20, tzinfo=UTC),
|
|
datetime(2026, 9, 21, tzinfo=UTC),
|
|
"UTC",
|
|
)[0]
|
|
assert event["description"] == "x" * 2_000
|
|
assert event["location"] == "y" * 500
|
|
|
|
|
|
def test_events_validate_window_and_disabled_sources_are_skipped(client, monkeypatch):
|
|
client = initialized(client)
|
|
monkeypatch.setattr(
|
|
"backend.calendar.validate_calendar_url",
|
|
lambda url: (url, "93.184.216.34", 443),
|
|
)
|
|
monkeypatch.setattr(
|
|
"backend.calendar.fetch_calendar",
|
|
lambda *args, **kwargs: FetchResult(ICS, None, None, False),
|
|
)
|
|
sub = client.post(
|
|
"/api/v1/calendar-subscriptions",
|
|
json={"name": "x", "url": "https://example.com/x.ics", "enabled": False},
|
|
).json()
|
|
assert sub["enabled"] is False
|
|
response = client.get(
|
|
"/api/v1/calendar-events",
|
|
params={"start": "2026-09-21T00:00:00Z", "end": "2026-09-20T00:00:00Z"},
|
|
)
|
|
assert response.status_code == 422
|
|
valid = client.get(
|
|
"/api/v1/calendar-events",
|
|
params={"start": "2026-09-20T00:00:00Z", "end": "2026-09-21T00:00:00Z"},
|
|
)
|
|
assert valid.json() == {"events": [], "sources": []}
|
|
|
|
|
|
def test_subscription_ownership_is_strict(client, monkeypatch):
|
|
client = initialized(client)
|
|
monkeypatch.setattr(
|
|
"backend.calendar.validate_calendar_url",
|
|
lambda url: (url, "93.184.216.34", 443),
|
|
)
|
|
monkeypatch.setattr(
|
|
"backend.calendar.fetch_calendar",
|
|
lambda *args, **kwargs: FetchResult(ICS, None, None, False),
|
|
)
|
|
sub = client.post(
|
|
"/api/v1/calendar-subscriptions",
|
|
json={"name": "private", "url": "https://example.com/private.ics"},
|
|
).json()
|
|
client.post("/api/v1/auth/logout")
|
|
assert client.patch(f"/api/v1/calendar-subscriptions/{sub['id']}", json={"name": "x"}).status_code == 401
|
|
assert client.delete(f"/api/v1/calendar-subscriptions/{sub['id']}").status_code == 401
|