uvos/vibe
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

v2.14.0 (#743)

Browse source 
Co-authored-by: Alexis Tacnet <alexis@mistral.ai>
Co-authored-by: Clément Drouin <clement.drouin@mistral.ai>
Co-authored-by: Guillaume LE GOFF <guillaume.lgf@gmail.com>
Co-authored-by: Lucas Marandat <31749711+lucasmrdt@users.noreply.github.com>
Co-authored-by: Maxime Dolores <maxime.dolores@ext.mistral.ai>
Co-authored-by: Pierre Rossinès <pierre.rossines@mistral.ai>
Co-authored-by: Quentin <quentin.torroba@mistral.ai>
Co-authored-by: Val <102326092+vdeva@users.noreply.github.com>
Co-authored-by: Vincent G <10739306+VinceOPS@users.noreply.github.com>
Co-authored-by: p.vezia <166131032+le-codeur-rapide@users.noreply.github.com>
Co-authored-by: Hiba Chaabnia <Hiba-Chaabnia@users.noreply.github.com>
Co-authored-by: Nikhil Bhima <nikhilbhima@users.noreply.github.com>
Co-authored-by: Nkipohcs <Nkipohcs@users.noreply.github.com>
Co-authored-by: Mistral Vibe <vibe@mistral.ai>
This commit is contained in:
Mathias Gesbert 2026-06-04 18:26:35 +02:00 • committed by GitHub
parent ad0d5c9520
commit 3f8487f761
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
197 changed files with 10819 additions and 2830 deletions
Download patch file Download diff file Expand all files Collapse all files

71
tests/core/session/test_image_snapshot.py Normal file
View file

@ -0,0 +1,71 @@
from __future__ import annotations
import hashlib
from pathlib import Path
import pytest
from vibe.core.session.image_snapshot import ImageSnapshotError, snapshot_image
PNG_BYTES = b"\x89PNG\r\n\x1a\n" + b"\x00" * 16
def test_snapshot_image_copies_to_attachments_dir(tmp_path: Path) -> None:
src = tmp_path / "screenshot.png"
src.write_bytes(PNG_BYTES)
session_dir = tmp_path / "session"
session_dir.mkdir()
att = snapshot_image(src, alias="screenshot.png", session_dir=session_dir)
digest = hashlib.sha1(PNG_BYTES, usedforsecurity=False).hexdigest()
assert att.path == (session_dir / "attachments" / f"{digest}.png").resolve()
assert att.alias == "screenshot.png"
assert att.mime_type == "image/png"
assert att.path.read_bytes() == PNG_BYTES
def test_snapshot_image_is_idempotent_on_same_bytes(tmp_path: Path) -> None:
src_a = tmp_path / "a.png"
src_b = tmp_path / "b.png"
src_a.write_bytes(PNG_BYTES)
src_b.write_bytes(PNG_BYTES)
session_dir = tmp_path / "session"
att_a = snapshot_image(src_a, alias="a.png", session_dir=session_dir)
att_b = snapshot_image(src_b, alias="b.png", session_dir=session_dir)
assert att_a.path == att_b.path
assert sum(1 for _ in (session_dir / "attachments").iterdir()) == 1
def test_snapshot_image_returns_source_when_session_dir_is_none(tmp_path: Path) -> None:
src = tmp_path / "screenshot.png"
src.write_bytes(PNG_BYTES)
att = snapshot_image(src, alias="screenshot.png", session_dir=None)
assert att.path == src.resolve()
assert att.alias == "screenshot.png"
def test_snapshot_image_rejects_non_image_extension(tmp_path: Path) -> None:
src = tmp_path / "readme.txt"
src.write_bytes(b"hi")
with pytest.raises(ImageSnapshotError):
snapshot_image(src, alias="readme.txt", session_dir=None)
def test_snapshot_image_rejects_missing_file(tmp_path: Path) -> None:
with pytest.raises(ImageSnapshotError):
snapshot_image(tmp_path / "missing.png", alias="missing.png", session_dir=None)
def test_snapshot_image_normalizes_jpg_to_jpeg_mime(tmp_path: Path) -> None:
src = tmp_path / "photo.jpg"
src.write_bytes(b"\xff\xd8\xff\xe0" + b"\x00" * 8)
att = snapshot_image(src, alias="photo.jpg", session_dir=None)
assert att.mime_type == "image/jpeg"

30
tests/core/session/test_last_session_pointer.py
View file

@ -80,6 +80,36 @@ def test_record_ignores_empty_session_id(
assert last_session_pointer.load(session_logging) is None
def test_clear_matching_removes_matching_pointers_only(
session_logging: SessionLoggingConfig,
) -> None:
pointer_dir = Path(session_logging.save_dir) / last_session_pointer.POINTER_DIR_NAME
pointer_dir.mkdir()
(pointer_dir / "ttys001").write_text("deleted-session\n", encoding="utf-8")
(pointer_dir / "ttys002").write_text("other-session\n", encoding="utf-8")
(pointer_dir / "ttys003").write_text("deleted-session\n", encoding="utf-8")
(pointer_dir / "nested").mkdir()
last_session_pointer.clear_matching(session_logging, "deleted-session")
assert not (pointer_dir / "ttys001").exists()
assert (pointer_dir / "ttys002").read_text(encoding="utf-8") == "other-session\n"
assert not (pointer_dir / "ttys003").exists()
assert (pointer_dir / "nested").is_dir()
def test_clear_matching_skips_when_logging_disabled(tmp_path: Path) -> None:
disabled = SessionLoggingConfig(save_dir=str(tmp_path), enabled=False)
pointer_dir = tmp_path / last_session_pointer.POINTER_DIR_NAME
pointer_dir.mkdir()
pointer_path = pointer_dir / "ttys001"
pointer_path.write_text("deleted-session\n", encoding="utf-8")
last_session_pointer.clear_matching(disabled, "deleted-session")
assert pointer_path.exists()
def test_current_tty_key_returns_none_when_ttyname_is_unavailable(
monkeypatch: pytest.MonkeyPatch,
) -> None:

9
tests/core/test_add_dir.py
View file

@ -344,13 +344,13 @@ class TestProjectPromptsDirsAdditionalDirs:
class TestProjectRootsNestedDedup:
def test_nested_add_dirs_collapse(self, tmp_path: Path) -> None:
def test_nested_add_dirs_preserved(self, tmp_path: Path) -> None:
outer = (tmp_path / "outer").resolve()
inner = outer / "inner"
inner.mkdir(parents=True)
mgr = HarnessFilesManager(sources=("user",), _additional_dirs=(outer, inner))
assert mgr.project_roots == [outer]
assert mgr.project_roots == [outer, inner]
def test_add_dir_containing_cwd_keeps_both(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
@ -380,10 +380,13 @@ class TestProjectRootsNestedDedup:
outer = (tmp_path / "outer").resolve()
inner = outer / "inner"
inner.mkdir(parents=True)
skills_dir = inner / ".vibe" / "skills"
skills_dir.mkdir(parents=True)
monkeypatch.chdir(outer)
trusted_folders_manager.trust_for_session(outer)
mgr = HarnessFilesManager(
sources=("user", "project"), _additional_dirs=(inner,)
)
assert mgr.project_roots == [outer]
assert mgr.project_roots == [outer, inner]
assert skills_dir in mgr.project_skills_dirs

2
tests/core/test_agents.py
View file

@ -20,7 +20,7 @@ class TestAgentProfile:
"""Test that EXPLORE agent has expected enabled tools."""
enabled_tools = EXPLORE.overrides.get("enabled_tools", [])
assert "grep" in enabled_tools
assert "read_file" in enabled_tools
assert "read" in enabled_tools
def test_builtin_agents_contains_explore(self) -> None:
"""Test that BUILTIN_AGENTS includes explore."""

3
tests/core/test_config_layer.py
View file

@ -13,6 +13,7 @@ from vibe.core.config.layer import (
TrustNotResolvedError,
UntrustedLayerError,
)
from vibe.core.config.patch import ConfigPatch
class StubLayer(ConfigLayer[BaseModel]):
@ -401,7 +402,7 @@ async def test_get_fingerprint_not_implemented() -> None:
async def test_apply_not_implemented() -> None:
layer = StubLayer()
with pytest.raises(NotImplementedError):
await layer.apply({"op": "set"})
await layer.apply(ConfigPatch(fingerprint="fp-1"))
@pytest.mark.asyncio

3
tests/core/test_config_orchestrator.py
View file

@ -7,6 +7,7 @@ import pytest
from vibe.core.config.layer import ConfigLayer, RawConfig
from vibe.core.config.orchestrator import ConfigOrchestrator
from vibe.core.config.patch import ConfigPatch
from vibe.core.config.schema import ConfigSchema, WithReplaceMerge
@ -76,7 +77,7 @@ async def test_origin_of_missing_key_returns_none() -> None:
async def test_apply_patch_raises_not_implemented() -> None:
orch = await ConfigOrchestrator.create(schema=SimpleSchema, layers=[])
with pytest.raises(NotImplementedError, match="M2"):
await orch.apply_patch({})
await orch.apply_patch(ConfigPatch(fingerprint="fp-1"))
@pytest.mark.asyncio

46
tests/core/test_config_paths.py
View file

@ -78,7 +78,7 @@ class TestProjectToolsDirs:
mgr = HarnessFilesManager(sources=("user", "project"))
assert mgr.project_tools_dirs == []
def test_finds_tools_dirs_recursively(
def test_does_not_find_nested_tools_dirs(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
@ -86,19 +86,6 @@ class TestProjectToolsDirs:
(tmp_path / ".vibe" / "tools").mkdir(parents=True)
(tmp_path / "sub" / ".vibe" / "tools").mkdir(parents=True)
mgr = HarnessFilesManager(sources=("user", "project"))
assert mgr.project_tools_dirs == [
tmp_path / ".vibe" / "tools",
tmp_path / "sub" / ".vibe" / "tools",
]
def test_does_not_descend_into_ignored_dirs(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(trusted_folders_manager, "is_trusted", lambda _: True)
(tmp_path / ".vibe" / "tools").mkdir(parents=True)
(tmp_path / ".git" / ".vibe" / "tools").mkdir(parents=True)
mgr = HarnessFilesManager(sources=("user", "project"))
assert mgr.project_tools_dirs == [tmp_path / ".vibe" / "tools"]
@ -147,7 +134,7 @@ class TestProjectAgentsDirs:
mgr = HarnessFilesManager(sources=("user", "project"))
assert mgr.project_agents_dirs == []
def test_finds_agents_dirs_recursively(
def test_does_not_find_nested_agents_dirs(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
@ -155,19 +142,6 @@ class TestProjectAgentsDirs:
(tmp_path / ".vibe" / "agents").mkdir(parents=True)
(tmp_path / "sub" / "deep" / ".vibe" / "agents").mkdir(parents=True)
mgr = HarnessFilesManager(sources=("user", "project"))
assert mgr.project_agents_dirs == [
tmp_path / ".vibe" / "agents",
tmp_path / "sub" / "deep" / ".vibe" / "agents",
]
def test_does_not_descend_into_ignored_dirs(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(trusted_folders_manager, "is_trusted", lambda _: True)
(tmp_path / ".vibe" / "agents").mkdir(parents=True)
(tmp_path / "__pycache__" / ".vibe" / "agents").mkdir(parents=True)
mgr = HarnessFilesManager(sources=("user", "project"))
assert mgr.project_agents_dirs == [tmp_path / ".vibe" / "agents"]
@ -544,7 +518,7 @@ class TestProjectSkillsDirs:
mgr = HarnessFilesManager(sources=("user", "project"))
assert mgr.project_skills_dirs == []
def test_finds_skills_dirs_recursively_in_trusted_folder(
def test_does_not_find_nested_skills_dirs(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
@ -553,20 +527,6 @@ class TestProjectSkillsDirs:
(tmp_path / "sub" / ".agents" / "skills").mkdir(parents=True)
(tmp_path / "sub" / "deep" / ".vibe" / "skills").mkdir(parents=True)
mgr = HarnessFilesManager(sources=("user", "project"))
assert mgr.project_skills_dirs == [
tmp_path / ".vibe" / "skills",
tmp_path / "sub" / ".agents" / "skills",
tmp_path / "sub" / "deep" / ".vibe" / "skills",
]
def test_does_not_descend_into_ignored_dirs(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(trusted_folders_manager, "is_trusted", lambda _: True)
(tmp_path / ".vibe" / "skills").mkdir(parents=True)
(tmp_path / "node_modules" / ".vibe" / "skills").mkdir(parents=True)
mgr = HarnessFilesManager(sources=("user", "project"))
assert mgr.project_skills_dirs == [tmp_path / ".vibe" / "skills"]

184
tests/core/test_config_resolution.py
View file

@ -309,6 +309,32 @@ class TestSetThinking:
assert result["models"][0].get("thinking") is None
assert result["models"][1]["thinking"] == "max"
def test_preserves_supports_images_when_materializing_defaults(
self, config_dir: Path
) -> None:
config_file = config_dir / "config.toml"
data = {"active_model": "mistral-medium-3.5"}
with config_file.open("wb") as f:
tomli_w.dump(data, f)
cfg = VibeConfig.load()
cfg.set_thinking("low")
reloaded = VibeConfig.load()
active = reloaded.get_active_model()
assert active.alias == "mistral-medium-3.5"
assert active.thinking == "low"
assert active.supports_images is True
with config_file.open("rb") as f:
result = tomllib.load(f)
active_entry = result["models"][0]
assert active_entry["supports_images"] is True
assert "temperature" not in active_entry
assert "input_price" not in active_entry
assert "output_price" not in active_entry
assert "auto_compact_threshold" not in active_entry
assert "supports_images" not in result["models"][1]
class TestMigrateLeavesFindInBashAllowlist:
def test_keeps_find_in_config_file(
@ -659,6 +685,62 @@ class TestMigrateMistralVibeCliLatestDefaults:
assert result["models"][0]["output_price"] == 7.5
assert result["models"][0]["thinking"] == "high"
def test_backfills_supports_images_on_existing_mistral_medium_entry(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("VIBE_HOME", str(tmp_path))
config_file = tmp_path / "config.toml"
data = {
"active_model": "mistral-medium-3.5",
"models": [
{
"name": "mistral-vibe-cli-latest",
"provider": "mistral",
"alias": "mistral-medium-3.5",
"temperature": 1.0,
"input_price": 1.5,
"output_price": 7.5,
"thinking": "high",
}
],
}
with config_file.open("wb") as f:
tomli_w.dump(data, f)
reset_harness_files_manager()
init_harness_files_manager("user")
VibeConfig._migrate()
with config_file.open("rb") as f:
result = tomllib.load(f)
assert result["models"][0]["supports_images"] is True
def test_preserves_explicit_supports_images_false(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("VIBE_HOME", str(tmp_path))
config_file = tmp_path / "config.toml"
data = {
"models": [
{
"name": "mistral-vibe-cli-latest",
"provider": "mistral",
"alias": "mistral-medium-3.5",
"supports_images": False,
}
]
}
with config_file.open("wb") as f:
tomli_w.dump(data, f)
reset_harness_files_manager()
init_harness_files_manager("user")
VibeConfig._migrate()
with config_file.open("rb") as f:
result = tomllib.load(f)
assert result["models"][0]["supports_images"] is False
class TestAutoCompactThresholdFallback:
def test_model_without_explicit_threshold_inherits_global(self) -> None:
@ -1320,3 +1402,105 @@ class TestIsActiveModelMistral:
active_model="llama-local",
)
assert cfg.is_active_model_mistral() is False
class TestMigrateRenamedTools:
def test_renames_read_file_and_search_replace_keys(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("VIBE_HOME", str(tmp_path))
config_file = tmp_path / "config.toml"
data = {
"tools": {
"read_file": {
"permission": "always",
"allowlist": ["src/**"],
"max_read_bytes": 64000,
},
"search_replace": {
"allowlist": ["src/**"],
"max_content_size": 100000,
"create_backup": True,
},
}
}
with config_file.open("wb") as f:
tomli_w.dump(data, f)
reset_harness_files_manager()
init_harness_files_manager("user")
VibeConfig._migrate()
with config_file.open("rb") as f:
result = tomllib.load(f)
tools = result["tools"]
assert "read_file" not in tools
assert "search_replace" not in tools
assert tools["read"] == {
"permission": "always",
"allowlist": ["src/**"],
"max_read_bytes": 64000,
}
# Common options carry over; edit-incompatible options are dropped.
assert tools["edit"] == {"allowlist": ["src/**"]}
def test_prefers_existing_new_key_and_drops_legacy(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("VIBE_HOME", str(tmp_path))
config_file = tmp_path / "config.toml"
data = {
"tools": {
"read_file": {"permission": "always"},
"read": {"permission": "ask"},
}
}
with config_file.open("wb") as f:
tomli_w.dump(data, f)
reset_harness_files_manager()
init_harness_files_manager("user")
VibeConfig._migrate()
with config_file.open("rb") as f:
result = tomllib.load(f)
assert "read_file" not in result["tools"]
assert result["tools"]["read"] == {"permission": "ask"}
def test_renames_entries_in_enabled_and_disabled_lists(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("VIBE_HOME", str(tmp_path))
config_file = tmp_path / "config.toml"
data = {
"enabled_tools": ["read_file", "grep"],
"disabled_tools": ["search_replace"],
}
with config_file.open("wb") as f:
tomli_w.dump(data, f)
reset_harness_files_manager()
init_harness_files_manager("user")
VibeConfig._migrate()
with config_file.open("rb") as f:
result = tomllib.load(f)
assert result["enabled_tools"] == ["read", "grep"]
assert result["disabled_tools"] == ["edit"]
def test_noop_when_no_legacy_tool_names(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("VIBE_HOME", str(tmp_path))
config_file = tmp_path / "config.toml"
data = {"tools": {"read": {"permission": "always"}}}
with config_file.open("wb") as f:
tomli_w.dump(data, f)
reset_harness_files_manager()
init_harness_files_manager("user")
VibeConfig._migrate()
with config_file.open("rb") as f:
result = tomllib.load(f)
assert result["tools"] == {"read": {"permission": "always"}}

86
tests/core/test_edit_encoding.py Normal file
View file

@ -0,0 +1,86 @@
from __future__ import annotations
import os
from pathlib import Path
import stat
import pytest
from tests.mock.utils import collect_result
from vibe.core.tools.base import BaseToolState, ToolError
from vibe.core.tools.builtins.edit import Edit, EditArgs, EditConfig
@pytest.mark.asyncio
async def test_edit_rewrites_with_detected_encoding(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
path = tmp_path / "utf16.txt"
original = "line one café\nline two été\n"
path.write_bytes(original.encode("utf-16"))
tool = Edit(config_getter=lambda: EditConfig(), state=BaseToolState())
await collect_result(
tool.run(
EditArgs(
file_path=str(path),
old_string="line one café",
new_string="LINE ONE CAFÉ",
)
)
)
assert path.read_bytes().startswith(b"\xff\xfe")
assert path.read_text(encoding="utf-16") == "LINE ONE CAFÉ\nline two été\n"
@pytest.mark.asyncio
async def test_edit_preserves_file_mode(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
path = tmp_path / "script.sh"
path.write_text("#!/bin/sh\necho old\n")
os.chmod(path, 0o755)
tool = Edit(config_getter=lambda: EditConfig(), state=BaseToolState())
await collect_result(
tool.run(EditArgs(file_path=str(path), old_string="old", new_string="new"))
)
assert stat.S_IMODE(path.stat().st_mode) == 0o755
@pytest.mark.asyncio
@pytest.mark.parametrize("newline", ["\r\n", "\r", "\n"])
async def test_edit_preserves_line_endings(
newline: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
path = tmp_path / "f.txt"
original = newline.join(["alpha", "beta", "gamma"])
path.write_bytes(original.encode("utf-8"))
tool = Edit(config_getter=lambda: EditConfig(), state=BaseToolState())
await collect_result(
tool.run(EditArgs(file_path=str(path), old_string="beta", new_string="BETA"))
)
assert path.read_bytes() == newline.join(["alpha", "BETA", "gamma"]).encode("utf-8")
@pytest.mark.asyncio
async def test_edit_binary_file_raises_tool_error(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
path = tmp_path / "blob.bin"
# PNG header + some non-text payload.
path.write_bytes(b"\x89PNG\r\n\x1a\n" + bytes(range(256)) * 4)
tool = Edit(config_getter=lambda: EditConfig(), state=BaseToolState())
with pytest.raises(ToolError, match="not valid text"):
await collect_result(
tool.run(EditArgs(file_path=str(path), old_string="PNG", new_string="JPG"))
)

48
tests/core/test_environment_layer.py Normal file
View file

@ -0,0 +1,48 @@
from __future__ import annotations
import os
from unittest.mock import patch
import pytest
from vibe.core.config.layers.environment import EnvironmentLayer
from vibe.core.config.vibe_schema import VibeConfigSchema
@pytest.mark.asyncio
async def test_reads_env_vars() -> None:
env = {
"MISTRAL_API_KEY": "test-key",
"VIBE_ACTIVE_MODEL": "mistral-large",
"VIBE_VIM_KEYBINDINGS": "true",
"VIBE_ENABLE_TELEMETRY": "0",
"VIBE_UNKNOWN_VAR": "ignored",
"VIBE_SESSION_LOGGING__ENABLED": "false",
"VIBE_SESSION_LOGGING__SESSION_PREFIX": "mysession",
"VIBE_API_TIMEOUT": ".12",
}
with patch.dict(os.environ, env, clear=True):
layer = EnvironmentLayer(schema=VibeConfigSchema)
data = await layer.load()
assert data.model_dump() == {
"active_model": "mistral-large",
"vim_keybindings": True,
"enable_telemetry": False,
"session_logging": {"enabled": False, "session_prefix": "mysession"},
"api_timeout": 0.12,
}
assert layer.name == "environment"
@pytest.mark.asyncio
async def test_no_vars_set_returns_empty() -> None:
with patch.dict(os.environ, {}, clear=True):
data = await EnvironmentLayer(schema=VibeConfigSchema).load()
assert data.model_dump() == {}
@pytest.mark.asyncio
async def test_always_trusted() -> None:
assert await EnvironmentLayer(schema=VibeConfigSchema).resolve_trust() is True

51
tests/core/test_llm_message_merge.py Normal file
View file

@ -0,0 +1,51 @@
from __future__ import annotations
from pathlib import Path
import pytest
from vibe.core.types import ImageAttachment, LLMMessage, Role
@pytest.fixture()
def image_a(tmp_path: Path) -> ImageAttachment:
p = tmp_path / "a.png"
p.write_bytes(b"\x89PNG")
return ImageAttachment(path=p, alias="a.png", mime_type="image/png")
@pytest.fixture()
def image_b(tmp_path: Path) -> ImageAttachment:
p = tmp_path / "b.png"
p.write_bytes(b"\x89PNG")
return ImageAttachment(path=p, alias="b.png", mime_type="image/png")
def _msg(content: str, images: list[ImageAttachment] | None = None) -> LLMMessage:
return LLMMessage(role=Role.assistant, content=content, images=images)
def test_merge_prefers_self_images_when_present(
image_a: ImageAttachment, image_b: ImageAttachment
) -> None:
merged = _msg("hi", images=[image_a]) + _msg(" there", images=[image_b])
assert merged.images == [image_a]
def test_merge_falls_back_to_other_when_self_is_none(image_b: ImageAttachment) -> None:
merged = _msg("hi") + _msg(" there", images=[image_b])
assert merged.images == [image_b]
def test_merge_preserves_explicit_empty_self_images_over_other(
image_b: ImageAttachment,
) -> None:
# An explicit `[]` (intentional clearing) must NOT silently inherit
# `other.images`. Truthy/falsy-based merging would have flipped this.
merged = _msg("hi", images=[]) + _msg(" there", images=[image_b])
assert merged.images == []
def test_merge_yields_none_when_both_sides_are_none() -> None:
merged = _msg("hi") + _msg(" there")
assert merged.images is None

114
tests/core/test_local_config_files.py Normal file
View file

@ -0,0 +1,114 @@
from __future__ import annotations
from pathlib import Path
from vibe.core.paths._local_config_files import LocalConfigDirs, find_local_config_dirs
class TestSubdirs:
def test_finds_config_at_root(self, tmp_path: Path) -> None:
(tmp_path / ".vibe" / "tools").mkdir(parents=True)
result = find_local_config_dirs(tmp_path)
assert tmp_path.resolve() / ".vibe" / "tools" in result.tools
def test_does_not_descend_into_subdirectories(self, tmp_path: Path) -> None:
(tmp_path / "sub" / ".vibe" / "tools").mkdir(parents=True)
(tmp_path / "a" / "b" / ".vibe" / "skills").mkdir(parents=True)
result = find_local_config_dirs(tmp_path)
assert result.tools == ()
assert result.skills == ()
assert result.agents == ()
assert result.config_dirs == ()
def test_finds_agents_skills_at_root(self, tmp_path: Path) -> None:
(tmp_path / ".agents" / "skills").mkdir(parents=True)
result = find_local_config_dirs(tmp_path)
assert tmp_path.resolve() / ".agents" / "skills" in result.skills
def test_finds_all_config_types_at_root(self, tmp_path: Path) -> None:
(tmp_path / ".vibe" / "tools").mkdir(parents=True)
(tmp_path / ".vibe" / "skills").mkdir(parents=True)
(tmp_path / ".vibe" / "agents").mkdir(parents=True)
(tmp_path / ".agents" / "skills").mkdir(parents=True)
result = find_local_config_dirs(tmp_path)
resolved = tmp_path.resolve()
assert resolved / ".vibe" / "tools" in result.tools
assert resolved / ".vibe" / "skills" in result.skills
assert resolved / ".vibe" / "agents" in result.agents
assert resolved / ".agents" / "skills" in result.skills
class TestConfigDirs:
def test_finds_vibe_with_tools(self, tmp_path: Path) -> None:
(tmp_path / ".vibe" / "tools").mkdir(parents=True)
result = find_local_config_dirs(tmp_path)
assert tmp_path.resolve() / ".vibe" in result.config_dirs
def test_finds_vibe_with_skills(self, tmp_path: Path) -> None:
(tmp_path / ".vibe" / "skills").mkdir(parents=True)
result = find_local_config_dirs(tmp_path)
assert tmp_path.resolve() / ".vibe" in result.config_dirs
def test_finds_agents_with_skills(self, tmp_path: Path) -> None:
(tmp_path / ".agents" / "skills").mkdir(parents=True)
result = find_local_config_dirs(tmp_path)
assert tmp_path.resolve() / ".agents" in result.config_dirs
def test_ignores_empty_vibe_dir(self, tmp_path: Path) -> None:
(tmp_path / ".vibe").mkdir()
result = find_local_config_dirs(tmp_path)
assert result.config_dirs == ()
def test_ignores_empty_agents_dir(self, tmp_path: Path) -> None:
(tmp_path / ".agents").mkdir()
result = find_local_config_dirs(tmp_path)
assert result.config_dirs == ()
def test_returns_empty_when_empty(self, tmp_path: Path) -> None:
result = find_local_config_dirs(tmp_path)
assert result.config_dirs == ()
def test_finds_vibe_with_prompts(self, tmp_path: Path) -> None:
(tmp_path / ".vibe" / "prompts").mkdir(parents=True)
result = find_local_config_dirs(tmp_path)
assert tmp_path.resolve() / ".vibe" in result.config_dirs
def test_finds_vibe_with_config_toml(self, tmp_path: Path) -> None:
(tmp_path / ".vibe").mkdir()
(tmp_path / ".vibe" / "config.toml").write_text("")
result = find_local_config_dirs(tmp_path)
assert tmp_path.resolve() / ".vibe" in result.config_dirs
def test_finds_vibe_and_agents_at_same_root(self, tmp_path: Path) -> None:
(tmp_path / ".vibe" / "skills").mkdir(parents=True)
(tmp_path / ".agents" / "skills").mkdir(parents=True)
result = find_local_config_dirs(tmp_path)
resolved = tmp_path.resolve()
assert resolved / ".vibe" in result.config_dirs
assert resolved / ".agents" in result.config_dirs
class TestLocalConfigDirsOr:
def test_or_concatenates_each_field(self) -> None:
a = LocalConfigDirs(
config_dirs=(Path("/a/.vibe"),),
tools=(Path("/a/.vibe/tools"),),
skills=(Path("/a/.vibe/skills"),),
agents=(Path("/a/.vibe/agents"),),
)
b = LocalConfigDirs(
config_dirs=(Path("/b/.vibe"),),
tools=(Path("/b/.vibe/tools"),),
skills=(Path("/b/.vibe/skills"),),
agents=(Path("/b/.vibe/agents"),),
)
merged = a | b
assert merged.config_dirs == (Path("/a/.vibe"), Path("/b/.vibe"))
assert merged.tools == (Path("/a/.vibe/tools"), Path("/b/.vibe/tools"))
assert merged.skills == (Path("/a/.vibe/skills"), Path("/b/.vibe/skills"))
assert merged.agents == (Path("/a/.vibe/agents"), Path("/b/.vibe/agents"))
def test_or_with_empty_is_identity(self) -> None:
a = LocalConfigDirs(tools=(Path("/a/.vibe/tools"),))
assert (a | LocalConfigDirs()) == a
assert (LocalConfigDirs() | a) == a

195
tests/core/test_local_config_walk.py
View file

@ -1,195 +0,0 @@
from __future__ import annotations
from pathlib import Path
from vibe.core.paths._local_config_walk import (
_MAX_DIRS,
WALK_MAX_DEPTH,
ConfigWalkResult,
walk_local_config_dirs,
)
class TestWalkTools:
def test_finds_config_at_root(self, tmp_path: Path) -> None:
(tmp_path / ".vibe" / "tools").mkdir(parents=True)
result = walk_local_config_dirs(tmp_path)
assert tmp_path.resolve() / ".vibe" / "tools" in result.tools
def test_finds_config_within_depth_limit(self, tmp_path: Path) -> None:
nested = tmp_path
for i in range(WALK_MAX_DEPTH):
nested = nested / f"level{i}"
(nested / ".vibe" / "skills").mkdir(parents=True)
result = walk_local_config_dirs(tmp_path)
assert nested.resolve() / ".vibe" / "skills" in result.skills
def test_does_not_find_config_beyond_depth_limit(self, tmp_path: Path) -> None:
nested = tmp_path
for i in range(WALK_MAX_DEPTH + 1):
nested = nested / f"level{i}"
(nested / ".vibe" / "tools").mkdir(parents=True)
result = walk_local_config_dirs(tmp_path)
assert not result.tools
assert not result.skills
assert not result.agents
def test_respects_dir_count_limit(self, tmp_path: Path) -> None:
for i in range(_MAX_DIRS + 10):
(tmp_path / f"dir{i:05d}").mkdir()
(tmp_path / "zzz_last" / ".vibe" / "tools").mkdir(parents=True)
result = walk_local_config_dirs(tmp_path)
assert isinstance(result.tools, tuple)
def test_skips_ignored_directories(self, tmp_path: Path) -> None:
(tmp_path / "node_modules" / ".vibe" / "tools").mkdir(parents=True)
(tmp_path / ".vibe" / "tools").mkdir(parents=True)
result = walk_local_config_dirs(tmp_path)
assert result.tools == (tmp_path.resolve() / ".vibe" / "tools",)
def test_skips_dot_directories(self, tmp_path: Path) -> None:
(tmp_path / ".hidden" / ".vibe" / "tools").mkdir(parents=True)
result = walk_local_config_dirs(tmp_path)
assert not result.tools
def test_preserves_alphabetical_ordering(self, tmp_path: Path) -> None:
(tmp_path / "bbb" / ".vibe" / "tools").mkdir(parents=True)
(tmp_path / "aaa" / ".vibe" / "tools").mkdir(parents=True)
(tmp_path / ".vibe" / "tools").mkdir(parents=True)
result = walk_local_config_dirs(tmp_path)
resolved = tmp_path.resolve()
assert result.tools == (
resolved / ".vibe" / "tools",
resolved / "aaa" / ".vibe" / "tools",
resolved / "bbb" / ".vibe" / "tools",
)
def test_finds_agents_skills(self, tmp_path: Path) -> None:
(tmp_path / ".agents" / "skills").mkdir(parents=True)
result = walk_local_config_dirs(tmp_path)
assert tmp_path.resolve() / ".agents" / "skills" in result.skills
def test_finds_all_config_types(self, tmp_path: Path) -> None:
(tmp_path / ".vibe" / "tools").mkdir(parents=True)
(tmp_path / ".vibe" / "skills").mkdir(parents=True)
(tmp_path / ".vibe" / "agents").mkdir(parents=True)
(tmp_path / ".agents" / "skills").mkdir(parents=True)
result = walk_local_config_dirs(tmp_path)
resolved = tmp_path.resolve()
assert resolved / ".vibe" / "tools" in result.tools
assert resolved / ".vibe" / "skills" in result.skills
assert resolved / ".vibe" / "agents" in result.agents
assert resolved / ".agents" / "skills" in result.skills
class TestWalkConfigDirs:
def test_finds_vibe_with_tools(self, tmp_path: Path) -> None:
(tmp_path / ".vibe" / "tools").mkdir(parents=True)
result = walk_local_config_dirs(tmp_path)
assert tmp_path.resolve() / ".vibe" in result.config_dirs
def test_finds_vibe_with_skills(self, tmp_path: Path) -> None:
(tmp_path / ".vibe" / "skills").mkdir(parents=True)
result = walk_local_config_dirs(tmp_path)
assert tmp_path.resolve() / ".vibe" in result.config_dirs
def test_finds_agents_with_skills(self, tmp_path: Path) -> None:
(tmp_path / ".agents" / "skills").mkdir(parents=True)
result = walk_local_config_dirs(tmp_path)
assert tmp_path.resolve() / ".agents" in result.config_dirs
def test_ignores_empty_vibe_dir(self, tmp_path: Path) -> None:
(tmp_path / ".vibe").mkdir()
result = walk_local_config_dirs(tmp_path)
assert result.config_dirs == ()
def test_ignores_empty_agents_dir(self, tmp_path: Path) -> None:
(tmp_path / ".agents").mkdir()
result = walk_local_config_dirs(tmp_path)
assert result.config_dirs == ()
def test_returns_empty_when_empty(self, tmp_path: Path) -> None:
result = walk_local_config_dirs(tmp_path)
assert result.config_dirs == ()
def test_finds_shallow_nested(self, tmp_path: Path) -> None:
(tmp_path / "sub" / ".vibe" / "skills").mkdir(parents=True)
result = walk_local_config_dirs(tmp_path)
assert tmp_path.resolve() / "sub" / ".vibe" in result.config_dirs
def test_finds_at_depth_2(self, tmp_path: Path) -> None:
(tmp_path / "a" / "b" / ".agents" / "skills").mkdir(parents=True)
result = walk_local_config_dirs(tmp_path)
assert tmp_path.resolve() / "a" / "b" / ".agents" in result.config_dirs
def test_returns_empty_beyond_default_depth(self, tmp_path: Path) -> None:
(tmp_path / "a" / "b" / "c" / "d" / "e" / ".vibe" / "tools").mkdir(parents=True)
result = walk_local_config_dirs(tmp_path)
assert result.config_dirs == ()
def test_custom_depth(self, tmp_path: Path) -> None:
(tmp_path / "a" / "b" / "c" / "d" / "e" / ".vibe" / "tools").mkdir(parents=True)
result = walk_local_config_dirs(tmp_path, max_depth=5)
assert (
tmp_path.resolve() / "a" / "b" / "c" / "d" / "e" / ".vibe"
in result.config_dirs
)
def test_finds_match_among_many_dirs(self, tmp_path: Path) -> None:
(tmp_path / ".vibe" / "tools").mkdir(parents=True)
for i in range(100):
(tmp_path / f"dir{i}").mkdir()
result = walk_local_config_dirs(tmp_path)
assert tmp_path.resolve() / ".vibe" in result.config_dirs
def test_skips_ignored_directories(self, tmp_path: Path) -> None:
(tmp_path / "node_modules" / ".vibe" / "skills").mkdir(parents=True)
result = walk_local_config_dirs(tmp_path)
assert result.config_dirs == ()
def test_finds_vibe_with_prompts(self, tmp_path: Path) -> None:
(tmp_path / ".vibe" / "prompts").mkdir(parents=True)
result = walk_local_config_dirs(tmp_path)
assert tmp_path.resolve() / ".vibe" in result.config_dirs
def test_finds_vibe_with_config_toml(self, tmp_path: Path) -> None:
(tmp_path / ".vibe").mkdir()
(tmp_path / ".vibe" / "config.toml").write_text("")
result = walk_local_config_dirs(tmp_path)
assert tmp_path.resolve() / ".vibe" in result.config_dirs
def test_finds_multiple_config_dirs(self, tmp_path: Path) -> None:
(tmp_path / ".vibe" / "skills").mkdir(parents=True)
(tmp_path / ".agents" / "skills").mkdir(parents=True)
(tmp_path / "sub" / ".vibe" / "tools").mkdir(parents=True)
result = walk_local_config_dirs(tmp_path)
resolved = tmp_path.resolve()
assert resolved / ".vibe" in result.config_dirs
assert resolved / ".agents" in result.config_dirs
assert resolved / "sub" / ".vibe" in result.config_dirs
class TestConfigWalkResultOr:
def test_or_concatenates_each_field(self) -> None:
a = ConfigWalkResult(
config_dirs=(Path("/a/.vibe"),),
tools=(Path("/a/.vibe/tools"),),
skills=(Path("/a/.vibe/skills"),),
agents=(Path("/a/.vibe/agents"),),
)
b = ConfigWalkResult(
config_dirs=(Path("/b/.vibe"),),
tools=(Path("/b/.vibe/tools"),),
skills=(Path("/b/.vibe/skills"),),
agents=(Path("/b/.vibe/agents"),),
)
merged = a | b
assert merged.config_dirs == (Path("/a/.vibe"), Path("/b/.vibe"))
assert merged.tools == (Path("/a/.vibe/tools"), Path("/b/.vibe/tools"))
assert merged.skills == (Path("/a/.vibe/skills"), Path("/b/.vibe/skills"))
assert merged.agents == (Path("/a/.vibe/agents"), Path("/b/.vibe/agents"))
def test_or_with_empty_is_identity(self) -> None:
a = ConfigWalkResult(tools=(Path("/a/.vibe/tools"),))
assert (a | ConfigWalkResult()) == a
assert (ConfigWalkResult() | a) == a

3
tests/core/test_overrides_layer.py
View file

@ -3,6 +3,7 @@ from __future__ import annotations
import pytest
from vibe.core.config.layers.overrides import OverridesLayer
from vibe.core.config.patch import ConfigPatch
@pytest.mark.asyncio
@ -23,7 +24,7 @@ async def test_always_trusted() -> None:
async def test_apply_raises_not_implemented() -> None:
layer = OverridesLayer(data={})
with pytest.raises(NotImplementedError, match="M2"):
await layer.apply({"op": "set"})
await layer.apply(ConfigPatch(fingerprint="fp-1"))
@pytest.mark.asyncio

3
tests/core/test_project_config_layer.py
View file

@ -6,6 +6,7 @@ import pytest
from vibe.core.config.layer import UntrustedLayerError
from vibe.core.config.layers.project import ProjectConfigLayer
from vibe.core.config.patch import ConfigPatch
from vibe.core.paths._vibe_home import GlobalPath
from vibe.core.trusted_folders import trusted_folders_manager
@ -66,7 +67,7 @@ async def test_default_name(tmp_working_directory: Path) -> None:
async def test_apply_raises_not_implemented(tmp_working_directory: Path) -> None:
layer = ProjectConfigLayer(path=tmp_working_directory)
with pytest.raises(NotImplementedError, match="M2"):
await layer.apply({"op": "set"})
await layer.apply(ConfigPatch(fingerprint="fp-1"))
@pytest.mark.asyncio

4
tests/core/test_remote_agent_loop.py
View file

@ -413,7 +413,6 @@ def test_working_task_promoted_to_real_tool_call_does_not_create_duplicate_row()
"output": {
"path": "/workspace/hello.py",
"bytes_written": 22,
"file_existed": False,
"content": 'print("Hello, World!")',
},
},
@ -456,7 +455,6 @@ def test_idle_boundary_waits_for_open_tool_results() -> None:
"output": {
"path": "/workspace/hello_world.js",
"bytes_written": 29,
"file_existed": False,
"content": "console.log('Hello, World!');",
},
},
@ -731,7 +729,7 @@ def test_working_completed_with_tool_call_id_emits_error_result() -> None:
"working",
{
"title": "Executing write_file",
"content": "Error: File exists. Set overwrite=True.",
"content": "Error: Permission denied.",
"toolUIState": {
"type": "file",
"toolCallId": "call-write-err",

131
tests/core/test_retry.py
View file

@ -1,9 +1,15 @@
from __future__ import annotations
from collections.abc import AsyncGenerator
import httpx
import pytest
from vibe.core.utils.retry import _is_retryable_http_error
from vibe.core.utils.retry import (
_is_retryable_http_error,
async_generator_retry,
async_retry,
)
def _make_http_status_error(status_code: int) -> httpx.HTTPStatusError:
@ -15,6 +21,10 @@ def _make_http_status_error(status_code: int) -> httpx.HTTPStatusError:
)
def _make_request(url: str = "https://example.com") -> httpx.Request:
return httpx.Request("POST", url)
class TestIsRetryableHttpError:
@pytest.mark.parametrize("code", [408, 409, 425, 429, 500, 502, 503, 504, 529])
def test_retryable_codes(self, code: int) -> None:
@ -24,8 +34,127 @@ class TestIsRetryableHttpError:
def test_non_retryable_codes(self, code: int) -> None:
assert _is_retryable_http_error(_make_http_status_error(code)) is False
@pytest.mark.parametrize(
"exc",
[
httpx.ConnectTimeout("connect timed out", request=_make_request()),
httpx.ReadTimeout("read timed out", request=_make_request()),
httpx.WriteTimeout("write timed out", request=_make_request()),
httpx.PoolTimeout("pool timed out", request=_make_request()),
httpx.ConnectError("connection refused", request=_make_request()),
httpx.ReadError("read failed", request=_make_request()),
httpx.WriteError("write failed", request=_make_request()),
httpx.RemoteProtocolError("server disconnected", request=_make_request()),
],
)
def test_retryable_network_errors(self, exc: Exception) -> None:
assert _is_retryable_http_error(exc) is True
def test_non_retryable_request_error(self) -> None:
assert _is_retryable_http_error(httpx.InvalidURL("bad url")) is False
def test_non_http_error_returns_false(self) -> None:
assert _is_retryable_http_error(ValueError("not http")) is False
def test_generic_exception_returns_false(self) -> None:
assert _is_retryable_http_error(RuntimeError("boom")) is False
class TestAsyncRetry:
@pytest.mark.asyncio
async def test_retries_network_error_then_succeeds(self) -> None:
attempts = 0
@async_retry(tries=3, delay_seconds=0.0, backoff_factor=1.0)
async def call() -> str:
nonlocal attempts
attempts += 1
if attempts < 2:
raise httpx.ConnectTimeout("timeout", request=_make_request())
return "ok"
result = await call()
assert result == "ok"
assert attempts == 2
@pytest.mark.asyncio
async def test_does_not_retry_non_retryable(self) -> None:
attempts = 0
@async_retry(tries=3, delay_seconds=0.0, backoff_factor=1.0)
async def call() -> str:
nonlocal attempts
attempts += 1
raise ValueError("nope")
with pytest.raises(ValueError):
await call()
assert attempts == 1
@pytest.mark.asyncio
async def test_exhausts_retries(self) -> None:
attempts = 0
@async_retry(tries=3, delay_seconds=0.0, backoff_factor=1.0)
async def call() -> str:
nonlocal attempts
attempts += 1
raise httpx.ReadTimeout("timeout", request=_make_request())
with pytest.raises(httpx.ReadTimeout):
await call()
assert attempts == 3
class TestAsyncGeneratorRetry:
@pytest.mark.asyncio
async def test_retries_before_first_yield(self) -> None:
attempts = 0
@async_generator_retry(tries=3, delay_seconds=0.0, backoff_factor=1.0)
async def gen() -> AsyncGenerator[int]:
nonlocal attempts
attempts += 1
if attempts < 2:
raise httpx.ConnectError("connect failed", request=_make_request())
yield 1
yield 2
items = [item async for item in gen()]
assert items == [1, 2]
assert attempts == 2
@pytest.mark.asyncio
async def test_does_not_retry_after_first_yield(self) -> None:
attempts = 0
@async_generator_retry(tries=3, delay_seconds=0.0, backoff_factor=1.0)
async def gen() -> AsyncGenerator[int]:
nonlocal attempts
attempts += 1
yield 1
raise httpx.ReadError("midstream", request=_make_request())
items: list[int] = []
with pytest.raises(httpx.ReadError):
async for item in gen():
items.append(item)
assert items == [1]
assert attempts == 1
@pytest.mark.asyncio
async def test_does_not_retry_non_retryable_before_yield(self) -> None:
attempts = 0
@async_generator_retry(tries=3, delay_seconds=0.0, backoff_factor=1.0)
async def gen() -> AsyncGenerator[int]:
nonlocal attempts
attempts += 1
raise ValueError("nope")
yield 0 # pragma: no cover
with pytest.raises(ValueError):
async for _ in gen():
pass
assert attempts == 1

47
tests/core/test_rewind_integration.py
View file

@ -25,23 +25,24 @@ async def _act_and_collect(agent_loop, prompt: str) -> list[BaseEvent]:
def _write_file_tool_call(
path: str, content: str, *, call_id: str = "call_1", overwrite: bool = False
path: str, content: str, *, call_id: str = "call_1"
) -> ToolCall:
args = json.dumps({"path": path, "content": content, "overwrite": overwrite})
args = json.dumps({"path": path, "content": content})
return ToolCall(
id=call_id, index=0, function=FunctionCall(name="write_file", arguments=args)
)
def _search_replace_tool_call(
file_path: str, search: str, replace: str, *, call_id: str = "call_1"
def _edit_tool_call(
file_path: str, old_string: str, new_string: str, *, call_id: str = "call_1"
) -> ToolCall:
content = f"<<<<<<< SEARCH\n{search}\n=======\n{replace}\n>>>>>>> REPLACE"
args = json.dumps({"file_path": file_path, "content": content})
args = json.dumps({
"file_path": file_path,
"old_string": old_string,
"new_string": new_string,
})
return ToolCall(
id=call_id,
index=0,
function=FunctionCall(name="search_replace", arguments=args),
id=call_id, index=0, function=FunctionCall(name="edit", arguments=args)
)
@ -54,10 +55,10 @@ def _bash_tool_call(command: str, *, call_id: str = "call_1") -> ToolCall:
def _make_agent_loop(backend: FakeBackend):
config = build_test_vibe_config(
enabled_tools=["write_file", "search_replace", "bash"],
enabled_tools=["write_file", "edit", "bash"],
tools={
"write_file": {"permission": "always"},
"search_replace": {"permission": "always"},
"edit": {"permission": "always"},
"bash": {"permission": "always"},
},
system_prompt_id="tests",
@ -98,10 +99,10 @@ class TestRewindIntegration:
assert not target.exists()
async def test_search_replace_rewind_restores_previous_version(
async def test_edit_rewind_restores_previous_version(
self, tmp_working_directory: Path
) -> None:
"""Edit a pre-existing file with search_replace, rewind restores original."""
"""Edit a pre-existing file with edit, rewind restores original."""
target = tmp_working_directory / "config.yaml"
target.write_text("key: original\n", encoding="utf-8")
@ -110,9 +111,7 @@ class TestRewindIntegration:
mock_llm_chunk(
content="Updating config.",
tool_calls=[
_search_replace_tool_call(
str(target), "key: original", "key: modified"
)
_edit_tool_call(str(target), "key: original", "key: modified")
],
)
],
@ -129,11 +128,11 @@ class TestRewindIntegration:
assert target.read_text() == "key: original\n"
async def test_write_then_search_replace_rewind_to_middle(
async def test_write_then_edit_rewind_to_middle(
self, tmp_working_directory: Path
) -> None:
"""Turn 1 creates a file with write_file, turn 2 patches it with
search_replace. Rewind to turn 2 restores the turn 1 version.
edit. Rewind to turn 2 restores the turn 1 version.
"""
target = tmp_working_directory / "app.py"
@ -148,12 +147,12 @@ class TestRewindIntegration:
)
],
[mock_llm_chunk(content="Created.")],
# Turn 2: patch with search_replace
# Turn 2: patch with edit
[
mock_llm_chunk(
content="Updating.",
tool_calls=[
_search_replace_tool_call(
_edit_tool_call(
str(target),
" pass",
' print("hello")',
@ -222,9 +221,7 @@ class TestRewindIntegration:
mock_llm_chunk(
content="v2.",
tool_calls=[
_write_file_tool_call(
str(target), "v2", call_id="call_2", overwrite=True
)
_edit_tool_call(str(target), "v1", "v2", call_id="call_2")
],
)
],
@ -234,9 +231,7 @@ class TestRewindIntegration:
mock_llm_chunk(
content="v2bis.",
tool_calls=[
_write_file_tool_call(
str(target), "v2bis", call_id="call_3", overwrite=True
)
_edit_tool_call(str(target), "v1", "v2bis", call_id="call_3")
],
)
],

34
tests/core/test_search_replace_encoding.py
View file

@ -1,34 +0,0 @@
from __future__ import annotations
from pathlib import Path
import pytest
from tests.mock.utils import collect_result
from vibe.core.tools.base import BaseToolState
from vibe.core.tools.builtins.search_replace import (
SearchReplace,
SearchReplaceArgs,
SearchReplaceConfig,
)
@pytest.mark.asyncio
async def test_search_replace_rewrites_with_detected_encoding(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
path = tmp_path / "utf16.txt"
original = "line one café\nline two été\n"
path.write_bytes(original.encode("utf-16"))
tool = SearchReplace(
config_getter=lambda: SearchReplaceConfig(), state=BaseToolState()
)
patch = "<<<<<<< SEARCH\nline one café\n=======\nLINE ONE CAFÉ\n>>>>>>> REPLACE"
await collect_result(
tool.run(SearchReplaceArgs(file_path=str(path), content=patch))
)
assert path.read_bytes().startswith(b"\xff\xfe")
assert path.read_text(encoding="utf-16") == "LINE ONE CAFÉ\nline two été\n"

29
tests/core/test_telemetry_send.py
View file

@ -28,9 +28,7 @@ def _make_resolved_tool_call(
tool_name: str, args_dict: dict[str, Any]
) -> ResolvedToolCall:
if tool_name == "write_file":
validated = WriteFileArgs(
path="foo.txt", content="x", overwrite=args_dict.get("overwrite", False)
)
validated = WriteFileArgs(path="foo.txt", content="x")
cls: type[BaseTool] = WriteFile
else:
validated = FakeToolArgs()
@ -169,12 +167,12 @@ class TestTelemetryClient:
assert telemetry_events[0]["properties"]["message_id"] == "msg-123"
def test_send_tool_call_finished_nb_files_created_write_file_new(
def test_send_tool_call_finished_nb_files_created_write_file(
self, telemetry_events: list[dict[str, Any]]
) -> None:
config = build_test_vibe_config(enable_telemetry=True)
client = TelemetryClient(config_getter=lambda: config)
tool_call = _make_resolved_tool_call("write_file", {"overwrite": False})
tool_call = _make_resolved_tool_call("write_file", {})
client.send_tool_call_finished(
tool_call=tool_call,
@ -182,31 +180,12 @@ class TestTelemetryClient:
decision=None,
agent_profile_name="default",
model="mistral-large",
result={"file_existed": False},
result={},
)
assert telemetry_events[0]["properties"]["nb_files_created"] == 1
assert telemetry_events[0]["properties"]["nb_files_modified"] == 0
def test_send_tool_call_finished_nb_files_modified_write_file_overwrite(
self, telemetry_events: list[dict[str, Any]]
) -> None:
config = build_test_vibe_config(enable_telemetry=True)
client = TelemetryClient(config_getter=lambda: config)
tool_call = _make_resolved_tool_call("write_file", {"overwrite": True})
client.send_tool_call_finished(
tool_call=tool_call,
status="success",
decision=None,
agent_profile_name="default",
model="mistral-large",
result={"file_existed": True},
)
assert telemetry_events[0]["properties"]["nb_files_created"] == 0
assert telemetry_events[0]["properties"]["nb_files_modified"] == 1
def test_send_tool_call_finished_decision_none(
self, telemetry_events: list[dict[str, Any]]
) -> None:

168
tests/core/test_trusted_folders.py
View file

@ -10,6 +10,8 @@ import tomli_w
from vibe.core.paths import AGENTS_MD_FILENAME, TRUSTED_FOLDERS_FILE
from vibe.core.trusted_folders import (
TrustedFoldersManager,
find_git_repo_ancestor,
find_repo_trustable_files_for_cwd,
find_trustable_files,
has_agents_md_file,
)
@ -291,15 +293,46 @@ class TestFindTrustRoot:
# child should find parent (closest), not tmp_path
assert manager.find_trust_root(child) == parent.resolve()
def test_ignores_untrusted_ancestors(self, tmp_path: Path) -> None:
def test_returns_none_when_closer_ancestor_is_untrusted(
self, tmp_path: Path
) -> None:
parent = tmp_path / "parent"
child = parent / "child"
child.mkdir(parents=True)
manager = TrustedFoldersManager()
manager.add_untrusted(parent)
manager.add_trusted(tmp_path)
# find_trust_root skips untrusted, finds tmp_path
assert manager.find_trust_root(child) == tmp_path.resolve()
# The closer untrusted ancestor blocks the higher trusted one.
assert manager.find_trust_root(child) is None
def test_returns_none_when_path_itself_is_untrusted(self, tmp_path: Path) -> None:
manager = TrustedFoldersManager()
manager.add_untrusted(tmp_path)
assert manager.find_trust_root(tmp_path) is None
class TestIsExplicitlyUntrusted:
def test_returns_true_for_path_in_untrusted_list(self, tmp_path: Path) -> None:
manager = TrustedFoldersManager()
manager.add_untrusted(tmp_path)
assert manager.is_explicitly_untrusted(tmp_path) is True
def test_returns_false_for_unknown_path(self, tmp_path: Path) -> None:
manager = TrustedFoldersManager()
assert manager.is_explicitly_untrusted(tmp_path) is False
def test_does_not_walk_ancestors(self, tmp_path: Path) -> None:
child = tmp_path / "child"
child.mkdir()
manager = TrustedFoldersManager()
manager.add_untrusted(tmp_path)
# parent is untrusted, but child itself is not in the list
assert manager.is_explicitly_untrusted(child) is False
def test_returns_false_for_trusted_path(self, tmp_path: Path) -> None:
manager = TrustedFoldersManager()
manager.add_trusted(tmp_path)
assert manager.is_explicitly_untrusted(tmp_path) is False
class TestHasAgentsMdFile:
@ -351,37 +384,132 @@ class TestFindTrustableFiles:
(tmp_path / "other.txt").write_text("", encoding="utf-8")
assert find_trustable_files(tmp_path) == []
def test_detects_vibe_config_in_subfolder(self, tmp_path: Path) -> None:
def test_ignores_vibe_config_in_subfolder(self, tmp_path: Path) -> None:
(tmp_path / "sub" / ".vibe" / "skills").mkdir(parents=True)
result = find_trustable_files(tmp_path)
assert "sub/.vibe/" in result
assert find_trustable_files(tmp_path) == []
def test_detects_agents_skills_in_subfolder(self, tmp_path: Path) -> None:
def test_ignores_agents_skills_in_subfolder(self, tmp_path: Path) -> None:
(tmp_path / "deep" / "nested" / ".agents" / "skills").mkdir(parents=True)
result = find_trustable_files(tmp_path)
assert "deep/nested/.agents/" in result
assert find_trustable_files(tmp_path) == []
def test_returns_empty_when_config_only_inside_ignored_dir(
self, tmp_path: Path
) -> None:
def test_returns_empty_when_config_only_in_subfolder(self, tmp_path: Path) -> None:
(tmp_path / "node_modules" / ".vibe" / "skills").mkdir(parents=True)
assert find_trustable_files(tmp_path) == []
def test_detects_nested_vibe_dir(self, tmp_path: Path) -> None:
(tmp_path / "pkg" / ".vibe" / "tools").mkdir(parents=True)
result = find_trustable_files(tmp_path)
assert "pkg/.vibe/" in result
def test_detects_multiple_files(self, tmp_path: Path) -> None:
def test_detects_multiple_files_at_root(self, tmp_path: Path) -> None:
(tmp_path / ".vibe" / "skills").mkdir(parents=True)
(tmp_path / ".agents" / "skills").mkdir(parents=True)
(tmp_path / "AGENTS.md").write_text("# Agent", encoding="utf-8")
(tmp_path / "sub" / ".agents" / "skills").mkdir(parents=True)
result = find_trustable_files(tmp_path)
assert ".vibe/" in result
assert ".agents/" in result
assert "AGENTS.md" in result
assert "sub/.agents/" in result
def test_no_duplicates_for_root_vibe_dir(self, tmp_path: Path) -> None:
(tmp_path / ".vibe" / "tools").mkdir(parents=True)
result = find_trustable_files(tmp_path)
assert result.count(".vibe/") == 1
def _make_git_repo(path: Path) -> None:
git_dir = path / ".git"
git_dir.mkdir()
(git_dir / "HEAD").write_text("ref: refs/heads/main\n", encoding="utf-8")
class TestFindRepoTrustableFilesForCwd:
def test_returns_empty_when_repo_root_is_none(self, tmp_path: Path) -> None:
assert find_repo_trustable_files_for_cwd(tmp_path, None) == []
def test_returns_empty_when_repo_root_is_not_ancestor(self, tmp_path: Path) -> None:
cwd = tmp_path / "cwd"
cwd.mkdir()
repo = tmp_path / "repo"
repo.mkdir()
_make_git_repo(repo)
assert find_repo_trustable_files_for_cwd(cwd, repo) == []
def test_includes_root_trustable_files(self, tmp_path: Path) -> None:
repo = tmp_path / "repo"
repo.mkdir()
_make_git_repo(repo)
(repo / ".vibe" / "skills").mkdir(parents=True)
(repo / ".agents" / "skills").mkdir(parents=True)
(repo / "AGENTS.md").write_text("# Agent", encoding="utf-8")
cwd = repo / "src" / "pkg"
cwd.mkdir(parents=True)
assert find_repo_trustable_files_for_cwd(cwd, repo) == [
".agents/",
".vibe/",
"AGENTS.md",
]
def test_includes_agents_md_between_cwd_and_repo_root(self, tmp_path: Path) -> None:
repo = tmp_path / "repo"
repo.mkdir()
_make_git_repo(repo)
cwd = repo / "src" / "pkg" / "deep"
cwd.mkdir(parents=True)
(repo / "src" / "AGENTS.md").write_text("# Source", encoding="utf-8")
(repo / "src" / "pkg" / "AGENTS.md").write_text("# Package", encoding="utf-8")
(cwd / "AGENTS.md").write_text("# Cwd", encoding="utf-8")
assert find_repo_trustable_files_for_cwd(cwd, repo) == [
"src/AGENTS.md",
"src/pkg/AGENTS.md",
]
class TestFindGitRepoAncestor:
def test_returns_path_when_directly_contains_git(self, tmp_path: Path) -> None:
_make_git_repo(tmp_path)
assert find_git_repo_ancestor(tmp_path) == tmp_path.resolve()
def test_ignores_git_file_pointer(self, tmp_path: Path) -> None:
(tmp_path / ".git").write_text("gitdir: /elsewhere", encoding="utf-8")
# Not a directory, so not treated as a repo root.
assert find_git_repo_ancestor(tmp_path) is None
def test_ignores_empty_git_directory(self, tmp_path: Path) -> None:
(tmp_path / ".git").mkdir()
# Missing HEAD, so not a real repo.
assert find_git_repo_ancestor(tmp_path) is None
def test_returns_closest_ancestor_with_git(self, tmp_path: Path) -> None:
_make_git_repo(tmp_path)
nested = tmp_path / "a" / "b" / "c"
nested.mkdir(parents=True)
assert find_git_repo_ancestor(nested) == tmp_path.resolve()
def test_returns_innermost_when_multiple_git_repos(self, tmp_path: Path) -> None:
_make_git_repo(tmp_path)
inner = tmp_path / "sub"
inner.mkdir()
_make_git_repo(inner)
child = inner / "deep"
child.mkdir()
assert find_git_repo_ancestor(child) == inner.resolve()
def test_returns_none_when_no_git_anywhere(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(Path, "home", classmethod(lambda _cls: tmp_path.parent))
assert find_git_repo_ancestor(tmp_path / "a") is None
def test_excludes_home_directory(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
home = tmp_path / "home"
home.mkdir()
_make_git_repo(home)
monkeypatch.setattr(Path, "home", classmethod(lambda _cls: home))
sub = home / "project"
sub.mkdir()
assert find_git_repo_ancestor(sub) is None
def test_terminates_at_filesystem_root_without_git(
self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.setattr(Path, "home", classmethod(lambda _cls: tmp_path / "nope"))
assert find_git_repo_ancestor(tmp_path) is None

3
tests/core/test_user_config_layer.py
View file

@ -6,6 +6,7 @@ import pytest
from vibe.core.config.layer import LayerImplementationError
from vibe.core.config.layers.user import UserConfigLayer
from vibe.core.config.patch import ConfigPatch
@pytest.mark.asyncio
@ -43,7 +44,7 @@ async def test_apply_raises_not_implemented(tmp_working_directory: Path) -> None
path = tmp_working_directory / "config.toml"
layer = UserConfigLayer(path=path, name="user-toml")
with pytest.raises(NotImplementedError, match="M2"):
await layer.apply({"op": "set"})
await layer.apply(ConfigPatch(fingerprint="fp-1"))
@pytest.mark.asyncio

135
tests/core/test_utils.py
View file

@ -1,12 +1,21 @@
from __future__ import annotations
import asyncio
from pathlib import Path
import pytest
from vibe.core.utils import compact_complete_display, get_server_url_from_api_base
import vibe.core.utils.io as io_utils
from vibe.core.utils.io import decode_safe, read_safe, read_safe_async
from vibe.core.utils.io import (
_FILE_WRITE_LOCKS,
decode_safe,
file_write_lock,
read_lines_safe,
read_lines_safe_async,
read_safe,
read_safe_async,
)
@pytest.mark.parametrize(
@ -201,3 +210,127 @@ class TestReadSafeAsync:
assert (await read_safe_async(f, raise_on_error=False)).text == "maf<EFBFBD>\n"
with pytest.raises(UnicodeDecodeError):
await read_safe_async(f, raise_on_error=True)
class TestReadLinesSafe:
def test_small_file_fully_read(self, tmp_path: Path) -> None:
f = tmp_path / "f.txt"
f.write_text("a\nb\nc\n", encoding="utf-8")
got = read_lines_safe(f, limit=100, max_bytes=1024)
assert got.lines == ["a", "b", "c"]
assert got.total_lines == 3
assert got.was_truncated is False
def test_no_trailing_newline(self, tmp_path: Path) -> None:
f = tmp_path / "f.txt"
f.write_text("a\nb", encoding="utf-8")
got = read_lines_safe(f, limit=100, max_bytes=1024)
assert got.lines == ["a", "b"]
assert got.total_lines == 2
assert got.was_truncated is False
def test_truncates_at_limit(self, tmp_path: Path) -> None:
f = tmp_path / "f.txt"
f.write_text("".join(f"line {i}\n" for i in range(1, 101)), encoding="utf-8")
got = read_lines_safe(f, limit=10, max_bytes=1024)
assert got.lines == [f"line {i}" for i in range(1, 11)]
assert got.total_lines is None
assert got.was_truncated is True
def test_offset_skips_leading_lines(self, tmp_path: Path) -> None:
f = tmp_path / "f.txt"
f.write_text("".join(f"line {i}\n" for i in range(1, 11)), encoding="utf-8")
got = read_lines_safe(f, start_line=3, limit=2, max_bytes=1024)
assert got.lines == ["line 3", "line 4"]
assert got.was_truncated is True
def test_offset_past_eof_reports_total(self, tmp_path: Path) -> None:
f = tmp_path / "f.txt"
f.write_text("a\nb\n", encoding="utf-8")
got = read_lines_safe(f, start_line=100, limit=10, max_bytes=1024)
assert got.lines == []
assert got.total_lines == 2
assert got.was_truncated is False
def test_does_not_load_whole_file(self, tmp_path: Path) -> None:
f = tmp_path / "big.txt"
f.write_text("".join(f"line {i}\n" for i in range(1_000_000)), encoding="utf-8")
got = read_lines_safe(f, limit=5, max_bytes=1024)
assert got.lines == [f"line {i}" for i in range(5)]
assert got.total_lines is None
assert got.was_truncated is True
def test_oversized_single_line_returns_partial(self, tmp_path: Path) -> None:
f = tmp_path / "f.txt"
f.write_text("x" * 5000 + "\n", encoding="utf-8")
got = read_lines_safe(f, limit=10, max_bytes=1024)
assert got.lines == ["x" * 1024]
assert got.was_truncated is True
def test_cumulative_byte_budget_truncates(self, tmp_path: Path) -> None:
f = tmp_path / "f.txt"
f.write_text("".join("x" * 200 + "\n" for _ in range(50)), encoding="utf-8")
got = read_lines_safe(f, limit=50, max_bytes=1024)
assert 0 < len(got.lines) < 50
assert got.total_lines is None
assert got.was_truncated is True
def test_oversized_unselected_line_is_skipped(self, tmp_path: Path) -> None:
f = tmp_path / "f.txt"
f.write_text("x" * 5000 + "\nkept\n", encoding="utf-8")
got = read_lines_safe(f, start_line=2, limit=10, max_bytes=1024)
assert got.lines == ["kept"]
@pytest.mark.parametrize("encoding", ["utf-16-le", "utf-16-be", "utf-16"])
def test_utf16_is_decoded(self, tmp_path: Path, encoding: str) -> None:
f = tmp_path / "u16.txt"
f.write_bytes("héllo\nwörld\n".encode(encoding))
got = read_lines_safe(f, limit=10, max_bytes=4096)
assert got.lines[-1] == "wörld"
# A leading BOM may remain as U+FEFF on the first line.
assert got.lines[0].endswith("héllo")
@pytest.mark.asyncio
async def test_async_matches_sync(self, tmp_path: Path) -> None:
f = tmp_path / "f.txt"
f.write_text("a\nb\nc\n", encoding="utf-8")
got = await read_lines_safe_async(f, limit=2, max_bytes=1024)
assert got.lines == ["a", "b"]
assert got.was_truncated is True
class TestFileWriteLock:
@pytest.mark.asyncio
async def test_same_lock_for_different_path_spellings(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
path = tmp_path / "f.txt"
path.touch()
_FILE_WRITE_LOCKS.clear()
order: list[str] = []
held = asyncio.Event()
release = asyncio.Event()
async def first() -> None:
async with file_write_lock(path):
order.append("first-acquired")
held.set()
await release.wait()
order.append("first-released")
async def second() -> None:
await held.wait()
# Same file, different spelling — must contend on the same lock.
async with file_write_lock(Path("f.txt")):
order.append("second-acquired")
t1 = asyncio.create_task(first())
t2 = asyncio.create_task(second())
await held.wait()
await asyncio.sleep(0)
assert order == ["first-acquired"]
release.set()
await asyncio.gather(t1, t2)
assert order == ["first-acquired", "first-released", "second-acquired"]

62
tests/core/test_vibe_config_schema.py Normal file
View file

@ -0,0 +1,62 @@
from __future__ import annotations
from pathlib import Path
import pytest
from vibe.core.config._settings import VibeConfig
from vibe.core.config.vibe_schema import VibeConfigSchema
def test_vibe_config_schema_covers_all_vibe_config_fields() -> None:
legacy_fields = set(VibeConfig.model_fields.keys())
schema_fields = set(VibeConfigSchema.model_fields.keys())
missing = legacy_fields - schema_fields
assert not missing, (
f"VibeConfigSchema is missing {len(missing)} field(s) that exist in VibeConfig: "
f"{sorted(missing)}. "
f"When you add a new field to VibeConfig, also add it to VibeConfigSchema "
f"(vibe/core/config/vibe_schema.py) with the appropriate merge annotation."
)
@pytest.mark.asyncio
async def test_full_toml_to_vibe_config_schema(tmp_path: Path) -> None:
toml_path = tmp_path / "config.toml"
toml_path.write_text(
"""\
vim_keybindings = true
api_timeout = 300.0
active_model = "codestral"
disabled_tools = ["bash"]
default_agent = "plan"
enabled_skills = ["search"]
enable_otel = true
[[models]]
alias = "codestral"
name = "codestral-latest"
provider = "mistral"
"""
)
from vibe.core.config.layers.user import UserConfigLayer
from vibe.core.config.orchestrator import ConfigOrchestrator
class VibeConfig(VibeConfigSchema):
pass
layer = UserConfigLayer(path=toml_path, name="user-toml")
orchestrator = await ConfigOrchestrator[VibeConfig].create(
schema=VibeConfig, layers=[layer]
)
config = orchestrator.config
assert config.vim_keybindings is True
assert config.api_timeout == 300.0
assert config.active_model == "codestral"
assert config.models[0].alias == "codestral"
assert "bash" in config.disabled_tools
assert config.default_agent == "plan"
assert "search" in config.enabled_skills
assert config.enable_otel is True

230
tests/core/tools/builtins/test_edit.py Normal file
View file

@ -0,0 +1,230 @@
from __future__ import annotations
import asyncio
from pathlib import Path
import pytest
from tests.mock.utils import collect_result
from vibe.core.tools.base import BaseToolState, ToolError
from vibe.core.tools.builtins.edit import Edit, EditArgs, EditConfig, EditResult
from vibe.core.tools.ui import ToolCallDisplay, ToolResultDisplay
from vibe.core.types import ToolResultEvent
def _make_edit() -> Edit:
return Edit(config_getter=lambda: EditConfig(), state=BaseToolState())
def _write(tmp_path: Path, name: str, content: str) -> Path:
p = tmp_path / name
p.write_text(content)
return p
@pytest.mark.asyncio
async def test_exact_match_replaces(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
_write(tmp_path, "f.txt", "hello world\n")
edit = _make_edit()
result = await collect_result(
edit.run(EditArgs(file_path="f.txt", old_string="hello", new_string="goodbye"))
)
assert (tmp_path / "f.txt").read_text() == "goodbye world\n"
assert result.file == str(tmp_path / "f.txt")
assert result.message == "The file has been updated successfully."
@pytest.mark.asyncio
async def test_replace_all(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.chdir(tmp_path)
_write(tmp_path, "f.txt", "aaa bbb aaa\n")
edit = _make_edit()
result = await collect_result(
edit.run(
EditArgs(
file_path="f.txt", old_string="aaa", new_string="ccc", replace_all=True
)
)
)
assert (tmp_path / "f.txt").read_text() == "ccc bbb ccc\n"
assert result.message == (
"The file has been updated. All occurrences were successfully replaced"
)
@pytest.mark.asyncio
async def test_not_found_raises(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
_write(tmp_path, "f.txt", "hello world\n")
edit = _make_edit()
with pytest.raises(ToolError, match="String to replace not found in file"):
await collect_result(
edit.run(EditArgs(file_path="f.txt", old_string="missing", new_string="x"))
)
@pytest.mark.asyncio
async def test_multiple_matches_without_replace_all_raises(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
_write(tmp_path, "f.txt", "aaa bbb aaa\n")
edit = _make_edit()
with pytest.raises(ToolError, match="Found 2 matches"):
await collect_result(
edit.run(EditArgs(file_path="f.txt", old_string="aaa", new_string="ccc"))
)
@pytest.mark.asyncio
async def test_old_equals_new_raises(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
_write(tmp_path, "f.txt", "hello\n")
edit = _make_edit()
with pytest.raises(ToolError, match="No changes to make"):
await collect_result(
edit.run(
EditArgs(file_path="f.txt", old_string="hello", new_string="hello")
)
)
@pytest.mark.asyncio
async def test_empty_old_string_raises(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
_write(tmp_path, "f.txt", "hello\n")
edit = _make_edit()
with pytest.raises(ToolError, match="old_string cannot be empty"):
await collect_result(
edit.run(EditArgs(file_path="f.txt", old_string="", new_string="x"))
)
@pytest.mark.asyncio
async def test_file_not_found_raises(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
edit = _make_edit()
with pytest.raises(ToolError, match="File does not exist"):
await collect_result(
edit.run(
EditArgs(
file_path="/nonexistent/file.py", old_string="x", new_string="y"
)
)
)
@pytest.mark.asyncio
async def test_empty_file_path_raises(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
edit = _make_edit()
with pytest.raises(ToolError, match="File path cannot be empty"):
await collect_result(
edit.run(EditArgs(file_path="", old_string="x", new_string="y"))
)
@pytest.mark.asyncio
async def test_deletion_removes_exact_string(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
_write(tmp_path, "f.txt", "line1\nline2\nline3\n")
edit = _make_edit()
await collect_result(
edit.run(EditArgs(file_path="f.txt", old_string="line2\n", new_string=""))
)
assert (tmp_path / "f.txt").read_text() == "line1\nline3\n"
@pytest.mark.asyncio
async def test_parallel_edits_same_file_all_land(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
_write(tmp_path, "f.txt", "A\nB\nC\nD\n")
edit = _make_edit()
await asyncio.gather(
collect_result(
edit.run(EditArgs(file_path="f.txt", old_string="A", new_string="A1"))
),
collect_result(
edit.run(EditArgs(file_path="f.txt", old_string="B", new_string="B1"))
),
collect_result(
edit.run(EditArgs(file_path="f.txt", old_string="C", new_string="C1"))
),
collect_result(
edit.run(EditArgs(file_path="f.txt", old_string="D", new_string="D1"))
),
)
assert (tmp_path / "f.txt").read_text() == "A1\nB1\nC1\nD1\n"
@pytest.mark.asyncio
async def test_relative_path_resolved_from_cwd(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
(tmp_path / "sub").mkdir(exist_ok=True)
_write(tmp_path / "sub", "f.txt", "old")
edit = _make_edit()
result = await collect_result(
edit.run(EditArgs(file_path="sub/f.txt", old_string="old", new_string="new"))
)
assert (tmp_path / "sub" / "f.txt").read_text() == "new"
assert result.file == str(tmp_path / "sub" / "f.txt")
def test_format_call_display() -> None:
args = EditArgs(file_path="/abs/foo.py", old_string="old", new_string="new")
display = Edit.format_call_display(args)
assert isinstance(display, ToolCallDisplay)
assert display.summary == "Editing foo.py"
def test_get_result_display() -> None:
result = EditResult(
file="/path/to/foo.py",
message="The file has been updated successfully.",
old_string="old",
new_string="new",
)
event = ToolResultEvent(
tool_call_id="test", tool_name="edit", tool_class=None, result=result
)
display = Edit.get_result_display(event)
assert isinstance(display, ToolResultDisplay)
assert display.success is True
assert "foo.py" in display.message

355
tests/core/tools/builtins/test_read.py Normal file
View file

@ -0,0 +1,355 @@
from __future__ import annotations
from collections.abc import Iterator
from pathlib import Path
import pytest
from tests.mock.utils import collect_result
from vibe.core.config.harness_files import (
init_harness_files_manager,
reset_harness_files_manager,
)
from vibe.core.tools.base import ToolError
from vibe.core.tools.builtins.read import (
DEFAULT_LINE_LIMIT,
MAX_BYTES,
Read,
ReadArgs,
ReadConfig,
ReadResult,
ReadState,
_add_line_numbers,
)
from vibe.core.tools.ui import ToolCallDisplay, ToolResultDisplay
from vibe.core.trusted_folders import trusted_folders_manager
from vibe.core.types import ToolResultEvent
from vibe.core.utils import VIBE_WARNING_TAG
def _make_read() -> Read:
return Read(config_getter=lambda: ReadConfig(), state=ReadState())
@pytest.mark.asyncio
async def test_reads_entire_small_file(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
(tmp_path / "hello.txt").write_text("line one\nline two\n", encoding="utf-8")
tool = _make_read()
result = await collect_result(
tool.run(ReadArgs(file_path=str(tmp_path / "hello.txt")))
)
assert result.num_lines == 2
assert result.total_lines == 2
assert result.start_line == 1
assert "line one" in result.content
assert "line two" in result.content
@pytest.mark.asyncio
async def test_reads_with_offset_and_limit(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
content = "".join(f"line {i}\n" for i in range(1, 11))
(tmp_path / "f.txt").write_text(content, encoding="utf-8")
tool = _make_read()
result = await collect_result(
tool.run(ReadArgs(file_path=str(tmp_path / "f.txt"), offset=3, limit=2))
)
assert result.num_lines == 2
assert result.start_line == 3
# Bounded read stops at the limit, so the true total is unknown.
assert result.total_lines is None
assert result.was_truncated is True
assert "line 3" in result.content
assert "line 4" in result.content
assert "line 2" not in result.content
assert "line 5" not in result.content
@pytest.mark.asyncio
async def test_empty_file_returns_warning(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
(tmp_path / "empty.txt").write_text("", encoding="utf-8")
tool = _make_read()
result = await collect_result(
tool.run(ReadArgs(file_path=str(tmp_path / "empty.txt")))
)
assert result.num_lines == 0
assert VIBE_WARNING_TAG in result.content
assert "empty" in result.content
@pytest.mark.asyncio
async def test_offset_beyond_file_returns_warning(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
(tmp_path / "short.txt").write_text("one\ntwo\n", encoding="utf-8")
tool = _make_read()
result = await collect_result(
tool.run(ReadArgs(file_path=str(tmp_path / "short.txt"), offset=100))
)
assert result.num_lines == 0
assert VIBE_WARNING_TAG in result.content
assert "shorter" in result.content
@pytest.mark.asyncio
async def test_exceeds_max_bytes_raises(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
# Create file that generates output exceeding MAX_BYTES
big_line = "x" * 200 + "\n"
lines_needed = (MAX_BYTES // len(big_line)) + 100
(tmp_path / "big.txt").write_text(big_line * lines_needed, encoding="utf-8")
tool = _make_read()
with pytest.raises(ToolError, match="exceeds maximum allowed size"):
await collect_result(tool.run(ReadArgs(file_path=str(tmp_path / "big.txt"))))
@pytest.mark.asyncio
async def test_truncated_when_more_lines_than_limit(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
content = "".join(f"line {i}\n" for i in range(1, 101))
(tmp_path / "f.txt").write_text(content, encoding="utf-8")
tool = _make_read()
result = await collect_result(
tool.run(ReadArgs(file_path=str(tmp_path / "f.txt"), limit=10))
)
assert result.num_lines == 10
assert result.was_truncated is True
assert result.total_lines is None
@pytest.mark.asyncio
async def test_single_oversized_line_raises(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
(tmp_path / "wide.txt").write_text("x" * (MAX_BYTES + 10), encoding="utf-8")
tool = _make_read()
with pytest.raises(ToolError, match="exceeds maximum allowed size"):
await collect_result(tool.run(ReadArgs(file_path=str(tmp_path / "wide.txt"))))
@pytest.mark.asyncio
async def test_file_not_found_raises(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
tool = _make_read()
with pytest.raises(ToolError, match="File not found"):
await collect_result(tool.run(ReadArgs(file_path=str(tmp_path / "nope.txt"))))
@pytest.mark.asyncio
async def test_empty_path_raises(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
tool = _make_read()
with pytest.raises(ToolError, match="file_path cannot be empty"):
await collect_result(tool.run(ReadArgs(file_path="")))
@pytest.mark.asyncio
async def test_directory_raises(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
(tmp_path / "adir").mkdir()
tool = _make_read()
with pytest.raises(ToolError, match="directory"):
await collect_result(tool.run(ReadArgs(file_path=str(tmp_path / "adir"))))
@pytest.mark.asyncio
async def test_relative_path_resolved(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
(tmp_path / "sub").mkdir()
(tmp_path / "sub" / "f.txt").write_text("ok\n", encoding="utf-8")
tool = _make_read()
result = await collect_result(tool.run(ReadArgs(file_path="sub/f.txt")))
assert result.num_lines == 1
assert str(tmp_path / "sub" / "f.txt") == result.file_path
def test_line_number_format() -> None:
formatted = _add_line_numbers(["hello", "world"], start=1)
lines = formatted.split("\n")
assert lines[0] == " 1\u2192hello"
assert lines[1] == " 2\u2192world"
def test_default_limit_is_2000() -> None:
assert DEFAULT_LINE_LIMIT == 2000
def test_format_call_display() -> None:
args = ReadArgs(file_path="/some/file.py")
display = Read.format_call_display(args)
assert isinstance(display, ToolCallDisplay)
assert "file.py" in display.summary
def test_format_call_display_with_offset_limit() -> None:
args = ReadArgs(file_path="/some/file.py", offset=10, limit=50)
display = Read.format_call_display(args)
assert "from line 10" in display.summary
assert "limit 50" in display.summary
def test_get_result_display() -> None:
result = ReadResult(
file_path="/path/to/foo.py",
content="...",
num_lines=10,
start_line=1,
total_lines=10,
)
event = ToolResultEvent(
tool_call_id="test", tool_name="read", tool_class=None, result=result
)
display = Read.get_result_display(event)
assert isinstance(display, ToolResultDisplay)
assert display.success is True
assert "10 lines" in display.message
assert "foo.py" in display.message
def test_get_result_display_truncated() -> None:
result = ReadResult(
file_path="/path/to/foo.py",
content="...",
num_lines=10,
start_line=1,
total_lines=100,
)
event = ToolResultEvent(
tool_call_id="test", tool_name="read", tool_class=None, result=result
)
display = Read.get_result_display(event)
assert "truncated" in display.message
def test_get_result_display_truncated_via_flag() -> None:
result = ReadResult(
file_path="/path/to/foo.py",
content="...",
num_lines=10,
start_line=1,
total_lines=None,
was_truncated=True,
)
event = ToolResultEvent(
tool_call_id="test", tool_name="read", tool_class=None, result=result
)
display = Read.get_result_display(event)
assert "truncated" in display.message
@pytest.fixture()
def _setup_manager(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(trusted_folders_manager, "is_trusted", lambda _: True)
monkeypatch.setattr(
trusted_folders_manager, "find_trust_root", lambda _: tmp_path.resolve()
)
reset_harness_files_manager()
init_harness_files_manager("user", "project")
yield
reset_harness_files_manager()
@pytest.mark.usefixtures("_setup_manager")
def test_agents_md_injection(tmp_path: Path) -> None:
sub = tmp_path / "sub"
sub.mkdir()
(sub / "AGENTS.md").write_text("# Sub instructions", encoding="utf-8")
target = sub / "file.py"
target.write_text("hello", encoding="utf-8")
tool = _make_read()
result = ReadResult(
file_path=str(target), content="hello", num_lines=1, start_line=1, total_lines=1
)
annotation = tool.get_result_extra(result)
assert annotation is not None
assert VIBE_WARNING_TAG in annotation
assert "# Sub instructions" in annotation
@pytest.mark.usefixtures("_setup_manager")
def test_agents_md_deduplicates(tmp_path: Path) -> None:
sub = tmp_path / "sub"
sub.mkdir()
(sub / "AGENTS.md").write_text("# Sub", encoding="utf-8")
(sub / "a.py").write_text("a", encoding="utf-8")
(sub / "b.py").write_text("b", encoding="utf-8")
tool = _make_read()
r1 = ReadResult(
file_path=str(sub / "a.py"),
content="a",
num_lines=1,
start_line=1,
total_lines=1,
)
assert tool.get_result_extra(r1) is not None
r2 = ReadResult(
file_path=str(sub / "b.py"),
content="b",
num_lines=1,
start_line=1,
total_lines=1,
)
assert tool.get_result_extra(r2) is None
def test_agents_md_returns_none_when_not_initialized(tmp_path: Path) -> None:
reset_harness_files_manager()
tool = _make_read()
result = ReadResult(
file_path=str(tmp_path / "file.py"),
content="",
num_lines=0,
start_line=1,
total_lines=0,
)
assert tool.get_result_extra(result) is None
reset_harness_files_manager()

217
tests/core/tools/builtins/test_read_file.py
View file

@ -1,217 +0,0 @@
from __future__ import annotations
from collections.abc import Iterator
from pathlib import Path
import pytest
from tests.mock.utils import collect_result
from vibe.core.config.harness_files import (
init_harness_files_manager,
reset_harness_files_manager,
)
from vibe.core.tools.builtins.read_file import (
ReadFile,
ReadFileArgs,
ReadFileResult,
ReadFileState,
ReadFileToolConfig,
)
from vibe.core.trusted_folders import trusted_folders_manager
from vibe.core.utils import VIBE_WARNING_TAG
@pytest.fixture()
def _setup_manager(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
"""Initialize harness files manager for tests, reset after."""
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(trusted_folders_manager, "is_trusted", lambda _: True)
monkeypatch.setattr(
trusted_folders_manager, "find_trust_root", lambda _: tmp_path.resolve()
)
reset_harness_files_manager()
init_harness_files_manager("user", "project")
yield
reset_harness_files_manager()
def _make_read_file() -> ReadFile:
return ReadFile(config_getter=lambda: ReadFileToolConfig(), state=ReadFileState())
class TestReadFileExecution:
@pytest.mark.asyncio
async def test_run_with_large_offset_still_reads_lines(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
test_file = tmp_path / "large_file.txt"
test_file.write_text(
"".join(f"line {i}\n" for i in range(200)), encoding="utf-8"
)
tool = ReadFile(
config_getter=lambda: ReadFileToolConfig(max_read_bytes=64),
state=ReadFileState(),
)
result = await collect_result(
tool.run(ReadFileArgs(path=str(test_file), offset=50, limit=2))
)
assert result.content == "line 50\nline 51\n"
assert result.lines_read == 2
assert result.was_truncated
@pytest.mark.asyncio
async def test_run_marks_truncated_when_max_read_bytes_exceeded(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
test_file = tmp_path / "big.txt"
test_file.write_text(
"".join(f"line {i}\n" for i in range(100)), encoding="utf-8"
)
tool = ReadFile(
config_getter=lambda: ReadFileToolConfig(max_read_bytes=20),
state=ReadFileState(),
)
result = await collect_result(tool.run(ReadFileArgs(path=str(test_file))))
assert result.was_truncated
assert result.lines_read > 0
@pytest.mark.asyncio
async def test_run_not_truncated_when_eof_reached(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
test_file = tmp_path / "small.txt"
test_file.write_text("a\nb\nc\n", encoding="utf-8")
tool = ReadFile(
config_getter=lambda: ReadFileToolConfig(max_read_bytes=1024),
state=ReadFileState(),
)
result = await collect_result(tool.run(ReadFileArgs(path=str(test_file))))
assert not result.was_truncated
assert result.lines_read == 3
@pytest.mark.asyncio
async def test_run_not_truncated_when_limit_matches_remaining_lines(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
test_file = tmp_path / "exact.txt"
test_file.write_text("a\nb\nc\n", encoding="utf-8")
tool = ReadFile(
config_getter=lambda: ReadFileToolConfig(max_read_bytes=1024),
state=ReadFileState(),
)
result = await collect_result(
tool.run(ReadFileArgs(path=str(test_file), limit=10))
)
assert not result.was_truncated
assert result.lines_read == 3
class TestGetResultExtra:
@pytest.mark.usefixtures("_setup_manager")
def test_returns_none_when_no_agents_md(self, tmp_path: Path) -> None:
sub = tmp_path / "sub"
sub.mkdir()
target = sub / "file.py"
target.write_text("hello", encoding="utf-8")
tool = _make_read_file()
result = ReadFileResult(
path=str(target), content="hello", lines_read=1, was_truncated=False
)
assert tool.get_result_extra(result) is None
@pytest.mark.usefixtures("_setup_manager")
def test_returns_tagged_content_when_agents_md_found(self, tmp_path: Path) -> None:
sub = tmp_path / "sub"
sub.mkdir()
(sub / "AGENTS.md").write_text("# Sub instructions", encoding="utf-8")
target = sub / "file.py"
target.write_text("hello", encoding="utf-8")
tool = _make_read_file()
result = ReadFileResult(
path=str(target), content="hello", lines_read=1, was_truncated=False
)
annotation = tool.get_result_extra(result)
assert annotation is not None
assert f"<{VIBE_WARNING_TAG}>" in annotation
assert f"</{VIBE_WARNING_TAG}>" in annotation
assert "# Sub instructions" in annotation
assert "project instructions for this directory" in annotation
@pytest.mark.usefixtures("_setup_manager")
def test_deduplicates_across_calls(self, tmp_path: Path) -> None:
sub = tmp_path / "sub"
sub.mkdir()
(sub / "AGENTS.md").write_text("# Sub", encoding="utf-8")
file1 = sub / "a.py"
file2 = sub / "b.py"
file1.write_text("a", encoding="utf-8")
file2.write_text("b", encoding="utf-8")
tool = _make_read_file()
result1 = ReadFileResult(
path=str(file1), content="a", lines_read=1, was_truncated=False
)
assert tool.get_result_extra(result1) is not None
# Second call for a different file in the same dir → no duplicate
result2 = ReadFileResult(
path=str(file2), content="b", lines_read=1, was_truncated=False
)
assert tool.get_result_extra(result2) is None
@pytest.mark.usefixtures("_setup_manager")
def test_injects_new_dir_after_dedup(self, tmp_path: Path) -> None:
sub_a = tmp_path / "a"
sub_b = tmp_path / "b"
sub_a.mkdir()
sub_b.mkdir()
(sub_a / "AGENTS.md").write_text("# A", encoding="utf-8")
(sub_b / "AGENTS.md").write_text("# B", encoding="utf-8")
file_a = sub_a / "f.py"
file_b = sub_b / "f.py"
file_a.write_text("", encoding="utf-8")
file_b.write_text("", encoding="utf-8")
tool = _make_read_file()
r1 = ReadFileResult(
path=str(file_a), content="", lines_read=0, was_truncated=False
)
ann1 = tool.get_result_extra(r1)
assert ann1 is not None
assert "# A" in ann1
# Different subdirectory → should inject its AGENTS.md
r2 = ReadFileResult(
path=str(file_b), content="", lines_read=0, was_truncated=False
)
ann2 = tool.get_result_extra(r2)
assert ann2 is not None
assert "# B" in ann2
def test_returns_none_when_manager_not_initialized(self, tmp_path: Path) -> None:
reset_harness_files_manager()
tool = _make_read_file()
result = ReadFileResult(
path=str(tmp_path / "file.py"),
content="",
lines_read=0,
was_truncated=False,
)
assert tool.get_result_extra(result) is None
reset_harness_files_manager()