forkless/strix
1
0
Fork 0
mirror of https://github.com/usestrix/strix.git synced 2026-08-24 11:52:38 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat(runtime): mount local targets instead of copying them in (#958)

Browse Source 
Co-authored-by: Ahmed Allam <ahmed39652003@gmail.com>
This commit is contained in:
devin-ai-integration[bot]
2026-08-02 07:45:10 -07:00
committed by GitHub
co-authored by Ahmed Allam
parent b6cf156e95
commit dbc427d816
21 changed files with 567 additions and 715 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_cli_target_list.py
+2 -7
View File
@@ -30,9 +30,7 @@ def test_parse_arguments_accepts_target_list_file(
) -> None:
target_list = tmp_path / "targets.txt"
target_list.write_text(
"https://test1.com/\n"
"\n"
"http://test2.com:5789/\n",
"https://test1.com/\n\nhttp://test2.com:5789/\n",
encoding="utf-8",
)
_stub_settings(monkeypatch)
@@ -84,7 +82,4 @@ def test_parse_arguments_rejects_resume_with_target_list(
with pytest.raises(SystemExit):
cli_main.parse_arguments()
assert (
"Cannot combine --resume with --target/--target-list/--mount"
in capsys.readouterr().err
)
assert "Cannot combine --resume with --target/--target-list" in capsys.readouterr().err
tests/test_config_loader.py
-1
View File
@@ -33,7 +33,6 @@ _LLM_ENV_KEYS = [
# RuntimeSettings
"STRIX_IMAGE",
"STRIX_RUNTIME_BACKEND",
"STRIX_MAX_LOCAL_COPY_MB",
# TelemetrySettings
"STRIX_TELEMETRY",
]
tests/test_local_dir_staging.py
-143
View File
@@ -1,143 +0,0 @@
"""Tests for symlink-safe LocalDir staging."""
from __future__ import annotations
from typing import TYPE_CHECKING
from strix.runtime.local_dir_staging import stage_symlink_safe_dir, tree_has_symlink
if TYPE_CHECKING:
from pathlib import Path
def _make_repo(tmp_path: Path) -> Path:
repo = tmp_path / "repo"
(repo / "pkg").mkdir(parents=True)
(repo / "pkg" / "mod.py").write_text("x = 1\n")
(repo / "README.md").write_text("readme\n")
return repo
def test_tree_without_symlinks_used_as_is(tmp_path: Path) -> None:
repo = _make_repo(tmp_path)
upload_path, staged = stage_symlink_safe_dir(repo)
assert staged is None
assert upload_path == repo.resolve()
assert not tree_has_symlink(repo)
def test_in_tree_file_symlink_is_dereferenced(tmp_path: Path) -> None:
repo = _make_repo(tmp_path)
(repo / "link.py").symlink_to(repo / "pkg" / "mod.py")
upload_path, staged = stage_symlink_safe_dir(repo)
assert staged is not None
assert upload_path == staged
assert not (staged / "link.py").is_symlink()
assert (staged / "link.py").read_text() == "x = 1\n"
assert (staged / "pkg" / "mod.py").read_text() == "x = 1\n"
assert not tree_has_symlink(staged)
def test_in_tree_relative_dir_symlink_is_dereferenced(tmp_path: Path) -> None:
repo = _make_repo(tmp_path)
(repo / "pkg_alias").symlink_to("pkg")
_upload, staged = stage_symlink_safe_dir(repo)
assert staged is not None
assert (staged / "pkg_alias" / "mod.py").read_text() == "x = 1\n"
assert not tree_has_symlink(staged)
def test_out_of_tree_symlink_is_dropped(tmp_path: Path) -> None:
repo = _make_repo(tmp_path)
outside = tmp_path / "outside.txt"
outside.write_text("secret\n")
(repo / "escape.txt").symlink_to(outside)
(repo / "abs_escape").symlink_to("/etc")
_upload, staged = stage_symlink_safe_dir(repo)
assert staged is not None
assert not (staged / "escape.txt").exists()
assert not (staged / "abs_escape").exists()
assert (staged / "README.md").exists()
def test_dangling_symlink_is_dropped(tmp_path: Path) -> None:
repo = _make_repo(tmp_path)
(repo / "dangling").symlink_to(repo / "does-not-exist")
_upload, staged = stage_symlink_safe_dir(repo)
assert staged is not None
assert not (staged / "dangling").exists()
assert not (staged / "dangling").is_symlink()
def test_cyclic_symlink_terminates(tmp_path: Path) -> None:
repo = _make_repo(tmp_path)
(repo / "self").symlink_to(repo)
(repo / "pkg" / "up").symlink_to("..")
_upload, staged = stage_symlink_safe_dir(repo)
assert staged is not None
assert (staged / "README.md").exists()
assert not tree_has_symlink(staged)
def test_nested_symlinks_inside_linked_dir(tmp_path: Path) -> None:
repo = _make_repo(tmp_path)
shared = repo / "shared"
shared.mkdir()
(shared / "conf.json").write_text("{}\n")
(shared / "escape").symlink_to("/etc/passwd")
(repo / "pkg" / "shared_link").symlink_to(shared)
_upload, staged = stage_symlink_safe_dir(repo)
assert staged is not None
assert (staged / "pkg" / "shared_link" / "conf.json").read_text() == "{}\n"
assert not (staged / "pkg" / "shared_link" / "escape").exists()
assert not (staged / "shared" / "escape").exists()
def test_staged_path_has_no_symlink_ancestor(tmp_path: Path, monkeypatch) -> None: # noqa: ANN001
"""The staging directory itself must never sit behind a symlink.
``tempfile.mkdtemp()`` honors ``$TMPDIR``, and on macOS the default
``$TMPDIR`` resolves through ``/var``, which is itself a symlink to
``/private/var``. ``LocalDir`` rejects any symlink component in its
source path, so returning the raw ``mkdtemp()`` result breaks every
local-dir upload on macOS whenever the source tree contains a symlink.
This reproduces that shape without depending on the host OS layout.
"""
repo = _make_repo(tmp_path)
(repo / "link.py").symlink_to(repo / "pkg" / "mod.py")
real_tmp_root = tmp_path / "real_tmp"
real_tmp_root.mkdir()
symlinked_tmp_root = tmp_path / "tmp_symlink"
symlinked_tmp_root.symlink_to(real_tmp_root)
def fake_mkdtemp(prefix: str = "") -> str:
real_dir = real_tmp_root / f"{prefix}fake"
real_dir.mkdir()
return str(symlinked_tmp_root / real_dir.name)
monkeypatch.setattr(
"strix.runtime.local_dir_staging.tempfile.mkdtemp", fake_mkdtemp
)
upload_path, staged = stage_symlink_safe_dir(repo)
assert staged is not None
assert upload_path == staged
for path in (staged, *staged.parents):
assert not path.is_symlink(), f"staged path has a symlink ancestor: {path}"
tests/test_local_sources.py
+105 -157
View File
@@ -1,171 +1,138 @@
"""Tests for local-source sizing and ``--mount`` target helpers in interface.utils."""
"""Tests for local-source collection and mount policy in interface.utils."""
from __future__ import annotations
import logging
import os
import sys
from typing import TYPE_CHECKING, Any
from pathlib import Path
from typing import Any
import pytest
if TYPE_CHECKING:
from pathlib import Path
from strix.interface.utils import (
build_mount_targets_info,
check_mountable_dir,
collect_local_sources,
dedupe_local_targets,
directory_size_bytes,
find_oversized_local_targets,
infer_target_type,
read_target_list_file,
)
def _write_file(path: Path, size: int) -> None:
path.write_bytes(b"x" * size)
def _local_target(target_path: str) -> dict[str, Any]:
return {
"type": "local_code",
"details": {"target_path": target_path, "workspace_subdir": "repo"},
"original": target_path,
}
def _local_target(target_path: str, *, mount: bool = False) -> dict[str, Any]:
details: dict[str, Any] = {"target_path": target_path, "workspace_subdir": "repo"}
if mount:
details["mount"] = True
return {"type": "local_code", "details": details, "original": target_path}
def test_collect_local_sources_protects_the_users_own_git() -> None:
sources = collect_local_sources([_local_target("/code")])
assert sources == [
{"source_path": "/code", "workspace_subdir": "repo", "protect_metadata": True}
]
def test_directory_size_empty_dir_is_zero(tmp_path: Path) -> None:
assert directory_size_bytes(tmp_path) == 0
def test_directory_size_sums_flat_and_nested_files(tmp_path: Path) -> None:
_write_file(tmp_path / "a.txt", 100)
nested = tmp_path / "sub" / "deep"
nested.mkdir(parents=True)
_write_file(nested / "b.txt", 250)
assert directory_size_bytes(tmp_path) == 350
def test_directory_size_skips_symlinks(tmp_path: Path) -> None:
_write_file(tmp_path / "real.txt", 100)
(tmp_path / "link.txt").symlink_to(tmp_path / "real.txt")
# The symlink target is counted once via the real file, not doubled.
assert directory_size_bytes(tmp_path) == 100
@pytest.mark.skipif(sys.platform == "win32", reason="relies on POSIX permissions")
def test_directory_size_logs_and_skips_unreadable_subdir(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
if hasattr(os, "geteuid") and os.geteuid() == 0:
pytest.skip("root bypasses directory permissions")
_write_file(tmp_path / "top.txt", 100)
locked = tmp_path / "locked"
locked.mkdir()
_write_file(locked / "secret.bin", 9999)
locked.chmod(0o000)
try:
with caplog.at_level(logging.WARNING):
size = directory_size_bytes(tmp_path)
finally:
locked.chmod(0o755)
# The unreadable subtree is excluded (not silently treated as readable) and
# the omission is logged rather than vanishing without a trace.
assert size == 100
assert any("Could not read" in record.message for record in caplog.records)
def test_find_oversized_returns_nothing_under_limit(tmp_path: Path) -> None:
_write_file(tmp_path / "a.txt", 100)
targets = [_local_target(str(tmp_path))]
assert find_oversized_local_targets(targets, max_bytes=1000) == []
def test_find_oversized_returns_target_over_limit(tmp_path: Path) -> None:
_write_file(tmp_path / "big.bin", 500)
targets = [_local_target(str(tmp_path))]
result = find_oversized_local_targets(targets, max_bytes=100)
assert result == [(str(tmp_path), 500)]
def test_find_oversized_ignores_mounted_targets(tmp_path: Path) -> None:
_write_file(tmp_path / "big.bin", 500)
targets = [_local_target(str(tmp_path), mount=True)]
assert find_oversized_local_targets(targets, max_bytes=100) == []
def test_find_oversized_ignores_non_local_targets() -> None:
targets = [{"type": "web_application", "details": {"target_url": "https://x"}}]
assert find_oversized_local_targets(targets, max_bytes=1) == []
@pytest.mark.parametrize("disabled", [0, -1])
def test_find_oversized_disabled_for_non_positive_limit(tmp_path: Path, disabled: int) -> None:
_write_file(tmp_path / "big.bin", 500)
targets = [_local_target(str(tmp_path))]
assert find_oversized_local_targets(targets, max_bytes=disabled) == []
def test_collect_local_sources_propagates_mount_flag() -> None:
copied = _local_target("/copied")
copied["details"]["workspace_subdir"] = "copied"
mounted = _local_target("/mounted", mount=True)
mounted["details"]["workspace_subdir"] = "mounted"
sources = collect_local_sources([copied, mounted])
by_path = {s["source_path"]: s for s in sources}
assert by_path["/copied"]["mount"] is False
assert by_path["/mounted"]["mount"] is True
def test_collect_local_sources_repository_is_never_mounted() -> None:
def test_collect_local_sources_leaves_a_clone_writable() -> None:
repo = {
"type": "repository",
"details": {"cloned_repo_path": "/clone", "workspace_subdir": "clone"},
}
sources = collect_local_sources([repo])
assert sources == [{"source_path": "/clone", "workspace_subdir": "clone", "mount": False}]
assert sources == [
{"source_path": "/clone", "workspace_subdir": "clone", "protect_metadata": False}
]
def test_build_mount_targets_info_for_valid_dir(tmp_path: Path) -> None:
result = build_mount_targets_info([str(tmp_path)])
assert len(result) == 1
entry = result[0]
assert entry["type"] == "local_code"
assert entry["details"]["mount"] is True
assert entry["details"]["target_path"] == str(tmp_path.resolve())
def test_check_mountable_dir_accepts_a_project_dir(tmp_path: Path) -> None:
check_mountable_dir(tmp_path)
def test_build_mount_targets_info_rejects_missing_path(tmp_path: Path) -> None:
missing = tmp_path / "does-not-exist"
def test_check_mountable_dir_rejects_missing_path(tmp_path: Path) -> None:
with pytest.raises(ValueError, match="not an existing directory"):
build_mount_targets_info([str(missing)])
check_mountable_dir(tmp_path / "nope")
def test_build_mount_targets_info_rejects_file(tmp_path: Path) -> None:
file_path = tmp_path / "a-file.txt"
_write_file(file_path, 10)
with pytest.raises(ValueError, match="not an existing directory"):
build_mount_targets_info([str(file_path)])
def test_check_mountable_dir_rejects_filesystem_root() -> None:
with pytest.raises(ValueError, match="Refusing to mount"):
check_mountable_dir(Path("/"))
@pytest.mark.parametrize("empty", ["", " "])
def test_build_mount_targets_info_rejects_empty_path(empty: str) -> None:
# An empty path would otherwise resolve to the current working directory
# and silently bind-mount it into the sandbox.
with pytest.raises(ValueError, match="must not be empty"):
build_mount_targets_info([empty])
def test_check_mountable_dir_rejects_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
home = tmp_path / "home"
home.mkdir()
monkeypatch.setenv("HOME", str(home))
monkeypatch.setattr(Path, "home", classmethod(lambda _cls: home))
with pytest.raises(ValueError, match="Refusing to mount"):
check_mountable_dir(home)
def test_check_mountable_dir_rejects_system_root() -> None:
etc = Path("/etc")
if not etc.is_dir():
pytest.skip("no /etc on this platform")
with pytest.raises(ValueError, match="Refusing to mount"):
check_mountable_dir(etc)
def test_check_mountable_dir_rejects_the_shared_home_root() -> None:
home_root = Path("/home")
if not home_root.is_dir():
pytest.skip("no /home on this platform")
with pytest.raises(ValueError, match="Refusing to mount"):
check_mountable_dir(home_root)
def test_check_mountable_dir_matches_forbidden_names_case_insensitively(tmp_path: Path) -> None:
ssh_dir = tmp_path / ".SSH"
ssh_dir.mkdir()
with pytest.raises(ValueError, match="holds credentials"):
check_mountable_dir(ssh_dir)
def test_check_mountable_dir_rejects_credential_dirs(tmp_path: Path) -> None:
ssh_dir = tmp_path / ".ssh"
ssh_dir.mkdir()
with pytest.raises(ValueError, match="holds credentials"):
check_mountable_dir(ssh_dir)
def test_check_mountable_dir_rejects_credential_subdirs(tmp_path: Path) -> None:
keys = tmp_path / ".ssh" / "keys"
keys.mkdir(parents=True)
with pytest.raises(ValueError, match="holds credentials"):
check_mountable_dir(keys)
def test_check_mountable_dir_rejects_system_subdirs() -> None:
system_subdir = next((p for p in (Path("/etc/ssl"), Path("/usr/bin")) if p.is_dir()), None)
if system_subdir is None:
pytest.skip("no system subdirectory on this platform")
with pytest.raises(ValueError, match="Refusing to mount"):
check_mountable_dir(system_subdir)
def test_check_mountable_dir_accepts_a_project_under_the_home_root(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
project = tmp_path / "home" / "dev" / "project"
project.mkdir(parents=True)
monkeypatch.setattr(Path, "home", classmethod(lambda _cls: tmp_path / "home" / "dev"))
check_mountable_dir(project)
def test_infer_target_type_applies_the_mount_policy() -> None:
with pytest.raises(ValueError, match="Refusing to mount"):
infer_target_type("/etc")
def test_read_target_list_file_strips_blank_lines(tmp_path: Path) -> None:
target_list = tmp_path / "targets.txt"
target_list.write_text(
"\n"
" https://test1.com/ \n"
"\n"
"http://test2.com:5789/\n"
" \n",
"\n https://test1.com/ \n\nhttp://test2.com:5789/\n \n",
encoding="utf-8",
)
@@ -178,10 +145,7 @@ def test_read_target_list_file_strips_blank_lines(tmp_path: Path) -> None:
def test_read_target_list_file_ignores_comment_lines(tmp_path: Path) -> None:
target_list = tmp_path / "targets.txt"
target_list.write_text(
"# production targets\n"
"https://test1.com/\n"
" # staging targets\n"
"http://test2.com:5789/\n",
"# production targets\nhttps://test1.com/\n # staging targets\nhttp://test2.com:5789/\n",
encoding="utf-8",
)
@@ -222,28 +186,12 @@ def test_dedupe_keeps_distinct_targets_in_order() -> None:
targets = [
_local_target("/a"),
{"type": "web_application", "details": {"target_url": "https://x"}},
_local_target("/b", mount=True),
_local_target("/b"),
]
assert dedupe_local_targets(targets) == targets
def test_dedupe_mount_supersedes_copied_same_path() -> None:
copied = _local_target("/repo")
mounted = _local_target("/repo", mount=True)
# Copied first, then mounted: the single surviving entry is the mount.
result = dedupe_local_targets([copied, mounted])
assert len(result) == 1
assert result[0]["details"]["mount"] is True
# Order-independent: mounted first, copied second also yields the mount.
result_rev = dedupe_local_targets([mounted, copied])
assert len(result_rev) == 1
assert result_rev[0]["details"]["mount"] is True
def test_dedupe_collapses_duplicate_mounts() -> None:
result = dedupe_local_targets(
[_local_target("/repo", mount=True), _local_target("/repo", mount=True)]
)
assert len(result) == 1
def test_dedupe_collapses_the_same_path() -> None:
assert dedupe_local_targets([_local_target("/repo"), _local_target("/repo")]) == [
_local_target("/repo")
]
tests/test_session_entries.py
+147 -54
View File
@@ -1,4 +1,4 @@
"""Tests for build_session_entries: splitting copied vs bind-mounted sources."""
"""Tests for how local sources reach the sandbox: bind mounts or manifest upload."""
from __future__ import annotations
@@ -6,82 +6,175 @@ from typing import TYPE_CHECKING, Any
from agents.sandbox.entries import LocalDir
from strix.runtime.session_manager import build_session_entries
from strix.runtime.backends import (
_BACKENDS,
_BIND_MOUNT_BACKENDS,
backend_supports_bind_mounts,
register_backend,
)
from strix.runtime.session_manager import build_bind_mounts, build_manifest_entries
if TYPE_CHECKING:
from pathlib import Path
def _source(subdir: str, path: str, *, mount: bool = False) -> dict[str, Any]:
return {"source_path": path, "workspace_subdir": subdir, "mount": mount}
def _source(subdir: str, path: str, *, protect_metadata: bool = False) -> dict[str, Any]:
return {"source_path": path, "workspace_subdir": subdir, "protect_metadata": protect_metadata}
def test_copied_source_becomes_localdir_entry(tmp_path: Path) -> None:
entries, bind_mounts, staged_dirs = build_session_entries([_source("repo", str(tmp_path))])
assert bind_mounts == []
assert staged_dirs == []
assert isinstance(entries["repo"], LocalDir)
assert entries["repo"].src == tmp_path.resolve()
def test_mounted_source_becomes_bind_mount(tmp_path: Path) -> None:
entries, bind_mounts, _staged = build_session_entries(
[_source("repo", str(tmp_path), mount=True)]
)
assert entries == {}
assert bind_mounts == [
def test_source_becomes_writable_bind_mount(tmp_path: Path) -> None:
assert build_bind_mounts([_source("repo", str(tmp_path))]) == [
{
"source": str(tmp_path.resolve()),
"target": "/workspace/repo",
"read_only": True,
"read_only": False,
}
]
def test_mixed_sources_split_correctly(tmp_path: Path) -> None:
copied = tmp_path / "copied"
mounted = tmp_path / "mounted"
copied.mkdir()
mounted.mkdir()
def test_git_dir_is_remounted_read_only_when_protected(tmp_path: Path) -> None:
(tmp_path / ".git").mkdir()
entries, bind_mounts, _staged = build_session_entries(
[
_source("copied", str(copied)),
_source("mounted", str(mounted), mount=True),
]
)
mounts = build_bind_mounts([_source("repo", str(tmp_path), protect_metadata=True)])
assert list(entries) == ["copied"]
assert isinstance(entries["copied"], LocalDir)
assert [m["target"] for m in bind_mounts] == ["/workspace/mounted"]
assert mounts == [
{"source": str(tmp_path.resolve()), "target": "/workspace/repo", "read_only": False},
{
"source": str((tmp_path / ".git").resolve()),
"target": "/workspace/repo/.git",
"read_only": True,
},
]
def test_agent_instruction_dirs_are_protected_too(tmp_path: Path) -> None:
(tmp_path / ".agents").mkdir()
(tmp_path / ".codex").mkdir()
mounts = build_bind_mounts([_source("repo", str(tmp_path), protect_metadata=True)])
assert [(m["target"], m["read_only"]) for m in mounts] == [
("/workspace/repo", False),
("/workspace/repo/.agents", True),
("/workspace/repo/.codex", True),
]
def test_worktree_git_pointer_file_is_protected(tmp_path: Path) -> None:
gitdir = tmp_path / "nested" / "gitdir"
gitdir.mkdir(parents=True)
(tmp_path / ".git").write_text(f"gitdir: {gitdir}\n", encoding="utf-8")
mounts = build_bind_mounts([_source("repo", str(tmp_path), protect_metadata=True)])
assert [(m["target"], m["read_only"]) for m in mounts] == [
("/workspace/repo", False),
("/workspace/repo/.git", True),
("/workspace/repo/nested/gitdir", True),
]
def test_git_pointer_to_a_missing_gitdir_is_not_mounted(tmp_path: Path) -> None:
(tmp_path / ".git").write_text(f"gitdir: {tmp_path / 'gone'}\n", encoding="utf-8")
mounts = build_bind_mounts([_source("repo", str(tmp_path), protect_metadata=True)])
assert [m["target"] for m in mounts] == ["/workspace/repo", "/workspace/repo/.git"]
def test_git_pointer_outside_the_tree_needs_no_nested_mount(tmp_path: Path) -> None:
tree = tmp_path / "worktree"
tree.mkdir()
(tree / ".git").write_text(f"gitdir: {tmp_path / 'main' / '.git'}\n", encoding="utf-8")
mounts = build_bind_mounts([_source("repo", str(tree), protect_metadata=True)])
assert [m["target"] for m in mounts] == ["/workspace/repo", "/workspace/repo/.git"]
def test_metadata_symlinked_outside_the_tree_is_not_mounted(tmp_path: Path) -> None:
outside = tmp_path / "elsewhere"
outside.mkdir()
tree = tmp_path / "repo"
tree.mkdir()
(tree / ".git").symlink_to(outside, target_is_directory=True)
mounts = build_bind_mounts([_source("repo", str(tree), protect_metadata=True)])
assert [m["target"] for m in mounts] == ["/workspace/repo"]
def test_no_git_guard_without_a_git_dir(tmp_path: Path) -> None:
mounts = build_bind_mounts([_source("repo", str(tmp_path), protect_metadata=True)])
assert [m["target"] for m in mounts] == ["/workspace/repo"]
def test_clone_keeps_its_git_writable(tmp_path: Path) -> None:
(tmp_path / ".git").mkdir()
mounts = build_bind_mounts([_source("clone", str(tmp_path), protect_metadata=False)])
assert [m["target"] for m in mounts] == ["/workspace/clone"]
def test_multiple_sources_each_get_a_mount(tmp_path: Path) -> None:
first = tmp_path / "first"
second = tmp_path / "second"
first.mkdir()
second.mkdir()
mounts = build_bind_mounts([_source("first", str(first)), _source("second", str(second))])
assert [m["target"] for m in mounts] == ["/workspace/first", "/workspace/second"]
assert all(m["read_only"] is False for m in mounts)
def test_incomplete_sources_are_skipped() -> None:
entries, bind_mounts, staged_dirs = build_session_entries(
[
{"source_path": "", "workspace_subdir": "x"},
{"source_path": "/p", "workspace_subdir": ""},
]
assert (
build_bind_mounts(
[
{"source_path": "", "workspace_subdir": "x"},
{"source_path": "/p", "workspace_subdir": ""},
]
)
== []
)
assert entries == {}
assert bind_mounts == []
assert staged_dirs == []
def test_symlink_tree_is_staged(tmp_path: Path) -> None:
repo = tmp_path / "repo"
repo.mkdir()
(repo / "real.txt").write_text("content")
(repo / "link.txt").symlink_to(repo / "real.txt")
def test_manifest_entries_upload_sources_for_backends_without_bind_mounts(
tmp_path: Path,
) -> None:
entries = build_manifest_entries([_source("repo", str(tmp_path), protect_metadata=True)])
entries, _mounts, staged_dirs = build_session_entries([_source("repo", str(repo))])
assert len(staged_dirs) == 1
assert set(entries) == {"repo"}
entry = entries["repo"]
assert isinstance(entry, LocalDir)
assert entry.src == staged_dirs[0]
assert not (staged_dirs[0] / "link.txt").is_symlink()
assert (staged_dirs[0] / "link.txt").read_text() == "content"
assert entry.src == tmp_path.resolve()
def test_manifest_entries_skip_incomplete_sources() -> None:
assert (
build_manifest_entries(
[
{"source_path": "", "workspace_subdir": "x"},
{"source_path": "/p", "workspace_subdir": ""},
]
)
== {}
)
def test_only_bind_mount_capable_backends_are_registered_as_such() -> None:
assert backend_supports_bind_mounts("docker")
assert not backend_supports_bind_mounts("e2b")
async def _remote_backend(**_kwargs: Any) -> tuple[Any, Any]:
return object(), object()
try:
register_backend("e2b", _remote_backend)
assert not backend_supports_bind_mounts("e2b")
register_backend("e2b", _remote_backend, supports_bind_mounts=True)
assert backend_supports_bind_mounts("e2b")
finally:
_BACKENDS.pop("e2b", None)
_BIND_MOUNT_BACKENDS.discard("e2b")