test: skip symlink-planting cases where the host cannot create symlinks (#5414)

* test: skip symlink-planting tests where the host lacks symlink privilege

* test: address symlink helper review feedback

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/tests/support/symlinks.py

@@ -0,0 +1,22 @@
+"""Symlink-planting helper shared by suites that exercise symlink confinement.
+
+Windows only grants symlink creation with Developer Mode enabled or an
+elevated process, so a plain ``symlink_to`` raises ``OSError: [WinError
+1314]`` on a stock Windows contributor machine. Suites that need real
+symlinks skip exactly at the creation point instead of erroring, while hosts
+with the privilege (CI, Linux, Developer Mode) run the full assertion.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+
+def symlink_or_skip(link: Path, target: Path, *, target_is_directory: bool = False) -> None:
+    """Create ``link`` -> ``target``, skipping the test when the host cannot."""
+    try:
+        link.symlink_to(target, target_is_directory=target_is_directory)
+    except OSError:
+        pytest.skip("symlink creation is unavailable on this host (on Windows it requires Developer Mode or elevation)")

backend/tests/test_channel_file_attachments.py

@@ -7,7 +7,7 @@ import os
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
-import pytest
+from support.symlinks import symlink_or_skip
 
 from app.channels.base import Channel
 from app.channels.message_bus import InboundMessage, MessageBus, OutboundMessage, ResolvedAttachment
@@ -237,10 +237,7 @@ class TestResolveAttachments:
         uploads_dir.mkdir(parents=True)
         victim = uploads_dir / "secret.pdf"
         victim.write_bytes(b"%PDF-1.4 secret")
-        try:
-            (outputs_dir / "report.pdf").symlink_to(victim)
-        except OSError:
-            pytest.skip("symlinks are unavailable on this platform")
+        symlink_or_skip(outputs_dir / "report.pdf", victim)
 
         with patch("app.gateway.path_utils.get_paths", return_value=paths):
             result = _resolve_attachments("t1", ["/mnt/user-data/outputs/report.pdf"], user_id="owner-1")
@@ -316,7 +313,7 @@ class TestInboundFileIngestion:
         uploads_dir = tmp_path / "uploads"
         uploads_dir.mkdir()
         outside_file = tmp_path / "outside-created.txt"
-        (uploads_dir / "victim.txt").symlink_to(outside_file)
+        symlink_or_skip(uploads_dir / "victim.txt", outside_file)
 
         msg = InboundMessage(
             channel_name="test-channel",
@@ -345,7 +342,7 @@ class TestInboundFileIngestion:
         uploads_dir = tmp_path / "uploads"
         uploads_dir.mkdir()
         missing_target = tmp_path / "missing-created.txt"
-        (uploads_dir / "victim.txt").symlink_to(missing_target)
+        symlink_or_skip(uploads_dir / "victim.txt", missing_target)
 
         msg = InboundMessage(
             channel_name="test-channel",

backend/tests/test_dingtalk_channel.py

@@ -8,6 +8,7 @@ from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
+from support.symlinks import symlink_or_skip
 
 from app.channels.commands import KNOWN_CHANNEL_COMMANDS
 from app.channels.dingtalk import (
@@ -2310,7 +2311,7 @@ class TestReceiveFile:
             uploads = tmp_path / "uploads"
             uploads.mkdir()
             outside = tmp_path / "outside.txt"
-            (uploads / "image.png").symlink_to(outside)
+            symlink_or_skip(uploads / "image.png", outside)
             _patch_uploads(monkeypatch, uploads)
             channel._download_by_code = AsyncMock(return_value=b"PWNED")
 

backend/tests/test_extension_manager.py

@@ -16,6 +16,7 @@ from pathlib import Path
 
 import pytest
 import yaml
+from support.symlinks import symlink_or_skip
 
 from deerflow.extensions.cli import find_project_root
 from deerflow.extensions.loader import ExtensionSpec
@@ -1636,7 +1637,7 @@ def test_local_install_rejects_symlinks_before_copying_or_resolving(tmp_path: Pa
     _write_local_extension(source)
     outside = tmp_path / "operator-secret.txt"
     outside.write_text("do not vendor me", encoding="utf-8")
-    (source / "linked-secret.txt").symlink_to(outside)
+    symlink_or_skip(source / "linked-secret.txt", outside)
     original_pyproject = (root / "backend" / "pyproject.toml").read_bytes()
 
     with pytest.raises(ValueError, match="symbolic links"):

backend/tests/test_feishu_parser.py

@@ -6,6 +6,7 @@ from pathlib import Path
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
+from support.symlinks import symlink_or_skip
 
 from app.channels import feishu as feishu_module
 from app.channels.commands import KNOWN_CHANNEL_COMMANDS
@@ -248,7 +249,7 @@ def test_feishu_receive_file_does_not_follow_planted_symlink(tmp_path, monkeypat
         uploads = paths.sandbox_uploads_dir("thread-1", user_id="ou-user")
         victim = tmp_path / "victim.txt"
         victim.write_bytes(b"SAFE")
-        (uploads / "report.txt").symlink_to(victim)
+        symlink_or_skip(uploads / "report.txt", victim)
 
         channel = _feishu_file_channel(_feishu_file_response("report.txt", b"PAYLOAD"))
         provider = MagicMock()
@@ -273,7 +274,7 @@ def test_feishu_receive_file_reserves_dangling_symlink_name(tmp_path, monkeypatc
         paths.ensure_thread_dirs("thread-1", user_id="ou-user")
         uploads = paths.sandbox_uploads_dir("thread-1", user_id="ou-user")
         missing_target = tmp_path / "missing.txt"
-        (uploads / "report.txt").symlink_to(missing_target)
+        symlink_or_skip(uploads / "report.txt", missing_target)
 
         channel = _feishu_file_channel(_feishu_file_response("report.txt", b"PAYLOAD"))
         provider = MagicMock()

backend/tests/test_gateway_path_utils.py

@@ -12,6 +12,7 @@ from pathlib import Path
 
 import pytest
 from fastapi import HTTPException
+from support.symlinks import symlink_or_skip
 
 from app.gateway.path_utils import OUTPUTS_VIRTUAL_ROOT, normalize_outputs_virtual_path, resolve_outputs_confined_path
 from deerflow.config.paths import Paths
@@ -32,13 +33,6 @@ def thread_dirs(tmp_path, monkeypatch) -> tuple[Path, Path]:
     return outputs, uploads
 
 
-def _symlink_or_skip(link: Path, target: Path) -> None:
-    try:
-        link.symlink_to(target)
-    except OSError:
-        pytest.skip("symlinks are unavailable on this platform")
-
-
 class TestNormalizeOutputsVirtualPath:
     def test_returns_canonical_absolute_virtual_path(self) -> None:
         assert normalize_outputs_virtual_path("mnt/user-data/outputs/report.md") == "/mnt/user-data/outputs/report.md"
@@ -100,7 +94,7 @@ class TestResolveOutputsConfinedPath:
         outputs, uploads = thread_dirs
         victim = uploads / "victim.txt"
         victim.write_text("before", encoding="utf-8")
-        _symlink_or_skip(outputs / "linked.txt", victim)
+        symlink_or_skip(outputs / "linked.txt", victim)
 
         with pytest.raises(HTTPException) as exc_info:
             resolve_outputs_confined_path(THREAD_ID, "/mnt/user-data/outputs/linked.txt", user_id=USER_ID)
@@ -113,7 +107,7 @@ class TestResolveOutputsConfinedPath:
         outputs, _ = thread_dirs
         outside = tmp_path / "outside.txt"
         outside.write_text("outside", encoding="utf-8")
-        _symlink_or_skip(outputs / "linked.txt", outside)
+        symlink_or_skip(outputs / "linked.txt", outside)
 
         with pytest.raises(HTTPException) as exc_info:
             resolve_outputs_confined_path(THREAD_ID, "/mnt/user-data/outputs/linked.txt", user_id=USER_ID)
@@ -129,7 +123,7 @@ class TestResolveOutputsConfinedPath:
         real_outputs.mkdir()
         (real_outputs / "report.md").write_text("hello", encoding="utf-8")
         outputs.rmdir()
-        _symlink_or_skip(outputs, real_outputs)
+        symlink_or_skip(outputs, real_outputs)
 
         with pytest.raises(HTTPException) as exc_info:
             resolve_outputs_confined_path(THREAD_ID, "/mnt/user-data/outputs/report.md", user_id=USER_ID)

backend/tests/test_local_skill_storage_write.py

@@ -2,11 +2,11 @@
 
 from __future__ import annotations
 
-import os
 import stat
 from unittest.mock import patch
 
 import pytest
+from support.symlinks import symlink_or_skip
 
 from deerflow.config.paths import Paths
 from deerflow.skills.storage import get_or_new_skill_storage, reset_skill_storage
@@ -151,7 +151,7 @@ def test_rejects_dotdot_only(storage):
 def test_rejects_symlink_pointing_outside(tmp_path, storage, skill_dir):
     outside = tmp_path / "outside.txt"
     link = skill_dir / "escape_link.txt"
-    os.symlink(outside, link)
+    symlink_or_skip(link, outside)
     with pytest.raises(ValueError, match="skill directory"):
         storage.write_custom_skill("demo-skill", "escape_link.txt", "x")
 
@@ -160,7 +160,7 @@ def test_rejects_symlink_dir_pointing_outside(tmp_path, storage, skill_dir):
     outside_dir = tmp_path / "outside_dir"
     outside_dir.mkdir()
     link_dir = skill_dir / "linked_dir"
-    os.symlink(outside_dir, link_dir)
+    symlink_or_skip(link_dir, outside_dir)
     with pytest.raises(ValueError, match="skill directory"):
         storage.write_custom_skill("demo-skill", "linked_dir/file.txt", "x")
 
@@ -175,7 +175,7 @@ def test_allows_symlink_within_skill_dir(tmp_path, storage, skill_dir):
     real_file = skill_dir / "real.md"
     real_file.write_text("real")
     link = skill_dir / "alias.md"
-    os.symlink(real_file, link)
+    symlink_or_skip(link, real_file)
     # Should not raise
     storage.write_custom_skill("demo-skill", "alias.md", "updated")
     # resolve() writes through to the real target file

backend/tests/test_sandbox_search_tools.py

@@ -2,6 +2,8 @@ import json
 from types import SimpleNamespace
 from unittest.mock import patch
 
+from support.symlinks import symlink_or_skip
+
 from deerflow.community.aio_sandbox.aio_sandbox import AioSandbox
 from deerflow.config.paths import Paths
 from deerflow.sandbox.local.local_sandbox import LocalSandbox, PathMapping
@@ -358,7 +360,7 @@ def test_find_grep_matches_skips_symlink_outside_root(tmp_path) -> None:
     workspace.mkdir()
     outside = tmp_path / "outside.txt"
     outside.write_text("TODO outside\n", encoding="utf-8")
-    (workspace / "outside-link.txt").symlink_to(outside)
+    symlink_or_skip(workspace / "outside-link.txt", outside)
 
     matches, truncated = find_grep_matches(workspace, "TODO")
 

backend/tests/test_skill_review_waivers.py

@@ -20,6 +20,7 @@ from skill_review_waivers import (
     parse_manifest,
     validate_manifest_against_facts,
 )
+from support.symlinks import symlink_or_skip
 
 from deerflow.skills.review.analyzer import analyze_skill_package
 from deerflow.skills.review.readers import LocalDirectoryReader
@@ -221,7 +222,7 @@ def test_matching_waiver_rejects_symlinked_package_outside_repository(tmp_path:
     external_target, digest = _write_target(external_root)
     public_root = tmp_path / "skills/public"
     public_root.mkdir(parents=True)
-    (public_root / "demo").symlink_to(external_target.parents[1], target_is_directory=True)
+    symlink_or_skip(public_root / "demo", external_target.parents[1], target_is_directory=True)
 
     assert (
         matching_waiver(

backend/tests/test_support_symlinks.py

@@ -0,0 +1,29 @@
+"""Pin the shared symlink-planting helper's two contracts (TDD for test infra)."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+from support.symlinks import symlink_or_skip
+
+
+def test_creates_real_symlink_when_the_host_allows_it(tmp_path: Path) -> None:
+    target = tmp_path / "real.txt"
+    target.write_text("payload", encoding="utf-8")
+    link = tmp_path / "link.txt"
+
+    symlink_or_skip(link, target)  # skips itself on hosts without the privilege
+
+    assert link.is_symlink()
+    assert link.read_text(encoding="utf-8") == "payload"
+
+
+def test_skips_when_symlink_creation_is_denied(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    def _denied(self: Path, target: Path, *, target_is_directory: bool = False) -> None:
+        raise OSError(1314, "privilege not held")
+
+    monkeypatch.setattr(Path, "symlink_to", _denied)
+
+    with pytest.raises(pytest.skip.Exception):
+        symlink_or_skip(tmp_path / "link.txt", tmp_path / "real.txt")