feat(utils): Implement VFSHelper for in-memory artifact management

This commit is contained in:
Yunxiao Xu
2026-02-23 04:36:51 -08:00
parent 8957e93f3d
commit 92d9288f38
2 changed files with 82 additions and 0 deletions

View File

@@ -0,0 +1,33 @@
from typing import Dict, Any, Optional, Tuple, List
class VFSHelper:
"""Helper class for managing in-memory Virtual File System (VFS) artifacts."""
def __init__(self, vfs_state: Dict[str, Any]):
"""Initialize with a reference to the VFS state from AgentState."""
self._vfs = vfs_state
def write(self, filename: str, content: Any, metadata: Optional[Dict[str, Any]] = None) -> None:
"""Write a file to the VFS."""
self._vfs[filename] = {
"content": content,
"metadata": metadata or {}
}
def read(self, filename: str) -> Tuple[Optional[Any], Optional[Dict[str, Any]]]:
"""Read a file and its metadata from the VFS. Returns (None, None) if not found."""
file_data = self._vfs.get(filename)
if file_data:
return file_data["content"], file_data["metadata"]
return None, None
def list(self) -> List[str]:
"""List all filenames in the VFS."""
return list(self._vfs.keys())
def delete(self, filename: str) -> bool:
"""Delete a file from the VFS. Returns True if deleted, False if not found."""
if filename in self._vfs:
del self._vfs[filename]
return True
return False

49
backend/tests/test_vfs.py Normal file
View File

@@ -0,0 +1,49 @@
import pytest
from ea_chatbot.utils.vfs import VFSHelper
def test_vfs_read_write():
"""Verify that we can write and read from the VFS."""
vfs = {}
helper = VFSHelper(vfs)
helper.write("test.txt", "Hello World", metadata={"type": "text"})
assert "test.txt" in vfs
assert vfs["test.txt"]["content"] == "Hello World"
assert vfs["test.txt"]["metadata"]["type"] == "text"
content, metadata = helper.read("test.txt")
assert content == "Hello World"
assert metadata["type"] == "text"
def test_vfs_list():
"""Verify that we can list files in the VFS."""
vfs = {}
helper = VFSHelper(vfs)
helper.write("a.txt", "content a")
helper.write("b.txt", "content b")
files = helper.list()
assert "a.txt" in files
assert "b.txt" in files
assert len(files) == 2
def test_vfs_delete():
"""Verify that we can delete files from the VFS."""
vfs = {}
helper = VFSHelper(vfs)
helper.write("test.txt", "Hello")
helper.delete("test.txt")
assert "test.txt" not in vfs
assert len(helper.list()) == 0
def test_vfs_not_found():
"""Verify that reading a non-existent file returns None."""
vfs = {}
helper = VFSHelper(vfs)
content, metadata = helper.read("missing.txt")
assert content is None
assert metadata is None