mirror of
https://github.com/hwchase17/langchain.git
synced 2026-03-18 11:07:36 +00:00
update sandbox integration tests
This commit is contained in:
@@ -32,6 +32,7 @@ class TestMySandboxStandard(SandboxIntegrationTests):
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
from abc import abstractmethod
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
@@ -42,6 +43,7 @@ deepagents = pytest.importorskip("deepagents")
|
||||
from deepagents.backends.protocol import (
|
||||
FileDownloadResponse,
|
||||
FileUploadResponse,
|
||||
ReadResult,
|
||||
SandboxBackendProtocol,
|
||||
)
|
||||
|
||||
@@ -110,8 +112,27 @@ class SandboxIntegrationTests(BaseStandardTests):
|
||||
content = "Line 1\nLine 2\nLine 3"
|
||||
sandbox_backend.write(test_path, content)
|
||||
result = sandbox_backend.read(test_path)
|
||||
assert "Error:" not in result
|
||||
assert all(line in result for line in ("Line 1", "Line 2", "Line 3"))
|
||||
assert isinstance(result, ReadResult)
|
||||
assert result.error is None
|
||||
assert result.file_data is not None
|
||||
assert all(
|
||||
line in result.file_data["content"]
|
||||
for line in ("Line 1", "Line 2", "Line 3")
|
||||
)
|
||||
|
||||
def test_read_binary_file(self, sandbox_backend: SandboxBackendProtocol) -> None:
|
||||
"""Upload a binary file and verify `read()` returns base64-encoded content."""
|
||||
if not self.has_sync:
|
||||
pytest.skip("Sync tests not supported.")
|
||||
test_path = "/tmp/test_sandbox_ops/binary.png"
|
||||
raw_bytes = bytes(range(256))
|
||||
sandbox_backend.upload_files([(test_path, raw_bytes)])
|
||||
result = sandbox_backend.read(test_path)
|
||||
assert isinstance(result, ReadResult)
|
||||
assert result.error is None
|
||||
assert result.file_data is not None
|
||||
assert result.file_data["encoding"] == "base64"
|
||||
assert base64.b64decode(result.file_data["content"]) == raw_bytes
|
||||
|
||||
def test_edit_single_occurrence(
|
||||
self, sandbox_backend: SandboxBackendProtocol
|
||||
@@ -125,39 +146,48 @@ class SandboxIntegrationTests(BaseStandardTests):
|
||||
result = sandbox_backend.edit(test_path, "Goodbye", "Farewell")
|
||||
assert result.error is None
|
||||
assert result.occurrences == 1
|
||||
file_content = sandbox_backend.read(test_path)
|
||||
assert "Farewell world" in file_content
|
||||
assert "Goodbye" not in file_content
|
||||
file_result = sandbox_backend.read(test_path)
|
||||
assert isinstance(file_result, ReadResult)
|
||||
assert file_result.error is None
|
||||
assert file_result.file_data is not None
|
||||
assert "Farewell world" in file_result.file_data["content"]
|
||||
assert "Goodbye" not in file_result.file_data["content"]
|
||||
|
||||
def test_ls_info_lists_files(self, sandbox_backend: SandboxBackendProtocol) -> None:
|
||||
"""Create files and verify `ls_info()` lists them."""
|
||||
def test_ls_lists_files(self, sandbox_backend: SandboxBackendProtocol) -> None:
|
||||
"""Create files and verify `ls()` lists them."""
|
||||
if not self.has_sync:
|
||||
pytest.skip("Sync tests not supported.")
|
||||
sandbox_backend.write("/tmp/test_sandbox_ops/a.txt", "a")
|
||||
sandbox_backend.write("/tmp/test_sandbox_ops/b.txt", "b")
|
||||
info = sandbox_backend.ls_info("/tmp/test_sandbox_ops")
|
||||
paths = sorted([i["path"] for i in info])
|
||||
result = sandbox_backend.ls("/tmp/test_sandbox_ops")
|
||||
assert result.error is None
|
||||
assert result.entries is not None
|
||||
paths = sorted([i["path"] for i in result.entries])
|
||||
assert "/tmp/test_sandbox_ops/a.txt" in paths
|
||||
assert "/tmp/test_sandbox_ops/b.txt" in paths
|
||||
|
||||
def test_glob_info(self, sandbox_backend: SandboxBackendProtocol) -> None:
|
||||
"""Create files and verify `glob_info()` returns expected matches."""
|
||||
def test_glob(self, sandbox_backend: SandboxBackendProtocol) -> None:
|
||||
"""Create files and verify `glob()` returns expected matches."""
|
||||
if not self.has_sync:
|
||||
pytest.skip("Sync tests not supported.")
|
||||
sandbox_backend.write("/tmp/test_sandbox_ops/x.py", "print('x')")
|
||||
sandbox_backend.write("/tmp/test_sandbox_ops/y.txt", "y")
|
||||
matches = sandbox_backend.glob_info("*.py", path="/tmp/test_sandbox_ops")
|
||||
assert [m["path"] for m in matches] == ["x.py"]
|
||||
result = sandbox_backend.glob("*.py", path="/tmp/test_sandbox_ops")
|
||||
assert result.error is None
|
||||
assert result.matches is not None
|
||||
assert [m["path"] for m in result.matches] == ["x.py"]
|
||||
|
||||
def test_grep_raw_literal(self, sandbox_backend: SandboxBackendProtocol) -> None:
|
||||
"""Verify `grep_raw()` performs literal matching on special characters."""
|
||||
def test_grep_literal(self, sandbox_backend: SandboxBackendProtocol) -> None:
|
||||
"""Verify `grep()` performs literal matching on special characters."""
|
||||
if not self.has_sync:
|
||||
pytest.skip("Sync tests not supported.")
|
||||
sandbox_backend.write("/tmp/test_sandbox_ops/grep.txt", "a (b)\nstr | int\n")
|
||||
matches = sandbox_backend.grep_raw("str | int", path="/tmp/test_sandbox_ops")
|
||||
assert isinstance(matches, list)
|
||||
assert matches[0]["path"].endswith("/grep.txt")
|
||||
assert matches[0]["text"].strip() == "str | int"
|
||||
result = sandbox_backend.grep("str | int", path="/tmp/test_sandbox_ops")
|
||||
assert result.error is None
|
||||
assert result.matches is not None
|
||||
assert len(result.matches) > 0
|
||||
assert result.matches[0]["path"].endswith("/grep.txt")
|
||||
assert result.matches[0]["text"].strip() == "str | int"
|
||||
|
||||
def test_upload_single_file(self, sandbox_backend: SandboxBackendProtocol) -> None:
|
||||
"""Upload one file and verify its contents on the sandbox."""
|
||||
|
||||
Reference in New Issue
Block a user