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

v2.17.0 (#822)

Browse source 
Co-authored-by: Clément Sirieix <clement.sirieix@mistral.ai>
Co-authored-by: Guillaume LE GOFF <guillaume.lgf@gmail.com>
Co-authored-by: Hdandria <henri.dandria@mistral.ai>
Co-authored-by: Ivana Dunisijevic <ivana.dunisijevic@mistral.ai>
Co-authored-by: Jean Burellier <sheplu@users.noreply.github.com>
Co-authored-by: Mathias Gesbert <mathias.gesbert@mistral.ai>
Co-authored-by: Mert Unsal <mert.unsal@mistral.ai>
Co-authored-by: Michel Thomazo <51709227+michelTho@users.noreply.github.com>
Co-authored-by: Paul VEZIA <166131032+le-codeur-rapide@users.noreply.github.com>
Co-authored-by: Pierre Rossinès <pierre.rossines@mistral.ai>
Co-authored-by: Val <102326092+vdeva@users.noreply.github.com>
Co-authored-by: Vincent G <10739306+VinceOPS@users.noreply.github.com>
Co-authored-by: renovate-mistral[bot] <253709520+renovate-mistral[bot]@users.noreply.github.com>
Co-authored-by: Mistral Vibe <vibe@mistral.ai>
This commit is contained in:
Clément Drouin 2026-06-19 11:01:24 +02:00 • committed by GitHub
parent 564a14365e
commit 6bedf271ce
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
223 changed files with 10533 additions and 6947 deletions
Download patch file Download diff file Expand all files Collapse all files

0
tests/core/nuage/__init__.py
View file

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

@ -1,301 +0,0 @@
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
from vibe.core.agent_loop import AgentLoopStateError
from vibe.core.nuage.exceptions import ErrorCode, WorkflowsException
from vibe.core.nuage.remote_events_source import RemoteEventsSource
from vibe.core.nuage.streaming import StreamEvent
_SESSION_ID = "test-session"
def _make_source(**kwargs) -> RemoteEventsSource:
config = build_test_vibe_config(enabled_tools=kwargs.pop("enabled_tools", []))
return RemoteEventsSource(session_id=_SESSION_ID, config=config, **kwargs)
def _make_retryable_exc(msg: str) -> WorkflowsException:
return WorkflowsException(message=msg, code=ErrorCode.GET_EVENTS_STREAM_ERROR)
def _make_stream_event(
broker_sequence: int | None = None, data: dict | None = None
) -> StreamEvent:
return StreamEvent(data=data or {}, broker_sequence=broker_sequence)
class TestIsRetryableStreamDisconnect:
def test_peer_closed_connection(self) -> None:
source = _make_source()
exc = _make_retryable_exc("Peer closed connection without response")
assert source._is_retryable_stream_disconnect(exc) is True
def test_incomplete_chunked_read(self) -> None:
source = _make_source()
exc = _make_retryable_exc("Incomplete chunked read during streaming")
assert source._is_retryable_stream_disconnect(exc) is True
def test_non_retryable_message(self) -> None:
source = _make_source()
exc = WorkflowsException(
message="some other error", code=ErrorCode.GET_EVENTS_STREAM_ERROR
)
assert source._is_retryable_stream_disconnect(exc) is False
def test_wrong_error_code(self) -> None:
source = _make_source()
exc = WorkflowsException(
message="peer closed connection",
code=ErrorCode.POST_EXECUTIONS_SIGNALS_ERROR,
)
assert source._is_retryable_stream_disconnect(exc) is False
async def _async_gen_from_list(items):
for item in items:
yield item
async def _async_gen_raise(items, exc):
for item in items:
yield item
raise exc
class _FakeStream:
def __init__(self, payloads=None, exc=None):
self._payloads = payloads or []
self._exc = exc
self._closed = False
def __aiter__(self):
return self._iterate().__aiter__()
async def _iterate(self):
for p in self._payloads:
yield p
if self._exc is not None:
raise self._exc
async def aclose(self):
self._closed = True
class TestStreamRemoteEventsRetry:
@pytest.mark.asyncio
async def test_retries_on_retryable_disconnect(self) -> None:
source = _make_source()
exc = _make_retryable_exc("peer closed connection")
call_count = 0
def make_stream(_params):
nonlocal call_count
call_count += 1
return _FakeStream(exc=exc)
mock_client = MagicMock()
mock_client.stream_events = make_stream
source._client = mock_client
with patch("asyncio.sleep", new_callable=AsyncMock):
events = [e async for e in source._stream_remote_events()]
assert events == []
assert call_count == 4 # 1 initial + 3 retries
@pytest.mark.asyncio
async def test_stops_after_max_retry_count(self) -> None:
source = _make_source()
exc = _make_retryable_exc("incomplete chunked read")
call_count = 0
def make_stream(_params):
nonlocal call_count
call_count += 1
return _FakeStream(exc=exc)
mock_client = MagicMock()
mock_client.stream_events = make_stream
source._client = mock_client
with patch("asyncio.sleep", new_callable=AsyncMock):
events = [e async for e in source._stream_remote_events()]
assert events == []
assert call_count == 4
@pytest.mark.asyncio
async def test_resets_retry_count_on_successful_event(self) -> None:
source = _make_source()
exc = _make_retryable_exc("peer closed connection")
successful_event = _make_stream_event(broker_sequence=0, data={})
call_count = 0
def make_stream(_params):
nonlocal call_count
call_count += 1
if call_count <= 2:
return _FakeStream(payloads=[successful_event], exc=exc)
return _FakeStream(exc=exc)
mock_client = MagicMock()
mock_client.stream_events = make_stream
source._client = mock_client
with (
patch("asyncio.sleep", new_callable=AsyncMock),
patch.object(source, "_normalize_stream_event", return_value=None),
):
events = [e async for e in source._stream_remote_events()]
assert events == []
# call 1: success + exc -> retry_count = 1
# call 2: success (reset) + exc -> retry_count = 1
# call 3: exc -> retry_count = 2
# call 4: exc -> retry_count = 3
# call 5: exc -> retry_count = 4 > 3 -> break
assert call_count == 5
@pytest.mark.asyncio
async def test_non_retryable_raises_agent_loop_state_error(self) -> None:
source = _make_source()
exc = WorkflowsException(
message="something bad", code=ErrorCode.TEMPORAL_CONNECTION_ERROR
)
mock_client = MagicMock()
mock_client.stream_events = lambda _: _FakeStream(exc=exc)
source._client = mock_client
with pytest.raises(AgentLoopStateError):
async for _ in source._stream_remote_events():
pass
class TestStreamRemoteEventsIdleBoundary:
@pytest.mark.asyncio
async def test_stops_on_idle_boundary(self) -> None:
source = _make_source()
event_data = _make_stream_event(broker_sequence=0, data={})
sentinel_event = MagicMock()
mock_client = MagicMock()
mock_client.stream_events = lambda _: _FakeStream(payloads=[event_data])
source._client = mock_client
workflow_event = MagicMock()
with (
patch.object(
source, "_normalize_stream_event", return_value=workflow_event
),
patch.object(
source, "_consume_workflow_event", return_value=[sentinel_event]
),
patch.object(source, "_is_idle_boundary", return_value=True) as mock_idle,
):
events = [
e
async for e in source._stream_remote_events(stop_on_idle_boundary=True)
]
assert events == [sentinel_event]
mock_idle.assert_called_once_with(workflow_event)
@pytest.mark.asyncio
async def test_continues_past_idle_boundary_when_disabled(self) -> None:
source = _make_source()
event1 = _make_stream_event(broker_sequence=0, data={})
event2 = _make_stream_event(broker_sequence=1, data={})
sentinel1 = MagicMock()
sentinel2 = MagicMock()
mock_client = MagicMock()
mock_client.stream_events = lambda _: _FakeStream(payloads=[event1, event2])
source._client = mock_client
workflow_event = MagicMock()
call_count = 0
def consume_side_effect(_evt):
nonlocal call_count
call_count += 1
return [sentinel1] if call_count == 1 else [sentinel2]
with (
patch.object(
source, "_normalize_stream_event", return_value=workflow_event
),
patch.object(
source, "_consume_workflow_event", side_effect=consume_side_effect
),
patch.object(source, "_is_idle_boundary", return_value=True),
):
events = [
e
async for e in source._stream_remote_events(stop_on_idle_boundary=False)
]
assert events == [sentinel1, sentinel2]
class TestBrokerSequenceTracking:
@pytest.mark.asyncio
async def test_next_start_seq_updated(self) -> None:
source = _make_source()
assert source._next_start_seq == 0
event1 = _make_stream_event(broker_sequence=5, data={})
event2 = _make_stream_event(broker_sequence=10, data={})
mock_client = MagicMock()
mock_client.stream_events = lambda _: _FakeStream(payloads=[event1, event2])
source._client = mock_client
with patch.object(source, "_normalize_stream_event", return_value=None):
events = [e async for e in source._stream_remote_events()]
assert events == []
assert source._next_start_seq == 11
@pytest.mark.asyncio
async def test_none_broker_sequence_not_updated(self) -> None:
source = _make_source()
source._next_start_seq = 5
event = _make_stream_event(broker_sequence=None, data={})
mock_client = MagicMock()
mock_client.stream_events = lambda _: _FakeStream(payloads=[event])
source._client = mock_client
with patch.object(source, "_normalize_stream_event", return_value=None):
events = [e async for e in source._stream_remote_events()]
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()

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

@ -1,248 +0,0 @@
from __future__ import annotations
import json
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:
return WorkflowsClient(base_url="http://localhost:8080", api_key="test-key")
def _valid_event_payload() -> dict:
return {
"stream": "test-stream",
"timestamp_unix_nano": 1000000,
"data": {"key": "value"},
}
class TestParseSSEData:
def test_valid_json_returns_stream_event(self) -> None:
client = _make_client()
payload = _valid_event_payload()
result = client._parse_sse_data(json.dumps(payload), event_type=None)
assert isinstance(result, StreamEvent)
assert result.stream == "test-stream"
assert result.data == {"key": "value"}
def test_error_event_type_raises(self) -> None:
client = _make_client()
payload = {"some": "data"}
with pytest.raises(WorkflowsException) as exc_info:
client._parse_sse_data(json.dumps(payload), event_type="error")
assert exc_info.value.code == ErrorCode.GET_EVENTS_STREAM_ERROR
assert "Stream error from server" in exc_info.value.message
def test_error_key_in_json_raises(self) -> None:
client = _make_client()
payload = {"error": "something went wrong"}
with pytest.raises(WorkflowsException) as exc_info:
client._parse_sse_data(json.dumps(payload), event_type=None)
assert exc_info.value.code == ErrorCode.GET_EVENTS_STREAM_ERROR
assert "something went wrong" in exc_info.value.message
def test_error_event_type_with_non_dict_parsed(self) -> None:
client = _make_client()
with pytest.raises(WorkflowsException) as exc_info:
client._parse_sse_data(json.dumps("a plain string"), event_type="error")
assert "a plain string" in exc_info.value.message
def test_malformed_json_raises(self) -> None:
client = _make_client()
with pytest.raises(json.JSONDecodeError):
client._parse_sse_data("{not valid json", event_type=None)
class TestIterSSEEvents:
@pytest.mark.asyncio
async def test_parses_data_lines(self) -> None:
client = _make_client()
payload = _valid_event_payload()
lines = [f"data: {json.dumps(payload)}"]
response = AsyncMock()
response.aiter_bytes = _async_byte_iter(lines)
events = [e async for e in client._iter_sse_events(response)]
assert len(events) == 1
assert events[0].stream == "test-stream"
@pytest.mark.asyncio
async def test_skips_empty_lines_and_comments(self) -> None:
client = _make_client()
payload = _valid_event_payload()
lines = ["", ": this is a comment", f"data: {json.dumps(payload)}", ""]
response = AsyncMock()
response.aiter_bytes = _async_byte_iter(lines)
events = [e async for e in client._iter_sse_events(response)]
assert len(events) == 1
@pytest.mark.asyncio
async def test_parses_event_type_and_passes_to_parse(self) -> None:
client = _make_client()
payload = {"error": "server broke"}
lines = ["event: error", f"data: {json.dumps(payload)}"]
response = AsyncMock()
response.aiter_bytes = _async_byte_iter(lines)
with pytest.raises(WorkflowsException) as exc_info:
_ = [e async for e in client._iter_sse_events(response)]
assert exc_info.value.code == ErrorCode.GET_EVENTS_STREAM_ERROR
@pytest.mark.asyncio
async def test_resets_event_type_after_data_line(self) -> None:
client = _make_client()
payload = _valid_event_payload()
lines = [
"event: custom_type",
f"data: {json.dumps(payload)}",
f"data: {json.dumps(payload)}",
]
response = AsyncMock()
response.aiter_bytes = _async_byte_iter(lines)
with patch.object(
client, "_parse_sse_data", wraps=client._parse_sse_data
) as mock_parse:
events = [e async for e in client._iter_sse_events(response)]
assert len(events) == 2
assert mock_parse.call_args_list[0].args[1] == "custom_type"
assert mock_parse.call_args_list[1].args[1] is None
@pytest.mark.asyncio
async def test_skips_non_data_non_event_lines(self) -> None:
client = _make_client()
payload = _valid_event_payload()
lines = ["id: 123", "retry: 5000", f"data: {json.dumps(payload)}"]
response = AsyncMock()
response.aiter_bytes = _async_byte_iter(lines)
events = [e async for e in client._iter_sse_events(response)]
assert len(events) == 1
@pytest.mark.asyncio
async def test_parse_failure_logs_warning_and_continues(self) -> None:
client = _make_client()
payload = _valid_event_payload()
lines = ["data: {not valid json}", f"data: {json.dumps(payload)}"]
response = AsyncMock()
response.aiter_bytes = _async_byte_iter(lines)
with patch("vibe.core.nuage.client.logger") as mock_logger:
events = [e async for e in client._iter_sse_events(response)]
assert len(events) == 1
mock_logger.warning.assert_called_once()
def _setup_mock_client(client: WorkflowsClient, mock_response: AsyncMock) -> None:
mock_stream = AsyncMock()
mock_stream.__aenter__ = AsyncMock(return_value=mock_response)
mock_stream.__aexit__ = AsyncMock(return_value=False)
mock_http = AsyncMock()
mock_http.stream = lambda *args, **kwargs: mock_stream
client._client = mock_http
class TestStreamEvents:
@pytest.mark.asyncio
async def test_yields_stream_events(self) -> None:
client = _make_client()
payload = _valid_event_payload()
lines = [f"data: {json.dumps(payload)}"]
mock_response = AsyncMock()
mock_response.raise_for_status = lambda: None
mock_response.aiter_bytes = _async_byte_iter(lines)
_setup_mock_client(client, mock_response)
params = StreamEventsQueryParams(workflow_exec_id="wf-1", start_seq=0)
events = [e async for e in client.stream_events(params)]
assert len(events) == 1
assert isinstance(events[0], StreamEvent)
@pytest.mark.asyncio
async def test_reraises_workflows_exception(self) -> None:
client = _make_client()
mock_response = AsyncMock()
mock_response.raise_for_status = lambda: None
mock_response.aiter_bytes = _async_byte_iter([
"event: error",
'data: {"error": "stream error"}',
])
_setup_mock_client(client, mock_response)
params = StreamEventsQueryParams(workflow_exec_id="wf-1")
with pytest.raises(WorkflowsException) as exc_info:
_ = [e async for e in client.stream_events(params)]
assert exc_info.value.code == ErrorCode.GET_EVENTS_STREAM_ERROR
@pytest.mark.asyncio
async def test_wraps_other_exceptions_in_workflows_exception(self) -> None:
client = _make_client()
mock_response = AsyncMock()
mock_response.raise_for_status.side_effect = RuntimeError("connection lost")
_setup_mock_client(client, mock_response)
params = StreamEventsQueryParams(workflow_exec_id="wf-1")
with pytest.raises(WorkflowsException) as exc_info:
_ = [e async for e in client.stream_events(params)]
assert exc_info.value.code == ErrorCode.GET_EVENTS_STREAM_ERROR
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_byte_iter(lines: list[str]):
async def _iter():
for line in lines:
yield f"{line}\n".encode()
return _iter

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

@ -1,11 +1,17 @@
from __future__ import annotations
import base64
import hashlib
from pathlib import Path
import pytest
from vibe.core.session.image_snapshot import ImageSnapshotError, snapshot_image
from vibe.core.session.image_snapshot import (
ImageSnapshotError,
snapshot_image,
snapshot_image_bytes,
)
from vibe.core.types import MAX_IMAGE_BYTES, FileImageSource, InlineImageSource
PNG_BYTES = b"\x89PNG\r\n\x1a\n" + b"\x00" * 16
@ -19,10 +25,11 @@ def test_snapshot_image_copies_to_attachments_dir(tmp_path: Path) -> None:
att = snapshot_image(src, alias="screenshot.png", session_dir=session_dir)
digest = hashlib.sha1(PNG_BYTES, usedforsecurity=False).hexdigest()
assert att.path == (session_dir / "attachments" / f"{digest}.png").resolve()
assert isinstance(att.source, FileImageSource)
assert att.source.path == (session_dir / "attachments" / f"{digest}.png").resolve()
assert att.alias == "screenshot.png"
assert att.mime_type == "image/png"
assert att.path.read_bytes() == PNG_BYTES
assert att.source.path.read_bytes() == PNG_BYTES
def test_snapshot_image_is_idempotent_on_same_bytes(tmp_path: Path) -> None:
@ -35,7 +42,9 @@ def test_snapshot_image_is_idempotent_on_same_bytes(tmp_path: Path) -> None:
att_a = snapshot_image(src_a, alias="a.png", session_dir=session_dir)
att_b = snapshot_image(src_b, alias="b.png", session_dir=session_dir)
assert att_a.path == att_b.path
assert isinstance(att_a.source, FileImageSource)
assert isinstance(att_b.source, FileImageSource)
assert att_a.source.path == att_b.source.path
assert sum(1 for _ in (session_dir / "attachments").iterdir()) == 1
@ -45,7 +54,8 @@ def test_snapshot_image_returns_source_when_session_dir_is_none(tmp_path: Path)
att = snapshot_image(src, alias="screenshot.png", session_dir=None)
assert att.path == src.resolve()
assert isinstance(att.source, FileImageSource)
assert att.source.path == src.resolve()
assert att.alias == "screenshot.png"
@ -69,3 +79,54 @@ def test_snapshot_image_normalizes_jpg_to_jpeg_mime(tmp_path: Path) -> None:
att = snapshot_image(src, alias="photo.jpg", session_dir=None)
assert att.mime_type == "image/jpeg"
def test_snapshot_image_bytes_inlines_when_session_dir_is_none() -> None:
att = snapshot_image_bytes(
PNG_BYTES, alias="pasted.png", mime_type="image/png", session_dir=None
)
assert isinstance(att.source, InlineImageSource)
assert att.source.data == base64.b64encode(PNG_BYTES).decode("ascii")
assert att.alias == "pasted.png"
assert att.mime_type == "image/png"
def test_snapshot_image_bytes_writes_file_when_session_dir_is_set(
tmp_path: Path,
) -> None:
session_dir = tmp_path / "session"
att = snapshot_image_bytes(
PNG_BYTES, alias="pasted.png", mime_type="image/png", session_dir=session_dir
)
digest = hashlib.sha1(PNG_BYTES, usedforsecurity=False).hexdigest()
assert isinstance(att.source, FileImageSource)
assert att.source.path == (session_dir / "attachments" / f"{digest}.png").resolve()
assert att.source.path.read_bytes() == PNG_BYTES
def test_snapshot_image_bytes_rejects_unsupported_mime() -> None:
with pytest.raises(ImageSnapshotError):
snapshot_image_bytes(
PNG_BYTES, alias="x", mime_type="image/tiff", session_dir=None
)
def test_snapshot_image_rejects_oversized_file(tmp_path: Path) -> None:
src = tmp_path / "big.png"
src.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * MAX_IMAGE_BYTES)
with pytest.raises(ImageSnapshotError):
snapshot_image(src, alias="big.png", session_dir=None)
def test_snapshot_image_bytes_rejects_oversized_data() -> None:
with pytest.raises(ImageSnapshotError):
snapshot_image_bytes(
b"x" * (MAX_IMAGE_BYTES + 1),
alias="big.png",
mime_type="image/png",
session_dir=None,
)

47
tests/core/test_agent_loop_refresh_config.py Normal file
View file

@ -0,0 +1,47 @@
from __future__ import annotations
import pytest
from tests.conftest import build_test_agent_loop, build_test_vibe_config
from tests.stubs.fake_mcp_registry import FakeMCPRegistry
from vibe.core.config import MCPHttp, MCPOAuth, MCPStreamableHttp, VibeConfig
from vibe.core.tools.mcp import AuthStatus
def test_refresh_config_reconciles_mcp_registry_status(
monkeypatch: pytest.MonkeyPatch,
) -> None:
kept = MCPHttp(name="kept", transport="http", url="http://kept:1")
removed = MCPHttp(name="removed", transport="http", url="http://removed:1")
registry = FakeMCPRegistry()
agent_loop = build_test_agent_loop(
config=build_test_vibe_config(mcp_servers=[kept, removed]),
mcp_registry=registry,
)
refreshed_config = build_test_vibe_config(mcp_servers=[kept])
monkeypatch.setattr(VibeConfig, "load", staticmethod(lambda: refreshed_config))
agent_loop.refresh_config()
assert registry.status() == {"kept": AuthStatus.STATIC}
def test_refresh_config_does_not_mark_undiscovered_oauth_server_ok(
monkeypatch: pytest.MonkeyPatch,
) -> None:
oauth = MCPStreamableHttp(
name="linear",
transport="streamable-http",
url="https://mcp.example.com/mcp",
auth=MCPOAuth(type="oauth", scopes=["read"]),
)
registry = FakeMCPRegistry()
agent_loop = build_test_agent_loop(
config=build_test_vibe_config(mcp_servers=[]), mcp_registry=registry
)
refreshed_config = build_test_vibe_config(mcp_servers=[oauth])
monkeypatch.setattr(VibeConfig, "load", staticmethod(lambda: refreshed_config))
agent_loop.refresh_config()
assert registry.status() == {"linear": AuthStatus.NEEDS_AUTH}

3
tests/core/test_backend_error.py
View file

@ -2,6 +2,7 @@ from __future__ import annotations
import pytest
from tests.constants import CHAT_COMPLETIONS_PATH
from vibe.core.llm.exceptions import BackendError, PayloadSummary
@ -24,7 +25,7 @@ def _make_error(
) -> BackendError:
return BackendError(
provider="test-provider",
endpoint="/v1/chat/completions",
endpoint=CHAT_COMPLETIONS_PATH,
status=status,
reason="some reason",
headers=headers or {},

6
tests/core/test_build_attachment_counts.py
View file

@ -4,7 +4,7 @@ from pathlib import Path
from vibe.core.telemetry.build_metadata import build_attachment_counts
from vibe.core.telemetry.types import AttachmentKind
from vibe.core.types import ImageAttachment, LLMMessage, Role
from vibe.core.types import FileImageSource, ImageAttachment, LLMMessage, Role
def _msg(images: list[ImageAttachment] | None) -> LLMMessage:
@ -13,7 +13,9 @@ def _msg(images: list[ImageAttachment] | None) -> LLMMessage:
def _image() -> ImageAttachment:
return ImageAttachment(
path=Path("/tmp/x.png"), alias="x.png", mime_type="image/png"
source=FileImageSource(path=Path("/tmp/x.png")),
alias="x.png",
mime_type="image/png",
)

109
tests/core/test_cache_store.py Normal file
View file

@ -0,0 +1,109 @@
from __future__ import annotations
from pathlib import Path
import tomllib
from vibe.core.cache_store import FileSystemVibeCodeCacheStore
class TestFileSystemVibeCodeCacheStore:
def test_reads_valid_toml_section(self, tmp_path: Path) -> None:
cache_path = tmp_path / "cache.toml"
cache_path.write_text('[update_cache]\nlatest_version = "1.0.0"\n')
store = FileSystemVibeCodeCacheStore(cache_path)
result = store.read_section("update_cache")
assert result["latest_version"] == "1.0.0"
def test_returns_empty_dict_when_file_is_missing(self, tmp_path: Path) -> None:
store = FileSystemVibeCodeCacheStore(tmp_path / "missing.toml")
assert store.read_section("update_cache") == {}
def test_returns_empty_dict_when_file_is_corrupted(self, tmp_path: Path) -> None:
cache_path = tmp_path / "cache.toml"
cache_path.write_text("{bad toml")
store = FileSystemVibeCodeCacheStore(cache_path)
assert store.read_section("update_cache") == {}
def test_writes_new_file(self, tmp_path: Path) -> None:
cache_path = tmp_path / "cache.toml"
store = FileSystemVibeCodeCacheStore(cache_path)
store.write_section("feedback", {"last_shown_at": 100.0})
with cache_path.open("rb") as f:
data = tomllib.load(f)
assert data["feedback"]["last_shown_at"] == 100.0
def test_merges_with_existing(self, tmp_path: Path) -> None:
cache_path = tmp_path / "cache.toml"
cache_path.write_text('[update_cache]\nlatest_version = "1.0.0"\n')
store = FileSystemVibeCodeCacheStore(cache_path)
store.write_section("feedback", {"last_shown_at": 200.0})
with cache_path.open("rb") as f:
data = tomllib.load(f)
assert data["update_cache"]["latest_version"] == "1.0.0"
assert data["feedback"]["last_shown_at"] == 200.0
def test_merges_within_section_and_leaves_other_sections_alone(
self, tmp_path: Path
) -> None:
cache_path = tmp_path / "cache.toml"
cache_path.write_text(
"[update_cache]\n"
'latest_version = "1.0.0"\n'
"stored_at_timestamp = 1\n"
'seen_whats_new_version = "1.0.0"\n\n'
"[feedback]\n"
"last_shown_at = 100.0\n"
)
store = FileSystemVibeCodeCacheStore(cache_path)
store.write_section(
"update_cache", {"latest_version": "2.0.0", "stored_at_timestamp": 2}
)
with cache_path.open("rb") as f:
data = tomllib.load(f)
assert data["update_cache"]["latest_version"] == "2.0.0"
assert data["update_cache"]["stored_at_timestamp"] == 2
assert data["update_cache"]["seen_whats_new_version"] == "1.0.0"
assert data["feedback"]["last_shown_at"] == 100.0
def test_reads_section(self, tmp_path: Path) -> None:
cache_path = tmp_path / "cache.toml"
cache_path.write_text("[feedback]\nlast_shown_at = 100\n")
store = FileSystemVibeCodeCacheStore(cache_path)
assert store.read_section("feedback") == {"last_shown_at": 100}
def test_returns_empty_dict_for_missing_section(self, tmp_path: Path) -> None:
store = FileSystemVibeCodeCacheStore(tmp_path / "cache.toml")
assert store.read_section("feedback") == {}
def test_writes_section(self, tmp_path: Path) -> None:
cache_path = tmp_path / "cache.toml"
store = FileSystemVibeCodeCacheStore(cache_path)
store.write_section("feedback", {"last_shown_at": 100})
with cache_path.open("rb") as f:
data = tomllib.load(f)
assert data["feedback"]["last_shown_at"] == 100
def test_replaces_non_table_section_when_writing(self, tmp_path: Path) -> None:
cache_path = tmp_path / "cache.toml"
cache_path.write_text('feedback = "not-a-table"\n')
store = FileSystemVibeCodeCacheStore(cache_path)
store.write_section("feedback", {"last_shown_at": 100})
with cache_path.open("rb") as f:
data = tomllib.load(f)
assert data["feedback"]["last_shown_at"] == 100

3
tests/core/test_config_builder.py
View file

@ -31,6 +31,9 @@ class FakeLayer(ConfigLayer[RawConfig]):
async def _build_config_snapshot(self) -> LayerConfigSnapshot:
return LayerConfigSnapshot(data=dict(self._data), fingerprint="fp")
async def _save_to_store(self, _next_config: RawConfig) -> str:
raise NotImplementedError
class UntrustedFakeLayer(FakeLayer):
async def _check_trust(self) -> bool:

206
tests/core/test_config_layer.py
View file

@ -3,18 +3,24 @@ from __future__ import annotations
import asyncio
from typing import Any
from jsonpointer import JsonPointerException
from pydantic import BaseModel, ValidationError
import pytest
from vibe.core.config.layer import (
ConfigLayer,
ConfigPatchApplicationError,
LayerImplementationError,
RawConfig,
TrustNotResolvedError,
UntrustedLayerError,
)
from vibe.core.config.patch import ConfigPatch
from vibe.core.config.types import ConcurrencyConflictError, LayerConfigSnapshot
from vibe.core.config.patch import AddOperationPatch, ConfigPatch, ReplaceOperationPatch
from vibe.core.config.types import (
ConcurrencyConflictError,
ConflictStrategy,
LayerConfigSnapshot,
)
class StubLayer(ConfigLayer[BaseModel]):
@ -45,6 +51,9 @@ class StubLayer(ConfigLayer[BaseModel]):
data=dict(self._data), fingerprint=f"fp-{self.read_count}"
)
async def _save_to_store(self, _next_config: BaseModel) -> str:
raise NotImplementedError("StubLayer.apply() is not implemented")
class ObservableStubLayer(StubLayer):
"""Stub that records _on_trust_changed calls."""
@ -57,6 +66,18 @@ class ObservableStubLayer(StubLayer):
self.trust_changes.append((old, new))
class WritableStubLayer(StubLayer):
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.writes: list[dict[str, Any]] = []
async def _save_to_store(self, _next_config: BaseModel) -> str:
next_data = _next_config.model_dump()
self.writes.append(next_data)
self._data = next_data
return f"write-fp-{len(self.writes)}"
class SampleSchema(BaseModel):
name: str
count: int = 0
@ -91,6 +112,9 @@ async def test_default_check_trust_returns_false() -> None:
async def _build_config_snapshot(self) -> LayerConfigSnapshot:
return LayerConfigSnapshot(data={}, fingerprint="fp")
async def _save_to_store(self, _next_config: BaseModel) -> str:
raise NotImplementedError
layer = DefaultTrustLayer(name="default")
result = await layer.resolve_trust()
assert result is False
@ -236,7 +260,7 @@ 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"}
assert result.model_dump() == {"key": "value"}
assert layer.fingerprint == "fp-1"
@ -246,7 +270,7 @@ async def test_load_auto_resolves_trust() -> None:
assert layer.is_trusted is None
result = await layer.load()
assert layer.is_trusted is True
assert result.model_extra == {"a": 1}
assert result.model_dump() == {"a": 1}
assert layer.fingerprint == "fp-1"
await layer.resolve_trust()
@ -311,21 +335,21 @@ async def test_invalidate_cache_causes_reload() -> None:
async def test_revoke_grant_cycle_refreshes_data() -> None:
layer = StubLayer(data={"v": 1})
result1 = await layer.load()
assert result1.model_extra == {"v": 1}
assert result1.model_dump() == {"v": 1}
await layer.revoke_trust()
layer._data = {"v": 2}
await layer.grant_trust()
result2 = await layer.load()
assert result2.model_extra == {"v": 2}
assert result2.model_dump() == {"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}
assert result1.model_dump() == {"v": 1}
# External revocation via resolve_trust (not revoke_trust)
layer._stub_trusted = False
@ -339,7 +363,7 @@ async def test_resolve_trust_clears_data_on_revocation() -> None:
await layer.resolve_trust()
result2 = await layer.load()
assert result2.model_extra == {"v": 2}
assert result2.model_dump() == {"v": 2}
@pytest.mark.asyncio
@ -350,7 +374,7 @@ async def test_load_returns_deep_copy() -> None:
result1.model_extra["items"].append("mutated")
result2 = await layer.load()
assert result2.model_extra == {"items": ["a", "b"]}
assert result2.model_dump() == {"items": ["a", "b"]}
assert layer.read_count == 1
@ -384,7 +408,7 @@ 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"}
assert result.model_dump() == {"anything": "goes"}
@pytest.mark.asyncio
@ -423,6 +447,9 @@ async def test_concurrent_loads_serialize() -> None:
data={"v": self.read_count}, fingerprint=f"fp-{self.read_count}"
)
async def _save_to_store(self, _next_config: BaseModel) -> str:
raise NotImplementedError
layer = SlowLayer()
results = await asyncio.gather(layer.load(), layer.load(), layer.load())
assert layer.read_count == 1
@ -438,8 +465,155 @@ async def test_fingerprint_returns_none_before_load() -> None:
@pytest.mark.asyncio
async def test_apply_not_implemented() -> None:
layer = StubLayer()
await layer.load()
fingerprint = layer.fingerprint
assert isinstance(fingerprint, str)
with pytest.raises(NotImplementedError):
await layer.apply(ConfigPatch(fingerprint="fp-1"))
await layer.apply(ConfigPatch(fingerprint=fingerprint))
@pytest.mark.asyncio
async def test_apply_operation_failure_wrapped() -> None:
original_data = {"tools": {"disabled_tools": ["bash"]}}
layer = WritableStubLayer(data=original_data)
await layer.load()
fingerprint = layer.fingerprint
assert isinstance(fingerprint, str)
with pytest.raises(
ConfigPatchApplicationError, match="Layer 'stub': failed to apply patch"
) as exc_info:
await layer.apply(
ConfigPatch(
AddOperationPatch(path="/tools/enabled_tools/-", value="read"),
fingerprint=fingerprint,
)
)
assert exc_info.value.layer_name == "stub"
assert isinstance(exc_info.value.__cause__, JsonPointerException)
assert "enabled_tools" in str(exc_info.value.__cause__)
assert layer.fingerprint == fingerprint
assert (await layer.load()).model_dump() == original_data
assert layer.writes == []
@pytest.mark.asyncio
async def test_apply_operation_failure_after_successful_operation_is_atomic() -> None:
original_data = {"active_model": "old", "tools": {"disabled_tools": ["bash"]}}
layer = WritableStubLayer(data=original_data)
await layer.load()
fingerprint = layer.fingerprint
assert isinstance(fingerprint, str)
with pytest.raises(ConfigPatchApplicationError):
await layer.apply(
ConfigPatch(
ReplaceOperationPatch(path="/active_model", value="new"),
AddOperationPatch(path="/tools/enabled_tools/-", value="read"),
fingerprint=fingerprint,
)
)
assert layer.fingerprint == fingerprint
assert (await layer.load()).model_dump() == original_data
assert layer.writes == []
@pytest.mark.asyncio
async def test_apply_schema_validation_failure_wrapped() -> None:
original_data = {"name": "test", "count": 1}
layer = WritableStubLayer(output_schema=SampleSchema, data=original_data)
await layer.load()
fingerprint = layer.fingerprint
assert isinstance(fingerprint, str)
with pytest.raises(
ConfigPatchApplicationError, match="Layer 'stub': failed to apply patch"
) as exc_info:
await layer.apply(
ConfigPatch(
ReplaceOperationPatch(path="/count", value={"bad": "value"}),
fingerprint=fingerprint,
)
)
assert exc_info.value.layer_name == "stub"
assert isinstance(exc_info.value.__cause__, ValidationError)
assert layer.fingerprint == fingerprint
assert (await layer.load()).model_dump() == original_data
assert layer.writes == []
@pytest.mark.asyncio
async def test_apply_cancel_rejects_stale_fingerprint() -> None:
layer = WritableStubLayer(data={"key": "old"})
await layer.load()
fingerprint = layer.fingerprint
assert isinstance(fingerprint, str)
await layer.apply(
ConfigPatch(
ReplaceOperationPatch(path="/key", value="first"), fingerprint=fingerprint
)
)
with pytest.raises(ConcurrencyConflictError) as exc_info:
await layer.apply(
ConfigPatch(
ReplaceOperationPatch(path="/key", value="second"),
fingerprint=fingerprint,
)
)
assert exc_info.value.expected_fp == fingerprint
assert exc_info.value.actual_fp == layer.fingerprint
assert (await layer.load()).model_dump() == {"key": "first"}
@pytest.mark.asyncio
async def test_apply_replace_accepts_stale_fingerprint() -> None:
layer = WritableStubLayer(data={"key": "old"})
await layer.load()
fingerprint = layer.fingerprint
assert isinstance(fingerprint, str)
await layer.apply(
ConfigPatch(
ReplaceOperationPatch(path="/key", value="first"), fingerprint=fingerprint
)
)
await layer.apply(
ConfigPatch(
ReplaceOperationPatch(path="/key", value="second"), fingerprint=fingerprint
),
on_conflict=ConflictStrategy.REPLACE,
)
assert (await layer.load()).model_dump() == {"key": "second"}
assert layer.writes == [{"key": "first"}, {"key": "second"}]
@pytest.mark.asyncio
async def test_save_to_store_failure_wrapped() -> None:
class BrokenApplyLayer(StubLayer):
async def _save_to_store(self, _next_config: BaseModel) -> str:
raise OSError("disk full")
layer = BrokenApplyLayer(data={"key": "old"})
await layer.load()
fingerprint = layer.fingerprint
assert isinstance(fingerprint, str)
with pytest.raises(LayerImplementationError, match="_save_to_store") as exc_info:
await layer.apply(
ConfigPatch(
ReplaceOperationPatch(path="/key", value="new"), fingerprint=fingerprint
)
)
assert isinstance(exc_info.value.__cause__, OSError)
@pytest.mark.asyncio
@ -499,6 +673,9 @@ class FakeLocalUserLayer(ConfigLayer[UserConfigSchema]):
async def _build_config_snapshot(self) -> LayerConfigSnapshot:
return LayerConfigSnapshot(data=dict(self._data), fingerprint="user-fp")
async def _save_to_store(self, _next_config: UserConfigSchema) -> str:
raise NotImplementedError
@pytest.mark.asyncio
async def test_scenario_local_user_layer_always_trusted() -> None:
@ -544,6 +721,9 @@ class FakeLocalProjectLayer(ConfigLayer[BaseModel]):
async def _build_config_snapshot(self) -> LayerConfigSnapshot:
return LayerConfigSnapshot(data=dict(self._data), fingerprint="project-fp")
async def _save_to_store(self, _next_config: BaseModel) -> str:
raise NotImplementedError
@pytest.mark.asyncio
async def test_scenario_local_project_layer_trust_lifecycle() -> None:
@ -561,7 +741,7 @@ async def test_scenario_local_project_layer_trust_lifecycle() -> None:
await layer.grant_trust()
assert trust_store == {"/tmp/my-project": True}
result = await layer.load()
assert result.model_extra == project_data
assert result.model_dump() == project_data
# 3. New instance with same trust store — loads directly
layer2 = FakeLocalProjectLayer(
@ -570,7 +750,7 @@ async def test_scenario_local_project_layer_trust_lifecycle() -> None:
assert layer2.is_trusted is None
result2 = await layer2.load()
assert layer2.is_trusted is True
assert result2.model_extra == project_data
assert result2.model_dump() == project_data
# 4. Revoke trust — removed from store, load raises
await layer2.revoke_trust()

22
tests/core/test_config_load_dotenv.py
View file

@ -20,7 +20,7 @@ def test_skips_missing_file(tmp_path: Path) -> None:
assert environ == {"EXISTING": "1"}
def test_sets_and_overrides_values(tmp_path: Path) -> None:
def test_adds_missing_values_without_overriding_existing(tmp_path: Path) -> None:
env_path = tmp_path / ".env"
_write_env_file(
env_path,
@ -42,11 +42,23 @@ def test_sets_and_overrides_values(tmp_path: Path) -> None:
load_dotenv_values(env_path=env_path, environ=environ)
assert environ["MISTRAL_API_KEY"] == "new-key"
assert environ["HTTPS_PROXY"] == "https://local-proxy:8080"
assert environ["OTHER"] == "from-env"
# An explicit process/shell value wins over the .env file.
assert environ["MISTRAL_API_KEY"] == "old-key"
assert environ["HTTPS_PROXY"] == "old-https"
assert environ["OTHER"] == "keep"
assert environ["FOO"] == "keep"
# Keys absent from the process env are still loaded from the .env file.
assert environ["NEW_KEY"] == "added"
assert environ["FOO"] == "replace"
def test_adds_dotenv_value_when_process_env_is_empty(tmp_path: Path) -> None:
env_path = tmp_path / ".env"
_write_env_file(env_path, "MISTRAL_API_KEY=file-key\n")
environ = {"MISTRAL_API_KEY": ""}
load_dotenv_values(env_path=env_path, environ=environ)
assert environ["MISTRAL_API_KEY"] == "file-key"
def test_ignores_empty_values(tmp_path: Path) -> None:

35
tests/core/test_config_mcp_auth.py
View file

@ -1,11 +1,12 @@
from __future__ import annotations
import logging
from unittest.mock import AsyncMock, MagicMock, patch
from pydantic import ValidationError
import pytest
from vibe.core.config import MCPHttp, MCPOAuth, MCPStaticAuth, MCPStreamableHttp
from vibe.core.tools.mcp import AuthStatus
from vibe.core.tools.mcp.registry import MCPRegistry
HTTP_TRANSPORTS = [
@ -183,10 +184,8 @@ def test_oauth_scopes_empty_list_allowed() -> None:
@pytest.mark.asyncio
@pytest.mark.parametrize(("cls", "transport"), HTTP_TRANSPORTS)
async def test_registry_skips_oauth_servers_with_gated_warning(
cls: type[MCPHttp | MCPStreamableHttp],
transport: str,
caplog: pytest.LogCaptureFixture,
async def test_registry_marks_oauth_servers_without_tokens_as_needing_auth(
cls: type[MCPHttp | MCPStreamableHttp], transport: str
) -> None:
srv = cls.model_validate({
"name": "linear",
@ -195,19 +194,23 @@ async def test_registry_skips_oauth_servers_with_gated_warning(
"auth": {"type": "oauth", "scopes": ["read"]},
})
registry = MCPRegistry()
storage = MagicMock()
storage.get_tokens = AsyncMock(return_value=None)
storage.delete_tokens = AsyncMock()
storage.delete_client_info = AsyncMock()
with caplog.at_level(logging.WARNING, logger="vibe"):
with (
patch("vibe.core.tools.mcp.registry.KeyringTokenStorage", return_value=storage),
patch(
"vibe.core.tools.mcp.registry.Fingerprint.load",
new=AsyncMock(return_value=None),
),
patch("vibe.core.tools.mcp.registry.Fingerprint.delete", new=AsyncMock()),
):
first = await registry.get_tools_async([srv])
assert first == {}
assert (
"OAuth support for MCP servers is not yet enabled; coming in a future release"
in caplog.text
)
caplog.clear()
with caplog.at_level(logging.WARNING, logger="vibe"):
second = await registry.get_tools_async([srv])
assert first == {}
assert second == {}
assert "OAuth support" not in caplog.text
assert registry.needs_auth == {"linear"}
assert registry.status()["linear"] == AuthStatus.NEEDS_AUTH

3
tests/core/test_config_orchestrator.py
View file

@ -23,6 +23,9 @@ class FakeLayer(ConfigLayer[RawConfig]):
async def _build_config_snapshot(self) -> LayerConfigSnapshot:
return LayerConfigSnapshot(data=dict(self._data), fingerprint="fp")
async def _save_to_store(self, _next_config: RawConfig) -> str:
raise NotImplementedError
class SimpleSchema(ConfigSchema):
value: Annotated[str, WithReplaceMerge()] = "default"

15
tests/core/test_config_otel.py
View file

@ -2,6 +2,7 @@ from __future__ import annotations
import pytest
from tests.constants import ANTHROPIC_BASE_URL
from vibe.core.config import OtelSpanExporterConfig, ProviderConfig, VibeConfig
from vibe.core.types import Backend
@ -62,7 +63,7 @@ class TestOtelSpanExporterConfig:
update={
"providers": [
ProviderConfig(
name="anthropic", api_base="https://api.anthropic.com/v1"
name="anthropic", api_base=f"{ANTHROPIC_BASE_URL}/v1"
)
]
}
@ -80,6 +81,18 @@ class TestOtelSpanExporterConfig:
assert result is not None
assert result.endpoint == "https://api.mistral.ai/telemetry/v1/traces"
def test_resolves_api_key_from_keyring(
self, vibe_config: VibeConfig, monkeypatch: pytest.MonkeyPatch
) -> None:
# Key stored only in the OS keyring (no env var) must still authenticate OTEL.
monkeypatch.delenv("MISTRAL_API_KEY", raising=False)
monkeypatch.setattr(
"keyring.get_password", lambda service, username: "sk-keyring"
)
result = vibe_config.otel_span_exporter_config
assert result is not None
assert result.headers == {"Authorization": "Bearer sk-keyring"}
def test_returns_none_and_warns_when_api_key_missing(
self,
vibe_config: VibeConfig,

252
tests/core/test_config_patch_ops.py
View file

@ -1,135 +1,84 @@
from __future__ import annotations
from collections.abc import Callable
from dataclasses import FrozenInstanceError
from typing import Any, get_args
from pydantic import ValidationError
import pytest
from vibe.core.config import (
AppendToList,
DeleteField,
PatchOp,
RemoveFromList,
SetField,
AddOperationPatch,
RemoveOperationPatch,
ReplaceOperationPatch,
)
from vibe.core.config.patch import ConfigPatch
from vibe.core.config.patch import ConfigPatch, PatchOp
@pytest.mark.parametrize(
("operation", "expected"),
[
(
AddOperationPatch(path="/tools/disabled_tools/0", value="bash"),
{"op": "add", "path": "/tools/disabled_tools/0", "value": "bash"},
),
(
ReplaceOperationPatch(path="/active_model", value="devstral-small"),
{"op": "replace", "path": "/active_model", "value": "devstral-small"},
),
(
RemoveOperationPatch(path="/tools/deprecated_setting"),
{"op": "remove", "path": "/tools/deprecated_setting"},
),
],
)
def test_json_patch_operations_convert_to_json_patch_payload(
operation: AddOperationPatch | ReplaceOperationPatch | RemoveOperationPatch,
expected: dict[str, Any],
) -> None:
assert operation.to_json_patch() == expected
def test_patch_op_union_contains_all_operations() -> None:
assert get_args(PatchOp) == (SetField, AppendToList, RemoveFromList, DeleteField)
def test_set_field_accepts_top_level_key() -> None:
op = SetField("active_model", "devstral-small")
assert op.key == "active_model"
assert op.value == "devstral-small"
def test_set_field_accepts_nested_key() -> None:
op = SetField("models.providers", {"mistral": {"region": "eu"}})
assert op.key == "models.providers"
def test_append_to_list_accepts_nested_key() -> None:
op = AppendToList("tools.disabled_tools", ("bash", "python"))
assert op.key == "tools.disabled_tools"
assert op.items == ("bash", "python")
def test_remove_from_list_accepts_nested_key() -> None:
op = RemoveFromList("models.available_models", ("codestral-latest",))
assert op.key == "models.available_models"
assert op.values == ("codestral-latest",)
def test_delete_field_accepts_nested_key() -> None:
op = DeleteField("tools.deprecated_setting")
assert op.key == "tools.deprecated_setting"
assert set(get_args(PatchOp.__value__)) == {
AddOperationPatch,
ReplaceOperationPatch,
RemoveOperationPatch,
}
@pytest.mark.parametrize(
"factory",
[
lambda key: SetField(key, "value"),
lambda key: AppendToList(key, ("value",)),
lambda key: RemoveFromList(key, ("value",)),
lambda key: DeleteField(key),
lambda path: AddOperationPatch(path=path, value="value"),
lambda path: ReplaceOperationPatch(path=path, value="value"),
lambda path: RemoveOperationPatch(path=path),
],
)
@pytest.mark.parametrize(
"invalid_key", ["", ".active_model", "active_model.", "tools..bash"]
)
def test_patch_operations_reject_invalid_key_paths(
factory: Callable[[str], object], invalid_key: str
) -> None:
with pytest.raises(ValueError, match="dot-separated path|must not be empty"):
factory(invalid_key)
def test_json_patch_operations_reject_non_pointer_paths(factory: Any) -> None:
with pytest.raises(ValidationError, match="valid JSON Pointer"):
factory("tools.disabled_tools")
@pytest.mark.parametrize(
"factory",
[
lambda key: SetField(key, "value"),
lambda key: AppendToList(key, ("value",)),
lambda key: RemoveFromList(key, ("value",)),
lambda key: DeleteField(key),
lambda path: AddOperationPatch(path=path, value="value"),
lambda path: ReplaceOperationPatch(path=path, value="value"),
lambda path: RemoveOperationPatch(path=path),
],
)
def test_patch_operations_reject_non_string_keys(
factory: Callable[[Any], object],
) -> None:
with pytest.raises(TypeError, match="Patch operation key must be a string"):
factory(1)
def test_json_patch_operations_reject_invalid_escapes(factory: Any) -> None:
with pytest.raises(ValidationError, match="valid JSON Pointer"):
factory("/tools/~2")
def test_append_to_list_rejects_non_tuple_items() -> None:
bad_items: Any = ["bash"]
def test_json_patch_operations_accept_slash_prefixed_paths() -> None:
op = ReplaceOperationPatch(path="/", value={"active_model": "devstral-small"})
with pytest.raises(TypeError, match="AppendToList.items must be a tuple"):
AppendToList("tools.disabled_tools", bad_items)
def test_remove_from_list_rejects_non_tuple_values() -> None:
bad_values: Any = ["bash"]
with pytest.raises(TypeError, match="RemoveFromList.values must be a tuple"):
RemoveFromList("tools.disabled_tools", bad_values)
def test_patch_operations_are_frozen() -> None:
op = SetField("active_model", "devstral-small")
with pytest.raises(FrozenInstanceError):
op.__setattr__("key", "models.active_model")
def test_scenario_mini_vibe_patch_operations() -> None:
operations: list[PatchOp] = [
SetField("active_model", "devstral-small"),
AppendToList("tools.disabled_tools", ("bash",)),
RemoveFromList("models.available_models", ("codestral-latest",)),
DeleteField("tools.deprecated_setting"),
]
assert operations == [
SetField("active_model", "devstral-small"),
AppendToList("tools.disabled_tools", ("bash",)),
RemoveFromList("models.available_models", ("codestral-latest",)),
DeleteField("tools.deprecated_setting"),
]
# --- ConfigPatch ---
assert op.path == "/"
def test_config_patch_stores_operations_and_metadata() -> None:
op = SetField("active_model", "devstral-small")
op = ReplaceOperationPatch(path="/active_model", value="devstral-small")
patch = ConfigPatch(op, fingerprint="fp-1", reason="test")
assert patch.operations == [op]
@ -145,9 +94,9 @@ def test_config_patch_defaults() -> None:
def test_config_patch_accepts_multiple_operations() -> None:
ops: list[PatchOp] = [
SetField("active_model", "devstral-small"),
AppendToList("tools.disabled_tools", ("bash",)),
ops = [
ReplaceOperationPatch(path="/active_model", value="devstral-small"),
AddOperationPatch(path="/tools/disabled_tools/-", value="bash"),
]
patch = ConfigPatch(*ops, fingerprint="fp-1")
@ -155,18 +104,23 @@ def test_config_patch_accepts_multiple_operations() -> None:
def test_config_patch_add_appends_operations() -> None:
patch = ConfigPatch(SetField("active_model", "devstral-small"), fingerprint="fp-1")
patch.add(DeleteField("tools.deprecated_setting"))
patch = ConfigPatch(
ReplaceOperationPatch(path="/active_model", value="devstral-small"),
fingerprint="fp-1",
)
patch.add(RemoveOperationPatch(path="/tools/deprecated_setting"))
assert patch.operations == [
SetField("active_model", "devstral-small"),
DeleteField("tools.deprecated_setting"),
ReplaceOperationPatch(path="/active_model", value="devstral-small"),
RemoveOperationPatch(path="/tools/deprecated_setting"),
]
def test_config_patch_add_returns_self() -> None:
patch = ConfigPatch(fingerprint="fp-1")
result = patch.add(SetField("active_model", "devstral-small"))
result = patch.add(
ReplaceOperationPatch(path="/active_model", value="devstral-small")
)
assert result is patch
@ -174,8 +128,8 @@ def test_config_patch_add_returns_self() -> None:
def test_config_patch_add_accepts_multiple_operations() -> None:
patch = ConfigPatch(fingerprint="fp-1")
patch.add(
SetField("active_model", "devstral-small"),
AppendToList("tools.disabled_tools", ("bash",)),
ReplaceOperationPatch(path="/active_model", value="devstral-small"),
AddOperationPatch(path="/tools/disabled_tools/-", value="bash"),
)
assert len(patch.operations) == 2
@ -184,42 +138,52 @@ def test_config_patch_add_accepts_multiple_operations() -> None:
def test_config_patch_add_is_chainable() -> None:
patch = (
ConfigPatch(fingerprint="fp-1")
.add(SetField("active_model", "devstral-small"))
.add(DeleteField("tools.deprecated_setting"))
.add(ReplaceOperationPatch(path="/active_model", value="devstral-small"))
.add(RemoveOperationPatch(path="/tools/deprecated_setting"))
)
assert len(patch.operations) == 2
def test_config_patch_describe_set_field() -> None:
patch = ConfigPatch(SetField("active_model", "devstral-small"), fingerprint="fp-1")
assert patch.describe() == ["set 'active_model' = 'devstral-small'"]
def test_config_patch_describe_append_to_list() -> None:
def test_config_patch_to_json_patch_from_wrappers() -> None:
patch = ConfigPatch(
AppendToList("tools.disabled_tools", ("bash", "python")), fingerprint="fp-1"
)
assert patch.describe() == ["append to 'tools.disabled_tools': ['bash', 'python']"]
def test_config_patch_describe_remove_from_list() -> None:
patch = ConfigPatch(
RemoveFromList("models.available_models", ("codestral-latest",)),
ReplaceOperationPatch(path="/active_model", value="devstral-small"),
AddOperationPatch(path="/tools/disabled_tools/-", value="bash"),
RemoveOperationPatch(path="/tools/deprecated_setting"),
fingerprint="fp-1",
)
assert patch.describe() == [
"remove from 'models.available_models': ['codestral-latest']"
assert patch.to_json_patch() == [
{"op": "replace", "path": "/active_model", "value": "devstral-small"},
{"op": "add", "path": "/tools/disabled_tools/-", "value": "bash"},
{"op": "remove", "path": "/tools/deprecated_setting"},
]
def test_config_patch_describe_delete_field() -> None:
patch = ConfigPatch(DeleteField("tools.deprecated_setting"), fingerprint="fp-1")
def test_config_patch_describe_add_operation() -> None:
patch = ConfigPatch(
AddOperationPatch(path="/tools/disabled_tools/-", value="bash"),
fingerprint="fp-1",
)
assert patch.describe() == ["delete 'tools.deprecated_setting'"]
assert patch.describe() == ["add '/tools/disabled_tools/-' = 'bash'"]
def test_config_patch_describe_replace_operation() -> None:
patch = ConfigPatch(
ReplaceOperationPatch(path="/active_model", value="devstral-small"),
fingerprint="fp-1",
)
assert patch.describe() == ["replace '/active_model' = 'devstral-small'"]
def test_config_patch_describe_remove_operation() -> None:
patch = ConfigPatch(
RemoveOperationPatch(path="/tools/deprecated_setting"), fingerprint="fp-1"
)
assert patch.describe() == ["remove '/tools/deprecated_setting'"]
def test_config_patch_describe_empty_returns_empty_list() -> None:
@ -230,28 +194,28 @@ def test_config_patch_describe_empty_returns_empty_list() -> None:
def test_config_patch_describe_multiple_operations() -> None:
patch = ConfigPatch(
SetField("active_model", "devstral-small"),
AppendToList("tools.disabled_tools", ("bash",)),
DeleteField("tools.deprecated_setting"),
ReplaceOperationPatch(path="/active_model", value="devstral-small"),
AddOperationPatch(path="/tools/disabled_tools/-", value="bash"),
RemoveOperationPatch(path="/tools/deprecated_setting"),
fingerprint="fp-1",
)
assert patch.describe() == [
"set 'active_model' = 'devstral-small'",
"append to 'tools.disabled_tools': ['bash']",
"delete 'tools.deprecated_setting'",
"replace '/active_model' = 'devstral-small'",
"add '/tools/disabled_tools/-' = 'bash'",
"remove '/tools/deprecated_setting'",
]
def test_scenario_build_patch_incrementally() -> None:
patch = ConfigPatch(fingerprint="fp-abc", reason="/model command")
patch.add(SetField("active_model", "devstral-small"))
patch.add(AppendToList("tools.disabled_tools", ("bash",)))
patch.add(ReplaceOperationPatch(path="/active_model", value="devstral-small"))
patch.add(AddOperationPatch(path="/tools/disabled_tools/-", value="bash"))
assert patch.fingerprint == "fp-abc"
assert patch.reason == "/model command"
assert len(patch.operations) == 2
assert patch.describe() == [
"set 'active_model' = 'devstral-small'",
"append to 'tools.disabled_tools': ['bash']",
"replace '/active_model' = 'devstral-small'",
"add '/tools/disabled_tools/-' = 'bash'",
]

32
tests/core/test_fingerprint.py
View file

@ -4,7 +4,11 @@ from pathlib import Path
import pytest
from vibe.core.config.fingerprint import capture_stable_file, create_dict_fingerprint
from vibe.core.config.fingerprint import (
capture_stable_file,
create_dict_fingerprint,
create_file_fingerprint,
)
from vibe.core.config.types import ConcurrencyConflictError
@ -67,6 +71,32 @@ class TestCaptureStableFile:
pass
class TestCreateFileFingerprint:
def test_captures_file_state(self, tmp_working_directory: Path) -> None:
path = tmp_working_directory / "config.toml"
path.write_text("key = 1")
with path.open("rb") as file:
first_fingerprint = create_file_fingerprint(file)
with path.open("rb") as file:
second_fingerprint = create_file_fingerprint(file)
assert isinstance(first_fingerprint, str)
assert first_fingerprint
assert first_fingerprint == second_fingerprint
def test_changes_when_file_changes(self, tmp_working_directory: Path) -> None:
path = tmp_working_directory / "config.toml"
path.write_text("key = 1")
with path.open("rb") as file:
first_fingerprint = create_file_fingerprint(file)
path.write_text("key = 2")
with path.open("rb") as file:
assert create_file_fingerprint(file) != first_fingerprint
class TestCreateDictFingerprint:
def test_empty_dict_returns_stable_non_empty_token(self) -> None:
first_fingerprint = create_dict_fingerprint({})

58
tests/core/test_image_attachment.py Normal file
View file

@ -0,0 +1,58 @@
from __future__ import annotations
from pathlib import Path
from pydantic import ValidationError
import pytest
from vibe.core.types import FileImageSource, ImageAttachment, InlineImageSource
def test_migrates_legacy_flat_path_shape() -> None:
att = ImageAttachment.model_validate({
"path": "/tmp/a.png",
"alias": "a.png",
"mime_type": "image/png",
})
assert isinstance(att.source, FileImageSource)
assert att.source.path == Path("/tmp/a.png")
def test_migrates_legacy_flat_data_shape() -> None:
att = ImageAttachment.model_validate({
"data": "Zm9v",
"alias": "pasted.png",
"mime_type": "image/png",
})
assert isinstance(att.source, InlineImageSource)
assert att.source.data == "Zm9v"
def test_source_path_construction() -> None:
att = ImageAttachment(
source=FileImageSource(path=Path("/tmp/a.png")),
alias="a.png",
mime_type="image/png",
)
assert isinstance(att.source, FileImageSource)
assert att.source.path == Path("/tmp/a.png")
def test_file_source_round_trips_through_json() -> None:
att = ImageAttachment(
source=FileImageSource(path=Path("/tmp/a.png")),
alias="a.png",
mime_type="image/png",
)
dumped = att.model_dump(exclude_none=True, mode="json")
assert dumped["source"] == {"kind": "file", "path": "/tmp/a.png"}
assert ImageAttachment.model_validate(dumped) == att
def test_rejects_attachment_without_source() -> None:
with pytest.raises(ValidationError):
ImageAttachment.model_validate({"alias": "a.png", "mime_type": "image/png"})

45
tests/core/test_llm_message_merge.py
View file

@ -4,27 +4,43 @@ from pathlib import Path
import pytest
from vibe.core.types import ImageAttachment, LLMMessage, Role
from vibe.core.types import (
FileImageSource,
ImageAttachment,
LLMMessage,
Role,
UserDisplayContentMetadata,
)
@pytest.fixture()
def image_a(tmp_path: Path) -> ImageAttachment:
p = tmp_path / "a.png"
p.write_bytes(b"\x89PNG")
return ImageAttachment(path=p, alias="a.png", mime_type="image/png")
return ImageAttachment(
source=FileImageSource(path=p), alias="a.png", mime_type="image/png"
)
@pytest.fixture()
def image_b(tmp_path: Path) -> ImageAttachment:
p = tmp_path / "b.png"
p.write_bytes(b"\x89PNG")
return ImageAttachment(path=p, alias="b.png", mime_type="image/png")
return ImageAttachment(
source=FileImageSource(path=p), alias="b.png", mime_type="image/png"
)
def _msg(content: str, images: list[ImageAttachment] | None = None) -> LLMMessage:
return LLMMessage(role=Role.assistant, content=content, images=images)
def _display_content(host: str) -> UserDisplayContentMetadata:
return UserDisplayContentMetadata(
version="1.0.0", host=host, content=[{"type": "text", "text": host}]
)
def test_merge_prefers_self_images_when_present(
image_a: ImageAttachment, image_b: ImageAttachment
) -> None:
@ -49,3 +65,26 @@ def test_merge_preserves_explicit_empty_self_images_over_other(
def test_merge_yields_none_when_both_sides_are_none() -> None:
merged = _msg("hi") + _msg(" there")
assert merged.images is None
def test_merge_prefers_self_user_display_content_when_present() -> None:
self_display_content = _display_content("mistral-vscode")
other_display_content = _display_content("mistral-jetbrains")
merged = LLMMessage(
role=Role.user, content="hi", user_display_content=self_display_content
) + LLMMessage(
role=Role.user, content=" there", user_display_content=other_display_content
)
assert merged.user_display_content == self_display_content
def test_merge_falls_back_to_other_user_display_content() -> None:
other_display_content = _display_content("mistral-vscode")
merged = LLMMessage(role=Role.user, content="hi") + LLMMessage(
role=Role.user, content=" there", user_display_content=other_display_content
)
assert merged.user_display_content == other_display_content

57
tests/core/test_mcp_oauth.py
View file

@ -4,6 +4,8 @@ import asyncio
from collections.abc import Callable, Iterator
from contextlib import suppress
import socket
from types import TracebackType
from unittest.mock import patch
import urllib.parse
import httpx
@ -11,6 +13,7 @@ import keyring
from keyring.backend import KeyringBackend
import keyring.backends.fail
import keyring.errors
from mcp.client.auth import OAuthFlowError
from mcp.shared.auth import OAuthClientInformationFull, OAuthToken
import pytest
import respx
@ -21,6 +24,7 @@ from vibe.core.auth.mcp_oauth import (
LoopbackCallbackHandler,
MCPOAuthError,
MCPOAuthHeadlessError,
MCPOAuthLoginFailed,
MCPOAuthPortInUse,
build_oauth_provider,
perform_oauth_login,
@ -366,6 +370,26 @@ class TestBuildOAuthProvider:
)
assert provider.context.client_metadata_url == "https://vibe.example/cm.json"
@pytest.mark.asyncio
async def test_client_id_is_exposed_as_fallback_client_info(
self, memory_keyring: _MemoryKeyring
) -> None:
srv = _oauth_server(client_id="pre-registered-client")
async def on_url(_url: str) -> None:
return None
async def cb() -> tuple[str, str | None]:
return "code", None
provider = build_oauth_provider(
srv, redirect_handler=on_url, callback_handler=cb
)
client_info = await provider.context.storage.get_client_info()
assert client_info is not None
assert client_info.client_id == "pre-registered-client"
@pytest.mark.asyncio
async def test_rejects_static_auth(self, memory_keyring: _MemoryKeyring) -> None:
from vibe.core.config import MCPStaticAuth
@ -388,6 +412,39 @@ class TestBuildOAuthProvider:
class TestPerformOAuthLogin:
@pytest.mark.asyncio
async def test_oauth_flow_error_becomes_login_failed(
self, memory_keyring: _MemoryKeyring
) -> None:
srv = _oauth_server(name="demo")
class OAuthFlowFailingClient:
def __init__(self, *args: object, **kwargs: object) -> None:
pass
async def __aenter__(self) -> OAuthFlowFailingClient:
return self
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
traceback: TracebackType | None,
) -> None:
pass
async def get(self, _url: str) -> None:
raise OAuthFlowError("cancelled")
async def on_url(_url: str) -> None:
pass
with patch(
"vibe.core.auth.mcp_oauth.httpx.AsyncClient", new=OAuthFlowFailingClient
):
with pytest.raises(MCPOAuthLoginFailed, match="cancelled"):
await perform_oauth_login(srv, on_url=on_url)
@pytest.mark.asyncio
async def test_full_flow_persists_tokens_and_fingerprint(
self, memory_keyring: _MemoryKeyring

56
tests/core/test_message_list.py Normal file
View file

@ -0,0 +1,56 @@
from __future__ import annotations
from vibe.core.types import LLMMessage, MessageList, Role
def test_update_system_prompt_replaces_existing_system_slot() -> None:
messages = MessageList(
initial=[
LLMMessage(role=Role.system, content="old"),
LLMMessage(role=Role.user, content="hi"),
]
)
messages.update_system_prompt("new")
assert len(messages) == 2
assert messages[0].role == Role.system
assert messages[0].content == "new"
assert messages[1].content == "hi"
def test_update_system_prompt_inserts_without_clobbering_when_no_system() -> None:
messages = MessageList(
initial=[
LLMMessage(role=Role.user, content="Hello"),
LLMMessage(role=Role.assistant, content="Hi there!"),
]
)
messages.update_system_prompt("system prompt")
assert len(messages) == 3
assert messages[0].role == Role.system
assert messages[1].content == "Hello"
assert messages[2].content == "Hi there!"
def test_update_system_prompt_inserts_into_empty_list() -> None:
messages = MessageList()
messages.update_system_prompt("system prompt")
assert len(messages) == 1
assert messages[0].role == Role.system
def test_update_system_prompt_notifies_only_when_requested() -> None:
observed: list[LLMMessage] = []
messages = MessageList(observer=observed.append)
messages.update_system_prompt("silent")
assert observed == []
messages.update_system_prompt("loud", notify=True)
assert len(observed) == 1
assert observed[0].content == "loud"

8
tests/core/test_overrides_layer.py
View file

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

8
tests/core/test_project_config_layer.py
View file

@ -6,7 +6,6 @@ import pytest
from vibe.core.config.layer import UntrustedLayerError
from vibe.core.config.layers.project import ProjectConfigLayer
from vibe.core.config.patch import ConfigPatch
from vibe.core.config.types import MISSING_CONFIG_FILE_FINGERPRINT
from vibe.core.paths._vibe_home import GlobalPath
from vibe.core.trusted_folders import trusted_folders_manager
@ -73,13 +72,6 @@ async def test_default_name(tmp_working_directory: Path) -> None:
assert layer.name == "project-toml"
@pytest.mark.asyncio
async def test_apply_raises_not_implemented(tmp_working_directory: Path) -> None:
layer = ProjectConfigLayer(path=tmp_working_directory)
with pytest.raises(NotImplementedError, match="M2"):
await layer.apply(ConfigPatch(fingerprint="fp-1"))
@pytest.mark.asyncio
async def test_trust_uses_path_parent_for_resolution(
tmp_working_directory: Path,

937
tests/core/test_remote_agent_loop.py
View file

@ -1,937 +0,0 @@
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 (
CustomTaskCanceled,
CustomTaskCanceledAttributes,
CustomTaskCompleted,
CustomTaskCompletedAttributes,
CustomTaskInProgress,
CustomTaskInProgressAttributes,
CustomTaskStarted,
CustomTaskStartedAttributes,
JSONPatchAdd,
JSONPatchAppend,
JSONPatchPayload,
JSONPatchReplace,
JSONPayload,
)
from vibe.core.nuage.remote_events_source import RemoteEventsSource
from vibe.core.types import (
AssistantEvent,
ReasoningEvent,
Role,
ToolCallEvent,
ToolResultEvent,
ToolStreamEvent,
UserMessageEvent,
WaitingForInputEvent,
)
_EXEC_ID = "session-123"
def _make_loop(enabled_tools: list[str] | None = None) -> RemoteEventsSource:
config = build_test_vibe_config(enabled_tools=enabled_tools or [])
return RemoteEventsSource(session_id=_EXEC_ID, config=config)
def _started(
task_id: str, task_type: str, payload: dict[str, Any]
) -> CustomTaskStarted:
return CustomTaskStarted(
event_id=f"evt-{task_id}-start",
workflow_exec_id=_EXEC_ID,
attributes=CustomTaskStartedAttributes(
custom_task_id=task_id,
custom_task_type=task_type,
payload=JSONPayload(value=payload),
),
)
def _completed(
task_id: str, task_type: str, payload: dict[str, Any]
) -> CustomTaskCompleted:
return CustomTaskCompleted(
event_id=f"evt-{task_id}-done",
workflow_exec_id=_EXEC_ID,
attributes=CustomTaskCompletedAttributes(
custom_task_id=task_id,
custom_task_type=task_type,
payload=JSONPayload(value=payload),
),
)
def _in_progress(
task_id: str, task_type: str, patches: list[Any]
) -> CustomTaskInProgress:
return CustomTaskInProgress(
event_id=f"evt-{task_id}-progress",
workflow_exec_id=_EXEC_ID,
attributes=CustomTaskInProgressAttributes(
custom_task_id=task_id,
custom_task_type=task_type,
payload=JSONPatchPayload(value=patches),
),
)
def _canceled(task_id: str, task_type: str, reason: str = "") -> CustomTaskCanceled:
return CustomTaskCanceled(
event_id=f"evt-{task_id}-cancel",
workflow_exec_id=_EXEC_ID,
attributes=CustomTaskCanceledAttributes(
custom_task_id=task_id, custom_task_type=task_type, reason=reason
),
)
def test_consume_wait_for_input_event_emits_waiting_event() -> None:
loop = _make_loop()
event = _started(
"wait-task-1",
"wait_for_input",
{
"task_id": "wait-task-1",
"input_schema": {"title": "ChatInput"},
"label": "What next?",
},
)
emitted_events = loop._consume_workflow_event(event)
assert len(emitted_events) == 2
assistant_event = emitted_events[0]
waiting_event = emitted_events[1]
assert isinstance(assistant_event, AssistantEvent)
assert assistant_event.content == "What next?"
assert isinstance(waiting_event, WaitingForInputEvent)
assert waiting_event.task_id == "wait-task-1"
assert waiting_event.label == "What next?"
assert waiting_event.predefined_answers is None
def test_consume_agent_input_keeps_repeated_text_across_distinct_turns() -> None:
loop = _make_loop()
first_event = _completed(
"input-1", "AgentInputState", {"input": {"message": [{"text": "continue"}]}}
)
second_event = _completed(
"input-2", "AgentInputState", {"input": {"message": [{"text": "continue"}]}}
)
assert loop._consume_workflow_event(first_event) == []
assert loop._consume_workflow_event(second_event) == []
assert [msg.content for msg in loop.messages if msg.role == Role.user] == [
"continue",
"continue",
]
def test_wait_for_input_emits_predefined_answers_and_user_message() -> None:
loop = _make_loop()
started = _started(
"wait-task-1",
"wait_for_input",
{
"input_schema": {
"title": "ChatInput",
"properties": {
"message": {
"examples": [
[{"type": "text", "text": "Python"}],
[{"type": "text", "text": "JavaScript"}],
[{"type": "text", "text": "Other"}],
]
}
},
},
"label": "Which language?",
},
)
completed = _completed(
"wait-task-1",
"wait_for_input",
{"input": {"message": [{"type": "text", "text": "Python"}]}},
)
started_events = loop._consume_workflow_event(started)
completed_events = loop._consume_workflow_event(completed)
assistant_event = next(
event for event in started_events if isinstance(event, AssistantEvent)
)
waiting_event = next(
event for event in started_events if isinstance(event, WaitingForInputEvent)
)
assert assistant_event.content == "Which language?"
assert waiting_event.predefined_answers == ["Python", "JavaScript"]
user_event = next(
event for event in completed_events if isinstance(event, UserMessageEvent)
)
assert user_event.content == "Python"
def test_tool_events_update_stats_and_messages() -> None:
loop = _make_loop(enabled_tools=["todo"])
started = _started(
"tool-task-1",
"AgentToolCallState",
{"name": "todo", "tool_call_id": "call-1", "kwargs": {"action": "read"}},
)
completed = _completed(
"tool-task-1",
"AgentToolCallState",
{
"name": "todo",
"tool_call_id": "call-1",
"kwargs": {"action": "read"},
"output": {"total_count": 0},
},
)
started_events = loop._consume_workflow_event(started)
completed_events = loop._consume_workflow_event(completed)
assert any(isinstance(event, ToolCallEvent) for event in started_events)
result_event = next(
event for event in completed_events if isinstance(event, ToolResultEvent)
)
assert result_event.error is None
assert result_event.cancelled is False
assert result_event.tool_call_id == "call-1"
assert loop.stats.tool_calls_agreed == 1
assert loop.stats.tool_calls_succeeded == 1
assert loop.stats.tool_calls_failed == 0
tool_messages = [msg for msg in loop.messages if msg.role == Role.tool]
assert len(tool_messages) == 1
assert tool_messages[0].tool_call_id == "call-1"
def test_ask_user_question_tool_emits_assistant_question() -> 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": [
{
"question": "Which file type should I create?",
"options": [{"label": "Python"}, {"label": "JavaScript"}],
}
]
},
},
)
events = loop._consume_workflow_event(started)
assistant_event = next(
event for event in events if isinstance(event, AssistantEvent)
)
tool_call_event = next(
event for event in events if isinstance(event, ToolCallEvent)
)
assert assistant_event.content == "Which file type should I create?"
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(
"tool-task-question",
"AgentToolCallState",
{
"name": "ask_user_question",
"tool_call_id": "call-question",
"kwargs": {
"questions": [{"question": "Which type of file?", "options": []}]
},
},
)
wait_started = _started(
"wait-task-1",
"wait_for_input",
{"input_schema": {"title": "ChatInput"}, "label": "Which type of file?"},
)
wait_completed = _completed(
"wait-task-1",
"wait_for_input",
{
"input_schema": {"title": "ChatInput"},
"label": "Which type of file?",
"input": {"message": [{"type": "text", "text": "Python"}]},
},
)
loop._consume_workflow_event(ask_started)
loop._consume_workflow_event(wait_started)
completed_events = loop._consume_workflow_event(wait_completed)
tool_result = next(
(e for e in completed_events if isinstance(e, ToolResultEvent)), None
)
assert tool_result is not None
assert tool_result.tool_call_id == "call-question"
user_message = next(e for e in completed_events if isinstance(e, UserMessageEvent))
assert user_message.content == "Python"
def test_working_events_without_tool_call_id_render_remote_progress_row() -> None:
loop = _make_loop()
started = _started(
"working-1",
"working",
{"title": "Creating sandbox", "content": "initializing", "toolUIState": None},
)
completed = _completed(
"working-1",
"working",
{
"title": "Creating sandbox",
"content": "sandbox created",
"toolUIState": None,
},
)
started_events = loop._consume_workflow_event(started)
completed_events = loop._consume_workflow_event(completed)
assert started_events == []
assert any(isinstance(event, ToolCallEvent) for event in completed_events)
assert any(isinstance(event, ToolStreamEvent) for event in completed_events)
result_event = next(
event for event in completed_events if isinstance(event, ToolResultEvent)
)
assert result_event.tool_name == "Creating sandbox"
assert result_event.tool_call_id == "working-1"
def test_working_events_with_tool_call_id_wait_for_real_tool_call() -> None:
loop = _make_loop()
working_started = _started(
"working-tool-1",
"working",
{
"title": "Executing write_file",
"content": "writing file",
"toolUIState": {"toolCallId": "call-write"},
},
)
tool_started = _started(
"tool-task-1",
"AgentToolCallState",
{
"name": "write_file",
"tool_call_id": "call-write",
"kwargs": {
"path": "hello_world.js",
"content": "console.log('Hello, World!');",
},
},
)
working_events = loop._consume_workflow_event(working_started)
tool_events = loop._consume_workflow_event(tool_started)
assert any(isinstance(event, ToolCallEvent) for event in working_events)
assert any(isinstance(event, ToolStreamEvent) for event in working_events)
assert not any(isinstance(event, ToolCallEvent) for event in tool_events)
assert not any(isinstance(event, ToolResultEvent) for event in tool_events)
def test_working_task_promoted_to_real_tool_call_does_not_create_duplicate_row() -> (
None
):
loop = _make_loop(enabled_tools=["write_file"])
working_started = _started(
"working-tool-1",
"working",
{
"title": "Writing file",
"content": '# hello.py\n\nprint("Hello, World!")',
"toolUIState": None,
},
)
working_promoted = _completed(
"working-tool-1",
"working",
{
"title": "Executing write_file",
"content": "",
"toolUIState": {
"type": "file",
"toolCallId": "call-write",
"operations": [
{
"type": "create",
"uri": "/workspace/hello.py",
"content": 'print("Hello, World!")',
}
],
},
},
)
agent_tool_completed = _completed(
"tool-task-1",
"AgentToolCallState",
{
"name": "write_file",
"tool_call_id": "call-write",
"kwargs": {"path": "hello.py", "content": 'print("Hello, World!")'},
"output": {
"path": "/workspace/hello.py",
"bytes_written": 22,
"content": 'print("Hello, World!")',
},
},
)
assert loop._consume_workflow_event(working_started) == []
promoted_events = loop._consume_workflow_event(working_promoted)
assert len([e for e in promoted_events if isinstance(e, ToolCallEvent)]) == 1
assert not any(isinstance(e, ToolStreamEvent) for e in promoted_events)
assert any(isinstance(e, ToolResultEvent) for e in promoted_events)
completed_events = loop._consume_workflow_event(agent_tool_completed)
assert not any(isinstance(e, ToolCallEvent) for e in completed_events)
assert not any(isinstance(e, ToolResultEvent) for e in completed_events)
def test_idle_boundary_waits_for_open_tool_results() -> None:
loop = _make_loop(enabled_tools=["write_file"])
working_started = _started(
"working-tool-1",
"working",
{
"title": "Executing write_file",
"content": "writing file",
"toolUIState": {"toolCallId": "call-write"},
},
)
idle_candidate = _completed("input-task-1", "AgentInputState", {"input": None})
tool_completed = _completed(
"tool-task-1",
"AgentToolCallState",
{
"name": "write_file",
"tool_call_id": "call-write",
"kwargs": {
"path": "hello_world.js",
"content": "console.log('Hello, World!');",
},
"output": {
"path": "/workspace/hello_world.js",
"bytes_written": 29,
"content": "console.log('Hello, World!');",
},
},
)
idle_after_tool = _completed("input-task-2", "AgentInputState", {"input": None})
working_events = loop._consume_workflow_event(working_started)
assert any(isinstance(event, ToolCallEvent) for event in working_events)
loop._consume_workflow_event(idle_candidate)
assert loop._is_idle_boundary(idle_candidate) is False
tool_events = loop._consume_workflow_event(tool_completed)
assert not any(isinstance(event, ToolCallEvent) for event in tool_events)
assert any(isinstance(event, ToolResultEvent) for event in tool_events)
loop._consume_workflow_event(idle_after_tool)
assert loop._is_idle_boundary(idle_after_tool) is True
def test_send_user_message_tool_is_not_rendered() -> None:
loop = _make_loop()
started = _started(
"tool-task-send-user-message",
"AgentToolCallState",
{
"name": "send_user_message",
"tool_call_id": "call-send",
"kwargs": {"message": "hello"},
},
)
completed = _completed(
"tool-task-send-user-message",
"AgentToolCallState",
{
"name": "send_user_message",
"tool_call_id": "call-send",
"kwargs": {"message": "hello"},
"output": {"success": True, "error": None},
},
)
assert loop._consume_workflow_event(started) == []
assert loop._consume_workflow_event(completed) == []
def test_send_user_message_working_events_are_not_rendered() -> None:
loop = _make_loop()
started = _started(
"working-send-user-message",
"working",
{
"title": "Executing send_user_message",
"content": "Hello!",
"toolUIState": {"toolCallId": "call-send-working"},
},
)
completed = _completed(
"working-send-user-message",
"working",
{
"title": "Executing send_user_message",
"content": "Hello!",
"toolUIState": {"toolCallId": "call-send-working"},
},
)
assert loop._consume_workflow_event(started) == []
assert loop._consume_workflow_event(completed) == []
def test_remote_bash_uses_known_tool_display_even_when_disabled_locally() -> None:
loop = _make_loop(enabled_tools=["write_file"])
started = _started(
"tool-task-bash",
"AgentToolCallState",
{
"name": "bash",
"tool_call_id": "call-bash",
"kwargs": {"command": "cat hello.py | wc -c"},
},
)
completed = _completed(
"tool-task-bash",
"AgentToolCallState",
{
"name": "bash",
"tool_call_id": "call-bash",
"kwargs": {"command": "cat hello.py | wc -c"},
"output": {
"command": "cat hello.py | wc -c",
"stdout": "22\n",
"stderr": "",
"returncode": 0,
},
},
)
started_events = loop._consume_workflow_event(started)
completed_events = loop._consume_workflow_event(completed)
tool_call_event = next(
event for event in started_events if isinstance(event, ToolCallEvent)
)
result_event = next(
event for event in completed_events if isinstance(event, ToolResultEvent)
)
assert tool_call_event.tool_name == "bash"
assert tool_call_event.tool_class.get_name() == "bash"
assert tool_call_event.args is not None
assert tool_call_event.args.command == "cat hello.py | wc -c" # type: ignore[attr-defined]
assert result_event.result is not None
assert result_event.result.command == "cat hello.py | wc -c" # type: ignore[attr-defined]
assert result_event.result.stdout == "22\n" # type: ignore[attr-defined]
def test_canceled_tool_marks_cancelled_and_failed_stats() -> None:
loop = _make_loop(enabled_tools=["todo"])
loop._task_state["tool-task-2"] = {
"name": "todo",
"tool_call_id": "call-2",
"kwargs": {"action": "read"},
}
canceled = _canceled(
"tool-task-2", "AgentToolCallState", reason="user interrupted tool"
)
events = loop._consume_workflow_event(canceled)
result_event = next(event for event in events if isinstance(event, ToolResultEvent))
assert result_event.cancelled is True
assert result_event.error == "Canceled: user interrupted tool"
assert loop.stats.tool_calls_failed == 1
assert loop.stats.tool_calls_succeeded == 0
def test_working_thinking_type_emits_assistant_events() -> None:
loop = _make_loop()
started = _started(
"thinking-1",
"working",
{"type": "thinking", "title": "Thinking", "content": "", "toolUIState": None},
)
in_progress = _in_progress(
"thinking-1", "working", [JSONPatchAppend(path="/content", value="Hello!")]
)
completed = _completed(
"thinking-1",
"working",
{
"type": "thinking",
"title": "Thinking",
"content": "Hello!",
"toolUIState": None,
},
)
started_events = loop._consume_workflow_event(started)
progress_events = loop._consume_workflow_event(in_progress)
completed_events = loop._consume_workflow_event(completed)
assert started_events == []
assert len(progress_events) == 1
assert isinstance(progress_events[0], ReasoningEvent)
assert progress_events[0].content == "Hello!"
assert completed_events == []
def test_working_bash_progress_without_tool_call_id_streams_command_output() -> None:
loop = _make_loop(enabled_tools=["write_file"])
started = _started(
"working-bash-1",
"working",
{"type": "tool", "title": "Planning", "content": "", "toolUIState": None},
)
in_progress = _in_progress(
"working-bash-1",
"working",
[
JSONPatchAdd(
path="/toolUIState",
value={
"type": "command",
"command": "ls -la /workspace",
"result": {
"status": "success",
"output": "total 4\ndrwxrwxrwx 2 root root 4096 Mar 20 10:18 .\ndrwxr-xr-x 1 root root 80 Mar 20 10:18 ..\n",
},
},
),
JSONPatchReplace(path="/title", value="Executing bash"),
JSONPatchReplace(path="/content", value=""),
],
)
started_events = loop._consume_workflow_event(started)
progress_events = loop._consume_workflow_event(in_progress)
assert started_events == []
tool_call_event = next(
event for event in progress_events if isinstance(event, ToolCallEvent)
)
tool_stream_event = next(
event for event in progress_events if isinstance(event, ToolStreamEvent)
)
assert tool_call_event.tool_name == "bash"
assert tool_call_event.tool_class.get_name() == "bash"
assert tool_call_event.tool_call_id == "working-bash-1"
assert tool_stream_event.tool_name == "bash"
assert tool_stream_event.tool_call_id == "working-bash-1"
assert "command: ls -la /workspace" in tool_stream_event.message
assert "total 4" in tool_stream_event.message
assert "drwxrwxrwx 2 root root 4096" in tool_stream_event.message
def test_working_completed_with_tool_call_id_emits_tool_result() -> None:
loop = _make_loop(enabled_tools=["write_file"])
working_started = _started(
"working-tool-1",
"working",
{
"title": "Executing write_file",
"content": "",
"toolUIState": {"toolCallId": "call-write-solo"},
},
)
working_completed = _completed(
"working-tool-1",
"working",
{
"title": "Executing write_file",
"content": "",
"toolUIState": {
"type": "file",
"toolCallId": "call-write-solo",
"operations": [
{
"type": "create",
"uri": "/workspace/hello.py",
"content": 'print("Hello, World!")',
}
],
},
},
)
started_events = loop._consume_workflow_event(working_started)
assert any(isinstance(e, ToolCallEvent) for e in started_events)
completed_events = loop._consume_workflow_event(working_completed)
result_events = [e for e in completed_events if isinstance(e, ToolResultEvent)]
assert len(result_events) == 1
assert result_events[0].error is None
assert result_events[0].tool_call_id == "call-write-solo"
def test_working_completed_with_tool_call_id_emits_error_result() -> None:
loop = _make_loop(enabled_tools=["write_file"])
working_started = _started(
"working-tool-2",
"working",
{
"title": "Executing write_file",
"content": "",
"toolUIState": {"toolCallId": "call-write-err"},
},
)
working_completed = _completed(
"working-tool-2",
"working",
{
"title": "Executing write_file",
"content": "Error: Permission denied.",
"toolUIState": {
"type": "file",
"toolCallId": "call-write-err",
"operations": [],
},
},
)
loop._consume_workflow_event(working_started)
completed_events = loop._consume_workflow_event(working_completed)
result_events = [e for e in completed_events if isinstance(e, ToolResultEvent)]
assert len(result_events) == 1
assert result_events[0].error is not None
assert result_events[0].tool_call_id == "call-write-err"
def test_json_patch_with_array_index_preserves_list_structure() -> None:
loop = _make_loop()
started = _started(
"msg-1",
"assistant_message",
{"contentChunks": [{"type": "text", "text": "Hello"}]},
)
in_progress = _in_progress(
"msg-1",
"assistant_message",
[JSONPatchReplace(path="/contentChunks/0/text", value="Hello world")],
)
loop._consume_workflow_event(started)
progress_events = loop._consume_workflow_event(in_progress)
assert len(progress_events) == 1
assert isinstance(progress_events[0], AssistantEvent)
assert progress_events[0].content == " world"
def test_steer_input_events_are_suppressed() -> None:
loop = _make_loop()
steer_started = _started(
"steer-1",
"wait_for_input",
{"input_schema": {"title": "ChatInput"}, "label": "Send a message to steer..."},
)
steer_completed = _completed(
"steer-1",
"wait_for_input",
{
"input_schema": {"title": "ChatInput"},
"label": "Send a message to steer...",
"input": 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"

138
tests/core/test_resolve_api_key.py Normal file
View file

@ -0,0 +1,138 @@
from __future__ import annotations
import keyring
from keyring.errors import KeyringError
import pytest
from tests.conftest import build_test_vibe_config
from vibe.core.config import MissingAPIKeyError, ProviderConfig, resolve_api_key
from vibe.core.llm.backend.mistral import MistralBackend
from vibe.core.types import Backend
def test_resolve_returns_env_value_without_consulting_keyring(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("CUSTOM_API_KEY", "env-key")
def _fail(service: str, username: str) -> str | None:
raise AssertionError("keyring must not be consulted when env is set")
monkeypatch.setattr(keyring, "get_password", _fail)
assert resolve_api_key("CUSTOM_API_KEY") == "env-key"
def test_resolve_falls_back_to_keyring_when_env_unset(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.delenv("CUSTOM_API_KEY", raising=False)
monkeypatch.setattr(
keyring, "get_password", lambda service, username: "keyring-key"
)
assert resolve_api_key("CUSTOM_API_KEY") == "keyring-key"
def test_resolve_returns_none_when_env_unset_and_keyring_empty(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.delenv("CUSTOM_API_KEY", raising=False)
monkeypatch.setattr(keyring, "get_password", lambda service, username: None)
assert resolve_api_key("CUSTOM_API_KEY") is None
def test_resolve_returns_none_when_keyring_raises(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.delenv("CUSTOM_API_KEY", raising=False)
def _unavailable(service: str, username: str) -> str | None:
raise KeyringError("no keyring")
monkeypatch.setattr(keyring, "get_password", _unavailable)
assert resolve_api_key("CUSTOM_API_KEY") is None
def test_resolve_returns_none_for_empty_env_key(
monkeypatch: pytest.MonkeyPatch,
) -> None:
def _fail(service: str, username: str) -> str | None:
raise AssertionError("keyring must not be consulted for an empty env key")
monkeypatch.setattr(keyring, "get_password", _fail)
assert resolve_api_key("") is None
def test_check_api_key_accepts_keyring_only_key(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.delenv("MISTRAL_API_KEY", raising=False)
monkeypatch.setattr(
keyring, "get_password", lambda service, username: "keyring-key"
)
# Should not raise MissingAPIKeyError despite the env var being unset.
config = build_test_vibe_config()
assert config.get_active_provider().api_key_env_var == "MISTRAL_API_KEY"
def test_check_api_key_raises_when_neither_env_nor_keyring(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.delenv("MISTRAL_API_KEY", raising=False)
monkeypatch.setattr(keyring, "get_password", lambda service, username: None)
with pytest.raises(MissingAPIKeyError):
build_test_vibe_config()
def test_mistral_backend_reads_keyring_only_key(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.delenv("MISTRAL_API_KEY", raising=False)
monkeypatch.setattr(
keyring, "get_password", lambda service, username: "keyring-key"
)
provider = ProviderConfig(
name="mistral",
api_base="https://api.mistral.ai/v1",
api_key_env_var="MISTRAL_API_KEY",
backend=Backend.MISTRAL,
)
backend = MistralBackend(provider)
assert backend._api_key == "keyring-key"
def test_vibe_code_api_key_resolves_from_keyring(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.delenv("MISTRAL_API_KEY", raising=False)
monkeypatch.setattr(
keyring, "get_password", lambda service, username: "keyring-key"
)
config = build_test_vibe_config()
assert config.vibe_code_api_key == "keyring-key"
def test_vibe_code_api_key_empty_when_unresolved(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.delenv("MISTRAL_API_KEY", raising=False)
monkeypatch.setattr(
keyring, "get_password", lambda service, username: "keyring-key"
)
config = build_test_vibe_config()
# Nothing resolves the key anymore; the property must return "" (not None).
monkeypatch.setattr(keyring, "get_password", lambda service, username: None)
assert config.vibe_code_api_key == ""

10
tests/core/test_telemetry_send.py
View file

@ -510,16 +510,6 @@ class TestTelemetryClient:
"nb_session_messages": 4,
}
def test_send_remote_resume_requested_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_remote_resume_requested(session_id="remote-123")
assert len(telemetry_events) == 1
assert telemetry_events[0]["event_name"] == "vibe.remote_resume_requested"
assert telemetry_events[0]["properties"] == {"session_id": "remote-123"}
def test_send_teleport_failed_payload_includes_error_details(
self, telemetry_events: list[dict[str, Any]]
) -> None:

3
tests/core/test_teleport_nuage.py
View file

@ -5,6 +5,7 @@ import json
import httpx
import pytest
from tests.constants import TELEPORT_SESSIONS_PATH
from vibe.core.teleport.errors import ServiceTeleportError
from vibe.core.teleport.nuage import (
NuageClient,
@ -52,7 +53,7 @@ async def test_start_posts_nuage_request() -> None:
response = await nuage.start(_request())
assert seen_request is not None
assert str(seen_request.url) == "https://chat.example.com/api/v1/code/sessions"
assert str(seen_request.url) == f"https://chat.example.com{TELEPORT_SESSIONS_PATH}"
assert seen_request.headers["authorization"] == "Bearer api-key"
assert seen_request.headers["content-type"] == "application/json"
assert json.loads(seen_request.content) == {

11
tests/core/test_teleport_service.py
View file

@ -14,6 +14,7 @@ import pytest
import zstandard
from tests.conftest import build_test_vibe_config
from tests.constants import TELEPORT_COMPLETE_URL, TELEPORT_SESSIONS_PATH
from vibe.core.teleport.errors import (
ServiceTeleportError,
ServiceTeleportNotSupportedError,
@ -59,7 +60,7 @@ def _mock_handler() -> Any:
"webSessionId": "web-session-id",
"projectId": "project-id",
"status": "running",
"url": "https://chat.example.com/code/project-id/web-session-id",
"url": TELEPORT_COMPLETE_URL,
},
)
@ -218,7 +219,7 @@ class TestTeleportServiceExecute:
"webSessionId": "web-session-id",
"projectId": "project-id",
"status": "running",
"url": "https://chat.example.com/code/project-id/web-session-id",
"url": TELEPORT_COMPLETE_URL,
},
)
@ -247,10 +248,8 @@ class TestTeleportServiceExecute:
assert isinstance(events[0], TeleportCheckingGitEvent)
assert isinstance(events[1], TeleportStartingWorkflowEvent)
assert isinstance(events[2], TeleportCompleteEvent)
assert (
events[2].url == "https://chat.example.com/code/project-id/web-session-id"
)
assert seen_url == "https://chat.example.com/api/v1/code/sessions"
assert events[2].url == TELEPORT_COMPLETE_URL
assert seen_url == f"https://chat.example.com{TELEPORT_SESSIONS_PATH}"
assert seen_body is not None
assert seen_body["message"] == {
"role": "user",

25
tests/core/test_text.py
View file

@ -1,6 +1,6 @@
from __future__ import annotations
from vibe.core.utils.text import snippet_start_line
from vibe.core.utils.text import snippet_start_line, snippet_start_lines
class TestSnippetStartLine:
@ -27,3 +27,26 @@ class TestSnippetStartLine:
def test_blank_snippet(self) -> None:
assert snippet_start_line("hello", "\n") is None
class TestSnippetStartLines:
def test_single_occurrence(self) -> None:
assert snippet_start_lines("a\nb\nc", "b") == [2]
def test_all_occurrences(self) -> None:
assert snippet_start_lines("x\ny\nx\nz\nx", "x") == [1, 3, 5]
def test_repeated_on_same_line(self) -> None:
assert snippet_start_lines("x x x", "x") == [1, 1, 1]
def test_non_overlapping(self) -> None:
assert snippet_start_lines("aaaa", "aa") == [1, 1]
def test_multiline_snippet_occurrences(self) -> None:
assert snippet_start_lines("a\nb\nc\na\nb", "a\nb") == [1, 4]
def test_not_found(self) -> None:
assert snippet_start_lines("hello\nworld", "missing") == []
def test_blank_snippet(self) -> None:
assert snippet_start_lines("hello", "\n") == []

264
tests/core/test_user_config_layer.py
View file

@ -1,18 +1,31 @@
from __future__ import annotations
import os
from pathlib import Path
import tomllib
from uuid import uuid4
import pytest
from vibe.core.config.layer import LayerImplementationError
from vibe.core.config.fingerprint import create_file_fingerprint
from vibe.core.config.layer import LayerImplementationError, LayerNotLoadedError
from vibe.core.config.layers.user import UserConfigLayer
from vibe.core.config.patch import ConfigPatch
from vibe.core.config.patch import (
AddOperationPatch,
ConfigPatch,
RemoveOperationPatch,
ReplaceOperationPatch,
)
from vibe.core.config.types import MISSING_CONFIG_FILE_FINGERPRINT
def random_config_file_name() -> str:
return f"config-{uuid4().hex}.toml"
@pytest.mark.asyncio
async def test_reads_toml_file(tmp_working_directory: Path) -> None:
path = tmp_working_directory / "config.toml"
path = tmp_working_directory / random_config_file_name()
path.write_text('active_model = "mistral-large"\ncount = 42\n')
layer = UserConfigLayer(path=path, name="user-toml")
@ -25,7 +38,7 @@ async def test_reads_toml_file(tmp_working_directory: Path) -> None:
@pytest.mark.asyncio
async def test_always_trusted(tmp_working_directory: Path) -> None:
path = tmp_working_directory / "config.toml"
path = tmp_working_directory / random_config_file_name()
path.write_text('key = "value"\n')
layer = UserConfigLayer(path=path, name="user-toml")
@ -37,7 +50,7 @@ async def test_always_trusted(tmp_working_directory: Path) -> None:
@pytest.mark.asyncio
async def test_missing_file_returns_empty(tmp_working_directory: Path) -> None:
path = tmp_working_directory / "nonexistent.toml"
path = tmp_working_directory / random_config_file_name()
layer = UserConfigLayer(path=path, name="user-toml")
data = await layer.load()
assert data.model_extra == {}
@ -45,16 +58,241 @@ async def test_missing_file_returns_empty(tmp_working_directory: Path) -> None:
@pytest.mark.asyncio
async def test_apply_raises_not_implemented(tmp_working_directory: Path) -> None:
path = tmp_working_directory / "config.toml"
async def test_apply_raises_when_file_does_not_exist(
tmp_working_directory: Path,
) -> None:
path = tmp_working_directory / random_config_file_name()
layer = UserConfigLayer(path=path, name="user-toml")
with pytest.raises(NotImplementedError, match="M2"):
await layer.apply(ConfigPatch(fingerprint="fp-1"))
await layer.load()
assert layer.fingerprint == MISSING_CONFIG_FILE_FINGERPRINT
with pytest.raises(LayerNotLoadedError, match="loaded before applying patches"):
await layer.apply(
ConfigPatch(
AddOperationPatch(path="/active_model", value="mistral-large"),
fingerprint=MISSING_CONFIG_FILE_FINGERPRINT,
)
)
assert not path.exists()
@pytest.mark.asyncio
async def test_apply_sets_field_and_refreshes_cache(
tmp_working_directory: Path,
) -> None:
path = tmp_working_directory / random_config_file_name()
path.write_text("""\
active_model = "old"
[tools]
disabled_tools = ["bash", "python"]
deprecated_setting = true
""")
layer = UserConfigLayer(path=path, name="user-toml")
await layer.load()
fingerprint = layer.fingerprint
assert isinstance(fingerprint, str)
await layer.apply(
ConfigPatch(
ReplaceOperationPatch(path="/active_model", value="new"),
AddOperationPatch(path="/tools/enabled_tools", value=["read"]),
AddOperationPatch(path="/tools/disabled_tools/-", value="node"),
RemoveOperationPatch(path="/tools/disabled_tools/0"),
RemoveOperationPatch(path="/tools/deprecated_setting"),
fingerprint=fingerprint,
)
)
expected_data = {
"active_model": "new",
"tools": {"disabled_tools": ["python", "node"], "enabled_tools": ["read"]},
}
with path.open("rb") as file:
assert tomllib.load(file) == expected_data
cached_data = layer._state.data
assert cached_data is not None
assert cached_data.model_extra == expected_data
assert layer.fingerprint != fingerprint
@pytest.mark.asyncio
async def test_apply_cache_fingerprint_matches_written_file(
tmp_working_directory: Path,
) -> None:
path = tmp_working_directory / random_config_file_name()
path.write_text("")
layer = UserConfigLayer(path=path, name="user-toml")
await layer.load()
fingerprint = layer.fingerprint
assert isinstance(fingerprint, str)
await layer.apply(
ConfigPatch(
AddOperationPatch(path="/active_model", value="mistral-large"),
fingerprint=fingerprint,
)
)
with path.open("rb") as file:
assert layer.fingerprint == create_file_fingerprint(file)
@pytest.mark.asyncio
async def test_apply_uses_unique_temp_file(tmp_working_directory: Path) -> None:
path = tmp_working_directory / random_config_file_name()
fixed_tmp_path = tmp_working_directory / f".{path.name}.tmp"
path.write_text("")
fixed_tmp_path.write_text("stale")
layer = UserConfigLayer(path=path, name="user-toml")
await layer.load()
fingerprint = layer.fingerprint
assert isinstance(fingerprint, str)
await layer.apply(
ConfigPatch(
AddOperationPatch(path="/active_model", value="mistral-large"),
fingerprint=fingerprint,
)
)
assert fixed_tmp_path.read_text() == "stale"
assert list(tmp_working_directory.glob(f".{path.name}.*.tmp")) == []
with path.open("rb") as file:
assert tomllib.load(file) == {"active_model": "mistral-large"}
def test_atomic_replace_preserves_replacement_fingerprint(
tmp_working_directory: Path,
) -> None:
path = tmp_working_directory / random_config_file_name()
replacement = tmp_working_directory / f".{path.name}.tmp"
path.write_text("key = 1")
replacement.write_text("key = 2")
with replacement.open("rb") as file:
replacement_fingerprint = create_file_fingerprint(file)
os.replace(replacement, path)
with path.open("rb") as file:
assert create_file_fingerprint(file) == replacement_fingerprint
@pytest.mark.asyncio
async def test_apply_raises_when_layer_is_not_loaded(
tmp_working_directory: Path,
) -> None:
path = tmp_working_directory / random_config_file_name()
layer = UserConfigLayer(path=path, name="user-toml")
with pytest.raises(LayerNotLoadedError, match="loaded before applying patches"):
await layer.apply(
ConfigPatch(
AddOperationPatch(path="/active_model", value="mistral-large"),
fingerprint=MISSING_CONFIG_FILE_FINGERPRINT,
)
)
@pytest.mark.asyncio
async def test_apply_raises_when_cache_is_invalidated(
tmp_working_directory: Path,
) -> None:
path = tmp_working_directory / random_config_file_name()
path.write_text('active_model = "old"\n')
layer = UserConfigLayer(path=path, name="user-toml")
await layer.load()
fingerprint = layer.fingerprint
assert isinstance(fingerprint, str)
await layer.invalidate_cache()
with pytest.raises(LayerNotLoadedError, match="loaded before applying patches"):
await layer.apply(
ConfigPatch(
ReplaceOperationPatch(path="/active_model", value="new"),
fingerprint=fingerprint,
)
)
@pytest.mark.asyncio
async def test_apply_raises_when_parent_directory_does_not_exist(
tmp_working_directory: Path,
) -> None:
path = tmp_working_directory / "nested" / random_config_file_name()
layer = UserConfigLayer(path=path, name="user-toml")
await layer.load()
with pytest.raises(LayerNotLoadedError, match="loaded before applying patches"):
await layer.apply(
ConfigPatch(
AddOperationPatch(path="/active_model", value="mistral-large"),
fingerprint=MISSING_CONFIG_FILE_FINGERPRINT,
)
)
assert not path.exists()
@pytest.mark.asyncio
async def test_commit_sets_missing_nested_field(tmp_working_directory: Path) -> None:
path = tmp_working_directory / random_config_file_name()
path.write_text("[models]\n")
layer = UserConfigLayer(path=path, name="user-toml")
await layer.load()
fingerprint = layer.fingerprint
assert isinstance(fingerprint, str)
await layer.apply(
ConfigPatch(
AddOperationPatch(path="/models/active_model", value="mistral-large"),
fingerprint=fingerprint,
)
)
with path.open("rb") as file:
assert tomllib.load(file) == {"models": {"active_model": "mistral-large"}}
@pytest.mark.asyncio
async def test_apply_overwrites_external_file_changes(
tmp_working_directory: Path,
) -> None:
path = tmp_working_directory / random_config_file_name()
path.write_text('active_model = "old"\n')
layer = UserConfigLayer(path=path, name="user-toml")
await layer.load()
fingerprint = layer.fingerprint
assert isinstance(fingerprint, str)
path.write_text('active_model = "external"\n')
await layer.apply(
ConfigPatch(
ReplaceOperationPatch(path="/active_model", value="new"),
fingerprint=fingerprint,
)
)
with path.open("rb") as file:
assert tomllib.load(file) == {"active_model": "new"}
data = await layer.load()
assert data.model_extra == {"active_model": "new"}
@pytest.mark.asyncio
async def test_nested_toml_structure(tmp_working_directory: Path) -> None:
path = tmp_working_directory / "config.toml"
path = tmp_working_directory / random_config_file_name()
path.write_text("""\
[models]
active_model = "test"
@ -72,7 +310,7 @@ provider = "p"
@pytest.mark.asyncio
async def test_invalid_toml_raises(tmp_working_directory: Path) -> None:
path = tmp_working_directory / "bad.toml"
path = tmp_working_directory / random_config_file_name()
path.write_text("this is not valid = = = toml [[[")
layer = UserConfigLayer(path=path, name="user-toml")
with pytest.raises(LayerImplementationError, match="_build_config_snapshot"):
@ -81,7 +319,7 @@ async def test_invalid_toml_raises(tmp_working_directory: Path) -> None:
@pytest.mark.asyncio
async def test_force_reload_reads_fresh_data(tmp_working_directory: Path) -> None:
path = tmp_working_directory / "config.toml"
path = tmp_working_directory / random_config_file_name()
path.write_text('value = "first"\n')
layer = UserConfigLayer(path=path, name="user-toml")
@ -107,7 +345,7 @@ async def test_force_reload_reads_fresh_data(tmp_working_directory: Path) -> Non
@pytest.mark.asyncio
async def test_empty_toml_file(tmp_working_directory: Path) -> None:
path = tmp_working_directory / "empty.toml"
path = tmp_working_directory / random_config_file_name()
path.write_text("")
layer = UserConfigLayer(path=path, name="user-toml")
data = await layer.load()

131
tests/core/test_user_display_content.py Normal file
View file

@ -0,0 +1,131 @@
from __future__ import annotations
import math
from types import SimpleNamespace
from pydantic import ValidationError
import pytest
from vibe.core.types import LLMMessage, Role, UserDisplayContentMetadata
def _metadata() -> UserDisplayContentMetadata:
return UserDisplayContentMetadata(
version="1.0.0",
host="mistral-vscode",
content=[
{"type": "text", "text": "Look at "},
{
"type": "workspace_mention",
"kind": "file",
"uri": "file:///repo/src/app.ts",
"name": "app.ts",
},
],
)
def test_accepts_valid_metadata_and_preserves_host_owned_content() -> None:
metadata = UserDisplayContentMetadata.model_validate({
"version": "1.0.0",
"host": "mistral-vscode",
"content": [
{"type": "text", "text": "Look at "},
{
"type": "workspace_mention",
"kind": "file",
"uri": "file:///repo/src/app.ts",
"name": "app.ts",
"automatic": False,
"nested": {"line": 12, "tags": ["source", None]},
},
],
})
assert metadata.version == "1.0.0"
assert metadata.host == "mistral-vscode"
assert metadata.content == [
{"type": "text", "text": "Look at "},
{
"type": "workspace_mention",
"kind": "file",
"uri": "file:///repo/src/app.ts",
"name": "app.ts",
"automatic": False,
"nested": {"line": 12, "tags": ["source", None]},
},
]
def test_strips_metadata_strings_without_stripping_host_owned_content() -> None:
metadata = UserDisplayContentMetadata.model_validate({
"version": " 1.0.0 ",
"host": " mistral-vscode ",
"content": [{"type": "text", "text": " keep spaces "}],
})
assert metadata.version == "1.0.0"
assert metadata.host == "mistral-vscode"
assert metadata.content == [{"type": "text", "text": " keep spaces "}]
@pytest.mark.parametrize(
"payload",
[
{"version": " ", "host": "mistral-vscode", "content": []},
{"version": 2, "host": "mistral-vscode", "content": []},
{"version": "1.0.0", "host": " ", "content": []},
{"version": "1.0.0", "host": "mistral-vscode", "content": ["plain text"]},
{"version": "1.0.0", "host": "mistral-vscode", "content": {"type": "text"}},
{
"version": "1.0.0",
"host": "mistral-vscode",
"content": [{"type": "text"}],
"unexpected": True,
},
{
"version": "1.0.0",
"host": "mistral-vscode",
"content": [{"type": "text", "value": object()}],
},
{
"version": "1.0.0",
"host": "mistral-vscode",
"content": [{"type": "text", "value": math.nan}],
},
],
)
def test_rejects_invalid_metadata(payload: object) -> None:
with pytest.raises(ValidationError):
UserDisplayContentMetadata.model_validate(payload)
def test_llm_message_round_trips_user_display_content() -> None:
metadata = _metadata()
message = LLMMessage(
role=Role.user, content="Look at app.ts", user_display_content=metadata
)
dumped = message.model_dump(exclude_none=True, mode="json")
loaded = LLMMessage.model_validate(dumped)
assert dumped["user_display_content"] == metadata.model_dump(mode="json")
assert loaded.user_display_content == metadata
def test_llm_message_keeps_old_sessions_without_user_display_content_valid() -> None:
message = LLMMessage.model_validate({"role": "user", "content": "hello"})
assert message.user_display_content is None
def test_llm_message_object_adapter_preserves_user_display_content() -> None:
metadata = _metadata()
message = LLMMessage.model_validate(
SimpleNamespace(
role=Role.user, content="Look at app.ts", user_display_content=metadata
)
)
assert message.user_display_content == metadata

2
tests/core/test_vibe_config_schema.py
View file

@ -27,6 +27,7 @@ async def test_full_toml_to_vibe_config_schema(tmp_path: Path) -> None:
"""\
vim_keybindings = true
api_timeout = 300.0
api_retry_max_elapsed_time = 120.0
active_model = "codestral"
disabled_tools = ["bash"]
default_agent = "plan"
@ -54,6 +55,7 @@ provider = "mistral"
assert config.vim_keybindings is True
assert config.api_timeout == 300.0
assert config.api_retry_max_elapsed_time == 120.0
assert config.active_model == "codestral"
assert config.models[0].alias == "codestral"
assert "bash" in config.disabled_tools

60
tests/core/tools/builtins/test_edit.py
View file

@ -230,21 +230,21 @@ def test_get_result_display() -> None:
assert "foo.py" in display.message
def test_ui_start_line_not_part_of_model_contract() -> None:
def test_ui_start_lines_not_part_of_model_contract() -> None:
result = EditResult(file="/x", message="m", old_string="a", new_string="b")
result._ui_start_line = 42
result._ui_start_lines = [42]
assert result.ui_start_line == 42
assert "ui_start_line" not in result.model_dump()
assert "_ui_start_line" not in result.model_dump()
assert "ui_start_line" not in result.model_dump_json()
assert "ui_start_line" not in EditResult.model_fields
assert "ui_start_line" not in EditResult.model_json_schema().get("properties", {})
assert "ui_start_line" not in dict(result)
assert result.ui_start_lines == [42]
assert "ui_start_lines" not in result.model_dump()
assert "_ui_start_lines" not in result.model_dump()
assert "ui_start_lines" not in result.model_dump_json()
assert "ui_start_lines" not in EditResult.model_fields
assert "ui_start_lines" not in EditResult.model_json_schema().get("properties", {})
assert "ui_start_lines" not in dict(result)
@pytest.mark.asyncio
async def test_ui_start_line_computed_at_edit_site(
async def test_ui_start_lines_computed_at_edit_site(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
@ -255,11 +255,11 @@ async def test_ui_start_line_computed_at_edit_site(
edit.run(EditArgs(file_path="f.txt", old_string="gamma", new_string="GAMMA"))
)
assert result.ui_start_line == 3
assert result.ui_start_lines == [3]
@pytest.mark.asyncio
async def test_ui_start_line_set_for_pure_deletion(
async def test_ui_start_lines_set_for_pure_deletion(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
@ -270,4 +270,38 @@ async def test_ui_start_line_set_for_pure_deletion(
edit.run(EditArgs(file_path="f.txt", old_string="remove\n", new_string=""))
)
assert result.ui_start_line == 3
assert result.ui_start_lines == [3]
@pytest.mark.asyncio
async def test_ui_start_lines_lists_all_occurrences_for_replace_all(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
_write(tmp_path, "f.txt", "x\ntgt\ny\ntgt\nz\ntgt\n")
edit = _make_edit()
result = await collect_result(
edit.run(
EditArgs(
file_path="f.txt", old_string="tgt", new_string="TGT", replace_all=True
)
)
)
assert result.ui_start_lines == [2, 4, 6]
@pytest.mark.asyncio
async def test_ui_start_lines_single_entry_without_replace_all(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
_write(tmp_path, "f.txt", "a\nuniq\nb\n")
edit = _make_edit()
result = await collect_result(
edit.run(EditArgs(file_path="f.txt", old_string="uniq", new_string="UNIQ"))
)
assert result.ui_start_lines == [2]