diff --git a/CHANGELOG.md b/CHANGELOG.md index f29d9ccb..8861dcac 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,7 +3,7 @@ ### Added -- New `haiku-ingester` service for continuous document ingestion: persistent SQLite job queue, async worker pool with retries and a dead-letter queue, FS/HTTP/S3 source adapters with per-source circuit breakers, and a FastAPI control plane (`/health`, `/jobs`, `/sources`, `/dlq`). Configured under `ingester:` in `haiku.rag.yaml`. Shipped behind the `[ingester]` extra, with Logfire spans (`ingester.poller.sweep` → `ingester.job` → `document.{fetch,convert,chunk,embed,store}`) for traceable ingestion. See [docs/ingester.md](docs/ingester.md). +- New `haiku-ingester` service for continuous document ingestion: persistent SQLite job queue, async worker pool with retries and a dead-letter queue, FS/HTTP/S3/WebDAV source adapters with per-source circuit breakers, and a FastAPI control plane (`/health`, `/jobs`, `/sources`, `/dlq`). Configured under `ingester:` in `haiku.rag.yaml`. Shipped behind the `[ingester]` extra, with Logfire spans (`ingester.poller.sweep` → `ingester.job` → `document.{fetch,convert,chunk,embed,store}`) for traceable ingestion. See [docs/ingester.md](docs/ingester.md). ### Removed diff --git a/docs/ingester.md b/docs/ingester.md index e119a074..dd439e9f 100644 --- a/docs/ingester.md +++ b/docs/ingester.md @@ -109,6 +109,39 @@ response from a configured URL triggers a delete event; other failure statuses fall through to UPSERT-with-no-revision so the worker can GET and decide. +### WebDAV + +```yaml +ingester: + sources: + - type: webdav + id: nextcloud + base_url: https://nextcloud.example.com/remote.php/dav/files/alice/Documents/ + username: alice + password: ${NEXTCLOUD_APP_PASSWORD} + ignore_patterns: ["**/Trash/**"] + poll_interval_s: 600 +``` + +Each sweep issues one `PROPFIND` with `Depth: infinity` against +`base_url` and parses the multistatus response. Files (non-collection +resources) are emitted as UPSERT / UNCHANGED based on the `getetag` +property (falling back to `getlastmodified` if the server omits it); +URIs that were in the previous snapshot but no longer appear under the +collection are emitted as DELETE. + +Fetches are plain HTTP GETs — any WebDAV server already supports them. + +Bearer-token auth can replace HTTP Basic via the standard `headers` map: + +```yaml + - type: webdav + id: kdrive + base_url: https://kdrive.infomaniak.com/app/drive/123/ + headers: + Authorization: Bearer ${KDRIVE_TOKEN} +``` + ## Workers and retry ```yaml diff --git a/haiku_rag_slim/haiku/rag/config/__init__.py b/haiku_rag_slim/haiku/rag/config/__init__.py index 0c6919ef..0d98b571 100644 --- a/haiku_rag_slim/haiku/rag/config/__init__.py +++ b/haiku_rag_slim/haiku/rag/config/__init__.py @@ -26,6 +26,7 @@ from haiku.rag.config.models import ( S3SourceConfig, SourceConfig, StorageConfig, + WebDAVSourceConfig, WorkerConfig, ) @@ -53,6 +54,7 @@ __all__ = [ "S3SourceConfig", "SourceConfig", "StorageConfig", + "WebDAVSourceConfig", "WorkerConfig", "find_config_file", "generate_default_config", diff --git a/haiku_rag_slim/haiku/rag/config/models.py b/haiku_rag_slim/haiku/rag/config/models.py index c585bfc7..ac5768db 100644 --- a/haiku_rag_slim/haiku/rag/config/models.py +++ b/haiku_rag_slim/haiku/rag/config/models.py @@ -323,8 +323,21 @@ class S3SourceConfig(_SourceBase): include_patterns: list[str] = [] +class WebDAVSourceConfig(_SourceBase): + """A WebDAV collection (Nextcloud, ownCloud, Apache mod_dav, etc.). Files + are discovered via PROPFIND on `base_url`; fetch is plain HTTP GET.""" + + type: Literal["webdav"] + base_url: str + username: str | None = None + password: str | None = None + headers: dict[str, str] = Field(default_factory=dict) + ignore_patterns: list[str] = [] + include_patterns: list[str] = [] + + SourceConfig = Annotated[ - FSSourceConfig | HTTPSourceConfig | S3SourceConfig, + FSSourceConfig | HTTPSourceConfig | S3SourceConfig | WebDAVSourceConfig, Field(discriminator="type"), ] diff --git a/haiku_rag_slim/haiku/rag/ingester/pollers/factory.py b/haiku_rag_slim/haiku/rag/ingester/pollers/factory.py index d4e37bc2..266b8277 100644 --- a/haiku_rag_slim/haiku/rag/ingester/pollers/factory.py +++ b/haiku_rag_slim/haiku/rag/ingester/pollers/factory.py @@ -3,8 +3,15 @@ from haiku.rag.config import ( HTTPSourceConfig, S3SourceConfig, SourceConfig, + WebDAVSourceConfig, +) +from haiku.rag.ingester.sources import ( + FSSource, + HTTPSource, + S3Source, + Source, + WebDAVSource, ) -from haiku.rag.ingester.sources import FSSource, HTTPSource, S3Source, Source def build_source( @@ -16,7 +23,7 @@ def build_source( Source IDs auto-derive from the target when the config didn't supply one, matching the conventions in the adapters themselves (fs:, - s3:/, http:). + s3:/, http:, webdav:). """ if isinstance(cfg, FSSourceConfig): return FSSource( @@ -39,4 +46,17 @@ def build_source( supported_extensions=supported_extensions, source_id=cfg.id, ) + if isinstance(cfg, WebDAVSourceConfig): + if cfg.id is None: + raise ValueError("WebDAVSourceConfig.id is required") + return WebDAVSource( + source_id=cfg.id, + base_url=cfg.base_url, + username=cfg.username, + password=cfg.password, + headers=cfg.headers, + ignore_patterns=cfg.ignore_patterns or None, + include_patterns=cfg.include_patterns or None, + supported_extensions=supported_extensions, + ) raise TypeError(f"Unsupported source config: {type(cfg).__name__}") diff --git a/haiku_rag_slim/haiku/rag/ingester/sources/__init__.py b/haiku_rag_slim/haiku/rag/ingester/sources/__init__.py index d6290d15..4ece8894 100644 --- a/haiku_rag_slim/haiku/rag/ingester/sources/__init__.py +++ b/haiku_rag_slim/haiku/rag/ingester/sources/__init__.py @@ -10,6 +10,7 @@ from haiku.rag.ingester.sources.fs import FSSource from haiku.rag.ingester.sources.http import HTTPSource from haiku.rag.ingester.sources.registry import resolve_fetcher from haiku.rag.ingester.sources.s3 import S3Source +from haiku.rag.ingester.sources.webdav import WebDAVSource __all__ = [ "FetchResult", @@ -21,5 +22,6 @@ __all__ = [ "Source", "SourceEvent", "SourceEventKind", + "WebDAVSource", "resolve_fetcher", ] diff --git a/haiku_rag_slim/haiku/rag/ingester/sources/webdav.py b/haiku_rag_slim/haiku/rag/ingester/sources/webdav.py new file mode 100644 index 00000000..5625b38d --- /dev/null +++ b/haiku_rag_slim/haiku/rag/ingester/sources/webdav.py @@ -0,0 +1,292 @@ +import hashlib +from collections.abc import AsyncIterator +from datetime import UTC, datetime +from urllib.parse import unquote, urljoin, urlparse +from xml.etree.ElementTree import Element, fromstring + +import httpx + +from haiku.rag.ingester.sources.base import ( + FetchResult, + RevisionSnapshot, + SourceEvent, + SourceEventKind, +) +from haiku.rag.ingester.sources.filter import ( + FileFilter, + _default_supported_extensions, +) + +# WebDAV PROPFIND uses XML with the DAV: namespace. Element tags arrive as +# Clark-notation strings like "{DAV:}response", so we precompile them once. +_DAV_NS = "DAV:" +_TAG_RESPONSE = f"{{{_DAV_NS}}}response" +_TAG_HREF = f"{{{_DAV_NS}}}href" +_TAG_PROPSTAT = f"{{{_DAV_NS}}}propstat" +_TAG_PROP = f"{{{_DAV_NS}}}prop" +_TAG_STATUS = f"{{{_DAV_NS}}}status" +_TAG_RESOURCETYPE = f"{{{_DAV_NS}}}resourcetype" +_TAG_COLLECTION = f"{{{_DAV_NS}}}collection" +_TAG_GETETAG = f"{{{_DAV_NS}}}getetag" +_TAG_GETLASTMODIFIED = f"{{{_DAV_NS}}}getlastmodified" +_TAG_GETCONTENTTYPE = f"{{{_DAV_NS}}}getcontenttype" + + +_PROPFIND_BODY = b""" + + + + + + + + +""" + + +class _PropfindEntry: + """One element decoded into the fields the source actually + uses. `is_collection` separates folders from files; revision is ETag + when present, otherwise the Last-Modified header value.""" + + __slots__ = ("href", "is_collection", "revision", "content_type") + + def __init__( + self, + href: str, + *, + is_collection: bool, + revision: str | None, + content_type: str | None, + ) -> None: + self.href = href + self.is_collection = is_collection + self.revision = revision + self.content_type = content_type + + +def _strip_etag(value: str | None) -> str | None: + if value is None: + return None + cleaned = value.strip().strip('"').strip("W/").strip().strip('"') + return cleaned or None + + +def _entry_from_response(response: Element) -> _PropfindEntry | None: + """Decode a single . Returns None if the prop block is missing + or the entry didn't return HTTP 200 (e.g. 404 for a known-bad path).""" + href_el = response.find(_TAG_HREF) + if href_el is None or not href_el.text: + return None + href = href_el.text + + is_collection = False + revision: str | None = None + last_modified: str | None = None + content_type: str | None = None + ok = False + + for propstat in response.findall(_TAG_PROPSTAT): + status_el = propstat.find(_TAG_STATUS) + if status_el is None or not status_el.text: + continue + if " 200 " not in status_el.text: + continue + ok = True + prop = propstat.find(_TAG_PROP) + if prop is None: + continue + resourcetype = prop.find(_TAG_RESOURCETYPE) + if resourcetype is not None and resourcetype.find(_TAG_COLLECTION) is not None: + is_collection = True + etag_el = prop.find(_TAG_GETETAG) + if etag_el is not None and etag_el.text: + revision = _strip_etag(etag_el.text) + lm_el = prop.find(_TAG_GETLASTMODIFIED) + if lm_el is not None and lm_el.text: + last_modified = lm_el.text.strip() or None + ct_el = prop.find(_TAG_GETCONTENTTYPE) + if ct_el is not None and ct_el.text: + content_type = ct_el.text.split(";")[0].strip().lower() or None + + if not ok: + return None + # Prefer ETag — stronger validator. Fall back to Last-Modified so revision + # detection still works against servers that don't return ETags on PROPFIND. + return _PropfindEntry( + href=href, + is_collection=is_collection, + revision=revision or last_modified, + content_type=content_type, + ) + + +def _parse_multistatus(body: bytes) -> list[_PropfindEntry]: + """Top-level multistatus parser. Raises ValueError on garbage XML so the + poller's circuit breaker can record a failure.""" + try: + root = fromstring(body) + except Exception as exc: # ParseError + any defensive surprise + raise ValueError(f"Invalid PROPFIND response XML: {exc}") from exc + return [ + entry + for response in root.findall(_TAG_RESPONSE) + if (entry := _entry_from_response(response)) is not None + ] + + +def _resolve_href(href: str, base_url: str) -> str: + """PROPFIND href values can be either absolute URLs or server-relative + paths. Resolve to absolute against base_url either way, then URL-decode + the path so the URI we store matches what a user would type.""" + absolute = urljoin(base_url, href) + parsed = urlparse(absolute) + decoded_path = unquote(parsed.path) + rebuilt = parsed._replace(path=decoded_path) + return rebuilt.geturl() + + +class WebDAVSource: + def __init__( + self, + *, + source_id: str, + base_url: str, + username: str | None = None, + password: str | None = None, + headers: dict[str, str] | None = None, + ignore_patterns: list[str] | None = None, + include_patterns: list[str] | None = None, + supported_extensions: list[str] | None = None, + transport: httpx.AsyncBaseTransport | None = None, + ) -> None: + self.source_id = source_id + # Trailing slash matters: urljoin treats path-without-slash as a sibling + # link, so "https://srv/dav" joined with "subdir/x" gives ".../x". + self.base_url = base_url if base_url.endswith("/") else base_url + "/" + self.username = username + self.password = password + self.headers = dict(headers or {}) + self.supported_extensions = ( + list(supported_extensions) + if supported_extensions is not None + else _default_supported_extensions() + ) + self.filter = FileFilter( + ignore_patterns=ignore_patterns, + include_patterns=include_patterns, + supported_extensions=self.supported_extensions, + ) + # transport is for testing — production callers leave it None. + self._transport = transport + + def supports(self, uri: str) -> bool: + return uri.startswith(self.base_url) + + def _client(self) -> httpx.AsyncClient: + auth = ( + (self.username, self.password) + if self.username is not None and self.password is not None + else None + ) + return httpx.AsyncClient( + auth=auth, headers=self.headers, transport=self._transport + ) + + async def head(self, uri: str) -> str | None: + async with self._client() as http: + response = await http.request( + "PROPFIND", + uri, + headers={"Depth": "0", "Content-Type": "application/xml"}, + content=_PROPFIND_BODY, + ) + if response.is_error: + return None + entries = _parse_multistatus(response.content) + if not entries: + return None + return entries[0].revision + + async def fetch(self, uri: str) -> FetchResult: + async with self._client() as http: + response = await http.get(uri) + response.raise_for_status() + body = response.content + content_type = ( + response.headers.get("content-type", "application/octet-stream") + .split(";")[0] + .strip() + .lower() + ) + # ETag from the GET response is the freshest revision; fall back to + # Last-Modified, matching HTTPSource's preference order. + revision = ( + _strip_etag(response.headers.get("etag")) + or (response.headers.get("last-modified") or "").strip() + or None + ) + extra: dict[str, str] = {} + last_modified = (response.headers.get("last-modified") or "").strip() + if last_modified: + extra["last_modified"] = last_modified + return FetchResult( + uri=uri, + body=body, + content_type=content_type, + content_hash=hashlib.md5(body, usedforsecurity=False).hexdigest(), + revision=revision, + extra_metadata=extra, + ) + + async def discover( + self, since: RevisionSnapshot | None = None + ) -> AsyncIterator[SourceEvent]: + snapshot: dict[str, str] = dict(since) if since else {} + now = datetime.now(UTC) + seen: set[str] = set() + + async with self._client() as http: + response = await http.request( + "PROPFIND", + self.base_url, + headers={"Depth": "infinity", "Content-Type": "application/xml"}, + content=_PROPFIND_BODY, + ) + response.raise_for_status() + entries = _parse_multistatus(response.content) + + for entry in entries: + if entry.is_collection: + continue + uri = _resolve_href(entry.href, self.base_url) + # The base URL itself sometimes appears as a non-collection on + # broken servers; skip anything that's not strictly under it. + if uri == self.base_url.rstrip("/") or not uri.startswith(self.base_url): + continue + if not self.filter.include_file(uri): + continue + seen.add(uri) + revision = entry.revision + if revision is not None and snapshot.get(uri) == revision: + kind = SourceEventKind.UNCHANGED + else: + kind = SourceEventKind.UPSERT + yield SourceEvent( + source_id=self.source_id, + uri=uri, + kind=kind, + revision=revision, + discovered_at=now, + ) + + for uri in snapshot: + if uri in seen: + continue + yield SourceEvent( + source_id=self.source_id, + uri=uri, + kind=SourceEventKind.DELETE, + revision=None, + discovered_at=now, + ) diff --git a/tests/ingester/test_config.py b/tests/ingester/test_config.py index 05b4977f..db49223b 100644 --- a/tests/ingester/test_config.py +++ b/tests/ingester/test_config.py @@ -48,6 +48,27 @@ def test_discriminator_picks_s3_source(): assert cfg.sources[0].uri == "s3://bucket/prefix/" +def test_discriminator_picks_webdav_source(): + cfg = IngesterConfig.model_validate( + { + "sources": [ + { + "type": "webdav", + "id": "nc", + "base_url": "https://nc.example.com/dav/", + "username": "alice", + "password": "hunter2", + } + ] + } + ) + from haiku.rag.config import WebDAVSourceConfig + + assert isinstance(cfg.sources[0], WebDAVSourceConfig) + assert cfg.sources[0].base_url == "https://nc.example.com/dav/" + assert cfg.sources[0].username == "alice" + + def test_discriminator_rejects_unknown_type(): with pytest.raises(ValidationError): IngesterConfig.model_validate({"sources": [{"type": "ftp", "uri": "x"}]}) diff --git a/tests/ingester/test_pollers.py b/tests/ingester/test_pollers.py index de36adb9..8950aeb8 100644 --- a/tests/ingester/test_pollers.py +++ b/tests/ingester/test_pollers.py @@ -10,6 +10,7 @@ from haiku.rag.config import ( FSSourceConfig, HTTPSourceConfig, S3SourceConfig, + WebDAVSourceConfig, ) from haiku.rag.ingester.pollers.circuit_breaker import CircuitBreaker from haiku.rag.ingester.pollers.manager import PollerManager @@ -278,6 +279,9 @@ async def test_manager_builds_pollers_per_source(tmp_path, jobs, sync): FSSourceConfig(type="fs", root=tmp_path), S3SourceConfig(type="s3", uri="s3://bucket/"), HTTPSourceConfig(type="http", id="urls", urls=[]), + WebDAVSourceConfig( + type="webdav", id="nc", base_url="https://nc.example.com/dav/" + ), ] manager = PollerManager( configs=configs, @@ -285,11 +289,12 @@ async def test_manager_builds_pollers_per_source(tmp_path, jobs, sync): sync_repo=sync, ) built = manager.build_pollers() - assert len(built) == 3 + assert len(built) == 4 assert {p.source_id for p in built} == { f"fs:{tmp_path.resolve()}", "s3:bucket/", "urls", + "nc", } diff --git a/tests/ingester/test_webdav_source.py b/tests/ingester/test_webdav_source.py new file mode 100644 index 00000000..e42ef54d --- /dev/null +++ b/tests/ingester/test_webdav_source.py @@ -0,0 +1,363 @@ +import hashlib + +import httpx +import pytest + +from haiku.rag.ingester.sources.base import SourceEventKind +from haiku.rag.ingester.sources.webdav import WebDAVSource + + +def _transport(handler) -> httpx.MockTransport: + return httpx.MockTransport(handler) + + +def _multistatus(*entries: dict) -> bytes: + """Build a response. Each entry is a dict like + {'href': '/dav/x.md', 'collection': False, 'etag': '"abc"', + 'last_modified': 'Wed, ...', 'content_type': 'text/markdown'}.""" + body = ['', ''] + for e in entries: + body.append(" ") + body.append(f" {e['href']}") + body.append(" ") + body.append(" HTTP/1.1 200 OK") + body.append(" ") + body.append(" ") + if e.get("collection"): + body.append(" ") + body.append(" ") + if "etag" in e: + body.append(f" {e['etag']}") + if "last_modified" in e: + body.append( + f" {e['last_modified']}" + ) + if "content_type" in e: + body.append( + f" {e['content_type']}" + ) + body.append(" ") + body.append(" ") + body.append(" ") + body.append("") + return "\n".join(body).encode() + + +def test_supports_uri_under_base_url(): + src = WebDAVSource(source_id="nc", base_url="https://nc.example.com/dav/") + assert src.supports("https://nc.example.com/dav/file.md") + assert src.supports("https://nc.example.com/dav/sub/file.md") + assert not src.supports("https://nc.example.com/other/file.md") + assert not src.supports("https://other.example.com/dav/file.md") + + +def test_base_url_trailing_slash_normalised(): + """base_url without trailing slash mustn't break urljoin during discovery.""" + src = WebDAVSource(source_id="nc", base_url="https://nc.example.com/dav") + assert src.base_url == "https://nc.example.com/dav/" + assert src.supports("https://nc.example.com/dav/x.md") + + +@pytest.mark.asyncio +async def test_fetch_returns_bytes_md5_revision_and_content_type(): + body = b"hello dav" + + def handler(request: httpx.Request) -> httpx.Response: + assert request.method == "GET" + return httpx.Response( + 200, + content=body, + headers={ + "content-type": "text/markdown; charset=utf-8", + "etag": '"rev-1"', + "last-modified": "Wed, 21 Oct 2025 07:28:00 GMT", + }, + ) + + src = WebDAVSource( + source_id="nc", + base_url="https://nc.example.com/dav/", + transport=_transport(handler), + ) + result = await src.fetch("https://nc.example.com/dav/a.md") + assert result.body == body + assert result.content_hash == hashlib.md5(body, usedforsecurity=False).hexdigest() + assert result.content_type == "text/markdown" + assert result.revision == "rev-1" + assert result.extra_metadata == {"last_modified": "Wed, 21 Oct 2025 07:28:00 GMT"} + + +@pytest.mark.asyncio +async def test_fetch_falls_back_to_last_modified_when_no_etag(): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + content=b"x", + headers={ + "content-type": "text/plain", + "last-modified": "Wed, 21 Oct 2025 07:28:00 GMT", + }, + ) + + src = WebDAVSource( + source_id="nc", + base_url="https://nc.example.com/dav/", + transport=_transport(handler), + ) + result = await src.fetch("https://nc.example.com/dav/a.txt") + assert result.revision == "Wed, 21 Oct 2025 07:28:00 GMT" + + +@pytest.mark.asyncio +async def test_head_returns_etag_from_propfind_depth_zero(): + def handler(request: httpx.Request) -> httpx.Response: + assert request.method == "PROPFIND" + assert request.headers["Depth"] == "0" + return httpx.Response( + 207, + content=_multistatus( + {"href": "/dav/a.md", "etag": '"rev-9"'}, + ), + ) + + src = WebDAVSource( + source_id="nc", + base_url="https://nc.example.com/dav/", + transport=_transport(handler), + ) + assert await src.head("https://nc.example.com/dav/a.md") == "rev-9" + + +@pytest.mark.asyncio +async def test_head_returns_none_on_404(): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(404) + + src = WebDAVSource( + source_id="nc", + base_url="https://nc.example.com/dav/", + transport=_transport(handler), + ) + assert await src.head("https://nc.example.com/dav/missing.md") is None + + +@pytest.mark.asyncio +async def test_discover_yields_upserts_and_skips_collections_and_unsupported(): + multistatus = _multistatus( + {"href": "/dav/", "collection": True}, + {"href": "/dav/sub/", "collection": True}, + {"href": "/dav/a.md", "etag": '"rev-a"', "content_type": "text/markdown"}, + {"href": "/dav/sub/b.txt", "etag": '"rev-b"', "content_type": "text/plain"}, + {"href": "/dav/skip.log", "etag": '"rev-log"', "content_type": "text/plain"}, + ) + + def handler(request: httpx.Request) -> httpx.Response: + assert request.method == "PROPFIND" + assert request.headers["Depth"] == "infinity" + return httpx.Response(207, content=multistatus) + + src = WebDAVSource( + source_id="nc", + base_url="https://nc.example.com/dav/", + transport=_transport(handler), + ) + + events = [event async for event in src.discover()] + by_uri = {e.uri: e for e in events} + assert set(by_uri) == { + "https://nc.example.com/dav/a.md", + "https://nc.example.com/dav/sub/b.txt", + } + assert all(e.kind is SourceEventKind.UPSERT for e in events) + assert by_uri["https://nc.example.com/dav/a.md"].revision == "rev-a" + + +@pytest.mark.asyncio +async def test_discover_emits_unchanged_when_snapshot_matches(): + multistatus = _multistatus( + {"href": "/dav/", "collection": True}, + {"href": "/dav/a.md", "etag": '"rev-a"', "content_type": "text/markdown"}, + ) + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(207, content=multistatus) + + src = WebDAVSource( + source_id="nc", + base_url="https://nc.example.com/dav/", + transport=_transport(handler), + ) + snapshot = {"https://nc.example.com/dav/a.md": "rev-a"} + events = [event async for event in src.discover(since=snapshot)] + assert [e.kind for e in events] == [SourceEventKind.UNCHANGED] + + +@pytest.mark.asyncio +async def test_discover_emits_delete_for_files_no_longer_listed(): + multistatus = _multistatus( + {"href": "/dav/", "collection": True}, + {"href": "/dav/a.md", "etag": '"rev-a"', "content_type": "text/markdown"}, + ) + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(207, content=multistatus) + + src = WebDAVSource( + source_id="nc", + base_url="https://nc.example.com/dav/", + transport=_transport(handler), + ) + snapshot = { + "https://nc.example.com/dav/a.md": "rev-a", + "https://nc.example.com/dav/gone.md": "rev-old", + } + events = [event async for event in src.discover(since=snapshot)] + kinds = {e.uri: e.kind for e in events} + assert kinds == { + "https://nc.example.com/dav/a.md": SourceEventKind.UNCHANGED, + "https://nc.example.com/dav/gone.md": SourceEventKind.DELETE, + } + + +@pytest.mark.asyncio +async def test_discover_uses_last_modified_when_etag_absent(): + multistatus = _multistatus( + { + "href": "/dav/a.md", + "last_modified": "Wed, 21 Oct 2025 07:28:00 GMT", + "content_type": "text/markdown", + }, + ) + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(207, content=multistatus) + + src = WebDAVSource( + source_id="nc", + base_url="https://nc.example.com/dav/", + transport=_transport(handler), + ) + events = [event async for event in src.discover()] + assert events[0].revision == "Wed, 21 Oct 2025 07:28:00 GMT" + + +@pytest.mark.asyncio +async def test_discover_raises_on_malformed_xml(): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(207, content=b" httpx.Response: + return httpx.Response(401) + + src = WebDAVSource( + source_id="nc", + base_url="https://nc.example.com/dav/", + transport=_transport(handler), + ) + with pytest.raises(httpx.HTTPStatusError): + [event async for event in src.discover()] + + +@pytest.mark.asyncio +async def test_basic_auth_sent_when_credentials_configured(): + seen_auth: list[str | None] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_auth.append(request.headers.get("authorization")) + return httpx.Response( + 207, + content=_multistatus({"href": "/dav/", "collection": True}), + ) + + src = WebDAVSource( + source_id="nc", + base_url="https://nc.example.com/dav/", + username="alice", + password="hunter2", + transport=_transport(handler), + ) + [event async for event in src.discover()] + assert seen_auth and seen_auth[0] is not None + assert seen_auth[0].startswith("Basic ") + + +@pytest.mark.asyncio +async def test_custom_headers_forwarded(): + seen: list[str | None] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request.headers.get("authorization")) + return httpx.Response( + 207, + content=_multistatus({"href": "/dav/", "collection": True}), + ) + + src = WebDAVSource( + source_id="nc", + base_url="https://nc.example.com/dav/", + headers={"Authorization": "Bearer tok-123"}, + transport=_transport(handler), + ) + [event async for event in src.discover()] + assert seen == ["Bearer tok-123"] + + +@pytest.mark.asyncio +async def test_discover_resolves_absolute_href(): + """Some servers return absolute URLs in href, others return server paths. + Both must produce the same stored URI.""" + multistatus = _multistatus( + { + "href": "https://nc.example.com/dav/a.md", + "etag": '"rev"', + "content_type": "text/markdown", + } + ) + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(207, content=multistatus) + + src = WebDAVSource( + source_id="nc", + base_url="https://nc.example.com/dav/", + transport=_transport(handler), + ) + events = [event async for event in src.discover()] + assert [e.uri for e in events] == ["https://nc.example.com/dav/a.md"] + + +@pytest.mark.asyncio +async def test_discover_url_decodes_href_path(): + """PROPFIND hrefs are percent-encoded per RFC 3986. We unquote them so + the stored URI matches what a user types in `add-src`.""" + multistatus = _multistatus( + { + "href": "/dav/my%20docs/Hello%20World.md", + "etag": '"rev"', + "content_type": "text/markdown", + } + ) + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(207, content=multistatus) + + src = WebDAVSource( + source_id="nc", + base_url="https://nc.example.com/dav/", + transport=_transport(handler), + ) + events = [event async for event in src.discover()] + assert [e.uri for e in events] == [ + "https://nc.example.com/dav/my docs/Hello World.md" + ]