From 5acd0b3ba881e896e3cb934166ef520c6bc35d1d Mon Sep 17 00:00:00 2001 From: AnoobFeng <109097565+AnoobFeng@users.noreply.github.com> Date: Sat, 4 Jul 2026 21:31:01 +0800 Subject: [PATCH] fix(gateway): offload gateway upload file IO (#3935) * fix(gateway): offload gateway upload file IO Move Gateway upload router filesystem work off the asyncio event loop by using a dedicated ContextVar-preserving file IO executor. Use async sandbox acquisition for non-mounted sandbox uploads and offload remote sandbox sync together with host file reads. Add blocking-IO regression coverage for upload, list, delete, and remote sandbox sync paths. * fix(gateway): align file IO worker env var prefix Rename the file IO executor worker-count environment variable from DEERFLOW_FILE_IO_WORKERS to DEER_FLOW_FILE_IO_WORKERS to match the repo's existing runtime configuration prefix convention. --- backend/AGENTS.md | 8 +- backend/app/gateway/routers/uploads.py | 145 ++++++++++++------ .../harness/deerflow/utils/file_io.py | 51 ++++++ .../tests/blocking_io/test_uploads_router.py | 135 ++++++++++++++++ .../support/detectors/thread_boundaries.py | 5 + .../tests/test_detect_thread_boundaries.py | 3 + backend/tests/test_file_io.py | 30 ++++ backend/tests/test_uploads_router.py | 41 +++-- 8 files changed, 362 insertions(+), 56 deletions(-) create mode 100644 backend/packages/harness/deerflow/utils/file_io.py create mode 100644 backend/tests/blocking_io/test_uploads_router.py create mode 100644 backend/tests/test_file_io.py diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 876889104..6f73e8a05 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -136,9 +136,12 @@ Blocking-IO runtime gate (`tests/blocking_io/`): for #1917); `test_sqlite_lifespan.py` (locks the offload around SQLite path resolution plus `ensure_sqlite_parent_dir`, fix for #1912); `test_jsonl_run_event_store.py` (locks `JsonlRunEventStore`'s async - API offloading its file IO via `asyncio.to_thread`, fix #3084); and + API offloading its file IO via `asyncio.to_thread`); `test_uploads_middleware.py` (locks `UploadsMiddleware.abefore_agent` - offloading the uploads-directory scan off the event loop). + offloading the uploads-directory scan off the event loop); and + `test_uploads_router.py` (locks Gateway upload/list/delete endpoints + offloading upload directory creation, staged writes, chmod/cleanup, + directory scans/deletes, and remote sandbox sync off the event loop). - `test_gate_smoke.py` is a meta-test asserting the gate actually catches unoffloaded blocking IO and that the `@pytest.mark.allow_blocking_io` opt-out works. @@ -766,6 +769,7 @@ Multi-file upload with automatic document conversion: - Files stored in thread-isolated directories under the resolving user's bucket (`users/{user_id}/threads/{thread_id}/user-data/uploads`). For IM channels the owner is threaded explicitly via the `user_id=` kwarg (see IM Channels → Owner-scoped file storage); HTTP/embedded callers resolve it from `get_effective_user_id()` - Duplicate filenames in a single upload request are auto-renamed with `_N` suffixes so later files do not truncate earlier files - Gateway HTTP uploads stage bytes as `.upload-*.part` files and atomically replace the destination only after size validation. These staging files are hidden from upload listings, agent upload context, and sandbox listing/search tools, and swept on Gateway startup if a hard crash leaves one behind. +- Gateway HTTP upload/list/delete handlers offload filesystem work through `deerflow.utils.file_io.run_file_io`, a dedicated ContextVar-preserving file IO executor. Non-mounted sandbox uploads acquire sandboxes with `SandboxProvider.acquire_async()` and offload `read_bytes()` plus `sandbox.update_file()` together. - Agent receives uploaded file list via `UploadsMiddleware` See [docs/FILE_UPLOAD.md](docs/FILE_UPLOAD.md) for details. diff --git a/backend/app/gateway/routers/uploads.py b/backend/app/gateway/routers/uploads.py index 267b9f8e1..f8452cc76 100644 --- a/backend/app/gateway/routers/uploads.py +++ b/backend/app/gateway/routers/uploads.py @@ -4,7 +4,9 @@ import logging import os import stat import tempfile +from dataclasses import dataclass from pathlib import Path +from typing import BinaryIO from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile from pydantic import BaseModel, Field @@ -32,6 +34,7 @@ from deerflow.uploads.manager import ( validate_upload_destination, ) from deerflow.utils.file_conversion import CONVERTIBLE_EXTENSIONS, convert_file_to_markdown +from deerflow.utils.file_io import run_file_io logger = logging.getLogger(__name__) @@ -43,6 +46,13 @@ DEFAULT_MAX_FILE_SIZE = 50 * 1024 * 1024 DEFAULT_MAX_TOTAL_SIZE = 100 * 1024 * 1024 +@dataclass(slots=True) +class _UploadTempFile: + file_path: Path + temp_path: Path + handle: BinaryIO + + class UploadedFileInfo(BaseModel): """Uploaded file metadata exposed by upload and list APIs.""" @@ -167,6 +177,78 @@ def _cleanup_uploaded_paths(paths: list[os.PathLike[str] | str]) -> None: logger.warning("Failed to clean up upload path after rejected request: %s", path, exc_info=True) +def _prepare_upload_destination(uploads_dir: os.PathLike[str] | str, display_filename: str) -> _UploadTempFile: + uploads_dir_path = Path(uploads_dir) + file_path = validate_upload_destination(uploads_dir_path, display_filename) + temp_fd, temp_path_str = tempfile.mkstemp(prefix=UPLOAD_STAGING_PREFIX, suffix=UPLOAD_STAGING_SUFFIX, dir=uploads_dir_path) + temp_path = Path(temp_path_str) + try: + handle = os.fdopen(temp_fd, "wb") + except Exception: + try: + os.close(temp_fd) + except OSError: + pass + try: + os.unlink(temp_path) + except FileNotFoundError: + pass + raise + return _UploadTempFile(file_path=file_path, temp_path=temp_path, handle=handle) + + +def _write_upload_chunk(upload_temp: _UploadTempFile, chunk: bytes) -> None: + upload_temp.handle.write(chunk) + + +def _abort_upload_temp(upload_temp: _UploadTempFile) -> None: + try: + upload_temp.handle.close() + finally: + try: + os.unlink(upload_temp.temp_path) + except FileNotFoundError: + pass + + +def _commit_upload_temp(upload_temp: _UploadTempFile) -> None: + upload_temp.handle.close() + try: + os.replace(upload_temp.temp_path, upload_temp.file_path) + except Exception: + try: + os.unlink(upload_temp.temp_path) + except FileNotFoundError: + pass + raise + + +def _make_uploaded_paths_sandbox_readable(paths: list[os.PathLike[str] | str]) -> None: + for file_path in paths: + _make_file_sandbox_readable(file_path) + + +def _sync_upload_to_sandbox(sandbox, file_path: os.PathLike[str] | str, virtual_path: str) -> None: + _make_file_sandbox_writable(file_path) + sandbox.update_file(virtual_path, Path(file_path).read_bytes()) + + +def _list_uploaded_files_for_thread(thread_id: str, user_id: str) -> dict: + uploads_dir = get_uploads_dir(thread_id, user_id=user_id) + result = list_files_in_dir(uploads_dir) + enrich_file_listing(result, thread_id) + + sandbox_uploads = get_paths().sandbox_uploads_dir(thread_id, user_id=user_id) + for f in result["files"]: + f["path"] = str(sandbox_uploads / f["filename"]) + return result + + +def _delete_uploaded_file_for_thread(thread_id: str, filename: str, user_id: str) -> dict: + uploads_dir = get_uploads_dir(thread_id, user_id=user_id) + return delete_file_safe(uploads_dir, filename, convertible_extensions=CONVERTIBLE_EXTENSIONS) + + async def _write_upload_file_with_limits( file: UploadFile, *, @@ -177,12 +259,9 @@ async def _write_upload_file_with_limits( total_size: int, ) -> tuple[os.PathLike[str] | str, int, int]: file_size = 0 - uploads_dir_path = Path(uploads_dir) - file_path = validate_upload_destination(uploads_dir_path, display_filename) - temp_fd, temp_path_str = tempfile.mkstemp(prefix=UPLOAD_STAGING_PREFIX, suffix=UPLOAD_STAGING_SUFFIX, dir=uploads_dir_path) - temp_path = Path(temp_path_str) - fh = os.fdopen(temp_fd, "wb") + upload_temp: _UploadTempFile | None = None try: + upload_temp = await run_file_io(_prepare_upload_destination, uploads_dir, display_filename) while chunk := await file.read(UPLOAD_CHUNK_SIZE): file_size += len(chunk) total_size += len(chunk) @@ -190,24 +269,15 @@ async def _write_upload_file_with_limits( raise HTTPException(status_code=413, detail=f"File too large: {display_filename}") if total_size > max_total_size: raise HTTPException(status_code=413, detail="Total upload size too large") - fh.write(chunk) + await run_file_io(_write_upload_chunk, upload_temp, chunk) + + await run_file_io(_commit_upload_temp, upload_temp) + file_path = upload_temp.file_path + upload_temp = None except Exception: - fh.close() - try: - os.unlink(temp_path) - except FileNotFoundError: - pass + if upload_temp is not None: + await run_file_io(_abort_upload_temp, upload_temp) raise - else: - fh.close() - try: - os.replace(temp_path, file_path) - except Exception: - try: - os.unlink(temp_path) - except FileNotFoundError: - pass - raise return file_path, file_size, total_size @@ -244,10 +314,10 @@ async def upload_files( try: effective_user_id = get_effective_user_id() - uploads_dir = ensure_uploads_dir(thread_id, user_id=effective_user_id) + uploads_dir = await run_file_io(ensure_uploads_dir, thread_id, user_id=effective_user_id) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) - sandbox_uploads = get_paths().sandbox_uploads_dir(thread_id, user_id=effective_user_id) + sandbox_uploads = uploads_dir uploaded_files = [] written_paths = [] sandbox_sync_targets = [] @@ -262,7 +332,7 @@ async def upload_files( sync_to_sandbox = not _uses_thread_data_mounts(sandbox_provider) sandbox = None if sync_to_sandbox: - sandbox_id = sandbox_provider.acquire(thread_id, user_id=effective_user_id) + sandbox_id = await sandbox_provider.acquire_async(thread_id, user_id=effective_user_id) sandbox = sandbox_provider.get(sandbox_id) if sandbox is None: raise HTTPException(status_code=500, detail="Failed to acquire sandbox") @@ -325,7 +395,7 @@ async def upload_files( uploaded_files.append(file_info) except HTTPException as e: - _cleanup_uploaded_paths(written_paths) + await run_file_io(_cleanup_uploaded_paths, written_paths) raise e except UnsafeUploadPathError as e: logger.warning("Skipping upload with unsafe destination %s: %s", file.filename, e) @@ -333,7 +403,7 @@ async def upload_files( continue except Exception as e: logger.error(f"Failed to upload {file.filename}: {e}") - _cleanup_uploaded_paths(written_paths) + await run_file_io(_cleanup_uploaded_paths, written_paths) raise HTTPException(status_code=500, detail=f"Failed to upload {file.filename}: {str(e)}") # Uploaded files are created with 0o600 permissions (owner read/write only). @@ -343,13 +413,11 @@ async def upload_files( # directory is bind-mounted into the container or synced via # sandbox.update_file. Always add group/other read bits so every sandbox # configuration can read the uploaded content. - for file_path in written_paths: - _make_file_sandbox_readable(file_path) + await run_file_io(_make_uploaded_paths_sandbox_readable, written_paths) if sync_to_sandbox: for file_path, virtual_path in sandbox_sync_targets: - _make_file_sandbox_writable(file_path) - sandbox.update_file(virtual_path, file_path.read_bytes()) + await run_file_io(_sync_upload_to_sandbox, sandbox, file_path, virtual_path) message = f"Successfully uploaded {len(uploaded_files)} file(s)" if skipped_files: @@ -379,16 +447,9 @@ async def get_upload_limits( async def list_uploaded_files(thread_id: str, request: Request) -> UploadListResponse: """List all files in a thread's uploads directory.""" try: - uploads_dir = get_uploads_dir(thread_id) + result = await run_file_io(_list_uploaded_files_for_thread, thread_id, get_effective_user_id()) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) - result = list_files_in_dir(uploads_dir) - enrich_file_listing(result, thread_id) - - # Gateway additionally includes the sandbox-relative path. - sandbox_uploads = get_paths().sandbox_uploads_dir(thread_id, user_id=get_effective_user_id()) - for f in result["files"]: - f["path"] = str(sandbox_uploads / f["filename"]) return UploadListResponse(**result) @@ -398,15 +459,13 @@ async def list_uploaded_files(thread_id: str, request: Request) -> UploadListRes async def delete_uploaded_file(thread_id: str, filename: str, request: Request) -> dict: """Delete a file from a thread's uploads directory.""" try: - uploads_dir = get_uploads_dir(thread_id) - except ValueError as e: - raise HTTPException(status_code=400, detail=str(e)) - try: - return delete_file_safe(uploads_dir, filename, convertible_extensions=CONVERTIBLE_EXTENSIONS) + return await run_file_io(_delete_uploaded_file_for_thread, thread_id, filename, get_effective_user_id()) except FileNotFoundError: raise HTTPException(status_code=404, detail=f"File not found: {filename}") except PathTraversalError: raise HTTPException(status_code=400, detail="Invalid path") + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) except Exception as e: logger.error(f"Failed to delete {filename}: {e}") raise HTTPException(status_code=500, detail=f"Failed to delete {filename}: {str(e)}") diff --git a/backend/packages/harness/deerflow/utils/file_io.py b/backend/packages/harness/deerflow/utils/file_io.py new file mode 100644 index 000000000..b8f31b466 --- /dev/null +++ b/backend/packages/harness/deerflow/utils/file_io.py @@ -0,0 +1,51 @@ +"""Dedicated async offload helper for filesystem work.""" + +from __future__ import annotations + +import asyncio +import atexit +import contextvars +import functools +import logging +import os +from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor + +logger = logging.getLogger(__name__) + + +def _default_file_io_workers() -> int: + raw = os.getenv("DEER_FLOW_FILE_IO_WORKERS") + if raw: + try: + workers = int(raw) + if workers > 0: + return workers + except ValueError: + pass + logger.warning("Invalid DEER_FLOW_FILE_IO_WORKERS value; using default file IO worker count") + return min(32, (os.cpu_count() or 1) + 4) + + +_FILE_IO_EXECUTOR = ThreadPoolExecutor(max_workers=_default_file_io_workers(), thread_name_prefix="file-io") + + +def _shutdown_file_io_executor() -> None: + _FILE_IO_EXECUTOR.shutdown(wait=False, cancel_futures=True) + + +atexit.register(_shutdown_file_io_executor) + + +async def run_file_io[**P, T](func: Callable[P, T], /, *args: P.args, **kwargs: P.kwargs) -> T: + """Run blocking filesystem-oriented work on the dedicated file IO pool. + + ``asyncio.to_thread`` copies ``ContextVar`` values automatically; raw + ``loop.run_in_executor`` does not. Copy the current context explicitly so + user-scoped helpers such as ``get_effective_user_id()`` keep working inside + the worker thread. + """ + loop = asyncio.get_running_loop() + ctx = contextvars.copy_context() + call = functools.partial(func, *args, **kwargs) + return await loop.run_in_executor(_FILE_IO_EXECUTOR, ctx.run, call) diff --git a/backend/tests/blocking_io/test_uploads_router.py b/backend/tests/blocking_io/test_uploads_router.py new file mode 100644 index 000000000..39dd91320 --- /dev/null +++ b/backend/tests/blocking_io/test_uploads_router.py @@ -0,0 +1,135 @@ +"""Regression anchor: uploads router must not block the event loop.""" + +from __future__ import annotations + +import asyncio +from io import BytesIO +from pathlib import Path +from types import SimpleNamespace + +import pytest +from _router_auth_helpers import call_unwrapped +from fastapi import UploadFile + +from app.gateway.routers import uploads +from deerflow.runtime.user_context import get_effective_user_id +from deerflow.uploads.manager import ensure_uploads_dir, get_uploads_dir + +pytestmark = pytest.mark.asyncio + + +class _SandboxRecorder: + def __init__(self) -> None: + self.updates: list[tuple[str, bytes]] = [] + + def update_file(self, path: str, content: bytes) -> None: + self.updates.append((path, content)) + + +class _MountedProvider: + uses_thread_data_mounts = True + + def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str: + raise AssertionError("mounted upload path must not acquire a sandbox") + + async def acquire_async(self, thread_id: str | None = None, *, user_id: str | None = None) -> str: + raise AssertionError("mounted upload path must not acquire a sandbox") + + def get(self, sandbox_id: str): + raise AssertionError("mounted upload path must not read a sandbox") + + +class _RemoteProvider: + uses_thread_data_mounts = False + + def __init__(self) -> None: + self.sandbox = _SandboxRecorder() + self.acquire_async_calls: list[tuple[str | None, str | None]] = [] + + def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str: + raise AssertionError("upload route should use acquire_async") + + async def acquire_async(self, thread_id: str | None = None, *, user_id: str | None = None) -> str: + self.acquire_async_calls.append((thread_id, user_id)) + return "remote-sandbox" + + def get(self, sandbox_id: str): + if sandbox_id == "remote-sandbox": + return self.sandbox + return None + + +def _reset_paths(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path)) + + import deerflow.config.paths as paths_mod + + monkeypatch.setattr(paths_mod, "_paths", None) + + +async def _thread_uploads_dir(thread_id: str, *, user_id: str | None = None) -> Path: + user_id = user_id or get_effective_user_id() + return await asyncio.to_thread(ensure_uploads_dir, thread_id, user_id=user_id) + + +async def test_upload_endpoint_mounted_provider_does_not_block_event_loop(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + _reset_paths(tmp_path, monkeypatch) + provider = _MountedProvider() + monkeypatch.setattr(uploads, "get_sandbox_provider", lambda: provider) + + result = await call_unwrapped( + uploads.upload_files, + "t-mounted", + request=None, + files=[UploadFile(filename="notes.txt", file=BytesIO(b"hello uploads"))], + config=SimpleNamespace(), + ) + + user_id = get_effective_user_id() + target = await asyncio.to_thread(lambda: get_uploads_dir("t-mounted", user_id=user_id) / "notes.txt") + assert result.success is True + assert result.files[0].filename == "notes.txt" + assert await asyncio.to_thread(target.read_bytes) == b"hello uploads" + + +async def test_upload_endpoint_remote_provider_syncs_without_blocking_event_loop(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + _reset_paths(tmp_path, monkeypatch) + provider = _RemoteProvider() + monkeypatch.setattr(uploads, "get_sandbox_provider", lambda: provider) + monkeypatch.setattr(uploads, "get_effective_user_id", lambda: "owner-upload") + + result = await call_unwrapped( + uploads.upload_files, + "t-remote", + request=None, + files=[UploadFile(filename="report.txt", file=BytesIO(b"remote bytes"))], + config=SimpleNamespace(), + ) + + assert result.success is True + assert provider.acquire_async_calls == [("t-remote", "owner-upload")] + assert provider.sandbox.updates == [("/mnt/user-data/uploads/report.txt", b"remote bytes")] + + +async def test_list_uploaded_files_does_not_block_event_loop(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + _reset_paths(tmp_path, monkeypatch) + uploads_dir = await _thread_uploads_dir("t-list") + await asyncio.to_thread((uploads_dir / "notes.txt").write_bytes, b"hello") + + result = await call_unwrapped(uploads.list_uploaded_files, "t-list", request=None) + + assert result.count == 1 + assert result.files[0].filename == "notes.txt" + assert result.files[0].size == len(b"hello") + + +async def test_delete_uploaded_file_does_not_block_event_loop(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + _reset_paths(tmp_path, monkeypatch) + uploads_dir = await _thread_uploads_dir("t-delete") + target = uploads_dir / "notes.txt" + await asyncio.to_thread(target.write_bytes, b"delete me") + + result = await call_unwrapped(uploads.delete_uploaded_file, "t-delete", "notes.txt", request=None) + + assert result == {"success": True, "message": "Deleted notes.txt"} + assert not await asyncio.to_thread(target.exists) diff --git a/backend/tests/support/detectors/thread_boundaries.py b/backend/tests/support/detectors/thread_boundaries.py index 6ea065064..42db69f9a 100644 --- a/backend/tests/support/detectors/thread_boundaries.py +++ b/backend/tests/support/detectors/thread_boundaries.py @@ -78,6 +78,11 @@ EXACT_CALL_RULES: dict[str, _CallRule] = { "ASYNC_THREAD_OFFLOAD", "Offloads synchronous work from an async context into a worker thread.", ), + "deerflow.utils.file_io.run_file_io": _CallRule( + "INFO", + "ASYNC_FILE_IO_OFFLOAD", + "Offloads filesystem-oriented work from an async context into the dedicated file IO thread pool.", + ), "asyncio.new_event_loop": _CallRule( "WARN", "NEW_EVENT_LOOP", diff --git a/backend/tests/test_detect_thread_boundaries.py b/backend/tests/test_detect_thread_boundaries.py index 102613e39..cc48eac38 100644 --- a/backend/tests/test_detect_thread_boundaries.py +++ b/backend/tests/test_detect_thread_boundaries.py @@ -20,6 +20,7 @@ def test_scan_file_detects_async_thread_and_tool_boundaries(tmp_path): import threading import time from concurrent.futures import ThreadPoolExecutor + from deerflow.utils.file_io import run_file_io from langchain.tools import tool from langchain_core.tools import StructuredTool @@ -29,6 +30,7 @@ def test_scan_file_detects_async_thread_and_tool_boundaries(tmp_path): async def handler(model): await asyncio.to_thread(str, "x") + await run_file_io(str, "y") model.invoke("blocking") time.sleep(1) @@ -53,6 +55,7 @@ def test_scan_file_detects_async_thread_and_tool_boundaries(tmp_path): assert async_tool_finding.function == "async_tool" assert async_tool_finding.async_context is True assert "ASYNC_THREAD_OFFLOAD" in categories + assert "ASYNC_FILE_IO_OFFLOAD" in categories assert "SYNC_INVOKE_IN_ASYNC" in categories assert "BLOCKING_CALL_IN_ASYNC" in categories assert "SYNC_ASYNC_BRIDGE" in categories diff --git a/backend/tests/test_file_io.py b/backend/tests/test_file_io.py new file mode 100644 index 000000000..45e649530 --- /dev/null +++ b/backend/tests/test_file_io.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +import contextvars +import threading + +import pytest + +from deerflow.utils.file_io import run_file_io + + +@pytest.mark.anyio +async def test_run_file_io_propagates_contextvars_to_worker() -> None: + marker: contextvars.ContextVar[str] = contextvars.ContextVar("marker", default="missing") + marker.set("owner-1") + + def read_marker() -> tuple[str, str]: + return marker.get(), threading.current_thread().name + + value, thread_name = await run_file_io(read_marker) + + assert value == "owner-1" + assert thread_name.startswith("file-io") + + +@pytest.mark.anyio +async def test_run_file_io_passes_args_and_kwargs() -> None: + def join_values(prefix: str, *, suffix: str) -> str: + return f"{prefix}:{suffix}" + + assert await run_file_io(join_values, "left", suffix="right") == "left:right" diff --git a/backend/tests/test_uploads_router.py b/backend/tests/test_uploads_router.py index 79d42f12a..23290c6e9 100644 --- a/backend/tests/test_uploads_router.py +++ b/backend/tests/test_uploads_router.py @@ -36,6 +36,15 @@ def _mounted_provider() -> MagicMock: return provider +def _symlink_to_or_skip(link_path: Path, target_path: Path) -> None: + try: + link_path.symlink_to(target_path) + except OSError as exc: + if getattr(exc, "winerror", None) == 1314: + pytest.skip("Windows symlink privilege is not available") + raise + + def test_upload_files_writes_thread_storage_and_skips_local_sandbox_sync(tmp_path): thread_uploads_dir = tmp_path / "uploads" thread_uploads_dir.mkdir(parents=True) @@ -178,7 +187,8 @@ def test_upload_files_syncs_non_local_sandbox_and_marks_markdown_file(tmp_path): provider = MagicMock() provider.uses_thread_data_mounts = False - provider.acquire.return_value = "aio-1" + provider.acquire.side_effect = AssertionError("upload route should use acquire_async") + provider.acquire_async = AsyncMock(return_value="aio-1") sandbox = MagicMock() provider.get.return_value = sandbox @@ -216,7 +226,8 @@ def test_upload_files_makes_non_local_files_sandbox_writable(tmp_path): provider = MagicMock() provider.uses_thread_data_mounts = False - provider.acquire.return_value = "aio-1" + provider.acquire.side_effect = AssertionError("upload route should use acquire_async") + provider.acquire_async = AsyncMock(return_value="aio-1") sandbox = MagicMock() provider.get.return_value = sandbox @@ -284,7 +295,8 @@ def test_upload_files_acquires_non_local_sandbox_before_writing(tmp_path): assert user_id == "owner-upload" return "aio-1" - provider.acquire.side_effect = acquire_before_writes + provider.acquire.side_effect = AssertionError("upload route should use acquire_async") + provider.acquire_async = AsyncMock(side_effect=acquire_before_writes) with ( patch.object(uploads, "get_effective_user_id", return_value="owner-upload"), @@ -295,7 +307,8 @@ def test_upload_files_acquires_non_local_sandbox_before_writing(tmp_path): result = asyncio.run(call_unwrapped(uploads.upload_files, "thread-aio", request=MagicMock(), files=[file], config=SimpleNamespace())) assert result.success is True - provider.acquire.assert_called_once_with("thread-aio", user_id="owner-upload") + provider.acquire.assert_not_called() + provider.acquire_async.assert_awaited_once_with("thread-aio", user_id="owner-upload") sandbox.update_file.assert_called_once_with("/mnt/user-data/uploads/notes.txt", b"hello uploads") @@ -305,7 +318,8 @@ def test_upload_files_fails_before_writing_when_non_local_sandbox_unavailable(tm provider = MagicMock() provider.uses_thread_data_mounts = False - provider.acquire.side_effect = RuntimeError("sandbox unavailable") + provider.acquire.side_effect = AssertionError("upload route should use acquire_async") + provider.acquire_async = AsyncMock(side_effect=RuntimeError("sandbox unavailable")) file = ChunkedUpload("notes.txt", [b"hello uploads"]) with ( @@ -317,6 +331,7 @@ def test_upload_files_fails_before_writing_when_non_local_sandbox_unavailable(tm assert list(thread_uploads_dir.iterdir()) == [] assert file.read_calls == [] + provider.acquire.assert_not_called() provider.get.assert_not_called() @@ -390,7 +405,8 @@ def test_upload_files_does_not_sync_non_local_sandbox_when_total_size_exceeds_li provider = MagicMock() provider.uses_thread_data_mounts = False - provider.acquire.return_value = "aio-1" + provider.acquire.side_effect = AssertionError("upload route should use acquire_async") + provider.acquire_async = AsyncMock(return_value="aio-1") sandbox = MagicMock() provider.get.return_value = sandbox @@ -408,7 +424,8 @@ def test_upload_files_does_not_sync_non_local_sandbox_when_total_size_exceeds_li asyncio.run(call_unwrapped(uploads.upload_files, "thread-aio", request=MagicMock(), files=files, config=SimpleNamespace())) assert exc_info.value.status_code == 413 - provider.acquire.assert_called_once_with("thread-aio", user_id="owner-upload") + provider.acquire.assert_not_called() + provider.acquire_async.assert_awaited_once_with("thread-aio", user_id="owner-upload") provider.get.assert_called_once_with("aio-1") sandbox.update_file.assert_not_called() @@ -419,7 +436,8 @@ def test_upload_files_does_not_sync_non_local_sandbox_when_conversion_fails(tmp_ provider = MagicMock() provider.uses_thread_data_mounts = False - provider.acquire.return_value = "aio-1" + provider.acquire.side_effect = AssertionError("upload route should use acquire_async") + provider.acquire_async = AsyncMock(return_value="aio-1") sandbox = MagicMock() provider.get.return_value = sandbox @@ -435,7 +453,8 @@ def test_upload_files_does_not_sync_non_local_sandbox_when_conversion_fails(tmp_ asyncio.run(call_unwrapped(uploads.upload_files, "thread-aio", request=MagicMock(), files=[file], config=SimpleNamespace())) assert exc_info.value.status_code == 500 - provider.acquire.assert_called_once_with("thread-aio", user_id="owner-upload") + provider.acquire.assert_not_called() + provider.acquire_async.assert_awaited_once_with("thread-aio", user_id="owner-upload") provider.get.assert_called_once_with("aio-1") sandbox.update_file.assert_not_called() assert not (thread_uploads_dir / "report.pdf").exists() @@ -559,7 +578,7 @@ def test_upload_files_rejects_preexisting_symlink_destination(tmp_path): thread_uploads_dir.mkdir(parents=True) outside_file = tmp_path / "outside.txt" outside_file.write_text("protected", encoding="utf-8") - (thread_uploads_dir / "victim.txt").symlink_to(outside_file) + _symlink_to_or_skip(thread_uploads_dir / "victim.txt", outside_file) provider = MagicMock() provider.uses_thread_data_mounts = True @@ -584,7 +603,7 @@ def test_upload_files_rejects_dangling_symlink_destination(tmp_path): thread_uploads_dir = tmp_path / "uploads" thread_uploads_dir.mkdir(parents=True) missing_target = tmp_path / "missing-target.txt" - (thread_uploads_dir / "victim.txt").symlink_to(missing_target) + _symlink_to_or_skip(thread_uploads_dir / "victim.txt", missing_target) provider = MagicMock() provider.uses_thread_data_mounts = True