diff --git a/haiku_rag_slim/haiku/rag/ingester/sources/webdav.py b/haiku_rag_slim/haiku/rag/ingester/sources/webdav.py index 5625b38d..290badbb 100644 --- a/haiku_rag_slim/haiku/rag/ingester/sources/webdav.py +++ b/haiku_rag_slim/haiku/rag/ingester/sources/webdav.py @@ -1,4 +1,5 @@ import hashlib +import re from collections.abc import AsyncIterator from datetime import UTC, datetime from urllib.parse import unquote, urljoin, urlparse @@ -65,10 +66,20 @@ class _PropfindEntry: self.content_type = content_type +# Matches an ETag value with optional leading whitespace, optional weak +# marker ``W/``, optional surrounding double quotes, and optional trailing +# whitespace. The non-greedy capture pulls out just the opaque inner value. +_ETAG_RE = re.compile(r'^\s*(?:W/)?"?(.*?)"?\s*$') + + def _strip_etag(value: str | None) -> str | None: + """Return the opaque part of an ETag header value (or ``getetag`` element): + strip surrounding whitespace, the optional ``W/`` weak marker, and + optional surrounding double quotes. Returns ``None`` for empty input.""" if value is None: return None - cleaned = value.strip().strip('"').strip("W/").strip().strip('"') + match = _ETAG_RE.match(value) + cleaned = match.group(1) if match else value.strip() return cleaned or None diff --git a/tests/ingester/test_webdav_source.py b/tests/ingester/test_webdav_source.py index e42ef54d..443d05cd 100644 --- a/tests/ingester/test_webdav_source.py +++ b/tests/ingester/test_webdav_source.py @@ -4,7 +4,29 @@ import httpx import pytest from haiku.rag.ingester.sources.base import SourceEventKind -from haiku.rag.ingester.sources.webdav import WebDAVSource +from haiku.rag.ingester.sources.webdav import WebDAVSource, _strip_etag + + +def test_strip_etag_strong_quoted(): + assert _strip_etag('"abc123"') == "abc123" + + +def test_strip_etag_weak_marker(): + assert _strip_etag('W/"abc123"') == "abc123" + + +def test_strip_etag_unquoted(): + assert _strip_etag("abc123") == "abc123" + + +def test_strip_etag_whitespace(): + assert _strip_etag(' W/"abc" ') == "abc" + + +def test_strip_etag_empty_returns_none(): + assert _strip_etag("") is None + assert _strip_etag('""') is None + assert _strip_etag(None) is None def _transport(handler) -> httpx.MockTransport: