219 lines
7.0 KiB
Python
219 lines
7.0 KiB
Python
"""End-to-end: a real CalDAV client against Radicale + this storage plugin.
|
|
|
|
SPEC 12 test 10. Exercises the actual HTTP surface (PROPFIND, REPORT, PUT,
|
|
DELETE) rather than calling the storage API directly.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
from pathlib import Path
|
|
from wsgiref.simple_server import WSGIRequestHandler, make_server
|
|
|
|
import pytest
|
|
|
|
import mdcaldav.storage as storage_module
|
|
from mdcaldav.config import Config
|
|
|
|
caldav = pytest.importorskip("caldav")
|
|
radicale = pytest.importorskip("radicale")
|
|
|
|
|
|
class _QuietHandler(WSGIRequestHandler):
|
|
def log_message(self, *args): # keep test output readable
|
|
pass
|
|
|
|
|
|
@pytest.fixture
|
|
def server(vault: Path, tmp_path: Path):
|
|
from radicale import Application
|
|
from radicale import config as radicale_config
|
|
|
|
cfg = Config()
|
|
cfg.vault.path = vault
|
|
cfg.index.db = tmp_path / "index.db"
|
|
storage_module.set_config(cfg)
|
|
|
|
configuration = radicale_config.load()
|
|
configuration.update(
|
|
{
|
|
"storage": {"type": "mdcaldav.storage"},
|
|
"auth": {"type": "none"},
|
|
"logging": {"level": "critical"},
|
|
},
|
|
"test",
|
|
)
|
|
|
|
httpd = make_server("localhost", 0, Application(configuration), handler_class=_QuietHandler)
|
|
thread = threading.Thread(target=httpd.serve_forever, daemon=True)
|
|
thread.start()
|
|
url = f"http://localhost:{httpd.server_port}"
|
|
try:
|
|
yield url, vault
|
|
finally:
|
|
httpd.shutdown()
|
|
thread.join(timeout=5)
|
|
storage_module.set_config(Config())
|
|
|
|
|
|
def _calendars(url: str):
|
|
client = caldav.DAVClient(url=url, username="test", password="test")
|
|
return client.principal().calendars()
|
|
|
|
|
|
def test_root_propfind_reports_current_user_principal(server):
|
|
"""Regression: the root collection's path must be "" or Radicale drops it.
|
|
|
|
A root item whose path was the storage's own user name got filtered out of
|
|
the multistatus, so clients could not discover the principal at all. Only
|
|
surfaced under htpasswd auth, hence this direct check.
|
|
"""
|
|
import base64
|
|
import urllib.request
|
|
|
|
url, _ = server
|
|
body = (
|
|
b'<?xml version="1.0"?><propfind xmlns="DAV:">'
|
|
b"<prop><current-user-principal/></prop></propfind>"
|
|
)
|
|
# Radicale answers 401 for current-user-principal when unauthenticated.
|
|
token = base64.b64encode(b"tyler:x").decode()
|
|
request = urllib.request.Request(
|
|
url + "/",
|
|
data=body,
|
|
method="PROPFIND",
|
|
headers={"Depth": "0", "Authorization": f"Basic {token}"},
|
|
)
|
|
payload = urllib.request.urlopen(request, timeout=10).read().decode()
|
|
|
|
assert "current-user-principal" in payload
|
|
assert "<href>/</href>" in payload
|
|
|
|
|
|
def test_collections_are_discoverable(server):
|
|
url, _ = server
|
|
names = {cal.url.path.rstrip("/").rsplit("/", 1)[-1] for cal in _calendars(url)}
|
|
assert {"daily", "projects"} <= names
|
|
|
|
|
|
def test_todos_are_listed_with_fields(server):
|
|
url, _ = server
|
|
daily = next(c for c in _calendars(url) if c.url.path.rstrip("/").endswith("daily"))
|
|
todos = daily.todos(include_completed=True)
|
|
assert todos
|
|
|
|
summaries = {str(t.icalendar_component.get("SUMMARY")) for t in todos}
|
|
assert "Water the seedlings" in summaries
|
|
assert "Book the soil test" in summaries
|
|
|
|
soil = next(
|
|
t
|
|
for t in todos
|
|
if str(t.icalendar_component.get("SUMMARY")) == "Book the soil test"
|
|
)
|
|
assert int(soil.icalendar_component.get("PRIORITY")) == 1
|
|
|
|
|
|
def test_subtasks_carry_related_to(server):
|
|
url, _ = server
|
|
projects = next(
|
|
c for c in _calendars(url) if c.url.path.rstrip("/").endswith("projects")
|
|
)
|
|
todos = projects.todos(include_completed=True)
|
|
by_summary = {str(t.icalendar_component.get("SUMMARY")): t for t in todos}
|
|
|
|
parent = by_summary["Trench the main line"]
|
|
child = by_summary["Rent the trencher"]
|
|
related = child.icalendar_component.get("RELATED-TO")
|
|
assert str(related) == str(parent.icalendar_component.get("UID"))
|
|
assert related.params.get("RELTYPE") == "PARENT"
|
|
|
|
|
|
def test_completing_a_todo_rewrites_the_markdown(server):
|
|
url, vault = server
|
|
daily = next(c for c in _calendars(url) if c.url.path.rstrip("/").endswith("daily"))
|
|
todo = next(
|
|
t
|
|
for t in daily.todos()
|
|
if str(t.icalendar_component.get("SUMMARY")) == "Book the soil test"
|
|
)
|
|
|
|
todo.complete()
|
|
|
|
text = (vault / "daily/2026-01-07.md").read_text()
|
|
assert "- [x] [#A] Book the soil test" in text
|
|
assert "- [ ] [#B] Call the arborist" in text # neighbour untouched
|
|
|
|
|
|
def test_description_edit_round_trips_to_markdown(server):
|
|
url, vault = server
|
|
daily = next(c for c in _calendars(url) if c.url.path.rstrip("/").endswith("daily"))
|
|
todo = next(
|
|
t
|
|
for t in daily.todos()
|
|
if str(t.icalendar_component.get("SUMMARY")) == "Call the arborist"
|
|
)
|
|
|
|
todo.icalendar_component["DESCRIPTION"] = "Ask about the oak"
|
|
todo.save()
|
|
|
|
text = (vault / "daily/2026-01-07.md").read_text()
|
|
assert "- [ ] [#B] Call the arborist\n - Ask about the oak\n" in text
|
|
|
|
|
|
def test_creating_a_todo_lands_in_the_collections_inbox(server):
|
|
"""A task created in `daily` must stay in `daily`, not jump to another list."""
|
|
url, vault = server
|
|
daily = next(c for c in _calendars(url) if c.url.path.rstrip("/").endswith("daily"))
|
|
daily.save_todo(summary="Buy netting", priority=1)
|
|
|
|
assert "- [ ] [#A] Buy netting" in (vault / "daily/inbox.md").read_text()
|
|
assert not (vault / "inbox.md").exists()
|
|
|
|
summaries = {
|
|
str(t.icalendar_component.get("SUMMARY"))
|
|
for t in daily.todos(include_completed=True)
|
|
}
|
|
assert "Buy netting" in summaries # visible in the list it was created in
|
|
|
|
|
|
def test_created_todo_keeps_the_client_uid(server):
|
|
url, _ = server
|
|
daily = next(c for c in _calendars(url) if c.url.path.rstrip("/").endswith("daily"))
|
|
created = daily.save_todo(summary="Mulch the beds")
|
|
uid = str(created.icalendar_component.get("UID"))
|
|
|
|
fetched = daily.todo_by_uid(uid)
|
|
assert str(fetched.icalendar_component.get("SUMMARY")) == "Mulch the beds"
|
|
|
|
|
|
def test_deleting_a_todo_removes_the_line(server):
|
|
url, vault = server
|
|
daily = next(c for c in _calendars(url) if c.url.path.rstrip("/").endswith("daily"))
|
|
todo = next(
|
|
t
|
|
for t in daily.todos()
|
|
if str(t.icalendar_component.get("SUMMARY")) == "Call the arborist"
|
|
)
|
|
|
|
todo.delete()
|
|
|
|
text = (vault / "daily/2026-01-07.md").read_text()
|
|
assert "Call the arborist" not in text
|
|
assert "Book the soil test" in text
|
|
|
|
|
|
def test_external_vim_edit_becomes_visible(server):
|
|
"""A change made on disk shows up on the next client read."""
|
|
url, vault = server
|
|
daily = next(c for c in _calendars(url) if c.url.path.rstrip("/").endswith("daily"))
|
|
|
|
rel = vault / "daily/2026-01-07.md"
|
|
rel.write_text(rel.read_text() + "- [ ] Order more compost\n")
|
|
|
|
summaries = {
|
|
str(t.icalendar_component.get("SUMMARY"))
|
|
for t in daily.todos(include_completed=True)
|
|
}
|
|
assert "Order more compost" in summaries
|