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

v2.15.0 (#773)

Browse source 
Co-authored-by: Kim-Adeline Miguel <51720070+kimadeline@users.noreply.github.com>
Co-authored-by: Mathias Gesbert <mathias.gesbert@mistral.ai>
Co-authored-by: Maxime Dolores <maxime.dolores@ext.mistral.ai>
Co-authored-by: Nelson PROIA <144663685+Nelson-PROIA@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: Quentin <quentin.torroba@mistral.ai>
Co-authored-by: Val <102326092+vdeva@users.noreply.github.com>
Co-authored-by: Vincent G <10739306+VinceOPS@users.noreply.github.com>
Co-authored-by: Mistral Vibe <vibe@mistral.ai>
This commit is contained in:
Guillaume LE GOFF 2026-06-12 13:16:04 +02:00 • committed by GitHub
parent 702d0f412e
commit cafb6d4147
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
247 changed files with 12401 additions and 3029 deletions
Download patch file Download diff file Expand all files Collapse all files

71
tests/core/test_agents.py
View file

@ -111,8 +111,25 @@ class TestAgentManager:
include_prompt_detail=False,
disabled_agents=["plan"],
)
with pytest.raises(ValueError, match="not available"):
with pytest.raises(ValueError, match="disabled_agents") as exc_info:
AgentManager(lambda: config, initial_agent="plan")
message = str(exc_info.value)
assert "default_agent" not in message
assert message.startswith("Agent 'plan'")
def test_explicit_agent_excluded_by_enabled_agents_does_not_blame_default(
self,
) -> None:
config = build_test_vibe_config(
include_project_context=False,
include_prompt_detail=False,
enabled_agents=["default"],
)
with pytest.raises(ValueError, match="enabled_agents") as exc_info:
AgentManager(lambda: config, initial_agent="plan")
message = str(exc_info.value)
assert "default_agent" not in message
assert message.startswith("Agent 'plan'")
def test_initial_agent_raises_when_agent_does_not_exist(self) -> None:
config = build_test_vibe_config(
@ -120,3 +137,55 @@ class TestAgentManager:
)
with pytest.raises(ValueError, match="not found"):
AgentManager(lambda: config, initial_agent="nonexistent-agent")
def test_default_agent_excluded_by_enabled_agents_raises_config_contradiction(
self,
) -> None:
config = build_test_vibe_config(
include_project_context=False,
include_prompt_detail=False,
enabled_agents=["plan"],
)
with pytest.raises(ValueError, match="enabled_agents") as exc_info:
AgentManager(lambda: config)
message = str(exc_info.value)
assert "default" in message
assert "default_agent" in message
def test_default_agent_excluded_by_disabled_agents_raises_config_contradiction(
self,
) -> None:
config = build_test_vibe_config(
include_project_context=False,
include_prompt_detail=False,
disabled_agents=["default"],
)
with pytest.raises(ValueError, match="disabled_agents") as exc_info:
AgentManager(lambda: config)
assert "default_agent" in str(exc_info.value)
def test_disabled_agents_ignored_entirely_when_enabled_agents_set(
self, caplog: pytest.LogCaptureFixture
) -> None:
config = build_test_vibe_config(
include_project_context=False,
include_prompt_detail=False,
enabled_agents=["plan"],
disabled_agents=["plan"],
)
with caplog.at_level("WARNING"):
manager = AgentManager(lambda: config, initial_agent="plan")
assert manager.active_profile.name == "plan"
assert caplog.text == ""
def test_install_required_agent_reports_install_not_disabled_agents(self) -> None:
# 'lean' is install_required and enabled but not installed: the message
# must point to installation, not blame disabled_agents.
config = build_test_vibe_config(
include_project_context=False,
include_prompt_detail=False,
enabled_agents=["lean"],
)
with pytest.raises(ValueError, match="requires installation") as exc_info:
AgentManager(lambda: config, initial_agent="lean")
assert "disabled_agents" not in str(exc_info.value)

46
tests/core/test_auth_crypto.py
View file

@ -1,46 +0,0 @@
from __future__ import annotations
from dataclasses import asdict
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from vibe.core.auth import EncryptedPayload, decrypt, encrypt
def _generate_test_key_pair() -> tuple[bytes, bytes]:
private_key = rsa.generate_private_key(public_exponent=65537, key_size=4096)
private_pem = private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption(),
)
public_pem = private_key.public_key().public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo,
)
return private_pem, public_pem
class TestEncryptDecrypt:
def test_encrypt_decrypt_roundtrip(self) -> None:
private_pem, public_pem = _generate_test_key_pair()
plaintext = "ghp_test_token_12345"
encrypted = encrypt(plaintext, public_pem)
assert encrypted.encrypted_key != plaintext
assert encrypted.ciphertext != plaintext
decrypted = decrypt(encrypted, private_pem)
assert decrypted == plaintext
def test_encrypted_payload_serialization(self) -> None:
payload = EncryptedPayload(
encrypted_key="enc_key", nonce="nonce123", ciphertext="cipher"
)
data = asdict(payload)
restored = EncryptedPayload(**data)
assert restored == payload

286
tests/core/test_auth_github.py
View file

