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

v1.1.3

Browse source 
Co-Authored-By: Quentin Torroba <quentin.torroba@mistral.ai>
Co-Authored-By: Michel Thomazo <michel.thomazo@mistral.ai>
Co-Authored-By: Vincent Guilloux <vincent.guilloux@mistral.ai>
This commit is contained in:
Mathias Gesbert 2025-12-12 17:47:20 +01:00 • committed by Mathias Gesbert
parent a340a721ea
commit 661588de0c
48 changed files with 1679 additions and 467 deletions
Download patch file Download diff file Expand all files Collapse all files

211
tests/acp/test_acp.py
View file

@ -4,6 +4,7 @@ import asyncio
from collections.abc import AsyncGenerator
import json
import os
from pathlib import Path
from typing import Any
from acp import (
@ -27,8 +28,10 @@ from acp.schema import (
)
from pydantic import BaseModel
import pytest
import tomli_w
from tests import TESTS_ROOT
from tests.conftest import get_base_config
from tests.mock.utils import get_mocking_env, mock_llm_chunk
from vibe.acp.utils import ToolOption
from vibe.core.types import FunctionCall, ToolCall
@ -38,6 +41,63 @@ MOCK_ENTRYPOINT_PATH = "tests/mock/mock_entrypoint.py"
PLAYGROUND_DIR = TESTS_ROOT / "playground"
def deep_merge(target: dict, source: dict) -> None:
for key, value in source.items():
if (
key in target
and isinstance(target.get(key), dict)
and isinstance(value, dict)
):
deep_merge(target[key], value)
elif (
key in target
and isinstance(target.get(key), list)
and isinstance(value, list)
):
if key in {"providers", "models"}:
target[key] = value
else:
target[key] = list(set(value + target[key]))
else:
target[key] = value
def _create_vibe_home_dir(tmp_path: Path, *sections: dict[str, Any]) -> Path:
"""Create a temporary vibe home directory with a minimal config file."""
vibe_home = tmp_path / ".vibe"
vibe_home.mkdir()
config_file = vibe_home / "config.toml"
base_config_dict = get_base_config()
base_config_dict["active_model"] = "devstral-latest"
if base_config_dict.get("models"):
for model in base_config_dict["models"]:
if model.get("name") == "mistral-vibe-cli-latest":
model["alias"] = "devstral-latest"
if sections:
for section_dict in sections:
deep_merge(base_config_dict, section_dict)
with config_file.open("wb") as f:
tomli_w.dump(base_config_dict, f)
return vibe_home
@pytest.fixture
def vibe_home_dir(tmp_path: Path) -> Path:
"""Create a temporary vibe home directory with a minimal config file."""
return _create_vibe_home_dir(tmp_path)
@pytest.fixture
def vibe_home_grep_ask(tmp_path: Path) -> Path:
"""Create a temporary vibe home directory with grep configured to ask permission."""
return _create_vibe_home_dir(tmp_path, {"tools": {"grep": {"permission": "ask"}}})
class JsonRpcRequest(BaseModel):
jsonrpc: str = "2.0"
id: int | str
@ -127,10 +187,15 @@ class WriteTextFileJsonRpcResponse(JsonRpcResponse):
async def get_acp_agent_process(
mock: bool = True, mock_env: dict[str, str] | None = None
mock_env: dict[str, str], vibe_home: Path
) -> AsyncGenerator[asyncio.subprocess.Process]:
current_env = os.environ.copy()
cmd = ["uv", "run", MOCK_ENTRYPOINT_PATH if mock else "vibe-acp"]
cmd = ["uv", "run", MOCK_ENTRYPOINT_PATH]
env = dict(current_env)
env.update(mock_env)
env["MISTRAL_API_KEY"] = "mock"
env["VIBE_HOME"] = str(vibe_home)
process = await asyncio.create_subprocess_exec(
*cmd,
@ -138,11 +203,7 @@ async def get_acp_agent_process(
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=TESTS_ROOT.parent,
env={
**current_env,
**(mock_env or {}),
**({"MISTRAL_API_KEY": "mock"} if mock else {}),
},
env=env,
)
try:
@ -322,9 +383,11 @@ async def initialize_session(acp_agent_process: asyncio.subprocess.Process) -> s
class TestSessionManagement:
@pytest.mark.asyncio
async def test_multiple_sessions_unique_ids(self) -> None:
async def test_multiple_sessions_unique_ids(self, vibe_home_dir: Path) -> None:
mock_env = get_mocking_env(mock_chunks=[mock_llm_chunk() for _ in range(3)])
async for process in get_acp_agent_process(mock_env=mock_env):
async for process in get_acp_agent_process(
mock_env=mock_env, vibe_home=vibe_home_dir
):
await send_json_rpc(
process,
InitializeJsonRpcRequest(
@ -359,9 +422,11 @@ class TestSessionManagement:
class TestSessionUpdates:
@pytest.mark.asyncio
async def test_agent_message_chunk_structure(self) -> None:
async def test_agent_message_chunk_structure(self, vibe_home_dir: Path) -> None:
mock_env = get_mocking_env([mock_llm_chunk(content="Hi") for _ in range(2)])
async for process in get_acp_agent_process(mock_env=mock_env):
async for process in get_acp_agent_process(
mock_env=mock_env, vibe_home=vibe_home_dir
):
# Check stderr for error details if process failed
if process.returncode is not None and process.stderr:
stderr_data = await process.stderr.read()
@ -395,7 +460,7 @@ class TestSessionUpdates:
assert response.params.update.content.text == "Hi"
@pytest.mark.asyncio
async def test_tool_call_update_structure(self) -> None:
async def test_tool_call_update_structure(self, vibe_home_dir: Path) -> None:
mock_env = get_mocking_env([
mock_llm_chunk(content="Hey"),
mock_llm_chunk(
@ -416,7 +481,9 @@ class TestSessionUpdates:
finish_reason="stop",
),
])
async for process in get_acp_agent_process(mock_env=mock_env):
async for process in get_acp_agent_process(
mock_env=mock_env, vibe_home=vibe_home_dir
):
session_id = await initialize_session(process)
await send_json_rpc(
@ -491,32 +558,32 @@ async def start_session_with_request_permission(
return last_response
@pytest.mark.skip(
reason="Disabled until we have a way to properly mock the fs and acp interactions"
)
class TestToolCallStructure:
@pytest.mark.asyncio
async def test_tool_call_request_permission_structure(self) -> None:
async def test_tool_call_request_permission_structure(
self, vibe_home_grep_ask: Path
) -> None:
custom_results = [
mock_llm_chunk(content="Hey"),
mock_llm_chunk(
tool_calls=[
ToolCall(
function=FunctionCall(
name="write_file",
arguments='{"path":"test.txt","content":"hello, world!"'
',"overwrite":true}',
name="grep",
arguments='{"pattern":"auth","path":".","max_matches":null,"use_default_ignore":true}',
),
type="function",
index=0,
)
],
name="write_file",
finish_reason="stop",
name="grep",
finish_reason="tool_calls",
),
]
mock_env = get_mocking_env(custom_results)
async for process in get_acp_agent_process(mock_env=mock_env):
async for process in get_acp_agent_process(
mock_env=mock_env, vibe_home=vibe_home_grep_ask
):
session_id = await initialize_session(process)
await send_json_rpc(
process,
@ -527,14 +594,15 @@ class TestToolCallStructure:
prompt=[
TextContentBlock(
type="text",
text="Create a new file named test.txt "
"with content 'hello, world!'",
text="Search for files containing the pattern 'auth'",
)
],
),
),
)
text_responses = await read_multiple_responses(process, max_count=3)
text_responses = await read_multiple_responses(
process, max_count=15, timeout_per_response=2.0
)
responses = parse_conversation(text_responses)
# Look for tool call request permission updates
@ -543,7 +611,8 @@ class TestToolCallStructure:
]
assert len(permission_requests) > 0, (
"No tool call permission requests found"
f"No tool call permission requests found. Got {len(responses)} responses: "
f"{[type(r).__name__ for r in responses]}"
)
first_request = permission_requests[0]
@ -552,32 +621,35 @@ class TestToolCallStructure:
assert first_request.params.toolCall.toolCallId is not None
@pytest.mark.asyncio
async def test_tool_call_update_approved_structure(self) -> None:
async def test_tool_call_update_approved_structure(
self, vibe_home_grep_ask: Path
) -> None:
custom_results = [
mock_llm_chunk(content="Hey"),
mock_llm_chunk(
tool_calls=[
ToolCall(
function=FunctionCall(
name="write_file",
arguments='{"path":"test.txt","content":"hello, world!"'
',"overwrite":true}',
name="grep",
arguments='{"pattern":"auth","path":".","max_matches":null,"use_default_ignore":true}',
),
type="function",
index=0,
)
],
name="write_file",
name="grep",
finish_reason="tool_calls",
),
mock_llm_chunk(
content="The file test.txt has been created", finish_reason="stop"
content="The search for 'auth' has been completed", finish_reason="stop"
),
]
mock_env = get_mocking_env(custom_results)
async for process in get_acp_agent_process(mock_env=mock_env):
async for process in get_acp_agent_process(
mock_env=mock_env, vibe_home=vibe_home_grep_ask
):
permission_request = await start_session_with_request_permission(
process, "Create a file named test.txt"
process, "Search for files containing the pattern 'auth'"
)
assert permission_request.params is not None
selected_option_id = ToolOption.ALLOW_ONCE
@ -613,34 +685,37 @@ class TestToolCallStructure:
assert approved_tool_call is not None
@pytest.mark.asyncio
async def test_tool_call_update_rejected_structure(self) -> None:
async def test_tool_call_update_rejected_structure(
self, vibe_home_grep_ask: Path
) -> None:
custom_results = [
mock_llm_chunk(content="Hey"),
mock_llm_chunk(
tool_calls=[
ToolCall(
function=FunctionCall(
name="write_file",
arguments='{"path":"test.txt","content":"hello, world!"'
',"overwrite":false}',
name="grep",
arguments='{"pattern":"auth","path":".","max_matches":null,"use_default_ignore":true}',
),
type="function",
index=0,
)
],
name="write_file",
name="grep",
finish_reason="tool_calls",
),
mock_llm_chunk(
content="The file test.txt has not been created, "
content="The search for 'auth' has not been performed, "
"because you rejected the permission request",
finish_reason="stop",
),
]
mock_env = get_mocking_env(custom_results)
async for process in get_acp_agent_process(mock_env=mock_env):
async for process in get_acp_agent_process(
mock_env=mock_env, vibe_home=vibe_home_grep_ask
):
permission_request = await start_session_with_request_permission(
process, "Create a file named test.txt"
process, "Search for files containing the pattern 'auth'"
)
assert permission_request.params is not None
selected_option_id = ToolOption.REJECT_ONCE
@ -677,28 +752,33 @@ class TestToolCallStructure:
@pytest.mark.skip(reason="Long running tool call updates are not implemented yet")
@pytest.mark.asyncio
async def test_tool_call_in_progress_update_structure(self) -> None:
async def test_tool_call_in_progress_update_structure(
self, vibe_home_grep_ask: Path
) -> None:
custom_results = [
mock_llm_chunk(content="Hey"),
mock_llm_chunk(
tool_calls=[
ToolCall(
function=FunctionCall(
name="bash",
arguments='{"command":"sleep 3","timeout":null}',
name="grep",
arguments='{"pattern":"auth","path":".","max_matches":null,"use_default_ignore":true}',
),
type="function",
index=0,
)
],
name="bash",
name="grep",
finish_reason="tool_calls",
),
mock_llm_chunk(
content="The command sleep 3 has been run", finish_reason="stop"
content="The search for 'auth' has been completed", finish_reason="stop"
),
]
mock_env = get_mocking_env(custom_results)
async for process in get_acp_agent_process(mock_env=mock_env):
async for process in get_acp_agent_process(
mock_env=mock_env, vibe_home=vibe_home_grep_ask
):
session_id = await initialize_session(process)
await send_json_rpc(
process,
@ -708,7 +788,8 @@ class TestToolCallStructure:
sessionId=session_id,
prompt=[
TextContentBlock(
type="text", text="Run sleep 3 in the current directory"
type="text",
text="Search for files containing the pattern 'auth'",
)
],
),
@ -732,34 +813,38 @@ class TestToolCallStructure:
)
@pytest.mark.asyncio
async def test_tool_call_result_update_failure_structure(self) -> None:
async def test_tool_call_result_update_failure_structure(
self, vibe_home_grep_ask: Path
) -> None:
custom_results = [
mock_llm_chunk(content="Hey"),
mock_llm_chunk(
tool_calls=[
ToolCall(
function=FunctionCall(
name="write_file",
arguments='{"path":"/test.txt","content":"hello, world!"'
',"overwrite":true}',
name="grep",
arguments='{"pattern":"auth","path":"/nonexistent","max_matches":null,"use_default_ignore":true}',
),
type="function",
index=0,
)
],
name="write_file",
name="grep",
finish_reason="tool_calls",
),
mock_llm_chunk(
content="The file /test.txt has not been created "
"because it's outside the project directory",
content="The search for 'auth' has failed "
"because the path does not exist",
finish_reason="stop",
),
]
mock_env = get_mocking_env(custom_results)
async for process in get_acp_agent_process(mock_env=mock_env):
async for process in get_acp_agent_process(
mock_env=mock_env, vibe_home=vibe_home_grep_ask
):
permission_request = await start_session_with_request_permission(
process, "Create a file named /test.txt"
process,
"Search for files containing the pattern 'auth' in /nonexistent",
)
assert permission_request.params is not None
selected_option_id = ToolOption.ALLOW_ONCE
@ -802,7 +887,9 @@ class TestCancellationStructure:
"(and not only at tool call time)"
)
@pytest.mark.asyncio
async def test_tool_call_update_cancelled_structure(self) -> None:
async def test_tool_call_update_cancelled_structure(
self, vibe_home_dir: Path
) -> None:
custom_results = [
mock_llm_chunk(content="Hey"),
mock_llm_chunk(
@ -827,7 +914,9 @@ class TestCancellationStructure:
),
]
mock_env = get_mocking_env(custom_results)
async for process in get_acp_agent_process(mock_env=mock_env):
async for process in get_acp_agent_process(
mock_env=mock_env, vibe_home=vibe_home_dir
):
permission_request = await start_session_with_request_permission(
process, "Create a file named test.txt"
)

4
tests/acp/test_initialize.py
View file

@ -41,7 +41,7 @@ class TestACPInitialize:
),
)
assert response.agentInfo == Implementation(
name="@mistralai/mistral-vibe", title="Mistral Vibe", version="1.1.2"
name="@mistralai/mistral-vibe", title="Mistral Vibe", version="1.1.3"
)
assert response.authMethods == []
@ -63,7 +63,7 @@ class TestACPInitialize:
),
)
assert response.agentInfo == Implementation(
name="@mistralai/mistral-vibe", title="Mistral Vibe", version="1.1.2"
name="@mistralai/mistral-vibe", title="Mistral Vibe", version="1.1.3"
)
assert response.authMethods is not None

2
tests/acp/test_set_model.py
View file

@ -50,6 +50,8 @@ def acp_agent(backend: FakeBackend) -> VibeAcpAgent:
],
)
VibeConfig.dump_config(config.model_dump())
class PatchedAgent(Agent):
def __init__(self, *args, **kwargs) -> None:
super().__init__(*args, **{**kwargs, "backend": backend})

54
tests/backend/test_backend.py
View file

@ -12,6 +12,8 @@ the tests will be. Always prefer real API data over manually constructed example
from __future__ import annotations
import json
import httpx
import pytest
import respx
@ -249,6 +251,58 @@ class TestBackend:
assert e.value.reason == response.reason_phrase
assert e.value.parsed_error is None
@pytest.mark.asyncio
@pytest.mark.parametrize(
"base_url,provider_name,expected_stream_options",
[
("https://api.fireworks.ai", "fireworks", {"include_usage": True}),
(
"https://api.mistral.ai",
"mistral",
{"include_usage": True, "stream_tool_calls": True},
),
],
)
async def test_backend_streaming_payload_includes_stream_options(
self, base_url: Url, provider_name: str, expected_stream_options: dict
):
with respx.mock(base_url=base_url) as mock_api:
route = mock_api.post("/v1/chat/completions").mock(
return_value=httpx.Response(
status_code=200,
stream=httpx.ByteStream(
b'data: {"choices": [{"delta": {"role": "assistant", "content": "hi"}, "finish_reason": "stop"}], "usage": {"prompt_tokens": 10, "completion_tokens": 5}}\n\ndata: [DONE]\n\n'
),
headers={"Content-Type": "text/event-stream"},
)
)
provider = ProviderConfig(
name=provider_name, api_base=f"{base_url}/v1", api_key_env_var="API_KEY"
)
backend = GenericBackend(provider=provider)
model = ModelConfig(
name="model_name", provider=provider_name, alias="model_alias"
)
messages = [LLMMessage(role=Role.user, content="hi")]
async for _ in backend.complete_streaming(
model=model,
messages=messages,
temperature=0.2,
tools=None,
max_tokens=None,
tool_choice=None,
extra_headers=None,
):
pass
assert route.called
request = route.calls.last.request
payload = json.loads(request.content)
assert payload["stream"] is True
assert payload["stream_options"] == expected_stream_options
@pytest.mark.asyncio
@pytest.mark.parametrize("backend_type", [Backend.MISTRAL, Backend.GENERIC])
async def test_backend_user_agent(self, backend_type: Backend):

228
tests/cli/test_clipboard.py Normal file
View file

@ -0,0 +1,228 @@
from __future__ import annotations
import base64
from types import SimpleNamespace
from typing import cast
from unittest.mock import MagicMock, mock_open, patch
import pytest
from textual.app import App
from vibe.cli.clipboard import _copy_osc52, copy_selection_to_clipboard
class MockWidget:
def __init__(
self,
text_selection: object | None = None,
get_selection_result: tuple[str, object] | None = None,
get_selection_raises: Exception | None = None,
) -> None:
self.text_selection = text_selection
self._get_selection_result = get_selection_result
self._get_selection_raises = get_selection_raises
def get_selection(self, selection: object) -> tuple[str, object]:
if self._get_selection_raises:
raise self._get_selection_raises
if self._get_selection_result is None:
return ("", None)
return self._get_selection_result
@pytest.fixture
def mock_app() -> App:
app = MagicMock(spec=App)
app.query = MagicMock(return_value=[])
app.notify = MagicMock()
app.copy_to_clipboard = MagicMock()
return cast(App, app)
@pytest.mark.parametrize(
"widgets,description",
[
([], "no widgets"),
([MockWidget(text_selection=None)], "no selection"),
([MockWidget()], "widget without text_selection attr"),
(
[
MockWidget(
text_selection=SimpleNamespace(),
get_selection_raises=ValueError("Error getting selection"),
)
],
"get_selection raises",
),
(
[MockWidget(text_selection=SimpleNamespace(), get_selection_result=None)],
"empty result",
),
(
[
MockWidget(
text_selection=SimpleNamespace(), get_selection_result=(" ", None)
)
],
"empty text",
),
],
)
def test_copy_selection_to_clipboard_no_notification(
mock_app: MagicMock, widgets: list[MockWidget], description: str
) -> None:
if description == "widget without text_selection attr":
del widgets[0].text_selection
mock_app.query.return_value = widgets
copy_selection_to_clipboard(mock_app)
mock_app.notify.assert_not_called()
@patch("vibe.cli.clipboard._copy_osc52")
@patch("vibe.cli.clipboard.pyperclip.copy")
def test_copy_selection_to_clipboard_success_with_osc52(
mock_pyperclip_copy: MagicMock, mock_osc52_copy: MagicMock, mock_app: MagicMock
) -> None:
widget = MockWidget(
text_selection=SimpleNamespace(), get_selection_result=("selected text", None)
)
mock_app.query.return_value = [widget]
copy_selection_to_clipboard(mock_app)
mock_osc52_copy.assert_called_once_with("selected text")
mock_pyperclip_copy.assert_not_called()
mock_app.copy_to_clipboard.assert_not_called()
mock_app.notify.assert_called_once_with(
'"selected text" copied to clipboard', severity="information", timeout=2
)
@patch("vibe.cli.clipboard._copy_osc52")
@patch("vibe.cli.clipboard.pyperclip.copy")
def test_copy_selection_to_clipboard_osc52_fails_success_with_pyperclip(
mock_pyperclip_copy: MagicMock, mock_osc52_copy: MagicMock, mock_app: MagicMock
) -> None:
widget = MockWidget(
text_selection=SimpleNamespace(),
get_selection_result=(" selected text ", None),
)
mock_app.query.return_value = [widget]
mock_osc52_copy.side_effect = Exception("osc52 failed")
copy_selection_to_clipboard(mock_app)
mock_osc52_copy.assert_called_once_with(" selected text ")
mock_pyperclip_copy.assert_called_once_with(" selected text ")
mock_app.notify.assert_called_once_with(
'" selected text " copied to clipboard', severity="information", timeout=2
)
mock_app.copy_to_clipboard.assert_not_called()
@patch("vibe.cli.clipboard._copy_osc52")
@patch("vibe.cli.clipboard.pyperclip.copy")
def test_copy_selection_to_clipboard_osc52_and_pyperclip_fail_success_with_app_copy(
mock_pyperclip_copy: MagicMock, mock_osc52_copy: MagicMock, mock_app: MagicMock
) -> None:
widget = MockWidget(
text_selection=SimpleNamespace(), get_selection_result=("selected text", None)
)
mock_app.query.return_value = [widget]
mock_osc52_copy.side_effect = Exception("osc52 failed")
mock_pyperclip_copy.side_effect = Exception("pyperclip failed")
copy_selection_to_clipboard(mock_app)
mock_osc52_copy.assert_called_once_with("selected text")
mock_pyperclip_copy.assert_called_once_with("selected text")
mock_app.copy_to_clipboard.assert_called_once_with("selected text")
mock_app.notify.assert_called_once_with(
'"selected text" copied to clipboard', severity="information", timeout=2
)
@patch("vibe.cli.clipboard._copy_osc52")
@patch("vibe.cli.clipboard.pyperclip.copy")
def test_copy_selection_to_clipboard_all_methods_fail(
mock_pyperclip_copy: MagicMock, mock_osc52_copy: MagicMock, mock_app: MagicMock
) -> None:
widget = MockWidget(
text_selection=SimpleNamespace(), get_selection_result=("selected text", None)
)
mock_app.query.return_value = [widget]
mock_osc52_copy.side_effect = Exception("osc52 failed")
mock_pyperclip_copy.side_effect = Exception("pyperclip failed")
mock_app.copy_to_clipboard.side_effect = Exception("app copy failed")
copy_selection_to_clipboard(mock_app)
mock_osc52_copy.assert_called_once_with("selected text")
mock_pyperclip_copy.assert_called_once_with("selected text")
mock_app.copy_to_clipboard.assert_called_once_with("selected text")
mock_app.notify.assert_called_once_with(
"Failed to copy - no clipboard method available", severity="warning", timeout=3
)
def test_copy_selection_to_clipboard_multiple_widgets(mock_app: MagicMock) -> None:
widget1 = MockWidget(
text_selection=SimpleNamespace(), get_selection_result=("first selection", None)
)
widget2 = MockWidget(
text_selection=SimpleNamespace(),
get_selection_result=("second selection", None),
)
widget3 = MockWidget(text_selection=None)
mock_app.query.return_value = [widget1, widget2, widget3]
with patch("vibe.cli.clipboard._copy_osc52") as mock_osc52_copy:
copy_selection_to_clipboard(mock_app)
mock_osc52_copy.assert_called_once_with("first selection\nsecond selection")
mock_app.notify.assert_called_once_with(
'"first selection⏎second selection" copied to clipboard',
severity="information",
timeout=2,
)
def test_copy_selection_to_clipboard_preview_shortening(mock_app: MagicMock) -> None:
long_text = "a" * 100
widget = MockWidget(
text_selection=SimpleNamespace(), get_selection_result=(long_text, None)
)
mock_app.query.return_value = [widget]
with (
patch("vibe.cli.clipboard._copy_osc52") as mock_osc52_copy,
patch("vibe.cli.clipboard.pyperclip.copy") as mock_pyperclip_copy,
):
mock_osc52_copy.side_effect = Exception("osc52 failed")
copy_selection_to_clipboard(mock_app)
mock_osc52_copy.assert_called_once_with(long_text)
mock_pyperclip_copy.assert_called_once_with(long_text)
notification_call = mock_app.notify.call_args
assert notification_call is not None
assert '"' in notification_call[0][0]
assert "copied to clipboard" in notification_call[0][0]
assert len(notification_call[0][0]) < len(long_text) + 30
@patch("builtins.open", new_callable=mock_open)
def test_copy_osc52_writes_correct_sequence(
mock_file: MagicMock, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.delenv("TMUX", raising=False)
test_text = "héllo wörld 🎉"
_copy_osc52(test_text)
encoded = base64.b64encode(test_text.encode("utf-8")).decode("ascii")
expected_seq = f"\033]52;c;{encoded}\a"
mock_file.assert_called_once_with("/dev/tty", "w")
handle = mock_file()
handle.write.assert_called_once_with(expected_seq)
handle.flush.assert_called_once()

90
tests/conftest.py
View file

@ -1,74 +1,48 @@
from __future__ import annotations
from pathlib import Path
import sys
from typing import Any
from pydantic.fields import FieldInfo
from pydantic_settings import BaseSettings, PydanticBaseSettingsSource
import pytest
import tomli_w
_in_mem_config: dict[str, Any] = {}
from vibe.core import config_path
class InMemSettingsSource(PydanticBaseSettingsSource):
def __init__(self, settings_cls: type[BaseSettings]) -> None:
super().__init__(settings_cls)
def get_field_value(
self, field: FieldInfo, field_name: str
) -> tuple[Any, str, bool]:
return _in_mem_config.get(field_name), field_name, False
def __call__(self) -> dict[str, Any]:
return _in_mem_config
@pytest.fixture(autouse=True, scope="session")
def _patch_vibe_config() -> None:
"""Patch VibeConfig.settings_customise_sources to only use init_settings in tests.
This ensures that even production code that creates VibeConfig instances
will only use init_settings and ignore environment variables and config files.
Runs once per test session before any tests execute.
"""
from vibe.core.config import VibeConfig
def patched_settings_customise_sources(
cls,
settings_cls: type[BaseSettings],
init_settings: PydanticBaseSettingsSource,
env_settings: PydanticBaseSettingsSource,
dotenv_settings: PydanticBaseSettingsSource,
file_secret_settings: PydanticBaseSettingsSource,
) -> tuple[PydanticBaseSettingsSource, ...]:
return (init_settings, InMemSettingsSource(settings_cls))
VibeConfig.settings_customise_sources = classmethod(
patched_settings_customise_sources
) # type: ignore[assignment]
def dump_config(cls, config: dict[str, Any]) -> None:
global _in_mem_config
_in_mem_config = config
VibeConfig.dump_config = classmethod(dump_config) # type: ignore[assignment]
def patched_load(cls, agent: str | None = None, **overrides: Any) -> Any:
return cls(**overrides)
VibeConfig.load = classmethod(patched_load) # type: ignore[assignment]
def get_base_config() -> dict[str, Any]:
return {
"active_model": "devstral-latest",
"providers": [
{
"name": "mistral",
"api_base": "https://api.mistral.ai/v1",
"api_key_env_var": "MISTRAL_API_KEY",
"backend": "mistral",
}
],
"models": [
{
"name": "mistral-vibe-cli-latest",
"provider": "mistral",
"alias": "devstral-latest",
}
],
}
@pytest.fixture(autouse=True)
def _reset_in_mem_config() -> None:
"""Reset in-memory config before each test to prevent test isolation issues.
def config_dir(
monkeypatch: pytest.MonkeyPatch, tmp_path_factory: pytest.TempPathFactory
) -> Path:
tmp_path = tmp_path_factory.mktemp("vibe")
config_dir = tmp_path / ".vibe"
config_dir.mkdir(parents=True, exist_ok=True)
config_file = config_dir / "config.toml"
config_file.write_text(tomli_w.dumps(get_base_config()), encoding="utf-8")
This ensures that each test starts with a clean configuration state,
preventing race conditions and test interference when tests run in parallel
or when VibeConfig.save_updates() modifies the shared _in_mem_config dict.
"""
global _in_mem_config
_in_mem_config = {}
monkeypatch.setattr(config_path, "_DEFAULT_VIBE_HOME", config_dir)
return config_dir
@pytest.fixture(autouse=True)

41
tests/core/test_config_resolution.py Normal file
View file

@ -0,0 +1,41 @@
from __future__ import annotations
from pathlib import Path
import pytest
from vibe.core.config_path import CONFIG_FILE, GLOBAL_CONFIG_FILE, VIBE_HOME
class TestResolveConfigFile:
def test_resolves_local_config_when_exists(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Test that local .vibe/config.toml is found when it exists."""
monkeypatch.chdir(tmp_path)
local_config_dir = tmp_path / ".vibe"
local_config_dir.mkdir()
local_config = local_config_dir / "config.toml"
local_config.write_text('active_model = "test"', encoding="utf-8")
assert CONFIG_FILE.path == local_config
assert CONFIG_FILE.path.is_file()
assert CONFIG_FILE.path.read_text(encoding="utf-8") == 'active_model = "test"'
def test_falls_back_to_global_config_when_local_missing(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Test that global config is returned when local config doesn't exist."""
monkeypatch.chdir(tmp_path)
# Ensure no local config exists
assert not (tmp_path / ".vibe" / "config.toml").exists()
assert CONFIG_FILE.path == GLOBAL_CONFIG_FILE.path
def test_respects_vibe_home_env_var(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Test that VIBE_HOME environment variable affects VIBE_HOME.path."""
assert VIBE_HOME.path != tmp_path
monkeypatch.setenv("VIBE_HOME", str(tmp_path))
assert VIBE_HOME.path == tmp_path

9
tests/onboarding/test_run_onboarding.py
View file

@ -20,12 +20,6 @@ class StubApp(App[str | None]):
return self._return_value
def _patch_env_file(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Path:
env_file = tmp_path / ".env"
monkeypatch.setattr(onboarding, "GLOBAL_ENV_FILE", env_file, raising=False)
return env_file
def _exit_raiser(code: int = 0) -> None:
raise SystemExit(code)
@ -33,7 +27,6 @@ def _exit_raiser(code: int = 0) -> None:
def test_exits_on_cancel(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path
) -> None:
_patch_env_file(monkeypatch, tmp_path)
monkeypatch.setattr(sys, "exit", _exit_raiser)
with pytest.raises(SystemExit) as excinfo:
@ -47,7 +40,6 @@ def test_exits_on_cancel(
def test_warns_on_save_error(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path
) -> None:
_patch_env_file(monkeypatch, tmp_path)
monkeypatch.setattr(sys, "exit", _exit_raiser)
onboarding.run_onboarding(StubApp("save_error:disk full"))
@ -60,7 +52,6 @@ def test_warns_on_save_error(
def test_successfully_completes(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path
) -> None:
_patch_env_file(monkeypatch, tmp_path)
monkeypatch.setattr(sys, "exit", _exit_raiser)
onboarding.run_onboarding(StubApp("completed"))

57
tests/onboarding/test_ui_onboarding.py
View file

@ -2,7 +2,7 @@ from __future__ import annotations
from collections.abc import Callable
from pathlib import Path
from typing import Any
import tomllib
import pytest
from textual.events import Resize
@ -10,9 +10,8 @@ from textual.geometry import Size
from textual.pilot import Pilot
from textual.widgets import Input
from vibe.core import config as core_config
from vibe.core.config_path import GLOBAL_CONFIG_FILE, GLOBAL_ENV_FILE
from vibe.setup.onboarding import OnboardingApp
import vibe.setup.onboarding.screens.api_key as api_key_module
from vibe.setup.onboarding.screens.api_key import ApiKeyScreen
from vibe.setup.onboarding.screens.theme_selection import THEMES, ThemeSelectionScreen
@ -31,32 +30,6 @@ async def _wait_for(
raise AssertionError(msg)
@pytest.fixture()
def onboarding_app(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> tuple[OnboardingApp, Path, dict[str, Any]]:
vibe_home = tmp_path / ".vibe"
env_file = vibe_home / ".env"
saved_updates: dict[str, Any] = {}
def record_updates(updates: dict[str, Any]) -> None:
saved_updates.update(updates)
monkeypatch.setenv("VIBE_HOME", str(vibe_home))
for module in (core_config, api_key_module):
monkeypatch.setattr(module, "GLOBAL_CONFIG_DIR", vibe_home, raising=False)
monkeypatch.setattr(module, "GLOBAL_ENV_FILE", env_file, raising=False)
monkeypatch.setattr(
core_config.VibeConfig,
"save_updates",
classmethod(lambda cls, updates: record_updates(updates)),
)
return OnboardingApp(), env_file, saved_updates
async def pass_welcome_screen(pilot: Pilot) -> None:
welcome_screen = pilot.app.get_screen("welcome")
await _wait_for(
@ -67,10 +40,8 @@ async def pass_welcome_screen(pilot: Pilot) -> None:
@pytest.mark.asyncio
async def test_ui_gets_through_the_onboarding_successfully(
onboarding_app: tuple[OnboardingApp, Path, dict[str, Any]],
) -> None:
app, env_file, config_updates = onboarding_app
async def test_ui_gets_through_the_onboarding_successfully() -> None:
app = OnboardingApp()
api_key_value = "sk-onboarding-test-key"
async with app.run_test() as pilot:
@ -88,19 +59,20 @@ async def test_ui_gets_through_the_onboarding_successfully(
assert app.return_value == "completed"
assert env_file.is_file()
env_contents = env_file.read_text(encoding="utf-8")
assert GLOBAL_ENV_FILE.path.is_file()
env_contents = GLOBAL_ENV_FILE.path.read_text(encoding="utf-8")
assert "MISTRAL_API_KEY" in env_contents
assert api_key_value in env_contents
assert config_updates.get("textual_theme") == app.theme
assert GLOBAL_CONFIG_FILE.path.is_file()
config_contents = GLOBAL_CONFIG_FILE.path.read_text(encoding="utf-8")
config_dict = tomllib.loads(config_contents)
assert config_dict.get("textual_theme") == app.theme
@pytest.mark.asyncio
async def test_ui_can_pick_a_theme_and_saves_selection(
onboarding_app: tuple[OnboardingApp, Path, dict[str, Any]],
) -> None:
app, _, config_updates = onboarding_app
async def test_ui_can_pick_a_theme_and_saves_selection(config_dir: Path) -> None:
app = OnboardingApp()
async with app.run_test() as pilot:
await pass_welcome_screen(pilot)
@ -121,4 +93,7 @@ async def test_ui_can_pick_a_theme_and_saves_selection(
await pilot.press("enter")
await _wait_for(lambda: isinstance(app.screen, ApiKeyScreen), pilot)
assert config_updates.get("textual_theme") == target_theme
assert GLOBAL_CONFIG_FILE.path.is_file()
config_contents = GLOBAL_CONFIG_FILE.path.read_text(encoding="utf-8")
config_dict = tomllib.loads(config_contents)
assert config_dict.get("textual_theme") == target_theme

10
tests/snapshots/test_ui_snapshot_release_update_notification.py
View file

@ -4,7 +4,13 @@ from textual.pilot import Pilot
from tests.snapshots.base_snapshot_test_app import BaseSnapshotTestApp, default_config
from tests.snapshots.snap_compare import SnapCompare
from vibe.cli.update_notifier import FakeVersionUpdateGateway, VersionUpdate
from tests.update_notifier.adapters.fake_update_cache_repository import (
FakeUpdateCacheRepository,
)
from tests.update_notifier.adapters.fake_version_update_gateway import (
FakeVersionUpdateGateway,
)
from vibe.cli.update_notifier import VersionUpdate
class SnapshotTestAppWithUpdate(BaseSnapshotTestApp):
@ -14,9 +20,11 @@ class SnapshotTestAppWithUpdate(BaseSnapshotTestApp):
version_update_notifier = FakeVersionUpdateGateway(
update=VersionUpdate(latest_version="1000.2.0")
)
update_cache_repository = FakeUpdateCacheRepository()
super().__init__(
config=config,
version_update_notifier=version_update_notifier,
update_cache_repository=update_cache_repository,
current_version="1.0.4",
)

24
tests/test_agent_observer_streaming.py
View file

@ -21,6 +21,7 @@ from vibe.core.middleware import (
from vibe.core.tools.base import BaseToolConfig, ToolPermission
from vibe.core.tools.builtins.todo import TodoArgs
from vibe.core.types import (
ApprovalResponse,
AssistantEvent,
FunctionCall,
LLMChunk,
@ -30,11 +31,7 @@ from vibe.core.types import (
ToolCallEvent,
ToolResultEvent,
)
from vibe.core.utils import (
ApprovalResponse,
CancellationReason,
get_user_cancellation_message,
)
from vibe.core.utils import CancellationReason, get_user_cancellation_message
class InjectBeforeMiddleware:
@ -147,23 +144,6 @@ async def test_act_emits_user_and_assistant_msgs(observer_capture) -> None:
assert observed[2][1] == "Pong!"
@pytest.mark.asyncio
async def test_act_yields_assistant_event_with_usage_stats() -> None:
backend = FakeBackend([mock_llm_chunk(content="Pong!")])
agent = Agent(make_config(), backend=backend)
events = [ev async for ev in agent.act("Ping?")]
assert len(events) == 1
ev = events[-1]
assert isinstance(ev, AssistantEvent)
assert ev.content == "Pong!"
# stats come from tests.mock.utils.mock_llm_result (prompt=10, completion=5)
assert ev.prompt_tokens == 10
assert ev.completion_tokens == 5
assert ev.session_total_tokens == 15
@pytest.mark.asyncio
async def test_act_streams_batched_chunks_in_order() -> None:
backend = FakeBackend([

25
tests/test_agent_tool_call.py
View file

@ -14,6 +14,7 @@ from vibe.core.config import SessionLoggingConfig, VibeConfig
from vibe.core.tools.base import BaseToolConfig, ToolPermission
from vibe.core.tools.builtins.todo import TodoItem
from vibe.core.types import (
ApprovalResponse,
AssistantEvent,
BaseEvent,
FunctionCall,
@ -24,7 +25,6 @@ from vibe.core.types import (
ToolCallEvent,
ToolResultEvent,
)
from vibe.core.utils import ApprovalResponse
async def act_and_collect_events(agent: Agent, prompt: str) -> list[BaseEvent]:
@ -140,7 +140,7 @@ async def test_tool_call_requires_approval_if_not_auto_approved() -> None:
async def test_tool_call_approved_by_callback() -> None:
def approval_callback(
_tool_name: str, _args: dict[str, Any], _tool_call_id: str
) -> tuple[str, str | None]:
) -> tuple[ApprovalResponse, str | None]:
return (ApprovalResponse.YES, None)
agent = make_agent(
@ -175,7 +175,7 @@ async def test_tool_call_rejected_when_auto_approve_disabled_and_rejected_by_cal
def approval_callback(
_tool_name: str, _args: dict[str, Any], _tool_call_id: str
) -> tuple[str, str | None]:
) -> tuple[ApprovalResponse, str | None]:
return (ApprovalResponse.NO, custom_feedback)
agent = make_agent(
@ -237,14 +237,20 @@ async def test_tool_call_skipped_when_permission_is_never() -> None:
@pytest.mark.asyncio
async def test_approval_always_flips_auto_approve_for_subsequent_calls() -> None:
async def test_approval_always_sets_tool_permission_for_subsequent_calls() -> None:
callback_invocations = []
agent_ref: Agent | None = None
def approval_callback(
tool_name: str, _args: dict[str, Any], _tool_call_id: str
) -> tuple[str, str | None]:
) -> tuple[ApprovalResponse, str | None]:
callback_invocations.append(tool_name)
return (ApprovalResponse.ALWAYS, None)
# Set permission to ALWAYS for this tool (simulating the new behavior)
assert agent_ref is not None
if tool_name not in agent_ref.config.tools:
agent_ref.config.tools[tool_name] = BaseToolConfig()
agent_ref.config.tools[tool_name].permission = ToolPermission.ALWAYS
return (ApprovalResponse.YES, None)
agent = make_agent(
auto_approve=False,
@ -261,11 +267,16 @@ async def test_approval_always_flips_auto_approve_for_subsequent_calls() -> None
mock_llm_chunk(content="Second done.", finish_reason="stop"),
]),
)
agent_ref = agent
events1 = await act_and_collect_events(agent, "First request")
events2 = await act_and_collect_events(agent, "Second request")
assert agent.auto_approve is True
tool_config_todo = agent.tool_manager.get_tool_config("todo")
assert tool_config_todo.permission is ToolPermission.ALWAYS
tool_config_help = agent.tool_manager.get_tool_config("bash")
assert tool_config_help.permission is not ToolPermission.ALWAYS
assert agent.auto_approve is False
assert len(callback_invocations) == 1
assert callback_invocations[0] == "todo"
assert isinstance(events1[2], ToolResultEvent)

17
tests/update_notifier/adapters/fake_update_cache_repository.py Normal file
View file

@ -0,0 +1,17 @@
from __future__ import annotations
from vibe.cli.update_notifier.ports.update_cache_repository import (
UpdateCache,
UpdateCacheRepository,
)
class FakeUpdateCacheRepository(UpdateCacheRepository):
def __init__(self, update_cache: UpdateCache | None = None) -> None:
self.update_cache: UpdateCache | None = update_cache
async def get(self) -> UpdateCache | None:
return self.update_cache
async def set(self, update_cache: UpdateCache) -> None:
self.update_cache = update_cache

24
tests/update_notifier/adapters/fake_version_update_gateway.py Normal file
View file

@ -0,0 +1,24 @@
from __future__ import annotations
from vibe.cli.update_notifier.ports.version_update_gateway import (
VersionUpdate,
VersionUpdateGateway,
VersionUpdateGatewayError,
)
class FakeVersionUpdateGateway(VersionUpdateGateway):
def __init__(
self,
update: VersionUpdate | None = None,
error: VersionUpdateGatewayError | None = None,
) -> None:
self._update: VersionUpdate | None = update
self._error = error
self.fetch_update_calls = 0
async def fetch_update(self) -> VersionUpdate | None:
self.fetch_update_calls += 1
if self._error is not None:
raise self._error
return self._update

78
tests/update_notifier/test_filesystem_update_cache_repository.py Normal file
View file

@ -0,0 +1,78 @@
from __future__ import annotations
import json
from pathlib import Path
import pytest
from vibe.cli.update_notifier.adapters.filesystem_update_cache_repository import (
FileSystemUpdateCacheRepository,
)
from vibe.cli.update_notifier.ports.update_cache_repository import UpdateCache
@pytest.mark.asyncio
async def test_reads_cache_from_file_when_present(tmp_path: Path) -> None:
cache_file = tmp_path / "update_cache.json"
cache_file.write_text(
json.dumps({"latest_version": "1.2.3", "stored_at_timestamp": 1_700_000_000})
)
repository = FileSystemUpdateCacheRepository(base_path=tmp_path)
cache = await repository.get()
assert cache is not None
assert cache.latest_version == "1.2.3"
assert cache.stored_at_timestamp == 1_700_000_000
@pytest.mark.asyncio
async def test_returns_none_when_cache_file_is_missing(tmp_path: Path) -> None:
repository = FileSystemUpdateCacheRepository(base_path=tmp_path)
cache = await repository.get()
assert cache is None
@pytest.mark.asyncio
async def test_returns_none_when_cache_file_is_corrupted(tmp_path: Path) -> None:
cache_dir = tmp_path / ".vibe"
cache_dir.mkdir()
(cache_dir / "update_cache.json").write_text("{not-json")
repository = FileSystemUpdateCacheRepository(base_path=tmp_path)
cache = await repository.get()
assert cache is None
@pytest.mark.asyncio
async def test_overwrites_existing_cache(tmp_path: Path) -> None:
cache_file = tmp_path / "update_cache.json"
cache_file.write_text(
json.dumps({"latest_version": "1.0.0", "stored_at_timestamp": 1_600_000_000})
)
repository = FileSystemUpdateCacheRepository(base_path=tmp_path)
await repository.set(
UpdateCache(latest_version="1.1.0", stored_at_timestamp=1_700_200_000)
)
content = json.loads(cache_file.read_text())
assert content["latest_version"] == "1.1.0"
assert content["stored_at_timestamp"] == 1_700_200_000
@pytest.mark.asyncio
async def test_silently_ignores_errors_when_writing_cache_fails(tmp_path: Path) -> None:
cache_dir = tmp_path / ".vibe"
cache_dir.mkdir()
(cache_dir / "update_cache.json").mkdir()
repository = FileSystemUpdateCacheRepository(base_path=tmp_path)
await repository.set(
UpdateCache(latest_version="1.2.0", stored_at_timestamp=1_700_300_000)
)
assert (cache_dir / "update_cache.json").is_dir()

4
tests/update_notifier/test_github_version_update_gateway.py
View file

@ -5,10 +5,8 @@ from collections.abc import Callable
import httpx
import pytest
from vibe.cli.update_notifier.github_version_update_gateway import (
from vibe.cli.update_notifier import (
GitHubVersionUpdateGateway,
)
from vibe.cli.update_notifier.version_update_gateway import (
VersionUpdateGatewayCause,
VersionUpdateGatewayError,
)

157
tests/update_notifier/test_pypi_version_update_gateway.py Normal file
View file

@ -0,0 +1,157 @@
from __future__ import annotations
from collections.abc import Callable
import httpx
import pytest
from vibe.cli.update_notifier.adapters.pypi_version_update_gateway import (
PyPIVersionUpdateGateway,
)
from vibe.cli.update_notifier.ports.version_update_gateway import (
VersionUpdate,
VersionUpdateGatewayCause,
VersionUpdateGatewayError,
)
Handler = Callable[[httpx.Request], httpx.Response]
PYPI_API_URL = "https://pypi.org"
@pytest.mark.asyncio
async def test_retrieves_nothing_when_no_versions_are_available() -> None:
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(
status_code=httpx.codes.OK, json={"versions": [], "files": []}
)
transport = httpx.MockTransport(handler)
async with httpx.AsyncClient(transport=transport, base_url=PYPI_API_URL) as client:
gateway = PyPIVersionUpdateGateway(project_name="mistral-vibe", client=client)
update = await gateway.fetch_update()
assert update is None
@pytest.mark.asyncio
async def test_retrieves_the_latest_non_yanked_version() -> None:
def handler(request: httpx.Request) -> httpx.Response:
assert request.headers["Accept"] == "application/vnd.pypi.simple.v1+json"
assert request.url.path == "/simple/mistral-vibe/"
return httpx.Response(
status_code=httpx.codes.OK,
json={
"versions": ["1.0.0", "1.0.1", "1.0.2"],
"files": [
{
"filename": "mistral_vibe-1.0.0-py3-none-any.whl",
"yanked": False,
},
{"filename": "mistral_vibe-1.0.1-py3-none-any.whl", "yanked": True},
{
"filename": "mistral_vibe-1.0.2-py3-none-any.whl",
"yanked": False,
},
],
},
)
transport = httpx.MockTransport(handler)
async with httpx.AsyncClient(transport=transport, base_url=PYPI_API_URL) as client:
gateway = PyPIVersionUpdateGateway(project_name="mistral-vibe", client=client)
update = await gateway.fetch_update()
assert update == VersionUpdate(latest_version="1.0.2")
@pytest.mark.asyncio
async def test_retrieves_nothing_when_only_yanked_versions_are_available() -> None:
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(
status_code=httpx.codes.OK,
json={
"versions": ["1.0.0"],
"files": [
{"filename": "mistral_vibe-1.0.0-py3-none-any.whl", "yanked": True}
],
},
)
transport = httpx.MockTransport(handler)
async with httpx.AsyncClient(transport=transport, base_url=PYPI_API_URL) as client:
gateway = PyPIVersionUpdateGateway(project_name="mistral-vibe", client=client)
update = await gateway.fetch_update()
assert update is None
@pytest.mark.asyncio
async def test_does_not_match_versions_by_substring() -> None:
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(
status_code=httpx.codes.OK,
json={
"versions": ["1.0.1"],
"files": [
{
"filename": "mistral_vibe-1.0.10-py3-none-any.whl",
"yanked": False,
}
],
},
)
transport = httpx.MockTransport(handler)
async with httpx.AsyncClient(transport=transport, base_url=PYPI_API_URL) as client:
gateway = PyPIVersionUpdateGateway(project_name="mistral-vibe", client=client)
update = await gateway.fetch_update()
assert update is None
def _raise_connect_timeout(request: httpx.Request) -> httpx.Response:
raise httpx.ConnectTimeout("boom", request=request)
@pytest.mark.parametrize(
("handler", "expected_cause", "expected_message"),
[
(
lambda _: httpx.Response(status_code=httpx.codes.NOT_FOUND),
VersionUpdateGatewayCause.NOT_FOUND,
None,
),
(
lambda _: httpx.Response(status_code=httpx.codes.FORBIDDEN),
VersionUpdateGatewayCause.FORBIDDEN,
None,
),
(
lambda _: httpx.Response(status_code=httpx.codes.INTERNAL_SERVER_ERROR),
VersionUpdateGatewayCause.ERROR_RESPONSE,
None,
),
(
lambda _: httpx.Response(status_code=httpx.codes.OK, content=b"{not-json"),
VersionUpdateGatewayCause.INVALID_RESPONSE,
None,
),
(_raise_connect_timeout, VersionUpdateGatewayCause.REQUEST_FAILED, None),
],
)
@pytest.mark.asyncio
async def test_retrieves_nothing_when_fetching_update_fails(
handler: Callable[[httpx.Request], httpx.Response],
expected_cause: VersionUpdateGatewayCause,
expected_message: str | None,
) -> None:
transport = httpx.MockTransport(handler)
async with httpx.AsyncClient(transport=transport, base_url=PYPI_API_URL) as client:
gateway = PyPIVersionUpdateGateway(project_name="mistral-vibe", client=client)
with pytest.raises(VersionUpdateGatewayError) as excinfo:
await gateway.fetch_update()
assert excinfo.value.cause == expected_cause
if expected_message is not None:
assert str(excinfo.value) == expected_message

66
tests/update_notifier/test_ui_version_update_notification.py
View file

@ -1,16 +1,21 @@
from __future__ import annotations
import asyncio
import time
from typing import Protocol
import pytest
from textual.app import Notification
from vibe.cli.textual_ui.app import VibeApp
from vibe.cli.update_notifier.fake_version_update_gateway import (
from tests.update_notifier.adapters.fake_update_cache_repository import (
FakeUpdateCacheRepository,
)
from tests.update_notifier.adapters.fake_version_update_gateway import (
FakeVersionUpdateGateway,
)
from vibe.cli.update_notifier.version_update_gateway import (
from vibe.cli.textual_ui.app import VibeApp
from vibe.cli.update_notifier import (
UpdateCache,
VersionUpdate,
VersionUpdateGatewayCause,
VersionUpdateGatewayError,
@ -59,6 +64,7 @@ class VibeAppFactory(Protocol):
self,
*,
notifier: FakeVersionUpdateGateway,
update_cache_repository: FakeUpdateCacheRepository | None = None,
config: VibeConfig | None = None,
auto_approve: bool = False,
current_version: str = "0.1.0",
@ -67,9 +73,13 @@ class VibeAppFactory(Protocol):
@pytest.fixture
def make_vibe_app(vibe_config_with_update_checks_enabled: VibeConfig) -> VibeAppFactory:
update_cache_repository = FakeUpdateCacheRepository()
def _make_app(
*,
notifier: FakeVersionUpdateGateway,
update_cache_repository: FakeUpdateCacheRepository
| None = update_cache_repository,
config: VibeConfig | None = None,
auto_approve: bool = False,
current_version: str = "0.1.0",
@ -78,6 +88,7 @@ def make_vibe_app(vibe_config_with_update_checks_enabled: VibeConfig) -> VibeApp
config=config or vibe_config_with_update_checks_enabled,
auto_approve=auto_approve,
version_update_notifier=notifier,
update_cache_repository=update_cache_repository,
current_version=current_version,
)
@ -159,3 +170,52 @@ async def test_ui_does_not_invoke_gateway_nor_show_update_notification_when_upda
await _assert_no_notifications(app, pilot, timeout=0.3)
assert notifier.fetch_update_calls == 0
@pytest.mark.asyncio
async def test_ui_does_not_show_toast_when_update_is_known_in_recent_cache_already(
make_vibe_app: VibeAppFactory,
) -> None:
timestamp_two_hours_ago = int(time.time()) - 2 * 60 * 60
notifier = FakeVersionUpdateGateway(update=VersionUpdate(latest_version="0.2.0"))
update_cache = UpdateCache(
latest_version="0.2.0", stored_at_timestamp=timestamp_two_hours_ago
)
update_cache_repository = FakeUpdateCacheRepository(update_cache=update_cache)
app = make_vibe_app(
notifier=notifier, update_cache_repository=update_cache_repository
)
async with app.run_test() as pilot:
await _assert_no_notifications(app, pilot, timeout=0.3)
assert notifier.fetch_update_calls == 0
@pytest.mark.asyncio
async def test_ui_does_show_toast_when_cache_entry_is_too_old(
make_vibe_app: VibeAppFactory,
) -> None:
timestamp_two_days_ago = int(time.time()) - 2 * 24 * 60 * 60
notifier = FakeVersionUpdateGateway(update=VersionUpdate(latest_version="0.2.0"))
update_cache = UpdateCache(
latest_version="0.2.0", stored_at_timestamp=timestamp_two_days_ago
)
update_cache_repository = FakeUpdateCacheRepository(update_cache=update_cache)
app = make_vibe_app(
notifier=notifier, update_cache_repository=update_cache_repository
)
async with app.run_test() as pilot:
await pilot.pause(0.3)
notifications = list(app._notifications)
assert notifications
notification = notifications[-1]
assert notification.severity == "information"
assert notification.title == "Update available"
assert (
notification.message
== '0.1.0 => 0.2.0\nRun "uv tool upgrade mistral-vibe" to update'
)
assert notifier.fetch_update_calls == 1

225
tests/update_notifier/test_version_update_use_case.py
View file

@ -2,18 +2,27 @@ from __future__ import annotations
import pytest
from vibe.cli.update_notifier.fake_version_update_gateway import (
from tests.update_notifier.adapters.fake_update_cache_repository import (
FakeUpdateCacheRepository,
)
from tests.update_notifier.adapters.fake_version_update_gateway import (
FakeVersionUpdateGateway,
)
from vibe.cli.update_notifier.version_update import (
VersionUpdateError,
is_version_update_available,
)
from vibe.cli.update_notifier.version_update_gateway import (
from vibe.cli.update_notifier import (
UpdateCache,
VersionUpdate,
VersionUpdateGatewayCause,
VersionUpdateGatewayError,
)
from vibe.cli.update_notifier.version_update import (
VersionUpdateError,
get_update_if_available,
)
@pytest.fixture
def current_timestamp() -> int:
return 1765278683
@pytest.mark.asyncio
@ -23,8 +32,10 @@ async def test_retrieves_the_latest_version_update_when_available() -> None:
update=VersionUpdate(latest_version=latest_update)
)
update = await is_version_update_available(
version_update_notifier, current_version="1.0.0"
update = await get_update_if_available(
version_update_notifier,
current_version="1.0.0",
update_cache_repository=FakeUpdateCacheRepository(),
)
assert update is not None
@ -39,8 +50,10 @@ async def test_retrieves_nothing_when_the_current_version_is_the_latest() -> Non
update=VersionUpdate(latest_version=latest_version)
)
update = await is_version_update_available(
version_update_notifier, current_version=current_version
update = await get_update_if_available(
version_update_notifier,
current_version=current_version,
update_cache_repository=FakeUpdateCacheRepository(),
)
assert update is None
@ -56,8 +69,10 @@ async def test_retrieves_nothing_when_the_current_version_is_greater_than_the_la
update=VersionUpdate(latest_version=latest_version)
)
update = await is_version_update_available(
version_update_notifier, current_version=current_version
update = await get_update_if_available(
version_update_notifier,
current_version=current_version,
update_cache_repository=FakeUpdateCacheRepository(),
)
assert update is None
@ -67,8 +82,10 @@ async def test_retrieves_nothing_when_the_current_version_is_greater_than_the_la
async def test_retrieves_nothing_when_no_version_is_available() -> None:
version_update_notifier = FakeVersionUpdateGateway(update=None)
update = await is_version_update_available(
version_update_notifier, current_version="1.0.0"
update = await get_update_if_available(
version_update_notifier,
current_version="1.0.0",
update_cache_repository=FakeUpdateCacheRepository(),
)
assert update is None
@ -80,8 +97,10 @@ async def test_retrieves_nothing_when_latest_version_is_invalid() -> None:
update=VersionUpdate(latest_version="invalid-version")
)
update = await is_version_update_available(
version_update_notifier, current_version="1.0.0"
update = await get_update_if_available(
version_update_notifier,
current_version="1.0.0",
update_cache_repository=FakeUpdateCacheRepository(),
)
assert update is None
@ -96,8 +115,10 @@ async def test_replaces_hyphens_with_plus_signs_in_latest_version_to_conform_wit
update=VersionUpdate(latest_version="1.6.1-jetbrains")
)
update = await is_version_update_available(
version_update_notifier, current_version="1.0.0"
update = await get_update_if_available(
version_update_notifier,
current_version="1.0.0",
update_cache_repository=FakeUpdateCacheRepository(),
)
assert update is not None
@ -110,8 +131,10 @@ async def test_retrieves_nothing_when_current_version_is_invalid() -> None:
update=VersionUpdate(latest_version="1.0.1")
)
update = await is_version_update_available(
version_update_notifier, current_version="invalid-version"
update = await get_update_if_available(
version_update_notifier,
current_version="invalid-version",
update_cache_repository=FakeUpdateCacheRepository(),
)
assert update is None
@ -139,8 +162,166 @@ async def test_raises_version_update_error(
)
with pytest.raises(VersionUpdateError) as excinfo:
await is_version_update_available(
version_update_notifier, current_version="1.0.0"
await get_update_if_available(
version_update_notifier,
current_version="1.0.0",
update_cache_repository=FakeUpdateCacheRepository(),
)
assert expected_message_substring in str(excinfo.value)
@pytest.mark.asyncio
async def test_notifies_and_updates_cache_when_repository_is_empty(
current_timestamp: int,
) -> None:
version_update_notifier = FakeVersionUpdateGateway(
update=VersionUpdate(latest_version="1.0.1")
)
update_cache_repository = FakeUpdateCacheRepository()
update = await get_update_if_available(
version_update_notifier,
current_version="1.0.0",
update_cache_repository=update_cache_repository,
get_current_timestamp=lambda: current_timestamp,
)
assert update is not None
assert update.latest_version == "1.0.1"
assert update.should_notify is True
assert version_update_notifier.fetch_update_calls == 1
assert update_cache_repository.update_cache is not None
assert update_cache_repository.update_cache.latest_version == "1.0.1"
assert update_cache_repository.update_cache.stored_at_timestamp == current_timestamp
@pytest.mark.asyncio
async def test_does_not_notify_when_an_available_update_has_been_recently_cached(
current_timestamp: int,
) -> None:
version_update_notifier = FakeVersionUpdateGateway(
update=VersionUpdate(latest_version="1.0.1")
)
timestamp_twelve_hours_ago = current_timestamp - 12 * 60 * 60
update_cache = UpdateCache(
latest_version="1.0.1", stored_at_timestamp=timestamp_twelve_hours_ago
)
update_cache_repository = FakeUpdateCacheRepository(update_cache=update_cache)
update = await get_update_if_available(
version_update_notifier,
current_version="1.0.0",
update_cache_repository=update_cache_repository,
get_current_timestamp=lambda: current_timestamp,
)
assert update is not None
assert update.latest_version == "1.0.1"
assert update.should_notify is False
assert version_update_notifier.fetch_update_calls == 0
@pytest.mark.asyncio
async def test_retrieves_nothing_when_the_recently_cached_update_is_the_one_currently_in_use(
current_timestamp: int,
) -> None:
version_update_notifier = FakeVersionUpdateGateway(
update=VersionUpdate(latest_version="1.0.1")
)
timestamp_twelve_hours_ago = current_timestamp - 12 * 60 * 60
update_cache = UpdateCache(
latest_version="1.0.1", stored_at_timestamp=timestamp_twelve_hours_ago
)
update_cache_repository = FakeUpdateCacheRepository(update_cache=update_cache)
update = await get_update_if_available(
version_update_notifier,
current_version="1.0.1",
update_cache_repository=update_cache_repository,
get_current_timestamp=lambda: current_timestamp,
)
assert update is None
assert version_update_notifier.fetch_update_calls == 0
@pytest.mark.asyncio
async def test_retrieves_fresh_update_and_notifies_and_updates_cache_when_cache_is_not_fresh(
current_timestamp: int,
) -> None:
version_update_notifier = FakeVersionUpdateGateway(
update=VersionUpdate(latest_version="1.0.2")
)
timestamp_two_days_ago = current_timestamp - 48 * 60 * 60
update_cache = UpdateCache(
latest_version="1.0.1", stored_at_timestamp=timestamp_two_days_ago
)
update_cache_repository = FakeUpdateCacheRepository(update_cache=update_cache)
update = await get_update_if_available(
version_update_notifier,
current_version="1.0.0",
update_cache_repository=update_cache_repository,
get_current_timestamp=lambda: current_timestamp,
)
assert update is not None
assert version_update_notifier.fetch_update_calls == 1
assert update.should_notify is True
assert update.latest_version == "1.0.2"
assert update_cache_repository.update_cache is not None
assert update_cache_repository.update_cache.stored_at_timestamp == current_timestamp
assert update_cache_repository.update_cache.latest_version == "1.0.2"
@pytest.mark.asyncio
async def test_updates_cache_timestamp_with_current_version_when_no_update_is_available(
current_timestamp: int,
) -> None:
version_update_notifier = FakeVersionUpdateGateway(update=None)
timestamp_two_days_ago = current_timestamp - 48 * 60 * 60
update_cache = UpdateCache(
latest_version="1.0.0", stored_at_timestamp=timestamp_two_days_ago
)
update_cache_repository = FakeUpdateCacheRepository(update_cache=update_cache)
update = await get_update_if_available(
version_update_notifier,
current_version="1.0.0",
update_cache_repository=update_cache_repository,
get_current_timestamp=lambda: current_timestamp,
)
assert update is None
assert version_update_notifier.fetch_update_calls == 1
assert update_cache_repository.update_cache is not None
assert update_cache_repository.update_cache.latest_version == "1.0.0"
assert update_cache_repository.update_cache.stored_at_timestamp == current_timestamp
@pytest.mark.asyncio
async def test_updates_cache_timestamp_with_current_version_when_gateway_errors(
current_timestamp: int,
) -> None:
version_update_notifier = FakeVersionUpdateGateway(
error=VersionUpdateGatewayError(cause=VersionUpdateGatewayCause.ERROR_RESPONSE)
)
timestamp_two_days_ago = current_timestamp - 48 * 60 * 60
update_cache = UpdateCache(
latest_version="1.0.0", stored_at_timestamp=timestamp_two_days_ago
)
update_cache_repository = FakeUpdateCacheRepository(update_cache=update_cache)
with pytest.raises(VersionUpdateError):
await get_update_if_available(
version_update_notifier,
current_version="1.0.0",
update_cache_repository=update_cache_repository,
get_current_timestamp=lambda: current_timestamp,
)
assert version_update_notifier.fetch_update_calls == 1
assert update_cache_repository.update_cache is not None
assert update_cache_repository.update_cache.latest_version == "1.0.0"
assert update_cache_repository.update_cache.stored_at_timestamp == current_timestamp