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

v2.9.0 (#641)

Browse source 
Co-authored-by: Antoine <33425718+anth2o@users.noreply.github.com>
Co-authored-by: Bastien <bastien.baret@gmail.com>
Co-authored-by: Clément Sirieix <clement.sirieix@mistral.ai>
Co-authored-by: Kim-Adeline Miguel <51720070+kimadeline@users.noreply.github.com>
Co-authored-by: Mathias Gesbert <mathias.gesbert@mistral.ai>
Co-authored-by: Maxime Dolores <maxime.dolores@ext.mistral.ai>
Co-authored-by: Michel Thomazo <51709227+michelTho@users.noreply.github.com>
Co-authored-by: Nelson PROIA <144663685+Nelson-PROIA@users.noreply.github.com>
Co-authored-by: Pierre Rossinès <pierre.rossines@mistral.ai>
Co-authored-by: Quentin <quentin.torroba@mistral.ai>
Co-authored-by: Robin Gullo <robin.gullo@mistral.ai>
Co-authored-by: Mistral Vibe <vibe@mistral.ai>
This commit is contained in:
Clément Drouin 2026-04-28 17:44:07 +02:00 • committed by GitHub
parent a83c81ecf5
commit 632ea8c032
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
253 changed files with 13965 additions and 2525 deletions
Download patch file Download diff file Expand all files Collapse all files

18
tests/core/nuage/test_remote_events_source.py
View file

@ -2,6 +2,7 @@ from __future__ import annotations
from unittest.mock import AsyncMock, MagicMock, patch
from pydantic import BaseModel, ValidationError
import pytest
from tests.conftest import build_test_vibe_config
@ -281,3 +282,20 @@ class TestBrokerSequenceTracking:
assert events == []
assert source._next_start_seq == 5
def test_consume_workflow_event_validation_error_is_logged_and_ignored() -> None:
class _InvalidPayload(BaseModel):
required: str
source = _make_source()
with pytest.raises(ValidationError) as exc_info:
_InvalidPayload.model_validate({})
source._translator.consume_workflow_event = MagicMock(side_effect=exc_info.value)
with patch("vibe.core.nuage.remote_events_source.logger.warning") as mock_warning:
events = source._consume_workflow_event(MagicMock())
assert events == []
mock_warning.assert_called_once()

44
tests/core/nuage/test_workflows_client.py
View file

@ -1,13 +1,14 @@
from __future__ import annotations
import json
from unittest.mock import AsyncMock, patch
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from vibe.core.nuage.client import WorkflowsClient
from vibe.core.nuage.exceptions import ErrorCode, WorkflowsException
from vibe.core.nuage.streaming import StreamEvent, StreamEventsQueryParams
from vibe.core.nuage.workflow import WorkflowExecutionStatus
def _make_client() -> WorkflowsClient:
@ -198,6 +199,47 @@ class TestStreamEvents:
assert "Failed to stream events" in exc_info.value.message
class TestGetWorkflowRuns:
@pytest.mark.asyncio
async def test_sends_current_user_filter_by_default(self) -> None:
client = _make_client()
mock_response = MagicMock()
mock_response.raise_for_status.return_value = None
mock_response.json.return_value = {"executions": [], "next_page_token": None}
mock_http = AsyncMock()
mock_http.get.return_value = mock_response
client._client = mock_http
await client.get_workflow_runs(
workflow_identifier="workflow-1",
page_size=10,
status=[WorkflowExecutionStatus.RUNNING],
)
mock_http.get.assert_awaited_once()
call_params = mock_http.get.call_args.kwargs["params"]
assert call_params["user_id"] == "current"
assert call_params["workflow_identifier"] == "workflow-1"
assert call_params["page_size"] == 10
@pytest.mark.asyncio
async def test_allows_overriding_user_id(self) -> None:
client = _make_client()
mock_response = MagicMock()
mock_response.raise_for_status.return_value = None
mock_response.json.return_value = {"executions": [], "next_page_token": None}
mock_http = AsyncMock()
mock_http.get.return_value = mock_response
client._client = mock_http
await client.get_workflow_runs(user_id="user-123")
call_params = mock_http.get.call_args.kwargs["params"]
assert call_params["user_id"] == "user-123"
def _async_line_iter(lines: list[str]):
async def _iter():
for line in lines:

79
tests/core/session/test_session_id.py Normal file
View file

@ -0,0 +1,79 @@
from __future__ import annotations
import re
import pytest
from vibe.core.session.session_id import extract_suffix, generate_session_id
UUID_RE = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$")
class TestGenerateSessionId:
def test_uuid_shape(self) -> None:
sid = generate_session_id()
assert UUID_RE.match(sid), f"Not UUID-shaped: {sid}"
assert len(sid) == 36
def test_unique_ids(self) -> None:
ids = {generate_session_id() for _ in range(100)}
assert len(ids) == 100
def test_suffix_preserved(self) -> None:
suffix = "aabbccddeeff"
sid = generate_session_id(suffix=suffix)
assert UUID_RE.match(sid)
assert sid.endswith(suffix)
def test_suffix_stable_across_regeneration(self) -> None:
suffix = "112233445566"
ids = [generate_session_id(suffix=suffix) for _ in range(50)]
# All share the same last segment
assert all(s.rsplit("-", 1)[-1] == suffix for s in ids)
# But first 8 chars are different (probabilistically)
prefixes = {s[:8] for s in ids}
assert len(prefixes) > 1
def test_default_suffix_is_12_hex(self) -> None:
sid = generate_session_id()
suffix = sid.rsplit("-", 1)[-1]
assert len(suffix) == 12
assert re.fullmatch(r"[0-9a-f]+", suffix)
def test_first_8_chars_are_unique_with_same_suffix(self) -> None:
suffix = "abcdef123456"
a = generate_session_id(suffix=suffix)
b = generate_session_id(suffix=suffix)
assert a[:8] != b[:8]
class TestExtractSuffix:
def test_extract_from_generated_id(self) -> None:
sid = generate_session_id()
suffix = extract_suffix(sid)
assert sid.endswith(suffix)
assert len(suffix) == 12
def test_extract_from_real_uuid(self) -> None:
uuid_str = "550e8400-e29b-41d4-a716-446655440000"
assert extract_suffix(uuid_str) == "446655440000"
def test_extract_roundtrip(self) -> None:
original_suffix = "deadbeef1234"
sid = generate_session_id(suffix=original_suffix)
assert extract_suffix(sid) == original_suffix
def test_extract_from_no_hyphen(self) -> None:
assert extract_suffix("abcdef") == "abcdef"
@pytest.mark.parametrize(
"session_id",
[
"550e8400-e29b-41d4-a716-446655440000",
"abcdef01-2345-6789-abcd-ef0123456789",
"12345678-aaaa-bbbb-cccc-aabbccddeeff",
],
)
def test_extract_always_returns_last_segment(self, session_id: str) -> None:
expected = session_id.rsplit("-", 1)[-1]
assert extract_suffix(session_id) == expected

493
tests/core/test_config_layer.py Normal file
View file

@ -0,0 +1,493 @@
from __future__ import annotations
import asyncio
from typing import Any
from pydantic import BaseModel, ValidationError
import pytest
from vibe.core.config.layer import (
ConfigLayer,
LayerImplementationError,
RawConfig,
UntrustedLayerError,
)
class StubLayer(ConfigLayer[BaseModel]):
"""Minimal concrete layer for testing."""
def __init__(
self,
*,
name: str = "stub",
output_schema: type[BaseModel] | None = None,
trusted: bool = True,
data: dict[str, Any] | None = None,
) -> None:
kwargs: dict[str, Any] = {"name": name}
if output_schema is not None:
kwargs["output_schema"] = output_schema
super().__init__(**kwargs)
self._stub_trusted = trusted
self._data = data or {}
self.read_count = 0
async def _check_trust(self) -> bool:
return self._stub_trusted
async def _read_config(self) -> dict[str, Any]:
self.read_count += 1
return dict(self._data)
class ObservableStubLayer(StubLayer):
"""Stub that records _on_trust_changed calls."""
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.trust_changes: list[tuple[bool | None, bool | None]] = []
async def _on_trust_changed(self, old: bool | None, new: bool | None) -> None:
self.trust_changes.append((old, new))
class SampleSchema(BaseModel):
name: str
count: int = 0
def test_abstract_read_config_enforced() -> None:
class IncompleteLayer(ConfigLayer[BaseModel]):
pass
with pytest.raises(TypeError):
IncompleteLayer(name="incomplete") # type: ignore[abstract]
def test_repr() -> None:
layer = StubLayer(name="my-layer")
assert repr(layer) == "StubLayer(name='my-layer')"
@pytest.mark.asyncio
async def test_default_check_trust_returns_false() -> None:
class DefaultTrustLayer(ConfigLayer[BaseModel]):
async def _read_config(self) -> dict[str, Any]:
return {}
layer = DefaultTrustLayer(name="default")
result = await layer.resolve_trust()
assert result is False
@pytest.mark.asyncio
async def test_trust_initially_none() -> None:
layer = StubLayer()
assert layer.is_trusted is None
@pytest.mark.asyncio
async def test_resolve_trust_trusted() -> None:
layer = StubLayer(trusted=True)
result = await layer.resolve_trust()
assert result is True
assert layer.is_trusted is True
@pytest.mark.asyncio
async def test_resolve_trust_untrusted() -> None:
layer = StubLayer(trusted=False)
result = await layer.resolve_trust()
assert result is False
assert layer.is_trusted is False
@pytest.mark.asyncio
async def test_resolve_trust_fires_on_trust_changed() -> None:
layer = ObservableStubLayer(trusted=True)
await layer.resolve_trust()
assert layer.trust_changes == [(None, True)]
@pytest.mark.asyncio
async def test_resolve_trust_no_callback_when_unchanged() -> None:
layer = ObservableStubLayer(trusted=True)
await layer.resolve_trust()
layer.trust_changes.clear()
await layer.resolve_trust()
assert layer.trust_changes == []
@pytest.mark.asyncio
async def test_grant_trust() -> None:
layer = StubLayer(trusted=False)
await layer.grant_trust()
assert layer.is_trusted is True
@pytest.mark.asyncio
async def test_grant_trust_fires_callback() -> None:
layer = ObservableStubLayer(trusted=False)
await layer.grant_trust()
assert layer.trust_changes == [(None, True)]
@pytest.mark.asyncio
async def test_grant_trust_noop_when_already_trusted() -> None:
layer = ObservableStubLayer(trusted=True)
await layer.resolve_trust()
layer.trust_changes.clear()
await layer.grant_trust()
assert layer.trust_changes == []
@pytest.mark.asyncio
async def test_revoke_trust() -> None:
layer = StubLayer(trusted=True)
await layer.resolve_trust()
await layer.revoke_trust()
assert layer.is_trusted is False
@pytest.mark.asyncio
async def test_revoke_trust_fires_callback() -> None:
layer = ObservableStubLayer(trusted=True)
await layer.resolve_trust()
layer.trust_changes.clear()
await layer.revoke_trust()
assert layer.trust_changes == [(True, False)]
@pytest.mark.asyncio
async def test_revoke_trust_clears_data() -> None:
layer = StubLayer(data={"k": "v"})
await layer.load()
assert layer.read_count == 1
await layer.revoke_trust()
await layer.grant_trust()
await layer.load()
assert layer.read_count == 2
@pytest.mark.asyncio
async def test_on_trust_changed_failure_preserves_state() -> None:
class FailingLayer(StubLayer):
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.should_fail = True
async def _on_trust_changed(self, old: bool | None, new: bool | None) -> None:
if self.should_fail:
raise RuntimeError("persistence failure")
layer = FailingLayer(trusted=False)
assert layer.is_trusted is None
with pytest.raises(LayerImplementationError, match="_on_trust_changed") as exc_info:
await layer.grant_trust()
assert isinstance(exc_info.value.__cause__, RuntimeError)
assert layer.is_trusted is None
layer.should_fail = False
await layer.grant_trust()
assert layer.is_trusted is True
@pytest.mark.asyncio
async def test_check_trust_failure_wrapped() -> None:
class BrokenTrustLayer(StubLayer):
async def _check_trust(self) -> bool:
raise OSError("trust store unavailable")
layer = BrokenTrustLayer()
with pytest.raises(LayerImplementationError, match="_check_trust") as exc_info:
await layer.resolve_trust()
assert isinstance(exc_info.value.__cause__, IOError)
@pytest.mark.asyncio
async def test_load_returns_data() -> None:
layer = StubLayer(data={"key": "value"})
result = await layer.load()
assert isinstance(result, RawConfig)
assert result.model_extra == {"key": "value"}
@pytest.mark.asyncio
async def test_load_auto_resolves_trust() -> None:
layer = StubLayer(trusted=True, data={"a": 1})
assert layer.is_trusted is None
result = await layer.load()
assert layer.is_trusted is True
assert result.model_extra == {"a": 1}
@pytest.mark.asyncio
async def test_load_caches_result() -> None:
layer = StubLayer()
await layer.load()
await layer.load()
await layer.load()
assert layer.read_count == 1
@pytest.mark.asyncio
async def test_load_force_bypasses_cache() -> None:
layer = StubLayer()
await layer.load()
assert layer.read_count == 1
await layer.load(force=True)
assert layer.read_count == 2
@pytest.mark.asyncio
async def test_load_untrusted_raises() -> None:
layer = StubLayer(trusted=False)
with pytest.raises(UntrustedLayerError, match="stub"):
await layer.load()
@pytest.mark.asyncio
async def test_load_after_grant_trust() -> None:
layer = StubLayer(trusted=False)
await layer.grant_trust()
result = await layer.load()
assert isinstance(result, RawConfig)
@pytest.mark.asyncio
async def test_load_after_revoke_trust_raises() -> None:
layer = StubLayer(trusted=True)
await layer.load()
await layer.revoke_trust()
with pytest.raises(UntrustedLayerError):
await layer.load()
@pytest.mark.asyncio
async def test_invalidate_cache_causes_reload() -> None:
layer = StubLayer()
await layer.load()
assert layer.read_count == 1
await layer.invalidate_cache()
await layer.load()
assert layer.read_count == 2
@pytest.mark.asyncio
async def test_revoke_grant_cycle_refreshes_data() -> None:
layer = StubLayer(data={"v": 1})
result1 = await layer.load()
assert result1.model_extra == {"v": 1}
await layer.revoke_trust()
layer._data = {"v": 2}
await layer.grant_trust()
result2 = await layer.load()
assert result2.model_extra == {"v": 2}
@pytest.mark.asyncio
async def test_resolve_trust_clears_data_on_revocation() -> None:
layer = StubLayer(data={"v": 1})
result1 = await layer.load()
assert result1.model_extra == {"v": 1}
# External revocation via resolve_trust (not revoke_trust)
layer._stub_trusted = False
await layer.resolve_trust()
assert layer.is_trusted is False
# Re-trust and update backing data while revoked
layer._stub_trusted = True
layer._data = {"v": 2}
await layer.resolve_trust()
result2 = await layer.load()
assert result2.model_extra == {"v": 2}
@pytest.mark.asyncio
async def test_load_returns_deep_copy() -> None:
layer = StubLayer(data={"items": ["a", "b"]})
result1 = await layer.load()
assert result1.model_extra is not None
result1.model_extra["items"].append("mutated")
result2 = await layer.load()
assert result2.model_extra == {"items": ["a", "b"]}
assert layer.read_count == 1
@pytest.mark.asyncio
async def test_read_config_failure_wrapped() -> None:
class BrokenReadLayer(StubLayer):
async def _read_config(self) -> dict[str, Any]:
raise OSError("config file missing")
layer = BrokenReadLayer()
with pytest.raises(LayerImplementationError, match="_read_config") as exc_info:
await layer.load()
assert isinstance(exc_info.value.__cause__, IOError)
@pytest.mark.asyncio
async def test_default_schema_preserves_extras() -> None:
layer = StubLayer(data={"anything": "goes"})
result = await layer.load()
assert isinstance(result, RawConfig)
assert result.model_extra == {"anything": "goes"}
@pytest.mark.asyncio
async def test_custom_schema_validates() -> None:
layer = StubLayer(output_schema=SampleSchema, data={"name": "test", "count": 3})
result = await layer.load()
assert isinstance(result, SampleSchema)
assert result.name == "test"
assert result.count == 3
@pytest.mark.asyncio
async def test_invalid_data_raises_validation_error() -> None:
layer = StubLayer(output_schema=SampleSchema, data={"count": "bad"})
with pytest.raises(ValidationError):
await layer.load()
@pytest.mark.asyncio
async def test_concurrent_loads_serialize() -> None:
class SlowLayer(ConfigLayer[BaseModel]):
def __init__(self) -> None:
super().__init__(name="slow")
self.read_count = 0
async def _check_trust(self) -> bool:
return True
async def _read_config(self) -> dict[str, Any]:
self.read_count += 1
await asyncio.sleep(0.05)
return {"v": self.read_count}
layer = SlowLayer()
results = await asyncio.gather(layer.load(), layer.load(), layer.load())
assert layer.read_count == 1
assert all(r == results[0] for r in results)
@pytest.mark.asyncio
async def test_get_fingerprint_not_implemented() -> None:
layer = StubLayer()
with pytest.raises(NotImplementedError):
await layer.get_fingerprint()
@pytest.mark.asyncio
async def test_apply_not_implemented() -> None:
layer = StubLayer()
with pytest.raises(NotImplementedError):
await layer.apply({"op": "set"})
# Scenario: LocalUserConfigLayer
class UserConfigSchema(BaseModel):
active_model: str
theme: str = "dark"
class FakeLocalUserLayer(ConfigLayer[UserConfigSchema]):
"""Simulates ~/.vibe/config.toml — always trusted, typed output."""
def __init__(self, data: dict[str, Any]) -> None:
super().__init__(name="user-toml", output_schema=UserConfigSchema)
self._data = data
async def _check_trust(self) -> bool:
return True
async def _read_config(self) -> dict[str, Any]:
return dict(self._data)
@pytest.mark.asyncio
async def test_scenario_local_user_layer_always_trusted() -> None:
layer = FakeLocalUserLayer({"active_model": "devstral-2", "theme": "light"})
assert layer.is_trusted is None
result = await layer.load()
assert layer.is_trusted is True
assert isinstance(result, UserConfigSchema)
assert result.active_model == "devstral-2"
assert result.theme == "light"
validated = layer.validate_output({
"active_model": "mistral-large",
"theme": "dark",
})
assert isinstance(validated, UserConfigSchema)
assert validated.active_model == "mistral-large"
# Scenario: LocalProjectConfigLayer
class FakeLocalProjectLayer(ConfigLayer[BaseModel]):
def __init__(
self, *, project_path: str, data: dict[str, Any], trust_store: dict[str, bool]
) -> None:
super().__init__(name=f"project-toml:{project_path}")
self._project_path = project_path
self._data = data
self._trust_store = trust_store
async def _check_trust(self) -> bool:
return self._trust_store.get(self._project_path, False)
async def _on_trust_changed(self, old: bool | None, new: bool | None) -> None:
if new:
self._trust_store[self._project_path] = True
else:
self._trust_store.pop(self._project_path, None)
async def _read_config(self) -> dict[str, Any]:
return dict(self._data)
@pytest.mark.asyncio
async def test_scenario_local_project_layer_trust_lifecycle() -> None:
trust_store: dict[str, bool] = {}
project_data = {"disabled_tools": ["rm"], "max_tokens": 4096}
# 1. Fresh layer with empty trust store — load raises
layer = FakeLocalProjectLayer(
project_path="/tmp/my-project", data=project_data, trust_store=trust_store
)
with pytest.raises(UntrustedLayerError):
await layer.load()
# 2. Grant trust — persisted in store, load succeeds
await layer.grant_trust()
assert trust_store == {"/tmp/my-project": True}
result = await layer.load()
assert result.model_extra == project_data
# 3. New instance with same trust store — loads directly
layer2 = FakeLocalProjectLayer(
project_path="/tmp/my-project", data=project_data, trust_store=trust_store
)
assert layer2.is_trusted is None
result2 = await layer2.load()
assert layer2.is_trusted is True
assert result2.model_extra == project_data
# 4. Revoke trust — removed from store, load raises
await layer2.revoke_trust()
assert "/tmp/my-project" not in trust_store
with pytest.raises(UntrustedLayerError):
await layer2.load()

116
tests/core/test_config_resolution.py
View file

@ -198,26 +198,50 @@ class TestSaveUpdates:
assert result == {"tools": {"bash": {"default_timeout": 600}}}
class TestMigrateRemovesFindFromBashAllowlist:
def test_removes_find_from_config_file(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("VIBE_HOME", str(tmp_path))
config_file = tmp_path / "config.toml"
data = {"tools": {"bash": {"allowlist": ["echo", "find", "ls"]}}}
class TestSetThinking:
def test_persists_thinking_to_toml(self, config_dir: Path) -> None:
config_file = config_dir / "config.toml"
data = {
"active_model": "my-model",
"models": [
{"name": "my-model", "provider": "mistral", "alias": "my-model"}
],
}
with config_file.open("wb") as f:
tomli_w.dump(data, f)
reset_harness_files_manager()
init_harness_files_manager("user")
VibeConfig._migrate()
cfg = VibeConfig.load()
cfg.set_thinking("high")
reloaded = VibeConfig.load()
assert reloaded.get_active_model().thinking == "high"
with config_file.open("rb") as f:
result = tomllib.load(f)
assert result["models"][0]["thinking"] == "high"
def test_persists_thinking_for_correct_model(self, config_dir: Path) -> None:
config_file = config_dir / "config.toml"
data = {
"active_model": "model-b",
"models": [
{"name": "model-a", "provider": "mistral", "alias": "model-a"},
{"name": "model-b", "provider": "mistral", "alias": "model-b"},
],
}
with config_file.open("wb") as f:
tomli_w.dump(data, f)
cfg = VibeConfig.load()
cfg.set_thinking("max")
with config_file.open("rb") as f:
result = tomllib.load(f)
assert "find" not in result["tools"]["bash"]["allowlist"]
assert result["tools"]["bash"]["allowlist"] == ["echo", "ls"]
assert result["models"][0].get("thinking") is None
assert result["models"][1]["thinking"] == "max"
def test_noop_when_find_not_present(
class TestMigrateLeavesFindInBashAllowlist:
def test_keeps_find_in_config_file(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("VIBE_HOME", str(tmp_path))
@ -232,7 +256,24 @@ class TestMigrateRemovesFindFromBashAllowlist:
with config_file.open("rb") as f:
result = tomllib.load(f)
assert result["tools"]["bash"]["allowlist"] == ["echo", "ls"]
assert result["tools"]["bash"]["allowlist"] == ["echo", "find", "ls"]
def test_noop_when_find_already_present(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("VIBE_HOME", str(tmp_path))
config_file = tmp_path / "config.toml"
data = {"tools": {"bash": {"allowlist": ["echo", "find", "ls"]}}}
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"]["bash"]["allowlist"] == ["echo", "find", "ls"]
def test_noop_when_no_bash_tools_section(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
@ -313,7 +354,7 @@ class TestDefaultProviderConfig:
class TestMistralBrowserAuthConfig:
def test_provider_browser_auth_urls_are_dumped_when_set(self) -> None:
cfg = build_test_vibe_config()
provider = cfg.get_provider_for_model(cfg.get_active_model())
provider = cfg.get_active_provider()
dumped = cfg.model_dump(mode="json")
assert provider.browser_auth_base_url == DEFAULT_MISTRAL_BROWSER_AUTH_BASE_URL
@ -808,7 +849,7 @@ class TestGetMistralProvider:
def test_returns_active_provider_when_it_is_mistral(self) -> None:
cfg = build_test_vibe_config()
provider = cfg.get_mistral_provider()
active = cfg.get_provider_for_model(cfg.get_active_model())
active = cfg.get_active_provider()
assert provider is active
assert provider is not None
assert provider.backend == Backend.MISTRAL
@ -867,3 +908,46 @@ class TestGetMistralProvider:
)
provider = cfg.get_mistral_provider()
assert provider is mistral_provider
class TestIsActiveModelMistral:
def test_returns_true_when_active_provider_is_mistral(self) -> None:
cfg = build_test_vibe_config()
assert cfg.is_active_model_mistral() is True
def test_returns_false_when_active_provider_is_not_mistral(self) -> None:
cfg = build_test_vibe_config(
providers=[
ProviderConfig(
name="llamacpp",
api_base="http://127.0.0.1:8080/v1",
api_key_env_var="",
)
],
models=[
ModelConfig(
name="llama-local", provider="llamacpp", alias="llama-local"
)
],
active_model="llama-local",
)
assert cfg.is_active_model_mistral() is False
def test_returns_false_when_active_model_resolution_fails(self) -> None:
cfg = build_test_vibe_config(
providers=[
ProviderConfig(
name="mistral",
api_base="https://api.mistral.ai/v1",
api_key_env_var="MISTRAL_API_KEY",
backend=Backend.MISTRAL,
)
],
models=[
ModelConfig(
name="llama-local", provider="llamacpp", alias="llama-local"
)
],
active_model="llama-local",
)
assert cfg.is_active_model_mistral() is False

504
tests/core/test_hooks.py Normal file
View file

@ -0,0 +1,504 @@
from __future__ import annotations
from pathlib import Path
import sys
import pytest
import tomli_w
from tests.conftest import build_test_agent_loop
from tests.mock.utils import mock_llm_chunk
from tests.stubs.fake_backend import FakeBackend
from vibe.core.config import VibeConfig
from vibe.core.hooks.config import (
HookConfig,
HookConfigResult,
_load_hooks_file,
load_hooks_from_fs,
)
from vibe.core.hooks.executor import HookExecutor
from vibe.core.hooks.manager import HooksManager
from vibe.core.hooks.models import (
HookEndEvent,
HookInvocation,
HookMessageSeverity,
HookStartEvent,
HookType,
HookUserMessage,
)
from vibe.core.types import BaseEvent
@pytest.fixture
def sample_invocation() -> HookInvocation:
return HookInvocation(
session_id="test-session",
transcript_path="",
cwd=str(Path.cwd()),
hook_event_name="post_agent_turn",
)
@pytest.fixture
def config_hooks_disabled() -> VibeConfig:
return VibeConfig(enable_experimental_hooks=False)
@pytest.fixture
def config_hooks_enabled() -> VibeConfig:
return VibeConfig(enable_experimental_hooks=True)
def _write_hooks_toml(path: Path, hooks: list[dict]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("wb") as f:
tomli_w.dump({"hooks": hooks}, f)
def _make_hook(
name: str = "test-hook", command: str = "echo ok", timeout: float = 30.0
) -> HookConfig:
return HookConfig(
name=name, type=HookType.POST_AGENT_TURN, command=command, timeout=timeout
)
class TestConfigLoading:
def test_load_from_global_file(
self, config_dir: Path, config_hooks_enabled: VibeConfig
) -> None:
_write_hooks_toml(
config_dir / "hooks.toml",
[
{
"name": "lint",
"type": HookType.POST_AGENT_TURN,
"command": "echo lint",
}
],
)
result = load_hooks_from_fs(config_hooks_enabled)
assert len(result.hooks) == 1
assert result.hooks[0].name == "lint"
assert result.issues == []
def test_load_from_both_global_and_project(
self,
config_dir: Path,
tmp_working_directory: Path,
config_hooks_enabled: VibeConfig,
) -> None:
_write_hooks_toml(
config_dir / "hooks.toml",
[
{
"name": "global-hook",
"type": "post_agent_turn",
"command": "echo global",
}
],
)
project_vibe = tmp_working_directory / ".vibe"
_write_hooks_toml(
project_vibe / "hooks.toml",
[
{
"name": "project-hook",
"type": "post_agent_turn",
"command": "echo project",
}
],
)
from vibe.core.trusted_folders import trusted_folders_manager
trusted_folders_manager.add_trusted(tmp_working_directory)
result = load_hooks_from_fs(config_hooks_enabled)
assert len(result.hooks) == 2
names = {h.name for h in result.hooks}
assert names == {"global-hook", "project-hook"}
def test_project_file_skipped_when_untrusted(
self, tmp_working_directory: Path, config_hooks_enabled: VibeConfig
) -> None:
project_vibe = tmp_working_directory / ".vibe"
_write_hooks_toml(
project_vibe / "hooks.toml",
[
{
"name": "sneaky-hook",
"type": "post_agent_turn",
"command": "echo sneaky",
}
],
)
result = load_hooks_from_fs(config_hooks_enabled)
assert not any(h.name == "sneaky-hook" for h in result.hooks)
def test_duplicate_hook_name_detection(
self,
config_dir: Path,
tmp_working_directory: Path,
config_hooks_enabled: VibeConfig,
) -> None:
_write_hooks_toml(
config_dir / "hooks.toml",
[{"name": "dup-hook", "type": "post_agent_turn", "command": "echo global"}],
)
project_vibe = tmp_working_directory / ".vibe"
_write_hooks_toml(
project_vibe / "hooks.toml",
[
{
"name": "dup-hook",
"type": "post_agent_turn",
"command": "echo project",
}
],
)
from vibe.core.trusted_folders import trusted_folders_manager
trusted_folders_manager.add_trusted(tmp_working_directory)
result = load_hooks_from_fs(config_hooks_enabled)
assert len(result.hooks) == 1
assert any("Duplicate" in i.message for i in result.issues)
def test_toml_parse_error_reported(
self, config_dir: Path, config_hooks_enabled: VibeConfig
) -> None:
hooks_file = config_dir / "hooks.toml"
hooks_file.write_text("this is not valid toml [[[", encoding="utf-8")
result = load_hooks_from_fs(config_hooks_enabled)
assert result.hooks == []
assert len(result.issues) == 1
assert (
"parse" in result.issues[0].message.lower()
or "Failed" in result.issues[0].message
)
def test_validation_error_reported(
self, config_dir: Path, config_hooks_enabled: VibeConfig
) -> None:
_write_hooks_toml(
config_dir / "hooks.toml",
[{"name": "bad", "type": "InvalidType", "command": "echo"}],
)
result = load_hooks_from_fs(config_hooks_enabled)
assert result.hooks == []
assert len(result.issues) == 1
def test_missing_command_reported(
self, config_dir: Path, config_hooks_enabled: VibeConfig
) -> None:
_write_hooks_toml(
config_dir / "hooks.toml", [{"name": "no-cmd", "type": "post_agent_turn"}]
)
result = load_hooks_from_fs(config_hooks_enabled)
assert result.hooks == []
assert len(result.issues) == 1
def test_empty_command_reported(
self, config_dir: Path, config_hooks_enabled: VibeConfig
) -> None:
_write_hooks_toml(
config_dir / "hooks.toml",
[{"name": "empty-cmd", "type": "post_agent_turn", "command": " "}],
)
result = load_hooks_from_fs(config_hooks_enabled)
assert result.hooks == []
assert len(result.issues) == 1
def test_default_timeout(
self, config_dir: Path, config_hooks_enabled: VibeConfig
) -> None:
_write_hooks_toml(
config_dir / "hooks.toml",
[{"name": "h", "type": "post_agent_turn", "command": "echo ok"}],
)
result = load_hooks_from_fs(config_hooks_enabled)
assert result.hooks[0].timeout == 30.0
def test_nonexistent_file_returns_empty(self, tmp_path: Path) -> None:
result = _load_hooks_file(tmp_path / "missing.toml")
assert result.hooks == []
assert result.issues == []
def test_hooks_disabled_returns_empty(
self, config_dir: Path, config_hooks_disabled: VibeConfig
) -> None:
_write_hooks_toml(
config_dir / "hooks.toml",
[
{
"name": "lint",
"type": HookType.POST_AGENT_TURN,
"command": "echo lint",
}
],
)
result = load_hooks_from_fs(config_hooks_disabled)
assert result.hooks == []
assert result.issues == []
class TestHookExecutor:
@pytest.mark.asyncio
async def test_exit_0_success(self, sample_invocation: HookInvocation) -> None:
hook = _make_hook(command="echo success")
result = await HookExecutor().run(hook, sample_invocation)
assert result.exit_code == 0
assert result.stdout == "success"
assert not result.timed_out
@pytest.mark.asyncio
async def test_exit_2_retry(self, sample_invocation: HookInvocation) -> None:
hook = _make_hook(command="echo 'fix this'; exit 2")
result = await HookExecutor().run(hook, sample_invocation)
assert result.exit_code == 2
assert "fix this" in result.stdout
assert not result.timed_out
@pytest.mark.asyncio
async def test_other_exit_code(self, sample_invocation: HookInvocation) -> None:
hook = _make_hook(command="echo 'oops'; exit 1")
result = await HookExecutor().run(hook, sample_invocation)
assert result.exit_code == 1
assert "oops" in result.stdout
@pytest.mark.asyncio
async def test_timeout(self, sample_invocation: HookInvocation) -> None:
hook = _make_hook(command="sleep 60", timeout=0.5)
result = await HookExecutor().run(hook, sample_invocation)
assert result.timed_out
assert result.exit_code is None
@pytest.mark.asyncio
async def test_stderr_captured_separately(
self, sample_invocation: HookInvocation
) -> None:
hook = _make_hook(command="echo out; echo err >&2")
result = await HookExecutor().run(hook, sample_invocation)
assert result.exit_code == 0
assert result.stdout == "out"
assert result.stderr == "err"
@pytest.mark.asyncio
async def test_stdin_json_received(self, sample_invocation: HookInvocation) -> None:
hook = _make_hook(
command=f"{sys.executable} -c \"import sys,json; d=json.load(sys.stdin); print(d['session_id'])\""
)
result = await HookExecutor().run(hook, sample_invocation)
assert result.exit_code == 0
assert result.stdout == "test-session"
class TestHooksManager:
@pytest.mark.asyncio
async def test_exit_0_emits_start_and_end(self) -> None:
handler = HooksManager([_make_hook(command="echo ok")])
from vibe.core.config import SessionLoggingConfig
from vibe.core.session.session_logger import SessionLogger
logger = SessionLogger(SessionLoggingConfig(enabled=False), "test-id")
events: list[BaseEvent | HookUserMessage] = []
async for ev in handler.run(HookType.POST_AGENT_TURN, "sess", logger):
events.append(ev)
event_types = [type(e).__name__ for e in events]
assert "HookStartEvent" in event_types
assert "HookEndEvent" in event_types
# Exit 0 with no retry = no HookUserMessage
assert not any(isinstance(e, HookUserMessage) for e in events)
@pytest.mark.asyncio
async def test_exit_2_emits_retry_message(self) -> None:
handler = HooksManager([_make_hook(command="echo 'fix it'; exit 2")])
from vibe.core.config import SessionLoggingConfig
from vibe.core.session.session_logger import SessionLogger
logger = SessionLogger(SessionLoggingConfig(enabled=False), "test-id")
events: list[BaseEvent | HookUserMessage] = []
async for ev in handler.run(HookType.POST_AGENT_TURN, "sess", logger):
events.append(ev)
retry_msgs = [e for e in events if isinstance(e, HookUserMessage)]
assert len(retry_msgs) == 1
assert "fix it" in retry_msgs[0].content
# Display message should be generic, not the stdout
end_msgs = [
e for e in events if isinstance(e, HookEndEvent) and e.content is not None
]
assert any("retrying" in m.content.lower() for m in end_msgs if m.content)
assert not any("fix it" in (m.content or "") for m in end_msgs)
@pytest.mark.asyncio
async def test_exit_2_without_output_emits_warning(self) -> None:
handler = HooksManager([_make_hook(command="exit 2")])
from vibe.core.config import SessionLoggingConfig
from vibe.core.session.session_logger import SessionLogger
logger = SessionLogger(SessionLoggingConfig(enabled=False), "test-id")
events: list[BaseEvent | HookUserMessage] = []
async for ev in handler.run(HookType.POST_AGENT_TURN, "sess", logger):
events.append(ev)
end_msgs = [e for e in events if isinstance(e, HookEndEvent)]
assert len(end_msgs) == 1
assert end_msgs[0].content == "Exited with code 2"
@pytest.mark.asyncio
async def test_max_retry_limit(self) -> None:
handler = HooksManager([_make_hook(command="echo retry; exit 2")])
from vibe.core.config import SessionLoggingConfig
from vibe.core.session.session_logger import SessionLogger
logger = SessionLogger(SessionLoggingConfig(enabled=False), "test-id")
# Run 3 times (should get retry each time)
for _ in range(3):
events = [
ev async for ev in handler.run(HookType.POST_AGENT_TURN, "sess", logger)
]
assert any(isinstance(e, HookUserMessage) for e in events)
# 4th time: max exceeded, no retry
events = [
ev async for ev in handler.run(HookType.POST_AGENT_TURN, "sess", logger)
]
assert not any(isinstance(e, HookUserMessage) for e in events)
# Should have error message about max retries
error_events = [
e
for e in events
if isinstance(e, HookEndEvent)
and e.content
and "exhausted" in e.content.lower()
]
assert len(error_events) == 1
@pytest.mark.asyncio
async def test_warning_on_nonzero_exit(self) -> None:
handler = HooksManager([_make_hook(command="echo warn; exit 1")])
from vibe.core.config import SessionLoggingConfig
from vibe.core.session.session_logger import SessionLogger
logger = SessionLogger(SessionLoggingConfig(enabled=False), "test-id")
events = [
ev async for ev in handler.run(HookType.POST_AGENT_TURN, "sess", logger)
]
warnings = [
e
for e in events
if isinstance(e, HookEndEvent) and e.status == HookMessageSeverity.WARNING
]
assert len(warnings) == 1
assert warnings[0].content and "warn" in warnings[0].content
@pytest.mark.asyncio
async def test_warning_falls_back_to_stderr(self) -> None:
hook = _make_hook(command="echo problem >&2; exit 1")
handler = HooksManager([hook])
from vibe.core.config import SessionLoggingConfig
from vibe.core.session.session_logger import SessionLogger
logger = SessionLogger(SessionLoggingConfig(enabled=False), "test-id")
events = [
ev async for ev in handler.run(HookType.POST_AGENT_TURN, "sess", logger)
]
warnings = [
e
for e in events
if isinstance(e, HookEndEvent) and e.status == HookMessageSeverity.WARNING
]
assert len(warnings) == 1
assert warnings[0].content and "problem" in warnings[0].content
@pytest.mark.asyncio
async def test_timeout_emits_warning(self) -> None:
handler = HooksManager([_make_hook(command="sleep 60", timeout=0.5)])
from vibe.core.config import SessionLoggingConfig
from vibe.core.session.session_logger import SessionLogger
logger = SessionLogger(SessionLoggingConfig(enabled=False), "test-id")
events = [
ev async for ev in handler.run(HookType.POST_AGENT_TURN, "sess", logger)
]
warnings = [
e
for e in events
if isinstance(e, HookEndEvent) and e.status == HookMessageSeverity.WARNING
]
assert len(warnings) == 1
assert warnings[0].content and "Timed out" in warnings[0].content
class TestAgentLoopIntegration:
@pytest.mark.asyncio
async def test_hooks_run_after_turn(self) -> None:
backend = FakeBackend(mock_llm_chunk(content="Hello!"))
hooks = [_make_hook(name="post-lint", command="echo ok")]
agent_loop = build_test_agent_loop(
backend=backend, hook_config_result=HookConfigResult(hooks=hooks, issues=[])
)
events = [ev async for ev in agent_loop.act("hi")]
event_types = [type(e).__name__ for e in events]
assert "HookStartEvent" in event_types
assert "HookEndEvent" in event_types
@pytest.mark.asyncio
async def test_hook_retry_reinjects_message(self) -> None:
# First call: LLM responds. Hook requests retry with "fix this".
# Second call (after retry injection): LLM responds again. Hook exits 0.
backend = FakeBackend([
[mock_llm_chunk(content="first response")],
[mock_llm_chunk(content="second response")],
])
# Create a script that exits 2 on first call, 0 on subsequent
counter_file = Path.cwd() / ".hook_counter"
script = (
f'{sys.executable} -c "'
f"from pathlib import Path; "
f"p = Path({str(counter_file)!r}); "
f"c = int(p.read_text()) if p.exists() else 0; "
f"p.write_text(str(c + 1)); "
f"import sys; "
f"print('fix this'); "
f"sys.exit(2 if c == 0 else 0)"
f'"'
)
hooks = [_make_hook(name="retry-hook", command=script)]
agent_loop = build_test_agent_loop(
backend=backend, hook_config_result=HookConfigResult(hooks=hooks, issues=[])
)
events = [ev async for ev in agent_loop.act("hi")]
# Should have two assistant events (two LLM turns)
from vibe.core.types import AssistantEvent
assistant_events = [e for e in events if isinstance(e, AssistantEvent)]
assert len(assistant_events) == 2
# Check that a retry user message was injected
user_messages = [
m for m in agent_loop.messages if m.role.value == "user" and m.injected
]
assert any("fix this" in (m.content or "") for m in user_messages)
@pytest.mark.asyncio
async def test_no_hooks_no_events(self) -> None:
backend = FakeBackend(mock_llm_chunk(content="Hello!"))
agent_loop = build_test_agent_loop(backend=backend)
events = [ev async for ev in agent_loop.act("hi")]
hook_events = [
e for e in events if isinstance(e, (HookStartEvent, HookEndEvent))
]
assert hook_events == []

170
tests/core/test_remote_agent_loop.py
View file

@ -1,6 +1,7 @@
from __future__ import annotations
from typing import Any
from unittest.mock import patch
from tests.conftest import build_test_vibe_config
from vibe.core.nuage.events import (
@ -244,6 +245,28 @@ def test_ask_user_question_tool_emits_assistant_question() -> None:
assert tool_call_event.tool_call_id == "call-question"
def test_ask_user_question_invalid_args_are_logged_and_ignored() -> None:
loop = _make_loop(enabled_tools=["ask_user_question"])
started = _started(
"tool-task-question",
"AgentToolCallState",
{
"name": "ask_user_question",
"tool_call_id": "call-question",
"kwargs": {"questions": [{}]},
},
)
with patch(
"vibe.core.nuage.remote_workflow_event_translator.logger.warning"
) as mock_warning:
events = loop._consume_workflow_event(started)
assert any(isinstance(event, ToolCallEvent) for event in events)
assert not any(isinstance(event, AssistantEvent) for event in events)
mock_warning.assert_called_once()
def test_ask_user_question_wait_for_input_completion_emits_tool_result() -> None:
loop = _make_loop(enabled_tools=["ask_user_question"])
ask_started = _started(
@ -765,5 +788,152 @@ def test_steer_input_events_are_suppressed() -> None:
)
assert loop._consume_workflow_event(steer_started) == []
assert loop._translator.pending_input_request is not None
assert loop._translator.pending_input_request.task_id == "steer-1"
assert loop._consume_workflow_event(steer_completed) == []
assert loop._translator.pending_input_request is None
def test_steer_input_allows_user_submission() -> None:
loop = _make_loop()
steer_started = _started(
"steer-1",
"wait_for_input",
{"input_schema": {"title": "ChatInput"}, "label": "Send a message to steer..."},
)
assert loop._consume_workflow_event(steer_started) == []
assert loop.is_waiting_for_input
loop._translator.pending_input_request = None
steer_completed = _completed(
"steer-1",
"wait_for_input",
{
"input_schema": {"title": "ChatInput"},
"label": "Send a message to steer...",
"input": {"message": [{"type": "text", "text": "do X instead"}]},
},
)
events = loop._consume_workflow_event(steer_completed)
assert any(isinstance(e, UserMessageEvent) for e in events)
user_event = next(e for e in events if isinstance(e, UserMessageEvent))
assert user_event.content == "do X instead"
def test_steer_does_not_overwrite_regular_pending_input() -> None:
loop = _make_loop()
regular_started = _started(
"regular-1",
"wait_for_input",
{"input_schema": {"title": "ChatInput"}, "label": "Enter your message"},
)
loop._consume_workflow_event(regular_started)
assert loop._translator.pending_input_request is not None
assert loop._translator.pending_input_request.task_id == "regular-1"
steer_started = _started(
"steer-1",
"wait_for_input",
{"input_schema": {"title": "ChatInput"}, "label": "Send a message to steer..."},
)
loop._consume_workflow_event(steer_started)
assert loop._translator.pending_input_request.task_id == "regular-1"
def test_invalid_steer_start_registers_task_for_terminal_handling() -> None:
loop = _make_loop()
steer_started = _started(
"steer-1", "wait_for_input", {"label": "Send a message to steer..."}
)
with patch("vibe.core.nuage.remote_events_source.logger.warning") as mock_warning:
assert loop._consume_workflow_event(steer_started) == []
mock_warning.assert_called_once()
assert "steer-1" in loop._translator._steer_task_ids
assert "steer-1" in loop._translator._invalid_steer_task_ids
assert loop._translator.pending_input_request is None
def test_invalid_steer_completion_does_not_clear_regular_prompt() -> None:
loop = _make_loop()
regular_started = _started(
"regular-1",
"wait_for_input",
{"input_schema": {"title": "ChatInput"}, "label": "Pick an option"},
)
loop._consume_workflow_event(regular_started)
steer_started = _started(
"steer-1", "wait_for_input", {"label": "Send a message to steer..."}
)
assert loop._consume_workflow_event(steer_started) == []
assert loop._translator.pending_input_request is not None
assert loop._translator.pending_input_request.task_id == "regular-1"
steer_completed = _completed(
"steer-1",
"wait_for_input",
{"label": "Send a message to steer...", "input": None},
)
events = loop._consume_workflow_event(steer_completed)
assert events == []
assert loop._translator.pending_input_request is not None
assert loop._translator.pending_input_request.task_id == "regular-1"
def test_invalid_steer_cancellation_does_not_clear_regular_prompt() -> None:
loop = _make_loop()
regular_started = _started(
"regular-1",
"wait_for_input",
{"input_schema": {"title": "ChatInput"}, "label": "Pick an option"},
)
loop._consume_workflow_event(regular_started)
steer_started = _started(
"steer-1", "wait_for_input", {"label": "Send a message to steer..."}
)
assert loop._consume_workflow_event(steer_started) == []
assert loop._translator.pending_input_request is not None
assert loop._translator.pending_input_request.task_id == "regular-1"
events = loop._consume_workflow_event(_canceled("steer-1", "wait_for_input"))
assert events == []
assert loop._translator.pending_input_request is not None
assert loop._translator.pending_input_request.task_id == "regular-1"
def test_steer_completion_is_preserved_while_regular_prompt_pending() -> None:
loop = _make_loop()
regular_started = _started(
"regular-1",
"wait_for_input",
{"input_schema": {"title": "ChatInput"}, "label": "Pick an option"},
)
loop._consume_workflow_event(regular_started)
steer_started = _started(
"steer-1",
"wait_for_input",
{"input_schema": {"title": "ChatInput"}, "label": "Send a message to steer..."},
)
loop._consume_workflow_event(steer_started)
steer_completed = _completed(
"steer-1",
"wait_for_input",
{
"input_schema": {"title": "ChatInput"},
"label": "Send a message to steer...",
"input": {"message": [{"type": "text", "text": "user steer msg"}]},
},
)
events = loop._consume_workflow_event(steer_completed)
user_event = next(e for e in events if isinstance(e, UserMessageEvent))
assert user_event.content == "user steer msg"
assert loop._translator.pending_input_request is not None
assert loop._translator.pending_input_request.task_id == "regular-1"

81
tests/core/test_scratchpad.py Normal file
View file

@ -0,0 +1,81 @@
from __future__ import annotations
from vibe.core.scratchpad import get_scratchpad_dir, init_scratchpad, is_scratchpad_path
class TestInitScratchpad:
def test_creates_directory(self):
result = init_scratchpad("test-session")
assert result is not None
assert result.is_dir()
def test_idempotent_same_session(self):
first = init_scratchpad("session-1")
second = init_scratchpad("session-1")
assert first == second
def test_different_sessions_get_different_dirs(self):
first = init_scratchpad("session-1")
second = init_scratchpad("session-2")
assert first != second
def test_session_id_in_dir_name(self):
result = init_scratchpad("abcdef123456")
assert result is not None
assert "abcdef12" in result.name
def test_sets_module_state(self):
init_scratchpad("test-session")
assert get_scratchpad_dir("test-session") is not None
class TestGetScratchpadDir:
def test_none_for_unknown_session(self):
assert get_scratchpad_dir("nonexistent") is None
def test_returns_path_after_init(self):
path = init_scratchpad("test-session")
assert get_scratchpad_dir("test-session") == path
class TestIsScratchpadPath:
def test_false_when_not_initialized(self):
assert not is_scratchpad_path("/tmp/anything")
def test_true_for_file_inside(self):
sp = init_scratchpad("test-session")
assert sp is not None
assert is_scratchpad_path(str(sp / "file.txt"))
def test_true_for_nested_file(self):
sp = init_scratchpad("test-session")
assert sp is not None
assert is_scratchpad_path(str(sp / "subdir" / "file.txt"))
def test_true_for_dir_itself(self):
sp = init_scratchpad("test-session")
assert sp is not None
assert is_scratchpad_path(str(sp))
def test_true_across_sessions(self):
sp1 = init_scratchpad("session-1")
sp2 = init_scratchpad("session-2")
assert sp1 is not None and sp2 is not None
assert is_scratchpad_path(str(sp1 / "file.txt"))
assert is_scratchpad_path(str(sp2 / "file.txt"))
def test_false_for_outside_path(self):
init_scratchpad("test-session")
assert not is_scratchpad_path("/etc/passwd")
def test_false_for_traversal_attack(self):
sp = init_scratchpad("test-session")
assert sp is not None
traversal = str(sp / ".." / ".." / ".." / "etc" / "passwd")
assert not is_scratchpad_path(traversal)
def test_false_for_sibling_directory(self):
sp = init_scratchpad("test-session")
assert sp is not None
sibling = str(sp.parent / "other-dir" / "file.txt")
assert not is_scratchpad_path(sibling)

188
tests/core/test_telemetry_send.py
View file

@ -10,7 +10,12 @@ from tests.conftest import build_test_vibe_config
from tests.stubs.fake_tool import FakeTool, FakeToolArgs
from vibe.core.agent_loop import ToolDecision, ToolExecutionResponse
from vibe.core.llm.format import ResolvedToolCall
from vibe.core.telemetry.build_metadata import (
build_base_metadata,
build_request_metadata,
)
from vibe.core.telemetry.send import TelemetryClient
from vibe.core.telemetry.types import EntrypointMetadata, TelemetryRequestMetadata
from vibe.core.tools.base import BaseTool, ToolPermission
from vibe.core.types import Backend
from vibe.core.utils import get_user_agent
@ -55,9 +60,7 @@ class TestTelemetryClient:
self, monkeypatch: pytest.MonkeyPatch
) -> None:
config = build_test_vibe_config(enable_telemetry=True)
env_key = config.get_provider_for_model(
config.get_active_model()
).api_key_env_var
env_key = config.get_active_provider().api_key_env_var
monkeypatch.delenv(env_key, raising=False)
client = TelemetryClient(config_getter=lambda: config)
assert client._get_mistral_api_key() is None
@ -76,9 +79,7 @@ class TestTelemetryClient:
TelemetryClient, "send_telemetry_event", _original_send_telemetry_event
)
config = build_test_vibe_config(enable_telemetry=False)
env_key = config.get_provider_for_model(
config.get_active_model()
).api_key_env_var
env_key = config.get_active_provider().api_key_env_var
monkeypatch.setenv(env_key, "sk-test")
client = TelemetryClient(config_getter=lambda: config)
client._client = MagicMock()
@ -97,9 +98,7 @@ class TestTelemetryClient:
TelemetryClient, "send_telemetry_event", _original_send_telemetry_event
)
config = build_test_vibe_config(enable_telemetry=True)
env_key = config.get_provider_for_model(
config.get_active_model()
).api_key_env_var
env_key = config.get_active_provider().api_key_env_var
monkeypatch.setenv(env_key, "sk-test")
client = TelemetryClient(config_getter=lambda: config)
mock_post = AsyncMock(return_value=MagicMock(status_code=204))
@ -150,6 +149,25 @@ class TestTelemetryClient:
assert properties["model"] == "mistral-large"
assert properties["nb_files_created"] == 0
assert properties["nb_files_modified"] == 0
assert properties["message_id"] is None
def test_send_tool_call_finished_with_message_id(
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("todo", {})
client.send_tool_call_finished(
tool_call=tool_call,
status="success",
decision=None,
agent_profile_name="default",
model="mistral-large",
message_id="msg-123",
)
assert telemetry_events[0]["properties"]["message_id"] == "msg-123"
def test_send_tool_call_finished_nb_files_created_write_file_new(
self, telemetry_events: list[dict[str, Any]]
@ -237,10 +255,21 @@ class TestTelemetryClient:
config = build_test_vibe_config(enable_telemetry=True)
client = TelemetryClient(config_getter=lambda: config)
client.send_auto_compact_triggered()
client.send_auto_compact_triggered(
nb_context_tokens_before=123,
nb_context_tokens_after=45,
auto_compact_threshold=100,
status="success",
)
assert len(telemetry_events) == 1
assert telemetry_events[0]["event_name"] == "vibe.auto_compact_triggered"
assert telemetry_events[0]["properties"] == {
"nb_context_tokens_before": 123,
"nb_context_tokens_after": 45,
"auto_compact_threshold": 100,
"status": "success",
}
def test_send_slash_command_used_payload(
self, telemetry_events: list[dict[str, Any]]
@ -289,8 +318,55 @@ class TestTelemetryClient:
assert properties["terminal_emulator"] == "vscode"
assert "version" in properties
def test_build_base_metadata_includes_entrypoint_and_session(self) -> None:
metadata = build_base_metadata(
entrypoint_metadata=EntrypointMetadata(
agent_entrypoint="cli",
agent_version="1.0.0",
client_name="vibe_cli",
client_version="1.0.0",
),
session_id="session-123",
parent_session_id="parent-session-456",
)
assert metadata == {
"agent_entrypoint": "cli",
"agent_version": "1.0.0",
"client_name": "vibe_cli",
"client_version": "1.0.0",
"session_id": "session-123",
"parent_session_id": "parent-session-456",
}
def test_build_request_metadata_includes_all_telemetry_metadata(self) -> None:
metadata = build_request_metadata(
entrypoint_metadata=EntrypointMetadata(
agent_entrypoint="cli",
agent_version="1.0.0",
client_name="vibe_cli",
client_version="1.0.0",
),
session_id="session-123",
parent_session_id="parent-session-456",
call_type="secondary_call",
message_id="message-456",
)
assert metadata == TelemetryRequestMetadata(
agent_entrypoint="cli",
agent_version="1.0.0",
client_name="vibe_cli",
client_version="1.0.0",
session_id="session-123",
parent_session_id="parent-session-456",
call_source="vibe_code",
call_type="secondary_call",
message_id="message-456",
)
@pytest.mark.asyncio
async def test_session_id_added_when_getter_provided(
async def test_parent_session_id_added_when_getter_provided(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(
@ -301,6 +377,46 @@ class TestTelemetryClient:
config.get_active_model()
).api_key_env_var
monkeypatch.setenv(env_key, "sk-test")
client = TelemetryClient(
config_getter=lambda: config,
session_id_getter=lambda: "session-123",
parent_session_id_getter=lambda: "parent-session-456",
)
mock_post = AsyncMock(return_value=MagicMock(status_code=204))
client._client = MagicMock()
client._client.post = mock_post
client._client.aclose = AsyncMock()
client.send_telemetry_event("vibe.test_event", {"key": "value"})
await client.aclose()
mock_post.assert_called_once_with(
"https://api.mistral.ai/v1/datalake/events",
json={
"event": "vibe.test_event",
"properties": {
"session_id": "session-123",
"parent_session_id": "parent-session-456",
"key": "value",
},
},
headers={
"Content-Type": "application/json",
"Authorization": "Bearer sk-test",
"User-Agent": get_user_agent(Backend.MISTRAL),
},
)
@pytest.mark.asyncio
async def test_session_id_added_when_getter_provided(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(
TelemetryClient, "send_telemetry_event", _original_send_telemetry_event
)
config = build_test_vibe_config(enable_telemetry=True)
env_key = config.get_active_provider().api_key_env_var
monkeypatch.setenv(env_key, "sk-test")
session_id = "test-session-uuid"
client = TelemetryClient(
config_getter=lambda: config, session_id_getter=lambda: session_id
@ -334,9 +450,7 @@ class TestTelemetryClient:
TelemetryClient, "send_telemetry_event", _original_send_telemetry_event
)
config = build_test_vibe_config(enable_telemetry=True)
env_key = config.get_provider_for_model(
config.get_active_model()
).api_key_env_var
env_key = config.get_active_provider().api_key_env_var
monkeypatch.setenv(env_key, "sk-test")
client = TelemetryClient(config_getter=lambda: config)
mock_post = AsyncMock(return_value=MagicMock(status_code=204))
@ -365,9 +479,7 @@ class TestTelemetryClient:
TelemetryClient, "send_telemetry_event", _original_send_telemetry_event
)
config = build_test_vibe_config(enable_telemetry=True)
env_key = config.get_provider_for_model(
config.get_active_model()
).api_key_env_var
env_key = config.get_active_provider().api_key_env_var
monkeypatch.setenv(env_key, "sk-test")
current_id = "first-session-id"
client = TelemetryClient(
@ -389,6 +501,40 @@ class TestTelemetryClient:
calls[1].kwargs["json"]["properties"]["session_id"] == "second-session-id"
)
def test_send_auto_compact_triggered_overrides_session_metadata(
self, telemetry_events: list[dict[str, Any]]
) -> None:
config = build_test_vibe_config(enable_telemetry=True)
client = TelemetryClient(
config_getter=lambda: config,
session_id_getter=lambda: "new-session-id",
parent_session_id_getter=lambda: "new-parent-session-id",
)
client.send_auto_compact_triggered(
nb_context_tokens_before=123,
nb_context_tokens_after=45,
auto_compact_threshold=100,
status="success",
session_id="original-session-id",
parent_session_id=None,
)
assert len(telemetry_events) == 1
properties = telemetry_events[0]["properties"]
assert properties["session_id"] == "original-session-id"
assert properties["parent_session_id"] is None
def test_send_ready_payload(self, telemetry_events: list[dict[str, Any]]) -> None:
config = build_test_vibe_config(enable_telemetry=True)
client = TelemetryClient(config_getter=lambda: config)
client.send_ready(init_duration_ms=1240)
assert len(telemetry_events) == 1
assert telemetry_events[0]["event_name"] == "vibe.ready"
assert telemetry_events[0]["properties"]["init_duration_ms"] == 1240
def test_send_request_sent_payload(
self, telemetry_events: list[dict[str, Any]]
) -> None:
@ -400,6 +546,7 @@ class TestTelemetryClient:
nb_context_chars=1234,
nb_context_messages=5,
nb_prompt_chars=42,
call_type="main_call",
)
assert len(telemetry_events) == 1
@ -409,6 +556,9 @@ class TestTelemetryClient:
assert properties["nb_context_chars"] == 1234
assert properties["nb_context_messages"] == 5
assert properties["nb_prompt_chars"] == 42
assert properties["call_source"] == "vibe_code"
assert properties["call_type"] == "main_call"
assert properties["message_id"] is None
def test_send_user_rating_feedback_payload(
self, telemetry_events: list[dict[str, Any]]
@ -522,9 +672,7 @@ class TestTelemetryClient:
self, monkeypatch: pytest.MonkeyPatch
) -> None:
config = build_test_vibe_config(enable_telemetry=True)
env_key = config.get_provider_for_model(
config.get_active_model()
).api_key_env_var
env_key = config.get_active_provider().api_key_env_var
monkeypatch.delenv(env_key, raising=False)
client = TelemetryClient(config_getter=lambda: config)

4
tests/core/test_teleport_nuage.py
View file

@ -134,7 +134,9 @@ class TestNuageClientStartWorkflow:
mock_client.post = AsyncMock(return_value=mock_response)
params = WorkflowParams(prompt="test", integrations=WorkflowIntegrations())
with pytest.raises(ServiceTeleportError, match="Nuage workflow trigger failed"):
with pytest.raises(
ServiceTeleportError, match="Vibe Code workflow trigger failed"
):
await nuage.start_workflow(params)

71
tests/core/test_teleport_service.py
View file

@ -45,9 +45,9 @@ class TestTeleportServiceCompressDiff:
mock_session_logger = MagicMock()
return TeleportService(
session_logger=mock_session_logger,
nuage_base_url="https://api.example.com",
nuage_workflow_id="workflow-id",
nuage_api_key="api-key",
vibe_code_base_url="https://api.example.com",
vibe_code_workflow_id="workflow-id",
vibe_code_api_key="api-key",
workdir=tmp_path,
)
@ -75,9 +75,9 @@ class TestTeleportServiceBuildGitHubParams:
mock_session_logger = MagicMock()
return TeleportService(
session_logger=mock_session_logger,
nuage_base_url="https://api.example.com",
nuage_workflow_id="workflow-id",
nuage_api_key="api-key",
vibe_code_base_url="https://api.example.com",
vibe_code_workflow_id="workflow-id",
vibe_code_api_key="api-key",
workdir=tmp_path,
)
@ -116,9 +116,9 @@ class TestTeleportServiceValidateConfig:
mock_session_logger = MagicMock()
service = TeleportService(
session_logger=mock_session_logger,
nuage_base_url="https://api.example.com",
nuage_workflow_id="workflow-id",
nuage_api_key="",
vibe_code_base_url="https://api.example.com",
vibe_code_workflow_id="workflow-id",
vibe_code_api_key="",
workdir=tmp_path,
)
with pytest.raises(ServiceTeleportError, match="MISTRAL_API_KEY not set"):
@ -128,9 +128,9 @@ class TestTeleportServiceValidateConfig:
mock_session_logger = MagicMock()
service = TeleportService(
session_logger=mock_session_logger,
nuage_base_url="https://api.example.com",
nuage_workflow_id="workflow-id",
nuage_api_key="valid-key",
vibe_code_base_url="https://api.example.com",
vibe_code_workflow_id="workflow-id",
vibe_code_api_key="valid-key",
workdir=tmp_path,
)
service._validate_config()
@ -138,12 +138,12 @@ class TestTeleportServiceValidateConfig:
def test_uses_custom_env_var_name_in_error(self, tmp_path: Path) -> None:
mock_session_logger = MagicMock()
mock_config = MagicMock()
mock_config.nuage_api_key_env_var = "CUSTOM_API_KEY"
mock_config.vibe_code_api_key_env_var = "CUSTOM_API_KEY"
service = TeleportService(
session_logger=mock_session_logger,
nuage_base_url="https://api.example.com",
nuage_workflow_id="workflow-id",
nuage_api_key="",
vibe_code_base_url="https://api.example.com",
vibe_code_workflow_id="workflow-id",
vibe_code_api_key="",
workdir=tmp_path,
vibe_config=mock_config,
)
@ -157,9 +157,9 @@ class TestTeleportServiceCheckSupported:
mock_session_logger = MagicMock()
return TeleportService(
session_logger=mock_session_logger,
nuage_base_url="https://api.example.com",
nuage_workflow_id="workflow-id",
nuage_api_key="api-key",
vibe_code_base_url="https://api.example.com",
vibe_code_workflow_id="workflow-id",
vibe_code_api_key="api-key",
workdir=tmp_path,
)
@ -199,9 +199,9 @@ class TestTeleportServiceIsSupported:
mock_session_logger = MagicMock()
return TeleportService(
session_logger=mock_session_logger,
nuage_base_url="https://api.example.com",
nuage_workflow_id="workflow-id",
nuage_api_key="api-key",
vibe_code_base_url="https://api.example.com",
vibe_code_workflow_id="workflow-id",
vibe_code_api_key="api-key",
workdir=tmp_path,
)
@ -222,9 +222,9 @@ class TestTeleportServiceExecute:
mock_session_logger = MagicMock()
service = TeleportService(
session_logger=mock_session_logger,
nuage_base_url="https://api.example.com",
nuage_workflow_id="workflow-id",
nuage_api_key="api-key",
vibe_code_base_url="https://api.example.com",
vibe_code_workflow_id="workflow-id",
vibe_code_api_key="api-key",
workdir=tmp_path,
)
service._git.fetch = AsyncMock()
@ -268,7 +268,7 @@ class TestTeleportServiceExecute:
mock_nuage.get_chat_assistant_url = AsyncMock(
return_value="https://chat.example.com/123"
)
service._nuage = mock_nuage
service._nuage_client_instance = mock_nuage
session = TeleportSession()
events = []
@ -282,6 +282,9 @@ class TestTeleportServiceExecute:
assert isinstance(events[3], TeleportFetchingUrlEvent)
assert isinstance(events[4], TeleportCompleteEvent)
assert events[4].url == "https://chat.example.com/123"
workflow_params = mock_nuage.start_workflow.call_args.args[0]
assert workflow_params.integrations.chat_assistant is not None
assert workflow_params.integrations.chat_assistant.project_name is None
@pytest.mark.asyncio
async def test_execute_requires_push_and_user_approves(
@ -303,7 +306,7 @@ class TestTeleportServiceExecute:
mock_nuage.get_chat_assistant_url = AsyncMock(
return_value="https://chat.example.com/123"
)
service._nuage = mock_nuage
service._nuage_client_instance = mock_nuage
session = TeleportSession()
events = []
@ -368,7 +371,7 @@ class TestTeleportServiceExecute:
mock_nuage.get_chat_assistant_url = AsyncMock(
return_value="https://chat.example.com/123"
)
service._nuage = mock_nuage
service._nuage_client_instance = mock_nuage
session = TeleportSession()
events = []
@ -400,7 +403,7 @@ class TestTeleportServiceExecute:
return_value=mock_github_connected
)
mock_nuage.get_chat_assistant_url = AsyncMock(return_value=None)
service._nuage = mock_nuage
service._nuage_client_instance = mock_nuage
session = TeleportSession()
gen = service.execute("test prompt", session)
@ -427,7 +430,7 @@ class TestTeleportServiceExecute:
mock_nuage.get_chat_assistant_url = AsyncMock(
return_value="https://chat.example.com/123"
)
service._nuage = mock_nuage
service._nuage_client_instance = mock_nuage
session = TeleportSession(
messages=[{"role": "user", "content": "help me refactor"}]
@ -446,15 +449,15 @@ class TestTeleportServiceContextManager:
mock_session_logger = MagicMock()
service = TeleportService(
session_logger=mock_session_logger,
nuage_base_url="https://api.example.com",
nuage_workflow_id="workflow-id",
nuage_api_key="api-key",
vibe_code_base_url="https://api.example.com",
vibe_code_workflow_id="workflow-id",
vibe_code_api_key="api-key",
workdir=tmp_path,
)
assert service._client is None
async with service:
assert service._client is not None
assert service._nuage is not None
assert service._nuage_client_instance is not None
assert service._client is None