tfaour
31adc34a19
64 tests covering: auth/setup and the CSRF gate, the "owner adds their own data, anyone linked can mute it" permission pattern shared across calendar-select/tasks-source/whiteboard-source, migration correctness (fresh install, idempotent re-run, expected columns), battery estimate outlier rejection, calendar_feed's fetch/merge/partial-failure handling, webdav_client's fetch/list-directory, the whiteboard force-refresh throttle bypass and browse endpoint, and render-size invariants across calendar views/orientations. No DB/HTTP fixtures need Docker, Node, or a real Immich/CalDAV/WebDAV server -- a fresh temp SQLite file plus a couple of small local HTTP servers as test doubles cover it all. Table data is wiped and reseeded between tests rather than relying on SQLAlchemy's transaction-rollback isolation pattern, which needs a pysqlite event-listener workaround app/db.py's engine doesn't have and has no reason to gain just for tests. Wired into .gitea/workflows/server-docker-build.yml as its own job that build-and-push now depends on, so a failing suite blocks the image push rather than just running alongside it for show.
158 lines
4.6 KiB
Python
158 lines
4.6 KiB
Python
"""calendar_feed.py's fetch/parse/merge against a local HTTP server
|
|
serving fixture .ics text -- pure functions, no ORM/FastAPI."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
from datetime import date
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
|
|
import pytest
|
|
|
|
from app.calendar_feed import CalendarSource, merge_events
|
|
|
|
_PLAIN_ICS = b"""BEGIN:VCALENDAR
|
|
VERSION:2.0
|
|
BEGIN:VEVENT
|
|
UID:[email protected]
|
|
SUMMARY:Dentist
|
|
DTSTART:20260801T140000Z
|
|
DTEND:20260801T150000Z
|
|
END:VEVENT
|
|
END:VCALENDAR
|
|
"""
|
|
|
|
_RECURRING_ICS = b"""BEGIN:VCALENDAR
|
|
VERSION:2.0
|
|
BEGIN:VEVENT
|
|
UID:[email protected]
|
|
SUMMARY:Standup
|
|
DTSTART:20260803T090000Z
|
|
DTEND:20260803T091500Z
|
|
RRULE:FREQ=WEEKLY;COUNT=4
|
|
END:VEVENT
|
|
END:VCALENDAR
|
|
"""
|
|
|
|
_SHARED_EVENT_ICS_A = b"""BEGIN:VCALENDAR
|
|
VERSION:2.0
|
|
BEGIN:VEVENT
|
|
UID:[email protected]
|
|
SUMMARY:Family Dinner
|
|
DTSTART:20260805T230000Z
|
|
DTEND:20260806T010000Z
|
|
END:VEVENT
|
|
END:VCALENDAR
|
|
"""
|
|
|
|
_SHARED_EVENT_ICS_B = b"""BEGIN:VCALENDAR
|
|
VERSION:2.0
|
|
BEGIN:VEVENT
|
|
UID:[email protected]
|
|
SUMMARY:Family Dinner
|
|
DTSTART:20260805T230000Z
|
|
DTEND:20260806T010000Z
|
|
END:VEVENT
|
|
END:VCALENDAR
|
|
"""
|
|
|
|
|
|
class _Handler(BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
body = self.server.feeds.get(self.path) # type: ignore[attr-defined]
|
|
if body is None:
|
|
self.send_response(404)
|
|
self.end_headers()
|
|
return
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/calendar")
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
|
|
@pytest.fixture
|
|
def ics_server():
|
|
server = HTTPServer(("127.0.0.1", 0), _Handler)
|
|
server.feeds = {
|
|
"/plain.ics": _PLAIN_ICS,
|
|
"/recurring.ics": _RECURRING_ICS,
|
|
"/shared-a.ics": _SHARED_EVENT_ICS_A,
|
|
"/shared-b.ics": _SHARED_EVENT_ICS_B,
|
|
}
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
port = server.server_address[1]
|
|
try:
|
|
yield f"http://127.0.0.1:{port}"
|
|
finally:
|
|
server.shutdown()
|
|
thread.join()
|
|
|
|
|
|
_WINDOW_START = date(2026, 7, 1)
|
|
_WINDOW_END = date(2026, 9, 1)
|
|
|
|
|
|
def test_single_source_fetches_its_event(ics_server):
|
|
sources = [CalendarSource("Alice", "ics", f"{ics_server}/plain.ics")]
|
|
events, summary = merge_events(sources, _WINDOW_START, _WINDOW_END)
|
|
assert summary == ""
|
|
assert len(events) == 1
|
|
assert events[0]["summary"] == "Dentist"
|
|
assert events[0]["sources"] == [{"owner_display_name": "Alice", "color_index": None}]
|
|
|
|
|
|
def test_recurring_event_expands_within_window(ics_server):
|
|
sources = [CalendarSource("Bob", "ics", f"{ics_server}/recurring.ics")]
|
|
events, summary = merge_events(sources, _WINDOW_START, _WINDOW_END)
|
|
assert summary == ""
|
|
assert len(events) == 4 # COUNT=4
|
|
assert all(e["summary"] == "Standup" for e in events)
|
|
# distinct occurrences, not the same one repeated
|
|
assert len({e["start"] for e in events}) == 4
|
|
|
|
|
|
def test_unreachable_source_does_not_blank_others(ics_server):
|
|
sources = [
|
|
CalendarSource("Alice", "ics", f"{ics_server}/plain.ics"),
|
|
CalendarSource("Bob", "ics", f"{ics_server}/does-not-exist.ics"),
|
|
]
|
|
events, summary = merge_events(sources, _WINDOW_START, _WINDOW_END)
|
|
assert len(events) == 1
|
|
assert events[0]["summary"] == "Dentist"
|
|
assert summary == "1 of 2 calendars unavailable"
|
|
|
|
|
|
def test_all_sources_unreachable(ics_server):
|
|
sources = [
|
|
CalendarSource("Alice", "ics", f"{ics_server}/nope1.ics"),
|
|
CalendarSource("Bob", "ics", f"{ics_server}/nope2.ics"),
|
|
]
|
|
events, summary = merge_events(sources, _WINDOW_START, _WINDOW_END)
|
|
assert events == []
|
|
assert summary == "2 of 2 calendars unavailable"
|
|
|
|
|
|
def test_duplicate_event_across_calendars_collapses_with_both_sources(ics_server):
|
|
"""A shared event synced onto two people's calendars (same summary/
|
|
start/end/all_day) should show up once, but carry both owners in
|
|
its `sources` list -- see merge_events' own docstring."""
|
|
sources = [
|
|
CalendarSource("Alice", "ics", f"{ics_server}/shared-a.ics"),
|
|
CalendarSource("Bob", "ics", f"{ics_server}/shared-b.ics"),
|
|
]
|
|
events, summary = merge_events(sources, _WINDOW_START, _WINDOW_END)
|
|
assert summary == ""
|
|
assert len(events) == 1
|
|
owners = {s["owner_display_name"] for s in events[0]["sources"]}
|
|
assert owners == {"Alice", "Bob"}
|
|
|
|
|
|
def test_events_outside_window_are_excluded(ics_server):
|
|
sources = [CalendarSource("Alice", "ics", f"{ics_server}/plain.ics")]
|
|
events, _ = merge_events(sources, date(2020, 1, 1), date(2020, 2, 1))
|
|
assert events == []
|