@ -1,286 +0,0 @@
from __future__ import annotations
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
import pytest
from vibe.core.auth.github import (
DeviceFlowHandle,
DeviceFlowInfo,
GitHubAuthError,
GitHubAuthProvider,
)
class TestDeviceFlowModels:
def test_device_flow_info(self) -> None:
info = DeviceFlowInfo(
user_code="ABC-123", verification_uri="https://example.com"
)
assert info.user_code == "ABC-123"
assert info.verification_uri == "https://example.com"
def test_device_flow_handle(self) -> None:
info = DeviceFlowInfo(
user_code="ABC-123", verification_uri="https://example.com"
)
handle = DeviceFlowHandle(device_code="dc_123", expires_in=900, info=info)
assert handle.device_code == "dc_123"
assert handle.expires_in == 900
assert handle.info.user_code == "ABC-123"
class TestGitHubAuthProviderContextManager:
@pytest.mark.asyncio
async def test_creates_client_on_enter(self) -> None:
provider = GitHubAuthProvider()
assert provider._client is None
async with provider:
assert provider._client is not None
assert provider._client is None
@pytest.mark.asyncio
async def test_uses_provided_client(self) -> None:
external_client = httpx.AsyncClient()
provider = GitHubAuthProvider(client=external_client)
async with provider:
assert provider._client is external_client
assert provider._client is external_client
await external_client.aclose()
class TestGitHubAuthProviderGetToken:
def test_returns_token_from_keyring(self) -> None:
with patch("vibe.core.auth.github.keyring") as mock_keyring:
mock_keyring.get_password.return_value = "ghp_test_token"
provider = GitHubAuthProvider()
token = provider.get_token()
assert token == "ghp_test_token"
def test_returns_none_on_keyring_error(self) -> None:
with patch("vibe.core.auth.github.keyring") as mock_keyring:
import keyring.errors
mock_keyring.errors = keyring.errors
mock_keyring.get_password.side_effect = keyring.errors.KeyringError("error")
provider = GitHubAuthProvider()
token = provider.get_token()
assert token is None
def test_returns_none_when_no_token(self) -> None:
with patch("vibe.core.auth.github.keyring") as mock_keyring:
mock_keyring.get_password.return_value = None
provider = GitHubAuthProvider()
token = provider.get_token()
assert token is None
class TestGitHubAuthProviderHasToken:
def test_returns_true_when_token_exists(self) -> None:
with patch("vibe.core.auth.github.keyring") as mock_keyring:
mock_keyring.get_password.return_value = "ghp_token"
provider = GitHubAuthProvider()
assert provider.has_token() is True
def test_returns_false_when_no_token(self) -> None:
with patch("vibe.core.auth.github.keyring") as mock_keyring:
mock_keyring.get_password.return_value = None
provider = GitHubAuthProvider()
assert provider.has_token() is False
class TestGitHubAuthProviderStartDeviceFlow:
@pytest.fixture
def mock_client(self) -> MagicMock:
return MagicMock(spec=httpx.AsyncClient)
@pytest.fixture
def provider(self, mock_client: MagicMock) -> GitHubAuthProvider:
return GitHubAuthProvider(client=mock_client)
@pytest.mark.asyncio
async def test_start_device_flow_success(
self, provider: GitHubAuthProvider, mock_client: MagicMock
) -> None:
mock_response = MagicMock()
mock_response.is_success = True
mock_response.json.return_value = {
"device_code": "dc_123",
"user_code": "ABC-123",
"verification_uri": "https://github.com/login/device",
"expires_in": 900,
}
mock_client.post = AsyncMock(return_value=mock_response)
with patch("vibe.core.auth.github.webbrowser") as mock_browser:
handle = await provider.start_device_flow(open_browser=True)
mock_browser.open.assert_called_once_with("https://github.com/login/device")
assert handle.device_code == "dc_123"
assert handle.info.user_code == "ABC-123"
assert handle.expires_in == 900
@pytest.mark.asyncio
async def test_start_device_flow_without_browser(
self, provider: GitHubAuthProvider, mock_client: MagicMock
) -> None:
mock_response = MagicMock()
mock_response.is_success = True
mock_response.json.return_value = {
"device_code": "dc_123",
"user_code": "ABC-123",
"verification_uri": "https://github.com/login/device",
"expires_in": 900,
}
mock_client.post = AsyncMock(return_value=mock_response)
with patch("vibe.core.auth.github.webbrowser") as mock_browser:
await provider.start_device_flow(open_browser=False)
mock_browser.open.assert_not_called()
@pytest.mark.asyncio
async def test_start_device_flow_failure(
self, provider: GitHubAuthProvider, mock_client: MagicMock
) -> None:
mock_response = MagicMock()
mock_response.is_success = False
mock_response.text = "Bad request"
mock_client.post = AsyncMock(return_value=mock_response)
with pytest.raises(GitHubAuthError, match="Failed to initiate device flow"):
await provider.start_device_flow()
class TestGitHubAuthProviderPollForToken:
@pytest.fixture
def mock_client(self) -> MagicMock:
return MagicMock(spec=httpx.AsyncClient)
@pytest.fixture
def provider(self, mock_client: MagicMock) -> GitHubAuthProvider:
return GitHubAuthProvider(client=mock_client)
@pytest.mark.asyncio
async def test_poll_returns_token_on_success(
self, provider: GitHubAuthProvider, mock_client: MagicMock
) -> None:
mock_response = MagicMock()
mock_response.json.return_value = {"access_token": "ghp_new_token"}
mock_client.post = AsyncMock(return_value=mock_response)
with patch("vibe.core.auth.github.asyncio.sleep", new_callable=AsyncMock):
token = await provider._poll_for_token(
mock_client, "dc_123", expires_in=10, interval=1
)
assert token == "ghp_new_token"
@pytest.mark.asyncio
async def test_poll_handles_slow_down(
self, provider: GitHubAuthProvider, mock_client: MagicMock
) -> None:
responses = [
MagicMock(
json=MagicMock(return_value={"error": "slow_down", "interval": 5})
),
MagicMock(json=MagicMock(return_value={"access_token": "ghp_token"})),
]
mock_client.post = AsyncMock(side_effect=responses)
with patch("vibe.core.auth.github.asyncio.sleep", new_callable=AsyncMock):
token = await provider._poll_for_token(
mock_client, "dc_123", expires_in=30, interval=1
)
assert token == "ghp_token"
@pytest.mark.asyncio
async def test_poll_raises_on_expired_token(
self, provider: GitHubAuthProvider, mock_client: MagicMock
) -> None:
mock_response = MagicMock()
mock_response.json.return_value = {"error": "expired_token"}
mock_client.post = AsyncMock(return_value=mock_response)
with patch("vibe.core.auth.github.asyncio.sleep", new_callable=AsyncMock):
with pytest.raises(GitHubAuthError, match="expired_token"):
await provider._poll_for_token(
mock_client, "dc_123", expires_in=10, interval=1
)
@pytest.mark.asyncio
async def test_poll_raises_on_access_denied(
self, provider: GitHubAuthProvider, mock_client: MagicMock
) -> None:
mock_response = MagicMock()
mock_response.json.return_value = {"error": "access_denied"}
mock_client.post = AsyncMock(return_value=mock_response)
with patch("vibe.core.auth.github.asyncio.sleep", new_callable=AsyncMock):
with pytest.raises(GitHubAuthError, match="access_denied"):
await provider._poll_for_token(
mock_client, "dc_123", expires_in=10, interval=1
)
@pytest.mark.asyncio
async def test_poll_raises_on_timeout(
self, provider: GitHubAuthProvider, mock_client: MagicMock
) -> None:
mock_response = MagicMock()
mock_response.json.return_value = {"error": "authorization_pending"}
mock_client.post = AsyncMock(return_value=mock_response)
with patch("vibe.core.auth.github.asyncio.sleep", new_callable=AsyncMock):
with pytest.raises(GitHubAuthError, match="timed out"):
await provider._poll_for_token(
mock_client, "dc_123", expires_in=2, interval=1
)
class TestGitHubAuthProviderSaveToken:
def test_save_token_success(self) -> None:
with patch("vibe.core.auth.github.keyring") as mock_keyring:
provider = GitHubAuthProvider()
provider._save_token("ghp_token")
mock_keyring.set_password.assert_called_once_with(
"vibe", "github_token", "ghp_token"
)
def test_save_token_raises_on_keyring_error(self) -> None:
with patch("vibe.core.auth.github.keyring") as mock_keyring:
import keyring.errors
mock_keyring.errors = keyring.errors
mock_keyring.set_password.side_effect = keyring.errors.KeyringError("error")
provider = GitHubAuthProvider()
with pytest.raises(GitHubAuthError, match="Failed to save token"):
provider._save_token("ghp_token")
class TestGitHubAuthProviderWaitForToken:
@pytest.fixture
def mock_client(self) -> MagicMock:
return MagicMock(spec=httpx.AsyncClient)
@pytest.fixture
def provider(self, mock_client: MagicMock) -> GitHubAuthProvider:
return GitHubAuthProvider(client=mock_client)
@pytest.mark.asyncio
async def test_wait_for_token_polls_and_saves(
self, provider: GitHubAuthProvider, mock_client: MagicMock
) -> None:
mock_response = MagicMock()
mock_response.json.return_value = {"access_token": "ghp_token"}
mock_client.post = AsyncMock(return_value=mock_response)
info = DeviceFlowInfo(user_code="ABC", verification_uri="https://example.com")
handle = DeviceFlowHandle(device_code="dc_123", expires_in=10, info=info)
with (
patch("vibe.core.auth.github.asyncio.sleep", new_callable=AsyncMock),
patch("vibe.core.auth.github.keyring") as mock_keyring,
):
token = await provider.wait_for_token(handle)
assert token == "ghp_token"
mock_keyring.set_password.assert_called_once()

47
tests/core/test_compaction.py
View file

