fix URL-decode file:// paths before stat/exists checks
This commit is contained in:
parent
7ea61a7b10
commit
6d15f1f247
2 changed files with 67 additions and 10 deletions
|
|
@ -1,7 +1,7 @@
|
||||||
import tempfile
|
import tempfile
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import TYPE_CHECKING
|
from typing import TYPE_CHECKING
|
||||||
from urllib.parse import urlparse
|
from urllib.parse import unquote, urlparse
|
||||||
|
|
||||||
from haiku.rag.client.processing import (
|
from haiku.rag.client.processing import (
|
||||||
ensure_chunks_embedded,
|
ensure_chunks_embedded,
|
||||||
|
|
@ -315,8 +315,10 @@ async def create_document_from_source(
|
||||||
# Directory case: recurse with the existing FS filter and produce one
|
# Directory case: recurse with the existing FS filter and produce one
|
||||||
# document per file. Remote schemes (http/s3) never hit this branch.
|
# document per file. Remote schemes (http/s3) never hit this branch.
|
||||||
if parsed_url.scheme in ("", "file"):
|
if parsed_url.scheme in ("", "file"):
|
||||||
|
# file:// URIs URL-encode special characters ([, ], spaces, etc.);
|
||||||
|
# unquote to get the real filesystem path before any stat/rglob.
|
||||||
local_path = (
|
local_path = (
|
||||||
Path(parsed_url.path)
|
Path(unquote(parsed_url.path))
|
||||||
if parsed_url.scheme == "file"
|
if parsed_url.scheme == "file"
|
||||||
else (Path(source) if isinstance(source, str) else source)
|
else (Path(source) if isinstance(source, str) else source)
|
||||||
)
|
)
|
||||||
|
|
@ -355,16 +357,15 @@ async def create_document_from_source(
|
||||||
fetcher = resolve_fetcher(source_str, storage_options=storage_options)
|
fetcher = resolve_fetcher(source_str, storage_options=storage_options)
|
||||||
|
|
||||||
# The stored URI is what we look up + persist by. For an explicit uri
|
# The stored URI is what we look up + persist by. For an explicit uri
|
||||||
# override, use it as-is. Otherwise canonicalize local paths to file://
|
# override, use it as-is. For a file:// input the source string is
|
||||||
# and leave remote URIs alone.
|
# already canonical (URL-encoded); round-tripping via Path.as_uri()
|
||||||
|
# would double-encode any escapes like %5B. For bare paths, canonicalize.
|
||||||
if uri is not None:
|
if uri is not None:
|
||||||
stored_uri = uri
|
stored_uri = uri
|
||||||
elif parsed_url.scheme in ("", "file"):
|
elif parsed_url.scheme == "file":
|
||||||
stored_uri = (
|
stored_uri = source_str
|
||||||
(Path(parsed_url.path) if parsed_url.scheme == "file" else Path(source_str))
|
elif parsed_url.scheme == "":
|
||||||
.absolute()
|
stored_uri = Path(source_str).absolute().as_uri()
|
||||||
.as_uri()
|
|
||||||
)
|
|
||||||
else:
|
else:
|
||||||
stored_uri = source_str
|
stored_uri = source_str
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -135,6 +135,62 @@ async def test_e2e_initial_sweep_lands_succeeded_jobs(tmp_path, jobs, sync):
|
||||||
assert row_b is not None and row_b.content_hash and row_b.last_ingested_at
|
assert row_b is not None and row_b.content_hash and row_b.last_ingested_at
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_e2e_handles_url_encoded_special_chars_in_path(tmp_path, jobs, sync):
|
||||||
|
"""File names containing characters that path.as_uri() URL-encodes (e.g.
|
||||||
|
Next.js dynamic-route brackets like `[chunk_id]`) must survive the
|
||||||
|
round-trip through the job queue without tripping the existence check
|
||||||
|
inside create_document_from_source."""
|
||||||
|
bracketed_dir = tmp_path / "[chunk_id]"
|
||||||
|
bracketed_dir.mkdir()
|
||||||
|
target = bracketed_dir / "route.ts"
|
||||||
|
target.write_text("export default {};")
|
||||||
|
|
||||||
|
client = _mock_client(tmp_path)
|
||||||
|
cfg = FSSourceConfig(
|
||||||
|
type="fs",
|
||||||
|
id="local",
|
||||||
|
root=tmp_path,
|
||||||
|
poll_interval_s=60.0,
|
||||||
|
)
|
||||||
|
|
||||||
|
manager = PollerManager(
|
||||||
|
configs=[cfg],
|
||||||
|
job_repo=jobs,
|
||||||
|
sync_repo=sync,
|
||||||
|
supported_extensions=[".ts"],
|
||||||
|
)
|
||||||
|
pool = WorkerPool(
|
||||||
|
client=client,
|
||||||
|
job_repo=jobs,
|
||||||
|
sync_repo=sync,
|
||||||
|
worker_count=1,
|
||||||
|
max_concurrent=1,
|
||||||
|
poll_idle_interval_s=0.05,
|
||||||
|
)
|
||||||
|
|
||||||
|
await pool.start()
|
||||||
|
await manager.start()
|
||||||
|
try:
|
||||||
|
|
||||||
|
async def _one_succeeded() -> bool:
|
||||||
|
counts = await jobs.counts_by_status()
|
||||||
|
return counts.get("succeeded", 0) == 1
|
||||||
|
|
||||||
|
await _wait_for(_one_succeeded, timeout=5.0)
|
||||||
|
finally:
|
||||||
|
await manager.stop()
|
||||||
|
await pool.stop()
|
||||||
|
|
||||||
|
counts = await jobs.counts_by_status()
|
||||||
|
assert counts.get("succeeded", 0) == 1
|
||||||
|
assert counts.get("dead", 0) == 0 # no PermanentError("File does not exist")
|
||||||
|
|
||||||
|
# The URI in the queue is URL-encoded; the worker still finds the file.
|
||||||
|
[call] = client.create_document_from_source.await_args_list
|
||||||
|
assert "%5Bchunk_id%5D" in call.args[0]
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_e2e_watchfiles_push_event_lands_as_job(tmp_path, jobs, sync):
|
async def test_e2e_watchfiles_push_event_lands_as_job(tmp_path, jobs, sync):
|
||||||
"""FSPoller's watchfiles loop: a file *added* after startup should land
|
"""FSPoller's watchfiles loop: a file *added* after startup should land
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue