Co-authored-by: Clément Sirieix <clement.sirieix@mistral.ai>
Co-authored-by: Kim-Adeline Miguel <51720070+kimadeline@users.noreply.github.com>
Co-authored-by: Mathias Gesbert <mathias.gesbert@mistral.ai>
Co-authored-by: Maxime Dolores <maxime.dolores@ext.mistral.ai>
Co-authored-by: Michel Thomazo <51709227+michelTho@users.noreply.github.com>
Co-authored-by: Paul Cacheux <paul.cacheux@mistral.ai>
Co-authored-by: Quentin <quentin.torroba@mistral.ai>
Co-authored-by: Simon <80467011+sorgfresser@users.noreply.github.com>
Co-authored-by: angelapopopo <angele.lenglemetz@mistral.ai>
Co-authored-by: Mistral Vibe <vibe@mistral.ai>
This commit is contained in:
Pierre Rossinès 2026-04-16 11:10:26 +02:00 committed by GitHub
parent e1a25caa52
commit 95336528f6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
59 changed files with 3992 additions and 403 deletions

View file

@ -0,0 +1,46 @@
from __future__ import annotations
from vibe.core.tools.base import BaseTool
from vibe.core.tools.connectors.connector_registry import (
ConnectorRegistry,
RemoteTool,
_normalize_name,
create_connector_proxy_tool_class,
)
class FakeConnectorRegistry(ConnectorRegistry):
"""Test double that returns canned connector tools without hitting the API."""
def __init__(self, connectors: dict[str, list[RemoteTool]] | None = None) -> None:
super().__init__(api_key="fake-key")
self._fake_connectors = connectors or {}
self._build_cache()
def _build_cache(self) -> None:
self._cache = {}
self._connector_names = []
self._connector_connected = {}
for connector_name, tools in self._fake_connectors.items():
alias = _normalize_name(connector_name)
connector_id = f"fake-id-{connector_name}"
tool_map: dict[str, type[BaseTool]] = {}
for remote in tools:
proxy_cls = create_connector_proxy_tool_class(
connector_name=connector_name,
connector_alias=alias,
connector_id=connector_id,
remote=remote,
api_key="fake-key",
)
tool_map[proxy_cls.get_name()] = proxy_cls
self._cache[connector_id] = tool_map
self._connector_names.append(alias)
self._connector_connected[alias] = bool(tool_map)
def get_tools(self) -> dict[str, type[BaseTool]]:
result: dict[str, type[BaseTool]] = {}
if self._cache:
for tools in self._cache.values():
result.update(tools)
return result