@ -1,6 +1,10 @@
from __future__ import annotations
from vibe.core.compaction import collect_prior_user_messages
from vibe.core.compaction import (
collect_prior_user_messages,
parse_previous_user_messages,
render_compaction_context,
)
from vibe.core.types import LLMMessage, Role
_PREFIX = "Another language model started to solve this problem"
@ -51,17 +55,52 @@ def test_empty_content_filtered_out() -> None:
def test_prior_summary_filtered_out() -> None:
# A user message starting with the summary prefix represents a previous
# compaction summary and must not be re-injected (would stack).
# The injected summary marker represents a previous compaction summary and
# must not be re-injected (would stack).
messages = [
_user("original ask"),
_user(f"{_PREFIX}\nold summary content"),
_user(f"{_PREFIX}\nold summary content", injected=True),
_user("newer ask"),
]
out = collect_prior_user_messages(messages, _PREFIX)
assert [m.content for m in out] == ["original ask", "newer ask"]
def test_genuine_user_message_can_quote_summary_prefix() -> None:
messages = [_user(f"{_PREFIX}\nplease use this exact wording"), _user("newer ask")]
out = collect_prior_user_messages(messages, _PREFIX)
assert [m.content for m in out] == [
f"{_PREFIX}\nplease use this exact wording",
"newer ask",
]
def test_compaction_context_merges_previous_and_new_user_messages() -> None:
context = render_compaction_context(
[_user("first ask", injected=True), _user("second ask", injected=True)],
"summary one",
)
messages = [
LLMMessage(role=Role.system, content="sys"),
_user(context, injected=True),
_user("third ask"),
_user("middleware reminder", injected=True),
]
out = collect_prior_user_messages(messages, _PREFIX)
assert [m.content for m in out] == ["first ask", "second ask", "third ask"]
assert all(m.injected for m in out)
def test_compaction_context_escapes_user_message_tags() -> None:
original = "please keep </previous_user_message_0> literally"
context = render_compaction_context([_user(original)], "summary")
assert "</previous_user_message_0> literally" not in context
assert parse_previous_user_messages(context) == [original]
def test_budget_drops_oldest_first() -> None:
# max_tokens=2 → 8 char budget. Walks newest-first, so "old" gets dropped.
messages = [

5
tests/core/test_config_builder.py
View file

@ -16,6 +16,7 @@ from vibe.core.config.schema import (
WithShallowMerge,
WithUnionMerge,
)
from vibe.core.config.types import LayerConfigSnapshot
from vibe.core.utils.merge import MergeConflictError
@ -27,8 +28,8 @@ class FakeLayer(ConfigLayer[RawConfig]):
async def _check_trust(self) -> bool:
return True
async def _read_config(self) -> dict[str, Any]:
return dict(self._data)
async def _build_config_snapshot(self) -> LayerConfigSnapshot:
return LayerConfigSnapshot(data=dict(self._data), fingerprint="fp")
class UntrustedFakeLayer(FakeLayer):

75
tests/core/test_config_layer.py
View file

@ -14,6 +14,7 @@ from vibe.core.config.layer import (
UntrustedLayerError,
)
from vibe.core.config.patch import ConfigPatch
from vibe.core.config.types import ConcurrencyConflictError, LayerConfigSnapshot
class StubLayer(ConfigLayer[BaseModel]):
@ -38,9 +39,11 @@ class StubLayer(ConfigLayer[BaseModel]):
async def _check_trust(self) -> bool:
return self._stub_trusted
async def _read_config(self) -> dict[str, Any]:
async def _build_config_snapshot(self) -> LayerConfigSnapshot:
self.read_count += 1
return dict(self._data)
return LayerConfigSnapshot(
data=dict(self._data), fingerprint=f"fp-{self.read_count}"
)
class ObservableStubLayer(StubLayer):
@ -59,7 +62,7 @@ class SampleSchema(BaseModel):
count: int = 0
def test_abstract_read_config_enforced() -> None:
def test_abstract_build_config_snapshot_enforced() -> None:
class IncompleteLayer(ConfigLayer[BaseModel]):
pass
@ -72,11 +75,21 @@ def test_repr() -> None:
assert repr(layer) == "StubLayer(name='my-layer')"
def test_layer_config_snapshot_strips_fingerprint() -> None:
snapshot = LayerConfigSnapshot(data={}, fingerprint=" fp ")
assert snapshot.fingerprint == "fp"
def test_layer_config_snapshot_rejects_empty_fingerprint() -> None:
with pytest.raises(ValidationError):
LayerConfigSnapshot(data={}, fingerprint=" ")
@pytest.mark.asyncio
async def test_default_check_trust_returns_false() -> None:
class DefaultTrustLayer(ConfigLayer[BaseModel]):
async def _read_config(self) -> dict[str, Any]:
return {}
async def _build_config_snapshot(self) -> LayerConfigSnapshot:
return LayerConfigSnapshot(data={}, fingerprint="fp")
layer = DefaultTrustLayer(name="default")
result = await layer.resolve_trust()
@ -224,6 +237,7 @@ async def test_load_returns_data() -> None:
result = await layer.load()
assert isinstance(result, RawConfig)
assert result.model_extra == {"key": "value"}
assert layer.fingerprint == "fp-1"
@pytest.mark.asyncio
@ -233,6 +247,10 @@ async def test_load_auto_resolves_trust() -> None:
result = await layer.load()
assert layer.is_trusted is True
assert result.model_extra == {"a": 1}
assert layer.fingerprint == "fp-1"
await layer.resolve_trust()
assert layer.fingerprint == "fp-1"
@pytest.mark.asyncio
@ -251,6 +269,7 @@ async def test_load_force_bypasses_cache() -> None:
assert layer.read_count == 1
await layer.load(force=True)
assert layer.read_count == 2
assert layer.fingerprint == "fp-2"
@pytest.mark.asyncio
@ -312,6 +331,7 @@ async def test_resolve_trust_clears_data_on_revocation() -> None:
layer._stub_trusted = False
await layer.resolve_trust()
assert layer.is_trusted is False
assert layer.fingerprint is None
# Re-trust and update backing data while revoked
layer._stub_trusted = True
@ -335,17 +355,30 @@ async def test_load_returns_deep_copy() -> None:
@pytest.mark.asyncio
async def test_read_config_failure_wrapped() -> None:
async def test_build_config_snapshot_failure_wrapped() -> None:
class BrokenReadLayer(StubLayer):
async def _read_config(self) -> dict[str, Any]:
async def _build_config_snapshot(self) -> LayerConfigSnapshot:
raise OSError("config file missing")
layer = BrokenReadLayer()
with pytest.raises(LayerImplementationError, match="_read_config") as exc_info:
with pytest.raises(
LayerImplementationError, match="_build_config_snapshot"
) as exc_info:
await layer.load()
assert isinstance(exc_info.value.__cause__, IOError)
@pytest.mark.asyncio
async def test_build_config_snapshot_concurrency_conflict_propagates() -> None:
class ConflictingReadLayer(StubLayer):
async def _build_config_snapshot(self) -> LayerConfigSnapshot:
raise ConcurrencyConflictError(expected_fp="before", actual_fp="after")
layer = ConflictingReadLayer()
with pytest.raises(ConcurrencyConflictError):
await layer.load()
@pytest.mark.asyncio
async def test_default_schema_preserves_extras() -> None:
layer = StubLayer(data={"anything": "goes"})
@ -364,10 +397,13 @@ async def test_custom_schema_validates() -> None:
@pytest.mark.asyncio
async def test_invalid_data_raises_validation_error() -> None:
async def test_invalid_data_raises_layer_implementation_error() -> None:
layer = StubLayer(output_schema=SampleSchema, data={"count": "bad"})
with pytest.raises(ValidationError):
with pytest.raises(
LayerImplementationError, match="_build_config_snapshot"
) as exc_info:
await layer.load()
assert isinstance(exc_info.value.__cause__, ValidationError)
@pytest.mark.asyncio
@ -380,10 +416,12 @@ async def test_concurrent_loads_serialize() -> None:
async def _check_trust(self) -> bool:
return True
async def _read_config(self) -> dict[str, Any]:
async def _build_config_snapshot(self) -> LayerConfigSnapshot:
self.read_count += 1
await asyncio.sleep(0.05)
return {"v": self.read_count}
return LayerConfigSnapshot(
data={"v": self.read_count}, fingerprint=f"fp-{self.read_count}"
)
layer = SlowLayer()
results = await asyncio.gather(layer.load(), layer.load(), layer.load())
@ -392,10 +430,9 @@ async def test_concurrent_loads_serialize() -> None:
@pytest.mark.asyncio
async def test_get_fingerprint_not_implemented() -> None:
async def test_fingerprint_returns_none_before_load() -> None:
layer = StubLayer()
with pytest.raises(NotImplementedError):
await layer.get_fingerprint()
assert layer.fingerprint is None
@pytest.mark.asyncio
@ -459,8 +496,8 @@ class FakeLocalUserLayer(ConfigLayer[UserConfigSchema]):
async def _check_trust(self) -> bool:
return True
async def _read_config(self) -> dict[str, Any]:
return dict(self._data)
async def _build_config_snapshot(self) -> LayerConfigSnapshot:
return LayerConfigSnapshot(data=dict(self._data), fingerprint="user-fp")
@pytest.mark.asyncio
@ -504,8 +541,8 @@ class FakeLocalProjectLayer(ConfigLayer[BaseModel]):
else:
self._trust_store.pop(self._project_path, None)
async def _read_config(self) -> dict[str, Any]:
return dict(self._data)
async def _build_config_snapshot(self) -> LayerConfigSnapshot:
return LayerConfigSnapshot(data=dict(self._data), fingerprint="project-fp")
@pytest.mark.asyncio

213
tests/core/test_config_mcp_auth.py Normal file
View file

@ -0,0 +1,213 @@
from __future__ import annotations
import logging
from pydantic import ValidationError
import pytest
from vibe.core.config import MCPHttp, MCPOAuth, MCPStaticAuth, MCPStreamableHttp
from vibe.core.tools.mcp.registry import MCPRegistry
HTTP_TRANSPORTS = [
pytest.param(MCPHttp, "http", id="http"),
pytest.param(MCPStreamableHttp, "streamable-http", id="streamable-http"),
]
@pytest.mark.parametrize(("cls", "transport"), HTTP_TRANSPORTS)
def test_legacy_top_level_keys_promote_to_static_auth(
cls: type[MCPHttp | MCPStreamableHttp],
transport: str,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("MCP_TOKEN", "secret-token")
srv = cls.model_validate({
"name": "remote",
"transport": transport,
"url": "https://mcp.example.com",
"api_key_env": "MCP_TOKEN",
})
assert isinstance(srv.auth, MCPStaticAuth)
assert srv.auth.api_key_env == "MCP_TOKEN"
assert srv.http_headers() == {"Authorization": "Bearer secret-token"}
@pytest.mark.parametrize(("cls", "transport"), HTTP_TRANSPORTS)
def test_legacy_custom_header_and_format(
cls: type[MCPHttp | MCPStreamableHttp],
transport: str,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("MCP_TOKEN", "k")
srv = cls.model_validate({
"name": "remote",
"transport": transport,
"url": "https://mcp.example.com",
"api_key_env": "MCP_TOKEN",
"api_key_header": "X-API-Key",
"api_key_format": "{token}",
})
assert srv.http_headers() == {"X-API-Key": "k"}
@pytest.mark.parametrize(("cls", "transport"), HTTP_TRANSPORTS)
def test_explicit_static_auth_round_trips(
cls: type[MCPHttp | MCPStreamableHttp], transport: str
) -> None:
srv = cls.model_validate({
"name": "remote",
"transport": transport,
"url": "https://mcp.example.com",
"auth": {"type": "static", "api_key_env": "X", "api_key_header": "X-API-Key"},
})
dumped = srv.model_dump()
rebuilt = cls.model_validate(dumped)
assert isinstance(rebuilt.auth, MCPStaticAuth)
assert rebuilt.auth.api_key_env == "X"
assert rebuilt.auth.api_key_header == "X-API-Key"
@pytest.mark.parametrize(("cls", "transport"), HTTP_TRANSPORTS)
def test_oauth_auth_parses(
cls: type[MCPHttp | MCPStreamableHttp], transport: str
) -> None:
srv = cls.model_validate({
"name": "linear",
"transport": transport,
"url": "https://mcp.linear.app/mcp",
"auth": {"type": "oauth", "scopes": ["read", "write"]},
})
assert isinstance(srv.auth, MCPOAuth)
assert srv.auth.scopes == ["read", "write"]
assert srv.auth.redirect_port == 47823
assert srv.http_headers() == {}
@pytest.mark.parametrize(("cls", "transport"), HTTP_TRANSPORTS)
def test_mixing_legacy_keys_with_auth_block_is_rejected(
cls: type[MCPHttp | MCPStreamableHttp], transport: str
) -> None:
with pytest.raises(ValidationError, match="cannot mix top-level"):
cls.model_validate({
"name": "remote",
"transport": transport,
"url": "https://mcp.example.com",
"api_key_env": "LEGACY",
"auth": {"type": "static", "api_key_env": "NEW"},
})
@pytest.mark.parametrize(("cls", "transport"), HTTP_TRANSPORTS)
def test_default_auth_is_static(
cls: type[MCPHttp | MCPStreamableHttp], transport: str
) -> None:
srv = cls.model_validate({
"name": "remote",
"transport": transport,
"url": "https://mcp.example.com",
})
assert isinstance(srv.auth, MCPStaticAuth)
assert srv.http_headers() == {}
def test_oauth_client_id_and_metadata_url_mutually_exclusive() -> None:
with pytest.raises(ValidationError, match="mutually exclusive"):
MCPOAuth.model_validate({
"type": "oauth",
"scopes": ["read"],
"client_id": "abc",
"client_metadata_url": "https://example.com/cm.json",
})
def test_oauth_client_metadata_url_must_be_http_url() -> None:
with pytest.raises(ValidationError):
MCPOAuth.model_validate({
"type": "oauth",
"scopes": ["read"],
"client_metadata_url": "not-a-url",
})
def test_oauth_client_id_rejects_empty_string() -> None:
with pytest.raises(ValidationError):
MCPOAuth.model_validate({"type": "oauth", "scopes": ["read"], "client_id": ""})
@pytest.mark.parametrize("port", [80, 1023, 0, 65536, 70000])
def test_oauth_redirect_port_out_of_range(port: int) -> None:
with pytest.raises(ValidationError):
MCPOAuth.model_validate({
"type": "oauth",
"scopes": ["read"],
"redirect_port": port,
})
def test_oauth_redirect_port_inside_range() -> None:
auth = MCPOAuth.model_validate({
"type": "oauth",
"scopes": ["read"],
"redirect_port": 1024,
})
assert auth.redirect_port == 1024
def test_static_auth_forbids_extra_keys() -> None:
with pytest.raises(ValidationError):
MCPStaticAuth.model_validate({"type": "static", "headerz": {}})
def test_oauth_forbids_extra_keys() -> None:
with pytest.raises(ValidationError):
MCPOAuth.model_validate({"type": "oauth", "scopes": ["read"], "scope": "x"})
def test_oauth_scopes_required() -> None:
with pytest.raises(ValidationError):
MCPOAuth.model_validate({"type": "oauth"})
def test_oauth_scopes_empty_list_allowed() -> None:
auth = MCPOAuth.model_validate({"type": "oauth", "scopes": []})
assert auth.scopes == []
@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,
) -> None:
srv = cls.model_validate({
"name": "linear",
"transport": transport,
"url": "https://mcp.linear.app/mcp",
"auth": {"type": "oauth", "scopes": ["read"]},
})
registry = MCPRegistry()
with caplog.at_level(logging.WARNING, logger="vibe"):
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 second == {}
assert "OAuth support" not in caplog.text

5
tests/core/test_config_orchestrator.py
View file

@ -9,6 +9,7 @@ from vibe.core.config.layer import ConfigLayer, RawConfig
from vibe.core.config.orchestrator import ConfigOrchestrator
from vibe.core.config.patch import ConfigPatch
from vibe.core.config.schema import ConfigSchema, WithReplaceMerge
from vibe.core.config.types import LayerConfigSnapshot
class FakeLayer(ConfigLayer[RawConfig]):
@ -19,8 +20,8 @@ class FakeLayer(ConfigLayer[RawConfig]):
async def _check_trust(self) -> bool:
return True
async def _read_config(self) -> dict[str, Any]:
return dict(self._data)
async def _build_config_snapshot(self) -> LayerConfigSnapshot:
return LayerConfigSnapshot(data=dict(self._data), fingerprint="fp")
class SimpleSchema(ConfigSchema):

83
tests/core/test_config_resolution.py
View file

@ -342,7 +342,10 @@ class TestMigrateLeavesFindInBashAllowlist:
) -> None:
monkeypatch.setenv("VIBE_HOME", str(tmp_path))
config_file = tmp_path / "config.toml"
data = {"tools": {"bash": {"allowlist": ["echo", "ls"]}}}
data = {
"applied_migrations": [VibeConfig._BASH_READ_ONLY_MIGRATION],
"tools": {"bash": {"allowlist": ["echo", "ls"]}},
}
with config_file.open("wb") as f:
tomli_w.dump(data, f)
@ -359,7 +362,10 @@ class TestMigrateLeavesFindInBashAllowlist:
) -> None:
monkeypatch.setenv("VIBE_HOME", str(tmp_path))
config_file = tmp_path / "config.toml"
data = {"tools": {"bash": {"allowlist": ["echo", "find", "ls"]}}}
data = {
"applied_migrations": [VibeConfig._BASH_READ_ONLY_MIGRATION],
"tools": {"bash": {"allowlist": ["echo", "find", "ls"]}},
}
with config_file.open("wb") as f:
tomli_w.dump(data, f)
@ -396,7 +402,8 @@ class TestMigrateStripsBashAllowlistWildcardSuffix:
monkeypatch.setenv("VIBE_HOME", str(tmp_path))
config_file = tmp_path / "config.toml"
data = {
"tools": {"bash": {"allowlist": ["git commit *", "npm install *", "echo"]}}
"applied_migrations": [VibeConfig._BASH_READ_ONLY_MIGRATION],
"tools": {"bash": {"allowlist": ["git commit *", "npm install *", "echo"]}},
}
with config_file.open("wb") as f:
tomli_w.dump(data, f)
@ -420,7 +427,8 @@ class TestMigrateStripsBashAllowlistWildcardSuffix:
monkeypatch.setenv("VIBE_HOME", str(tmp_path))
config_file = tmp_path / "config.toml"
data = {
"tools": {"bash": {"allowlist": ["git commit *", "git commit", "find"]}}
"applied_migrations": [VibeConfig._BASH_READ_ONLY_MIGRATION],
"tools": {"bash": {"allowlist": ["git commit *", "git commit", "find"]}},
}
with config_file.open("wb") as f:
tomli_w.dump(data, f)
@ -438,7 +446,10 @@ class TestMigrateStripsBashAllowlistWildcardSuffix:
) -> None:
monkeypatch.setenv("VIBE_HOME", str(tmp_path))
config_file = tmp_path / "config.toml"
data = {"tools": {"bash": {"allowlist": ["echo", "find", "ls"]}}}
data = {
"applied_migrations": [VibeConfig._BASH_READ_ONLY_MIGRATION],
"tools": {"bash": {"allowlist": ["echo", "find", "ls"]}},
}
with config_file.open("wb") as f:
tomli_w.dump(data, f)
@ -451,6 +462,68 @@ class TestMigrateStripsBashAllowlistWildcardSuffix:
assert result["tools"]["bash"]["allowlist"] == ["echo", "find", "ls"]
class TestMigrateBashReadOnlyDefaults:
def test_merges_read_only_commands_into_existing_allowlist(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
from vibe.core.tools.builtins.bash import default_read_only_commands
monkeypatch.setenv("VIBE_HOME", str(tmp_path))
config_file = tmp_path / "config.toml"
data = {"tools": {"bash": {"allowlist": ["echo", "git commit"]}}}
with config_file.open("wb") as f:
tomli_w.dump(data, f)
reset_harness_files_manager()
init_harness_files_manager("user")
VibeConfig._migrate()
with config_file.open("rb") as f:
result = tomllib.load(f)
allowlist = result["tools"]["bash"]["allowlist"]
assert "git commit" in allowlist
for cmd in default_read_only_commands():
assert cmd in allowlist
assert VibeConfig._BASH_READ_ONLY_MIGRATION in result["applied_migrations"]
def test_does_not_readd_removed_command_after_migration(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("VIBE_HOME", str(tmp_path))
config_file = tmp_path / "config.toml"
data = {
"applied_migrations": [VibeConfig._BASH_READ_ONLY_MIGRATION],
"tools": {"bash": {"allowlist": ["echo", "find"]}},
}
with config_file.open("wb") as f:
tomli_w.dump(data, f)
reset_harness_files_manager()
init_harness_files_manager("user")
VibeConfig._migrate()
with config_file.open("rb") as f:
result = tomllib.load(f)
assert result["tools"]["bash"]["allowlist"] == ["echo", "find"]
def test_noop_when_no_bash_allowlist(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("VIBE_HOME", str(tmp_path))
config_file = tmp_path / "config.toml"
data = {"active_model": "test"}
with config_file.open("wb") as f:
tomli_w.dump(data, f)
reset_harness_files_manager()
init_harness_files_manager("user")
VibeConfig._migrate()
with config_file.open("rb") as f:
result = tomllib.load(f)
assert result == {"active_model": "test"}
class TestMigrateMistralVibeCliLatestDefaults:
def test_updates_alias_temperature_and_thinking_for_default_model(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch

20
tests/core/test_environment_layer.py
View file

@ -43,6 +43,26 @@ async def test_no_vars_set_returns_empty() -> None:
assert data.model_dump() == {}
@pytest.mark.asyncio
async def test_fingerprint_changes_when_env_changes() -> None:
with patch.dict(os.environ, {"VIBE_ACTIVE_MODEL": "first-model"}, clear=True):
layer = EnvironmentLayer(schema=VibeConfigSchema)
data1 = await layer.load()
fp1 = layer.fingerprint
os.environ["VIBE_ACTIVE_MODEL"] = "second-model"
data2 = await layer.load(force=True)
fp2 = layer.fingerprint
assert data1.model_dump() == {"active_model": "first-model"}
assert data2.model_dump() == {"active_model": "second-model"}
assert isinstance(fp1, str)
assert fp1
assert isinstance(fp2, str)
assert fp2
assert fp1 != fp2
@pytest.mark.asyncio
async def test_always_trusted() -> None:
assert await EnvironmentLayer(schema=VibeConfigSchema).resolve_trust() is True

118
tests/core/test_fingerprint.py Normal file
View file

@ -0,0 +1,118 @@
from __future__ import annotations
from pathlib import Path
import pytest
from vibe.core.config.fingerprint import capture_stable_file, create_dict_fingerprint
from vibe.core.config.types import ConcurrencyConflictError
class TestCaptureStableFile:
def test_captures_unchanged_file(self, tmp_working_directory: Path) -> None:
path = tmp_working_directory / "config.toml"
path.write_text("key = 1")
with capture_stable_file(path) as (file, first_fingerprint):
assert file.read() == b"key = 1"
with capture_stable_file(path) as (file, second_fingerprint):
assert file.read() == b"key = 1"
assert isinstance(first_fingerprint, str)
assert first_fingerprint
assert first_fingerprint == second_fingerprint
def test_raises_when_file_changes(self, tmp_working_directory: Path) -> None:
path = tmp_working_directory / "config.toml"
path.write_text("key = 1")
with pytest.raises(ConcurrencyConflictError) as exc_info:
with capture_stable_file(path):
path.write_text("key = 123")
assert exc_info.value.actual_fp != exc_info.value.expected_fp
def test_raises_when_path_is_replaced_after_open(
self, tmp_working_directory: Path
) -> None:
path = tmp_working_directory / "config.toml"
replacement = tmp_working_directory / "replacement.toml"
path.write_text("key = 1")
replacement.write_text("key = 2")
with pytest.raises(ConcurrencyConflictError) as exc_info:
with capture_stable_file(path) as (file, _):
replacement.replace(path)
assert file.read() == b"key = 1"
assert exc_info.value.actual_fp != exc_info.value.expected_fp
def test_raises_when_file_disappears_after_read(
self, tmp_working_directory: Path
) -> None:
path = tmp_working_directory / "config.toml"
path.write_text("key = 1")
with pytest.raises(FileNotFoundError):
with capture_stable_file(path) as (file, _):
assert file.read() == b"key = 1"
path.unlink()
def test_raises_when_file_is_missing(self, tmp_working_directory: Path) -> None:
path = tmp_working_directory / "missing.toml"
with pytest.raises(FileNotFoundError):
with capture_stable_file(path):
pass
class TestCreateDictFingerprint:
def test_empty_dict_returns_stable_non_empty_token(self) -> None:
first_fingerprint = create_dict_fingerprint({})
second_fingerprint = create_dict_fingerprint({})
assert isinstance(first_fingerprint, str)
assert first_fingerprint
assert first_fingerprint == second_fingerprint
def test_stable_for_same_dict(self) -> None:
data = {
"VIBE_MODEL": "mistral-large",
"VIBE_THEME": "dark",
"VIBE_TOOLS": ["read", "write"],
}
fp1 = create_dict_fingerprint(data)
fp2 = create_dict_fingerprint(data)
assert fp1 == fp2
def test_order_independent(self) -> None:
fp1 = create_dict_fingerprint({"a": "1", "b": "2"})
fp2 = create_dict_fingerprint({"b": "2", "a": "1"})
assert fp1 == fp2
def test_serializes_path_values(self) -> None:
fp1 = create_dict_fingerprint({
"tool_paths": [Path("/tmp/custom-tools")],
"agent_paths": [Path("agents")],
})
fp2 = create_dict_fingerprint({
"tool_paths": ["/tmp/custom-tools"],
"agent_paths": ["agents"],
})
assert fp1 == fp2
def test_changes_when_list_order_changes(self) -> None:
fp1 = create_dict_fingerprint({"tools": ["read", "write"]})
fp2 = create_dict_fingerprint({"tools": ["write", "read"]})
assert fp1 != fp2
def test_changes_when_value_changes(self) -> None:
fp1 = create_dict_fingerprint({"VIBE_MODEL": "mistral-large"})
fp2 = create_dict_fingerprint({"VIBE_MODEL": "devstral-2"})
assert fp1 != fp2
def test_changes_when_key_added(self) -> None:
fp1 = create_dict_fingerprint({"a": "1"})
fp2 = create_dict_fingerprint({"a": "1", "b": "2"})
assert fp1 != fp2

1644
tests/core/test_hooks.py
View file

File diff suppressed because it is too large Load diff

17
tests/core/test_local_config_files.py
View file

@ -2,6 +2,8 @@ from __future__ import annotations
from pathlib import Path
import pytest
from vibe.core.paths._local_config_files import LocalConfigDirs, find_local_config_dirs
@ -87,6 +89,21 @@ class TestConfigDirs:
assert resolved / ".vibe" in result.config_dirs
assert resolved / ".agents" in result.config_dirs
def test_unreadable_config_dirs_do_not_crash(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
def fake_is_dir(self: Path) -> bool:
raise PermissionError(13, "Permission denied")
def fake_is_file(self: Path) -> bool:
raise PermissionError(13, "Permission denied")
monkeypatch.setattr(Path, "is_dir", fake_is_dir)
monkeypatch.setattr(Path, "is_file", fake_is_file)
result = find_local_config_dirs(tmp_path)
assert result == LocalConfigDirs()
class TestLocalConfigDirsOr:
def test_or_concatenates_each_field(self) -> None:

496
tests/core/test_mcp_oauth.py Normal file
View file

@ -0,0 +1,496 @@
from __future__ import annotations
import asyncio
from collections.abc import Callable, Iterator
from contextlib import suppress
import socket
import urllib.parse
import httpx
import keyring
from keyring.backend import KeyringBackend
import keyring.backends.fail
import keyring.errors
from mcp.shared.auth import OAuthClientInformationFull, OAuthToken
import pytest
import respx
from vibe.core.auth.mcp_oauth import (
Fingerprint,
KeyringTokenStorage,
LoopbackCallbackHandler,
MCPOAuthError,
MCPOAuthHeadlessError,
MCPOAuthPortInUse,
build_oauth_provider,
perform_oauth_login,
)
from vibe.core.config import MCPOAuth, MCPStreamableHttp
class _MemoryKeyring(KeyringBackend):
priority = 100 # type: ignore[assignment]
def __init__(self) -> None:
self.store: dict[tuple[str, str], str] = {}
def get_password(self, service: str, username: str) -> str | None:
return self.store.get((service, username))
def set_password(self, service: str, username: str, password: str) -> None:
self.store[(service, username)] = password
def delete_password(self, service: str, username: str) -> None:
if (service, username) not in self.store:
raise keyring.errors.PasswordDeleteError(username)
del self.store[(service, username)]
@pytest.fixture
def memory_keyring() -> Iterator[_MemoryKeyring]:
original = keyring.get_keyring()
fake = _MemoryKeyring()
keyring.set_keyring(fake)
try:
yield fake
finally:
keyring.set_keyring(original)
@pytest.fixture
def headless_keyring() -> Iterator[None]:
original = keyring.get_keyring()
keyring.set_keyring(keyring.backends.fail.Keyring())
try:
yield
finally:
keyring.set_keyring(original)
def _oauth_server(
*,
name: str = "demo",
url: str = "https://mcp.example.com/mcp",
scopes: list[str] | None = None,
client_id: str | None = None,
client_metadata_url: str | None = None,
redirect_port: int = 47823,
) -> MCPStreamableHttp:
auth = MCPOAuth(
type="oauth",
scopes=scopes if scopes is not None else ["read", "write"],
client_id=client_id,
client_metadata_url=client_metadata_url, # type: ignore[arg-type]
redirect_port=redirect_port,
)
return MCPStreamableHttp(transport="streamable-http", name=name, url=url, auth=auth)
def _free_port() -> int:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("127.0.0.1", 0))
return int(s.getsockname()[1])
async def _send_callback(port: int, query: str, *, timeout: float = 5.0) -> bytes:
deadline = asyncio.get_event_loop().time() + timeout
last_err: BaseException | None = None
while asyncio.get_event_loop().time() < deadline:
try:
reader, writer = await asyncio.open_connection("127.0.0.1", port)
except (ConnectionRefusedError, OSError) as exc:
last_err = exc
await asyncio.sleep(0.02)
continue
try:
request = (
f"GET /callback?{query} HTTP/1.0\r\n"
"Host: 127.0.0.1\r\n"
"Connection: close\r\n"
"\r\n"
)
writer.write(request.encode("ascii"))
await writer.drain()
return await reader.read()
finally:
writer.close()
with suppress(Exception):
await writer.wait_closed()
raise RuntimeError(f"loopback never bound on port {port}: {last_err}")
class TestKeyringTokenStorage:
@pytest.mark.asyncio
async def test_round_trip_tokens(self, memory_keyring: _MemoryKeyring) -> None:
storage = KeyringTokenStorage(alias="linear")
assert await storage.get_tokens() is None
tokens = OAuthToken(
access_token="at",
token_type="Bearer",
expires_in=3600,
refresh_token="rt",
scope="read write",
)
await storage.set_tokens(tokens)
loaded = await storage.get_tokens()
assert loaded is not None
assert loaded.access_token == "at"
assert loaded.refresh_token == "rt"
assert loaded.scope == "read write"
assert ("vibe", "mcp-oauth:linear:tokens") in memory_keyring.store
@pytest.mark.asyncio
async def test_round_trip_client_info(self, memory_keyring: _MemoryKeyring) -> None:
storage = KeyringTokenStorage(alias="linear")
assert await storage.get_client_info() is None
info = OAuthClientInformationFull(
client_id="abc123",
redirect_uris=["http://127.0.0.1:47823/callback"], # type: ignore[list-item]
token_endpoint_auth_method="none",
)
await storage.set_client_info(info)
loaded = await storage.get_client_info()
assert loaded is not None
assert loaded.client_id == "abc123"
assert ("vibe", "mcp-oauth:linear:client_info") in memory_keyring.store
@pytest.mark.asyncio
async def test_per_alias_isolation(self, memory_keyring: _MemoryKeyring) -> None:
a = KeyringTokenStorage(alias="linear")
b = KeyringTokenStorage(alias="notion")
await a.set_tokens(
OAuthToken(access_token="A", token_type="Bearer", expires_in=60)
)
await b.set_tokens(
OAuthToken(access_token="B", token_type="Bearer", expires_in=60)
)
loaded_a = await a.get_tokens()
loaded_b = await b.get_tokens()
assert loaded_a is not None and loaded_a.access_token == "A"
assert loaded_b is not None and loaded_b.access_token == "B"
def test_headless_init_raises(self, headless_keyring: None) -> None:
with pytest.raises(MCPOAuthHeadlessError) as exc_info:
KeyringTokenStorage(alias="linear")
msg = str(exc_info.value)
assert "linear" in msg
assert "api_key_env" in msg
assert exc_info.value.server_alias == "linear"
class TestFingerprint:
def test_compute_stable_across_scope_order(
self, memory_keyring: _MemoryKeyring
) -> None:
a = _oauth_server(scopes=["read", "write", "admin"])
b = _oauth_server(scopes=["admin", "write", "read"])
assert Fingerprint.compute(a) == Fingerprint.compute(b)
def test_compute_strips_whitespace_and_dedupes(
self, memory_keyring: _MemoryKeyring
) -> None:
a = _oauth_server(scopes=["read", "write"])
b = _oauth_server(scopes=[" read ", "write", "write", ""])
assert Fingerprint.compute(a) == Fingerprint.compute(b)
def test_compute_marker_for_client_id(self, memory_keyring: _MemoryKeyring) -> None:
srv = _oauth_server(client_id="pre-registered-id")
assert Fingerprint.compute(srv).client_marker == "pre-registered-id"
def test_compute_marker_for_client_metadata_url(
self, memory_keyring: _MemoryKeyring
) -> None:
srv = _oauth_server(client_metadata_url="https://vibe.example/cm.json")
fp = Fingerprint.compute(srv)
assert fp.client_marker.startswith("https://vibe.example/cm.json")
def test_compute_marker_for_dcr(self, memory_keyring: _MemoryKeyring) -> None:
assert Fingerprint.compute(_oauth_server()).client_marker == "<dcr>"
def test_compute_rejects_static_auth(self, memory_keyring: _MemoryKeyring) -> None:
from vibe.core.config import MCPStaticAuth
srv = MCPStreamableHttp(
transport="streamable-http",
name="x",
url="https://x/mcp",
auth=MCPStaticAuth(),
)
with pytest.raises(TypeError, match="OAuth"):
Fingerprint.compute(srv)
def test_matches_detects_url_change(self, memory_keyring: _MemoryKeyring) -> None:
a = Fingerprint.compute(_oauth_server(url="https://a/mcp"))
b = Fingerprint.compute(_oauth_server(url="https://b/mcp"))
assert a != b
def test_matches_detects_scope_change(self, memory_keyring: _MemoryKeyring) -> None:
a = Fingerprint.compute(_oauth_server(scopes=["read"]))
b = Fingerprint.compute(_oauth_server(scopes=["read", "write"]))
assert a != b
def test_matches_detects_marker_change(
self, memory_keyring: _MemoryKeyring
) -> None:
a = Fingerprint.compute(_oauth_server(client_id="x"))
b = Fingerprint.compute(_oauth_server(client_id="y"))
assert a != b
@pytest.mark.asyncio
async def test_load_returns_none_when_missing(
self, memory_keyring: _MemoryKeyring
) -> None:
assert await Fingerprint.load("nope") is None
@pytest.mark.asyncio
async def test_save_and_load_round_trip(
self, memory_keyring: _MemoryKeyring
) -> None:
fp = Fingerprint.compute(_oauth_server(name="linear"))
await fp.save("linear")
loaded = await Fingerprint.load("linear")
assert loaded == fp
class TestLoopbackCallbackHandler:
@pytest.mark.asyncio
async def test_happy_path(self) -> None:
port = _free_port()
handler = LoopbackCallbackHandler(port=port, server_alias="demo")
async def driver() -> bytes:
return await _send_callback(port, "code=AUTH_CODE_123&state=STATE_XYZ")
serve_task = asyncio.create_task(handler.serve_once())
driver_task = asyncio.create_task(driver())
code, state = await serve_task
response = await driver_task
assert code == "AUTH_CODE_123"
assert state == "STATE_XYZ"
assert b"200 OK" in response
assert b"Login complete" in response
@pytest.mark.asyncio
async def test_port_in_use_raises(self) -> None:
port = _free_port()
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 0)
sock.bind(("127.0.0.1", port))
sock.listen(1)
try:
handler = LoopbackCallbackHandler(port=port, server_alias="demo")
with pytest.raises(MCPOAuthPortInUse) as exc_info:
await handler.serve_once()
assert exc_info.value.port == port
assert exc_info.value.server_alias == "demo"
assert "redirect_port" in str(exc_info.value)
finally:
sock.close()
@pytest.mark.asyncio
async def test_missing_code_raises(self) -> None:
port = _free_port()
handler = LoopbackCallbackHandler(port=port, server_alias="demo")
async def driver() -> bytes:
return await _send_callback(port, "error=access_denied&state=S")
serve_task = asyncio.create_task(handler.serve_once())
driver_task = asyncio.create_task(driver())
with pytest.raises(MCPOAuthError, match="missing 'code'"):
await serve_task
response = await driver_task
assert b"400 Bad Request" in response
class TestBuildOAuthProvider:
@pytest.mark.asyncio
async def test_metadata_matches_config(
self, memory_keyring: _MemoryKeyring
) -> None:
srv = _oauth_server(scopes=["read", "write"], redirect_port=51234)
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
)
md = provider.context.client_metadata
assert md.scope == "read write"
assert md.client_name == "Mistral Vibe"
assert md.token_endpoint_auth_method == "none"
assert md.grant_types == ["authorization_code", "refresh_token"]
assert md.redirect_uris is not None
assert str(md.redirect_uris[0]) == "http://127.0.0.1:51234/callback"
@pytest.mark.asyncio
async def test_empty_scopes_becomes_none(
self, memory_keyring: _MemoryKeyring
) -> None:
srv = _oauth_server(scopes=[])
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
)
assert provider.context.client_metadata.scope is None
@pytest.mark.asyncio
async def test_client_metadata_url_forwarded(
self, memory_keyring: _MemoryKeyring
) -> None:
srv = _oauth_server(client_metadata_url="https://vibe.example/cm.json")
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
)
assert provider.context.client_metadata_url == "https://vibe.example/cm.json"
@pytest.mark.asyncio
async def test_rejects_static_auth(self, memory_keyring: _MemoryKeyring) -> None:
from vibe.core.config import MCPStaticAuth
srv = MCPStreamableHttp(
transport="streamable-http",
name="x",
url="https://x/mcp",
auth=MCPStaticAuth(),
)
async def on_url(_url: str) -> None:
return None
async def cb() -> tuple[str, str | None]:
return "code", None
with pytest.raises(TypeError, match="OAuth"):
build_oauth_provider(srv, redirect_handler=on_url, callback_handler=cb)
class TestPerformOAuthLogin:
@pytest.mark.asyncio
async def test_full_flow_persists_tokens_and_fingerprint(
self, memory_keyring: _MemoryKeyring
) -> None:
port = _free_port()
server_url = "https://mcp.example.com/mcp"
as_url = "https://as.example.com"
srv = _oauth_server(
name="demo", url=server_url, scopes=["read"], redirect_port=port
)
async def on_url(url: str) -> None:
qs = urllib.parse.urlparse(url).query
state = urllib.parse.parse_qs(qs)["state"][0]
async def fire() -> None:
await _send_callback(port, f"code=THE_CODE&state={state}")
asyncio.get_event_loop().create_task(fire())
async with respx.mock(assert_all_called=False) as router:
router.get(server_url).mock(side_effect=_mcp_responses())
router.get(
"https://mcp.example.com/.well-known/oauth-protected-resource"
).mock(
return_value=httpx.Response(
200,
json={"resource": server_url, "authorization_servers": [as_url]},
)
)
router.get(
"https://as.example.com/.well-known/oauth-authorization-server"
).mock(
return_value=httpx.Response(
200,
json={
"issuer": as_url,
"authorization_endpoint": f"{as_url}/authorize",
"token_endpoint": f"{as_url}/token",
"registration_endpoint": f"{as_url}/register",
"response_types_supported": ["code"],
"code_challenge_methods_supported": ["S256"],
"grant_types_supported": [
"authorization_code",
"refresh_token",
],
},
)
)
router.post(f"{as_url}/register").mock(
return_value=httpx.Response(
201,
json={
"client_id": "dcr-client-id",
"redirect_uris": [f"http://127.0.0.1:{port}/callback"],
"token_endpoint_auth_method": "none",
"grant_types": ["authorization_code", "refresh_token"],
"response_types": ["code"],
},
)
)
router.post(f"{as_url}/token").mock(
return_value=httpx.Response(
200,
json={
"access_token": "ACCESS_TOKEN",
"token_type": "Bearer",
"expires_in": 3600,
"refresh_token": "REFRESH_TOKEN",
"scope": "read",
},
)
)
router.route(host="127.0.0.1").pass_through()
await perform_oauth_login(srv, on_url=on_url)
storage = KeyringTokenStorage(alias="demo")
tokens = await storage.get_tokens()
assert tokens is not None
assert tokens.access_token == "ACCESS_TOKEN"
assert tokens.refresh_token == "REFRESH_TOKEN"
fp = await Fingerprint.load("demo")
assert fp is not None
assert fp == Fingerprint.compute(srv)
def _mcp_responses() -> Callable[[httpx.Request], httpx.Response]:
state = {"calls": 0}
def _factory(_request: httpx.Request) -> httpx.Response:
state["calls"] += 1
if state["calls"] == 1:
return httpx.Response(
401,
headers={
"WWW-Authenticate": (
"Bearer resource_metadata="
'"https://mcp.example.com/.well-known/oauth-protected-resource"'
)
},
)
return httpx.Response(200, json={"ok": True})
return _factory

