ClaudeandClaude Fable 5
01f5e54a66
Adds files_mcp/models.py (FileEntry, OperationResult) and files_mcp/core/fs_ops.py with 9 pure file operations on already-resolved Path objects: create/read/update/delete/list_dir/stat/search/move/copy. All writes go through tempfile.mkstemp + os.replace for atomicity; read enforces utf-8 + 1MB cap; create/update enforce strict presence semantics (create: absent or overwrite; update: present). 39 unit tests in tests/unit/test_fs_ops.py. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
299 lines
11 KiB
Python
299 lines
11 KiB
Python
# coding=utf-8
|
|
"""
|
|
@Time :2026/6/30
|
|
@Author :tao.chen
|
|
"""
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from files_mcp.core import fs_ops
|
|
from files_mcp.models import FileEntry, OperationResult
|
|
|
|
|
|
# --- create ------------------------------------------------------------
|
|
|
|
class TestCreate:
|
|
def test_creates_new_file(self, tmp_path: Path):
|
|
target = tmp_path / "new.txt"
|
|
result = fs_ops.create(target, "hello", overwrite=False)
|
|
assert isinstance(result, OperationResult)
|
|
assert result.status == "CREATED"
|
|
assert result.size == 5
|
|
assert target.read_text(encoding="utf-8") == "hello"
|
|
|
|
def test_overwrite_true_replaces(self, tmp_path: Path):
|
|
target = tmp_path / "f.txt"
|
|
target.write_text("old", encoding="utf-8")
|
|
result = fs_ops.create(target, "new", overwrite=True)
|
|
assert result.status == "CREATED"
|
|
assert target.read_text(encoding="utf-8") == "new"
|
|
|
|
def test_overwrite_false_raises(self, tmp_path: Path):
|
|
target = tmp_path / "f.txt"
|
|
target.write_text("old", encoding="utf-8")
|
|
with pytest.raises(FileExistsError, match="already exists"):
|
|
fs_ops.create(target, "new", overwrite=False)
|
|
|
|
def test_too_large_raises(self, tmp_path: Path):
|
|
target = tmp_path / "big.txt"
|
|
with pytest.raises(ValueError, match="content too large"):
|
|
fs_ops.create(target, "x" * (fs_ops.MAX_BYTES + 1), overwrite=False)
|
|
|
|
def test_atomic_write_leaves_no_tmp_on_success(self, tmp_path: Path):
|
|
target = tmp_path / "f.txt"
|
|
fs_ops.create(target, "hi", overwrite=False)
|
|
# No leftover .tmp_* in the directory
|
|
leftovers = [p for p in tmp_path.iterdir() if p.name.startswith(".tmp_")]
|
|
assert leftovers == []
|
|
|
|
|
|
# --- read --------------------------------------------------------------
|
|
|
|
class TestRead:
|
|
def test_reads_existing(self, tmp_path: Path):
|
|
f = tmp_path / "f.txt"
|
|
f.write_text("hello", encoding="utf-8")
|
|
out = fs_ops.read(f)
|
|
assert out == {
|
|
"path": str(f),
|
|
"content": "hello",
|
|
"size": 5,
|
|
"encoding": "utf-8",
|
|
}
|
|
|
|
def test_missing_raises_keyerror(self, tmp_path: Path):
|
|
with pytest.raises(KeyError, match="not found"):
|
|
fs_ops.read(tmp_path / "nope.txt")
|
|
|
|
def test_directory_raises(self, tmp_path: Path):
|
|
with pytest.raises(ValueError, match="not a regular file"):
|
|
fs_ops.read(tmp_path)
|
|
|
|
def test_too_large_raises(self, tmp_path: Path):
|
|
f = tmp_path / "big.bin"
|
|
f.write_bytes(b"x" * (fs_ops.MAX_BYTES + 1))
|
|
with pytest.raises(ValueError, match="file too large"):
|
|
fs_ops.read(f)
|
|
|
|
def test_non_utf8_raises(self, tmp_path: Path):
|
|
f = tmp_path / "binary.bin"
|
|
f.write_bytes(b"\xff\xfe\x00\x01")
|
|
with pytest.raises(ValueError, match="not valid utf-8"):
|
|
fs_ops.read(f)
|
|
|
|
|
|
# --- update ------------------------------------------------------------
|
|
|
|
class TestUpdate:
|
|
def test_updates_existing(self, tmp_path: Path):
|
|
f = tmp_path / "f.txt"
|
|
f.write_text("old", encoding="utf-8")
|
|
result = fs_ops.update(f, "new")
|
|
assert result.status == "UPDATED"
|
|
assert result.size == 3
|
|
assert f.read_text(encoding="utf-8") == "new"
|
|
|
|
def test_missing_raises_keyerror(self, tmp_path: Path):
|
|
with pytest.raises(KeyError, match="not found"):
|
|
fs_ops.update(tmp_path / "nope.txt", "x")
|
|
|
|
def test_too_large_raises(self, tmp_path: Path):
|
|
f = tmp_path / "f.txt"
|
|
f.write_text("old", encoding="utf-8")
|
|
with pytest.raises(ValueError, match="content too large"):
|
|
fs_ops.update(f, "x" * (fs_ops.MAX_BYTES + 1))
|
|
|
|
|
|
# --- delete ------------------------------------------------------------
|
|
|
|
class TestDelete:
|
|
def test_deletes_file(self, tmp_path: Path):
|
|
f = tmp_path / "f.txt"
|
|
f.write_text("x", encoding="utf-8")
|
|
result = fs_ops.delete(f, recursive=False)
|
|
assert result.status == "DELETED"
|
|
assert result.size is None
|
|
assert not f.exists()
|
|
|
|
def test_missing_raises_keyerror(self, tmp_path: Path):
|
|
with pytest.raises(KeyError, match="not found"):
|
|
fs_ops.delete(tmp_path / "nope.txt", recursive=False)
|
|
|
|
def test_non_empty_dir_raises_without_recursive(self, tmp_path: Path):
|
|
d = tmp_path / "d"
|
|
d.mkdir()
|
|
(d / "child").write_text("x", encoding="utf-8")
|
|
with pytest.raises(ValueError, match="not empty"):
|
|
fs_ops.delete(d, recursive=False)
|
|
|
|
def test_recursive_deletes_tree(self, tmp_path: Path):
|
|
d = tmp_path / "d"
|
|
d.mkdir()
|
|
(d / "child").write_text("x", encoding="utf-8")
|
|
(d / "sub").mkdir()
|
|
(d / "sub" / "leaf").write_text("y", encoding="utf-8")
|
|
fs_ops.delete(d, recursive=True)
|
|
assert not d.exists()
|
|
|
|
def test_empty_dir_ok_without_recursive(self, tmp_path: Path):
|
|
d = tmp_path / "empty"
|
|
d.mkdir()
|
|
fs_ops.delete(d, recursive=False)
|
|
assert not d.exists()
|
|
|
|
|
|
# --- list_dir ----------------------------------------------------------
|
|
|
|
class TestListDir:
|
|
def test_single_level(self, tmp_path: Path):
|
|
(tmp_path / "a.txt").write_text("a", encoding="utf-8")
|
|
(tmp_path / "b").mkdir()
|
|
entries = fs_ops.list_dir(tmp_path, recursive=False, max_depth=1)
|
|
names = sorted(e.name for e in entries)
|
|
assert names == ["a.txt", "b"]
|
|
assert all(isinstance(e, FileEntry) for e in entries)
|
|
|
|
def test_recursive_respects_max_depth(self, tmp_path: Path):
|
|
(tmp_path / "a.txt").write_text("a", encoding="utf-8")
|
|
(tmp_path / "sub").mkdir()
|
|
(tmp_path / "sub" / "b.txt").write_text("b", encoding="utf-8")
|
|
(tmp_path / "sub" / "deep").mkdir()
|
|
(tmp_path / "sub" / "deep" / "c.txt").write_text("c", encoding="utf-8")
|
|
depth0 = fs_ops.list_dir(tmp_path, recursive=True, max_depth=0)
|
|
assert sorted(e.name for e in depth0) == ["a.txt", "sub"]
|
|
depth1 = fs_ops.list_dir(tmp_path, recursive=True, max_depth=1)
|
|
assert sorted(e.name for e in depth1) == ["a.txt", "b.txt", "deep", "sub"]
|
|
|
|
def test_missing_raises(self, tmp_path: Path):
|
|
with pytest.raises(KeyError, match="not found"):
|
|
fs_ops.list_dir(tmp_path / "nope", recursive=False, max_depth=1)
|
|
|
|
def test_file_raises(self, tmp_path: Path):
|
|
f = tmp_path / "f.txt"
|
|
f.write_text("x", encoding="utf-8")
|
|
with pytest.raises(ValueError, match="not a directory"):
|
|
fs_ops.list_dir(f, recursive=False, max_depth=1)
|
|
|
|
def test_negative_depth_raises(self, tmp_path: Path):
|
|
with pytest.raises(ValueError, match="max_depth"):
|
|
fs_ops.list_dir(tmp_path, recursive=True, max_depth=-1)
|
|
|
|
|
|
# --- stat --------------------------------------------------------------
|
|
|
|
class TestStat:
|
|
def test_file(self, tmp_path: Path):
|
|
f = tmp_path / "f.txt"
|
|
f.write_text("hello", encoding="utf-8")
|
|
e = fs_ops.stat(f)
|
|
assert e.name == "f.txt"
|
|
assert e.type == "file"
|
|
assert e.size == 5
|
|
|
|
def test_directory(self, tmp_path: Path):
|
|
e = fs_ops.stat(tmp_path)
|
|
assert e.type == "directory"
|
|
assert e.size == 0
|
|
|
|
def test_missing_raises(self, tmp_path: Path):
|
|
with pytest.raises(KeyError, match="not found"):
|
|
fs_ops.stat(tmp_path / "nope")
|
|
|
|
|
|
# --- search ------------------------------------------------------------
|
|
|
|
class TestSearch:
|
|
def test_simple_glob(self, tmp_path: Path):
|
|
(tmp_path / "a.py").write_text("a", encoding="utf-8")
|
|
(tmp_path / "b.txt").write_text("b", encoding="utf-8")
|
|
(tmp_path / "c.py").write_text("c", encoding="utf-8")
|
|
matches = fs_ops.search(tmp_path, "*.py", recursive=False)
|
|
assert sorted(Path(m).name for m in matches) == ["a.py", "c.py"]
|
|
|
|
def test_recursive_glob(self, tmp_path: Path):
|
|
(tmp_path / "sub").mkdir()
|
|
(tmp_path / "sub" / "x.py").write_text("x", encoding="utf-8")
|
|
(tmp_path / "a.py").write_text("a", encoding="utf-8")
|
|
matches = fs_ops.search(tmp_path, "*.py", recursive=True)
|
|
assert sorted(Path(m).name for m in matches) == ["a.py", "x.py"]
|
|
|
|
def test_no_match_returns_empty(self, tmp_path: Path):
|
|
(tmp_path / "a.txt").write_text("a", encoding="utf-8")
|
|
assert fs_ops.search(tmp_path, "*.py", recursive=False) == []
|
|
|
|
def test_missing_raises(self, tmp_path: Path):
|
|
with pytest.raises(KeyError, match="not found"):
|
|
fs_ops.search(tmp_path / "nope", "*", recursive=False)
|
|
|
|
def test_non_directory_raises(self, tmp_path: Path):
|
|
f = tmp_path / "f.txt"
|
|
f.write_text("x", encoding="utf-8")
|
|
with pytest.raises(ValueError, match="not a directory"):
|
|
fs_ops.search(f, "*", recursive=False)
|
|
|
|
|
|
# --- move / copy -------------------------------------------------------
|
|
|
|
class TestMove:
|
|
def test_rename(self, tmp_path: Path):
|
|
src = tmp_path / "a.txt"
|
|
src.write_text("hi", encoding="utf-8")
|
|
result = fs_ops.move(src, tmp_path / "b.txt", overwrite=False)
|
|
assert result.status == "MOVED"
|
|
assert result.path == str(tmp_path / "b.txt")
|
|
assert not src.exists()
|
|
assert (tmp_path / "b.txt").read_text(encoding="utf-8") == "hi"
|
|
|
|
def test_missing_src_raises(self, tmp_path: Path):
|
|
with pytest.raises(KeyError, match="src not found"):
|
|
fs_ops.move(tmp_path / "nope", tmp_path / "dst", overwrite=False)
|
|
|
|
def test_existing_dst_raises_without_overwrite(self, tmp_path: Path):
|
|
src = tmp_path / "a.txt"
|
|
src.write_text("a", encoding="utf-8")
|
|
dst = tmp_path / "b.txt"
|
|
dst.write_text("b", encoding="utf-8")
|
|
with pytest.raises(FileExistsError, match="dst already exists"):
|
|
fs_ops.move(src, dst, overwrite=False)
|
|
|
|
def test_overwrite_replaces(self, tmp_path: Path):
|
|
src = tmp_path / "a.txt"
|
|
src.write_text("new", encoding="utf-8")
|
|
dst = tmp_path / "b.txt"
|
|
dst.write_text("old", encoding="utf-8")
|
|
fs_ops.move(src, dst, overwrite=True)
|
|
assert dst.read_text(encoding="utf-8") == "new"
|
|
assert not src.exists()
|
|
|
|
|
|
class TestCopy:
|
|
def test_copies_file(self, tmp_path: Path):
|
|
src = tmp_path / "a.txt"
|
|
src.write_text("hi", encoding="utf-8")
|
|
result = fs_ops.copy(src, tmp_path / "b.txt", overwrite=False)
|
|
assert result.status == "COPIED"
|
|
assert result.size == 2
|
|
assert src.read_text(encoding="utf-8") == "hi" # source unchanged
|
|
assert (tmp_path / "b.txt").read_text(encoding="utf-8") == "hi"
|
|
|
|
def test_missing_src_raises(self, tmp_path: Path):
|
|
with pytest.raises(KeyError, match="src not found"):
|
|
fs_ops.copy(tmp_path / "nope", tmp_path / "dst", overwrite=False)
|
|
|
|
def test_existing_dst_raises_without_overwrite(self, tmp_path: Path):
|
|
src = tmp_path / "a.txt"
|
|
src.write_text("a", encoding="utf-8")
|
|
dst = tmp_path / "b.txt"
|
|
dst.write_text("b", encoding="utf-8")
|
|
with pytest.raises(FileExistsError, match="dst already exists"):
|
|
fs_ops.copy(src, dst, overwrite=False)
|
|
|
|
def test_copies_directory(self, tmp_path: Path):
|
|
src = tmp_path / "d"
|
|
src.mkdir()
|
|
(src / "child").write_text("x", encoding="utf-8")
|
|
result = fs_ops.copy(src, tmp_path / "d2", overwrite=False)
|
|
assert result.status == "COPIED"
|
|
assert (tmp_path / "d2" / "child").read_text(encoding="utf-8") == "x"
|