deer-flow/backend/tests/test_extended_length_paths.py
Lym308758 346ff7b253
fix(memory): rank capacity eviction per record, not per fact_id (#7454)
* fix(memory): rank capacity eviction per record, not per fact_id

select_facts_for_capacity keyed every score by fact_id, so an imported
batch that repeats an id collapsed into a single dict entry: the last
record's score won, and both the ranking pass and the evicted audit
read that one value for every position. A 0.9-confidence record could
therefore be evicted while its own 0.1-confidence duplicate survived.
The import path keeps caller-supplied ids without checking them for
uniqueness, and any over-cap import lands here.

Score by position instead: a parallel list next to the indexed facts,
so ranking and the audit read each record's own score. The public
``scores`` dict stays keyed by fact_id with last-write-wins
semantics, keeping the DeerMem benchmark contract and the callers
that never read it unchanged.

Red on main: test_confidence_policy_scores_duplicate_fact_ids_per_record
fails with the highest-confidence record evicted; green on this branch.

* fix(utils): keep share-less UNC roots out of the extended-length prefix

extended_length_path prefixed every absolute UNC path with
\\?\UNC\, but Win32 only accepts \\?\UNC\server\share\... — a
share-less root (\\server, \\server\, or a bare \) is not a valid
device path, so every Win32 call handed the prefixed spelling fails.
Only prefix when both the server and the share are present;
otherwise return the ordinary spelling unchanged.

Red on main: the \\server, \\server\, //, and //server cases of
test_windows_extended_spelling assert the forged prefix; green on
this branch.
2026-10-25 13:20:15 +08:00

77 lines
2.9 KiB
Python

"""Windows filesystem spelling and document confinement regressions."""
import ntpath
import os
from pathlib import Path
from types import SimpleNamespace
import pytest
from deerflow.config.paths import Paths
from deerflow.utils import host_paths
@pytest.mark.parametrize(
("raw", "expected"),
[
(r"C:\shelf\..\documents", r"\\?\C:\documents"),
("C:/shelf/documents", r"\\?\C:\shelf\documents"),
(r"\\server\share\shelf", r"\\?\UNC\server\share\shelf"),
(r"\\server", r"\\server"),
("\\\\server\\", "\\\\server\\"),
(r"//", r"\\"),
(r"//server", r"\\server"),
(r"\\?\C:\shelf", r"\\?\C:\shelf"),
(r"\\?\UNC\server\share\shelf", r"\\?\UNC\server\share\shelf"),
(r"\\.\C:\shelf", r"\\.\C:\shelf"),
(r"\\.\pipe\deerflow", r"\\.\pipe\deerflow"),
(r"\\?\C:\shelf\..\documents", r"\\?\C:\shelf\..\documents"),
],
)
def test_windows_extended_spelling(monkeypatch, raw, expected):
# Patch this module's OS binding, never the process-wide os.name (pathlib
# and pytest need to keep using the actual host's path implementation).
monkeypatch.setattr(host_paths, "os", SimpleNamespace(name="nt", path=ntpath))
assert str(host_paths.extended_length_path(Path(raw))) == expected
def test_posix_path_is_unchanged(monkeypatch):
monkeypatch.setattr(host_paths, "os", SimpleNamespace(name="posix"))
path = Path("relative/documents")
assert host_paths.extended_length_path(path) is path
@pytest.mark.skipif(os.name != "nt", reason="native Windows relative-path normalization")
def test_windows_relative_path_is_absolutized(monkeypatch, tmp_path):
monkeypatch.chdir(tmp_path)
actual = host_paths.extended_length_path(Path("shelf/../documents"))
assert str(actual) == "\\\\?\\" + str(tmp_path / "documents")
def test_document_traversal_is_still_rejected(tmp_path):
paths = Paths(base_dir=tmp_path)
with pytest.raises(ValueError, match="path traversal"):
paths.project_document_path("alice", "../outside")
with pytest.raises(ValueError, match="path traversal"):
paths.project_document_path("alice", str(tmp_path / "outside"))
@pytest.mark.parametrize("outside_root", [False, True])
def test_document_symlink_confinement(tmp_path, outside_root):
paths = Paths(base_dir=tmp_path)
base = paths.user_projects_dir("alice")
base.mkdir(parents=True)
target = tmp_path / "outside" if outside_root else base / "target"
target.mkdir()
try:
(base / "link").symlink_to(target, target_is_directory=True)
except OSError:
if os.name == "nt":
pytest.skip("Windows symlink privilege is not available")
raise
if outside_root:
with pytest.raises(ValueError, match="path traversal"):
paths.project_document_path("alice", "link/content")
else:
assert paths.project_document_path("alice", "link/content") == (target / "content").resolve()