4
tests/core/test_overrides_layer.py
View file

@ -80,6 +80,10 @@ async def test_live_reference_picks_up_caller_mutation() -> None:
data: dict[str, object] = {"key": "original"}
layer = OverridesLayer(data=data)
await layer.load()
fp1 = layer.fingerprint
data["key"] = "updated"
result = await layer.load(force=True)
fp2 = layer.fingerprint
assert result.model_extra == {"key": "updated"}
assert fp1 != fp2

10
tests/core/test_project_config_layer.py
View file

@ -7,6 +7,7 @@ import pytest
from vibe.core.config.layer import UntrustedLayerError
from vibe.core.config.layers.project import ProjectConfigLayer
from vibe.core.config.patch import ConfigPatch
from vibe.core.config.types import MISSING_CONFIG_FILE_FINGERPRINT
from vibe.core.paths._vibe_home import GlobalPath
from vibe.core.trusted_folders import trusted_folders_manager
@ -21,6 +22,14 @@ async def test_reads_toml_when_trusted(tmp_working_directory: Path) -> None:
layer = ProjectConfigLayer(path=tmp_working_directory)
data = await layer.load()
assert data.model_extra == {"active_model": "project-model"}
fp1 = layer.fingerprint
assert isinstance(fp1, str)
assert fp1
config_path.unlink()
data = await layer.load(force=True)
assert data.model_extra == {}
assert layer.fingerprint == MISSING_CONFIG_FILE_FINGERPRINT
@pytest.mark.asyncio
@ -55,6 +64,7 @@ async def test_missing_file_returns_empty(tmp_working_directory: Path) -> None:
layer = ProjectConfigLayer(path=tmp_working_directory)
data = await layer.load()
assert data.model_extra == {}
assert layer.fingerprint == MISSING_CONFIG_FILE_FINGERPRINT
@pytest.mark.asyncio

181
tests/core/test_telemetry_send.py
View file

@ -1,9 +1,11 @@
from __future__ import annotations
import asyncio
from pathlib import Path
from typing import Any
from unittest.mock import AsyncMock, MagicMock
from pydantic import BaseModel
import pytest
from tests.conftest import build_test_vibe_config
@ -14,7 +16,7 @@ from vibe.core.telemetry.build_metadata import (
build_base_metadata,
build_request_metadata,
)
from vibe.core.telemetry.send import TelemetryClient
from vibe.core.telemetry.send import TelemetryClient, _extract_file_extension
from vibe.core.telemetry.types import (
AttachmentKind,
EntrypointMetadata,
@ -25,18 +27,35 @@ from vibe.core.types import Backend
from vibe.core.utils import get_user_agent
_original_send_telemetry_event = TelemetryClient.send_telemetry_event
from vibe.core.tools.builtins.edit import Edit, EditArgs
from vibe.core.tools.builtins.read import Read, ReadArgs
from vibe.core.tools.builtins.write_file import WriteFile, WriteFileArgs
def _make_resolved_tool_call(
tool_name: str, args_dict: dict[str, Any]
) -> ResolvedToolCall:
if tool_name == "write_file":
validated = WriteFileArgs(path="foo.txt", content="x")
cls: type[BaseTool] = WriteFile
else:
validated = FakeToolArgs()
cls = FakeTool
validated: BaseModel
cls: type[BaseTool]
match tool_name:
case "write_file":
validated = WriteFileArgs(
path=args_dict.get("path", "foo.txt"), content="x"
)
cls = WriteFile
case "edit":
validated = EditArgs(
file_path=args_dict.get("file_path", "foo.txt"),
old_string="a",
new_string="b",
)
cls = Edit
case "read":
validated = ReadArgs(file_path=args_dict.get("file_path", "foo.txt"))
cls = Read
case _:
validated = FakeToolArgs()
cls = FakeTool
return ResolvedToolCall(
tool_name=tool_name, tool_class=cls, validated_args=validated, call_id="call_1"
)
@ -50,6 +69,37 @@ def _run_telemetry_tasks() -> None:
loop.close()
class TestExtractFileExtension:
@pytest.mark.parametrize(
("path", "expected"),
[
("/tmp/foo.py", ".py"),
("foo.TSX", ".tsx"),
("/tmp/Makefile", None),
("/tmp/.bashrc", None),
("archive.tar.gz", ".gz"),
("", None),
],
)
def test_string_inputs(self, path: str, expected: str | None) -> None:
assert _extract_file_extension(path) == expected
@pytest.mark.parametrize(
("path", "expected"),
[
(Path("/tmp/foo.py"), ".py"),
(Path("foo.TSX"), ".tsx"),
(Path("/tmp/Makefile"), None),
],
)
def test_path_inputs(self, path: Path, expected: str | None) -> None:
assert _extract_file_extension(path) == expected
@pytest.mark.parametrize("path", [None, 42, ["foo.py"], {"path": "foo.py"}])
def test_non_path_like_inputs_return_none(self, path: object) -> None:
assert _extract_file_extension(path) is None
class TestTelemetryClient:
def test_send_telemetry_event_swallows_config_getter_value_error(self) -> None:
def _raise_config_error() -> Any:
@ -151,6 +201,7 @@ class TestTelemetryClient:
assert properties["model"] == "mistral-large"
assert properties["nb_files_created"] == 0
assert properties["nb_files_modified"] == 0
assert properties["file_extension"] is None
assert properties["message_id"] is None
def test_send_tool_call_finished_with_message_id(
@ -176,7 +227,7 @@ class TestTelemetryClient:
) -> None:
config = build_test_vibe_config(enable_telemetry=True)
client = TelemetryClient(config_getter=lambda: config)
tool_call = _make_resolved_tool_call("write_file", {})
tool_call = _make_resolved_tool_call("write_file", {"path": "/tmp/foo.PY"})
client.send_tool_call_finished(
tool_call=tool_call,
@ -189,6 +240,86 @@ class TestTelemetryClient:
assert telemetry_events[0]["properties"]["nb_files_created"] == 1
assert telemetry_events[0]["properties"]["nb_files_modified"] == 0
assert telemetry_events[0]["properties"]["file_extension"] == ".py"
def test_send_tool_call_finished_file_extension_edit(
self, telemetry_events: list[dict[str, Any]]
) -> None:
config = build_test_vibe_config(enable_telemetry=True)
client = TelemetryClient(config_getter=lambda: config)
tool_call = _make_resolved_tool_call(
"edit", {"file_path": "/tmp/component.tsx"}
)
client.send_tool_call_finished(
tool_call=tool_call,
status="success",
decision=None,
agent_profile_name="default",
model="mistral-large",
result={},
)
assert telemetry_events[0]["properties"]["nb_files_modified"] == 1
assert telemetry_events[0]["properties"]["nb_files_created"] == 0
assert telemetry_events[0]["properties"]["file_extension"] == ".tsx"
def test_send_tool_call_finished_file_extension_read(
self, telemetry_events: list[dict[str, Any]]
) -> None:
config = build_test_vibe_config(enable_telemetry=True)
client = TelemetryClient(config_getter=lambda: config)
tool_call = _make_resolved_tool_call("read", {"file_path": "/tmp/lib.rs"})
client.send_tool_call_finished(
tool_call=tool_call,
status="success",
decision=None,
agent_profile_name="default",
model="mistral-large",
result={},
)
assert telemetry_events[0]["properties"]["nb_files_created"] == 0
assert telemetry_events[0]["properties"]["nb_files_modified"] == 0
assert telemetry_events[0]["properties"]["file_extension"] == ".rs"
def test_send_tool_call_finished_file_extension_none_when_no_suffix(
self, telemetry_events: list[dict[str, Any]]
) -> None:
config = build_test_vibe_config(enable_telemetry=True)
client = TelemetryClient(config_getter=lambda: config)
tool_call = _make_resolved_tool_call("write_file", {"path": "/tmp/Makefile"})
client.send_tool_call_finished(
tool_call=tool_call,
status="success",
decision=None,
agent_profile_name="default",
model="mistral-large",
result={},
)
assert telemetry_events[0]["properties"]["file_extension"] is None
def test_send_tool_call_finished_file_extension_none_on_failure(
self, telemetry_events: list[dict[str, Any]]
) -> None:
config = build_test_vibe_config(enable_telemetry=True)
client = TelemetryClient(config_getter=lambda: config)
tool_call = _make_resolved_tool_call("write_file", {"path": "/tmp/foo.py"})
client.send_tool_call_finished(
tool_call=tool_call,
status="failure",
decision=None,
agent_profile_name="default",
model="mistral-large",
result={},
)
assert telemetry_events[0]["properties"]["nb_files_created"] == 0
assert telemetry_events[0]["properties"]["file_extension"] is None
def test_send_tool_call_finished_decision_none(
self, telemetry_events: list[dict[str, Any]]
@ -378,6 +509,40 @@ 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:
config = build_test_vibe_config(enable_telemetry=True)
client = TelemetryClient(config_getter=lambda: config)
client.send_teleport_failed(
stage="workflow_start",
error_class="ServiceTeleportError",
push_required=False,
nb_session_messages=4,
error_details={"failure_kind": "http_error", "http_status_code": 502},
)
assert telemetry_events[0]["event_name"] == "vibe.teleport_failed"
assert telemetry_events[0]["properties"] == {
"stage": "workflow_start",
"error_class": "ServiceTeleportError",
"push_required": False,
"nb_session_messages": 4,
"failure_kind": "http_error",
"http_status_code": 502,
}
def test_send_new_session_payload(
self, telemetry_events: list[dict[str, Any]]
) -> None:

34
tests/core/test_teleport_nuage.py
View file

@ -174,9 +174,14 @@ async def test_start_raises_for_unsuccessful_response() -> None:
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
nuage = NuageClient("https://chat.example.com", "api-key", client=client)
with pytest.raises(ServiceTeleportError, match="Nuage start"):
with pytest.raises(ServiceTeleportError, match="status 401") as exc_info:
await nuage.start(_request())
assert exc_info.value.telemetry_details == {
"failure_kind": "http_error",
"http_status_code": 401,
}
@pytest.mark.asyncio
async def test_start_raises_for_invalid_response() -> None:
@ -185,5 +190,30 @@ async def test_start_raises_for_invalid_response() -> None:
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
nuage = NuageClient("https://chat.example.com", "api-key", client=client)
with pytest.raises(ServiceTeleportError, match="response was invalid"):
with pytest.raises(
ServiceTeleportError, match="response was invalid"
) as exc_info:
await nuage.start(_request())
assert exc_info.value.telemetry_details == {
"failure_kind": "invalid_schema",
"http_status_code": 200,
}
@pytest.mark.asyncio
async def test_start_raises_for_invalid_json_response() -> None:
async def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(
200, text="not-json", headers={"content-type": "text/plain"}
)
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
nuage = NuageClient("https://chat.example.com", "api-key", client=client)
with pytest.raises(ServiceTeleportError, match="not valid JSON") as exc_info:
await nuage.start(_request())
assert exc_info.value.telemetry_details == {
"failure_kind": "invalid_json",
"http_status_code": 200,
}

6
tests/core/test_teleport_telemetry.py
View file

@ -87,7 +87,10 @@ class TestTeleportAgentLoopTelemetry:
) -> AsyncGenerator[object, object]:
yield TeleportCheckingGitEvent()
yield TeleportStartingWorkflowEvent()
raise ServiceTeleportError("Workflow api-key-123 could not be started.")
raise ServiceTeleportError(
"Workflow api-key-123 could not be started.",
telemetry_details={"http_status_code": 502},
)
agent_loop.messages.append(LLMMessage(role=Role.user, content="hello"))
_set_teleport_service(agent_loop, FakeTeleportService())
@ -103,6 +106,7 @@ class TestTeleportAgentLoopTelemetry:
"error_class": "ServiceTeleportError",
"push_required": False,
"nb_session_messages": 1,
"http_status_code": 502,
"session_id": agent_loop.session_id,
}
assert "api-key-123" not in str(telemetry_events[-1]["properties"])

34
tests/core/test_trusted_folders.py
View file

@ -351,6 +351,19 @@ class TestHasAgentsMdFile:
def test_agents_md_filename_constant(self) -> None:
assert AGENTS_MD_FILENAME == "AGENTS.md"
def test_unreadable_agents_md_does_not_crash(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
denied = (tmp_path / "AGENTS.md").resolve()
def fake_is_file(self: Path) -> bool:
if self.resolve() == denied:
raise PermissionError(13, "Permission denied")
return False
monkeypatch.setattr(Path, "is_file", fake_is_file)
assert has_agents_md_file(tmp_path) is False
class TestFindTrustableFiles:
def test_returns_empty_for_clean_directory(self, tmp_path: Path) -> None:
@ -513,3 +526,24 @@ class TestFindGitRepoAncestor:
) -> None:
monkeypatch.setattr(Path, "home", classmethod(lambda _cls: tmp_path / "nope"))
assert find_git_repo_ancestor(tmp_path) is None
def test_unreadable_ancestor_git_head_does_not_crash(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
ancestor = tmp_path / "mnt" / "vast"
(ancestor / ".git").mkdir(parents=True)
cwd = ancestor / "project"
cwd.mkdir()
real_is_file = Path.is_file
denied = (ancestor / ".git" / "HEAD").resolve()
def fake_is_file(self: Path) -> bool:
if self.resolve() == denied:
raise PermissionError(13, "Permission denied")
return real_is_file(self)
monkeypatch.setattr(Path, "is_file", fake_is_file)
monkeypatch.setattr(Path, "home", classmethod(lambda _cls: tmp_path / "home"))
assert find_git_repo_ancestor(cwd) is None

19
tests/core/test_user_config_layer.py
View file

@ -7,6 +7,7 @@ import pytest
from vibe.core.config.layer import LayerImplementationError
from vibe.core.config.layers.user import UserConfigLayer
from vibe.core.config.patch import ConfigPatch
from vibe.core.config.types import MISSING_CONFIG_FILE_FINGERPRINT
@pytest.mark.asyncio
@ -17,6 +18,9 @@ async def test_reads_toml_file(tmp_working_directory: Path) -> None:
layer = UserConfigLayer(path=path, name="user-toml")
data = await layer.load()
assert data.model_extra == {"active_model": "mistral-large", "count": 42}
fingerprint = layer.fingerprint
assert isinstance(fingerprint, str)
assert fingerprint
@pytest.mark.asyncio
@ -37,6 +41,7 @@ async def test_missing_file_returns_empty(tmp_working_directory: Path) -> None:
layer = UserConfigLayer(path=path, name="user-toml")
data = await layer.load()
assert data.model_extra == {}
assert layer.fingerprint == MISSING_CONFIG_FILE_FINGERPRINT
@pytest.mark.asyncio
@ -70,7 +75,7 @@ async def test_invalid_toml_raises(tmp_working_directory: Path) -> None:
path = tmp_working_directory / "bad.toml"
path.write_text("this is not valid = = = toml [[[")
layer = UserConfigLayer(path=path, name="user-toml")
with pytest.raises(LayerImplementationError, match="_read_config"):
with pytest.raises(LayerImplementationError, match="_build_config_snapshot"):
await layer.load()
@ -81,11 +86,23 @@ async def test_force_reload_reads_fresh_data(tmp_working_directory: Path) -> Non
layer = UserConfigLayer(path=path, name="user-toml")
data1 = await layer.load()
fp1 = layer.fingerprint
assert data1.model_extra == {"value": "first"}
assert isinstance(fp1, str)
assert fp1
path.write_text('value = "second"\n')
data2 = await layer.load(force=True)
fp2 = layer.fingerprint
assert data2.model_extra == {"value": "second"}
assert isinstance(fp2, str)
assert fp2
assert fp1 != fp2
path.unlink()
data3 = await layer.load(force=True)
assert data3.model_extra == {}
assert layer.fingerprint == MISSING_CONFIG_FILE_FINGERPRINT
@pytest.mark.asyncio

22
tests/core/test_utils.py
View file

@ -106,6 +106,28 @@ class TestReadSafe:
with pytest.raises(FileNotFoundError):
read_safe(tmp_path / "nope.txt")
def test_from_subprocess_prefers_oem_over_locale_ansi(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
# \x82 is invalid UTF-8 and decodes differently across single-byte
# encodings: cp1252 (Windows ANSI) -> "‚" (low-9 quote);
# cp850 (Windows OEM) -> "é". Subprocess output prefers OEM over the
# ANSI locale; file reads (from_subprocess=False) must not.
raw = "café\n".encode("cp850")
monkeypatch.setattr(
io_utils.locale, "getpreferredencoding", lambda _do_setlocale: "cp1252"
)
monkeypatch.setattr(io_utils, "_encoding_from_best_match", lambda _raw: None)
monkeypatch.setattr(io_utils, "_windows_oem_encoding", lambda: "cp850")
from_file = decode_safe(raw)
assert from_file.encoding == "cp1252"
assert from_file.text == raw.decode("cp1252")
from_subprocess = decode_safe(raw, from_subprocess=True)
assert from_subprocess.encoding == "cp850"
assert from_subprocess.text == "café\n"
class TestReadSafeNewlines:
def test_lf(self, tmp_path: Path) -> None:

5
tests/core/tools/builtins/test_read.py
View file

@ -244,7 +244,6 @@ def test_get_result_display() -> None:
assert isinstance(display, ToolResultDisplay)
assert display.success is True
assert "10 lines" in display.message
assert "foo.py" in display.message
@ -261,7 +260,7 @@ def test_get_result_display_truncated() -> None:
)
display = Read.get_result_display(event)
assert "truncated" in display.message
assert "truncated" in display.suffix
def test_get_result_display_truncated_via_flag() -> None:
@ -278,7 +277,7 @@ def test_get_result_display_truncated_via_flag() -> None:
)
display = Read.get_result_display(event)
assert "truncated" in display.message
assert "truncated" in display.suffix
@pytest.fixture()