tests/test_embedding_providers.py
Rama main · réplica de lectura del repositorio original.
from __future__ import annotations
import os
import sys
import unittest
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parent))
from _isolation import clean_env
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from tools import embedding_providers as providers # noqa: E402
class EmbeddingConfigurationTests(unittest.TestCase):
def test_explicit_env_file_is_used_below_process_environment(self) -> None:
import tempfile
with tempfile.TemporaryDirectory() as tmp:
env_file = Path(tmp) / "embedding.env"
env_file.write_text(
"EMBEDDING_PROVIDER=gemini\nEMBEDDING_MODEL=gemini-file-model\nGEMINI_API_KEY=file-key\n",
encoding="utf-8",
)
with mock.patch.dict(os.environ, {**clean_env(), "GEMINI_API_KEY": "process-key"}, clear=True):
provider = providers.get_provider(env_file=env_file)
self.assertEqual(provider.api_key, "process-key")
self.assertEqual(provider.model, "gemini-file-model")
def test_explicit_env_file_supplies_credential_without_mutating_process(self) -> None:
import tempfile
with tempfile.TemporaryDirectory() as tmp:
env_file = Path(tmp) / "embedding.env"
env_file.write_text("GEMINI_API_KEY=file-key\n", encoding="utf-8")
with mock.patch.dict(os.environ, {**clean_env(), }, clear=True):
provider = providers.get_provider(env_file=env_file)
self.assertNotIn("GEMINI_API_KEY", os.environ)
self.assertEqual(provider.api_key, "file-key")
def test_missing_explicit_env_file_is_actionable(self) -> None:
with mock.patch.dict(os.environ, {**clean_env(), }, clear=True):
with self.assertRaisesRegex(FileNotFoundError, "env file"):
providers.get_provider(env_file=Path("/does/not/exist.env"))
def test_defaults_to_gemini_and_reads_runtime_credential(self) -> None:
with mock.patch.dict(os.environ, {**clean_env(), "GEMINI_API_KEY": "runtime-only"}, clear=True):
config = providers.load_config()
self.assertEqual(config.provider, "gemini")
self.assertEqual(config.model, "gemini-embedding-2-preview")
provider = providers.get_provider()
self.assertIsInstance(provider, providers.GeminiProvider)
self.assertEqual(provider.api_key, "runtime-only")
def test_missing_credential_fails_without_private_path_fallback(self) -> None:
with mock.patch.dict(os.environ, {**clean_env(), "EMBEDDING_PROVIDER": "gemini"}, clear=True):
with self.assertRaisesRegex(EnvironmentError, "GEMINI_API_KEY"):
providers.get_provider()
def test_unknown_provider_is_actionable(self) -> None:
with mock.patch.dict(os.environ, {**clean_env(), "EMBEDDING_PROVIDER": "wat"}, clear=True):
with self.assertRaisesRegex(ValueError, "Supported: gemini, openai, ollama"):
providers.get_provider()
def test_incompatible_model_is_rejected_before_provider_creation(self) -> None:
with mock.patch.dict(os.environ, {**clean_env(),
"EMBEDDING_PROVIDER": "gemini",
"EMBEDDING_MODEL": "text-embedding-3-small",
"GEMINI_API_KEY": "runtime-only",
}, clear=True):
with self.assertRaisesRegex(ValueError, "not compatible"):
providers.get_provider()
def test_endpoint_must_be_http_url(self) -> None:
with mock.patch.dict(os.environ, {**clean_env(),
"EMBEDDING_PROVIDER": "gemini",
"GEMINI_API_KEY": "runtime-only",
"EMBEDDING_ENDPOINT": "file:///tmp/secret",
}, clear=True):
with self.assertRaisesRegex(ValueError, r"HTTP\(S\)"):
providers.get_provider()
def test_malformed_config_types_are_actionable(self) -> None:
malformed = {
"embedding": {
"provider": ["gemini"],
"model": "gemini-embedding-2-preview",
"endpoint": "https://example.test/models",
"workspace_root": "",
}
}
with mock.patch.dict(os.environ, {**clean_env(), }, clear=True), \
mock.patch.object(providers, "_load_xkb_config", return_value=malformed):
with self.assertRaisesRegex(ValueError, r"embedding\.provider.*string"):
providers.load_config()
def test_malformed_workspace_root_type_is_actionable(self) -> None:
malformed = {"embedding": {"workspace_root": 123}}
with mock.patch.dict(os.environ, {**clean_env(), }, clear=True), \
mock.patch.object(providers, "_load_xkb_config", return_value=malformed):
with self.assertRaisesRegex(ValueError, r"embedding\.workspace_root.*string"):
providers.load_config()
def test_falsy_malformed_config_types_are_not_defaulted(self) -> None:
for setting, value in {
"provider": [],
"model": {},
"endpoint": 0,
"workspace_root": False,
}.items():
with self.subTest(setting=setting):
malformed = {"embedding": {setting: value}}
with mock.patch.dict(os.environ, {**clean_env(), }, clear=True), \
mock.patch.object(providers, "_load_xkb_config", return_value=malformed):
with self.assertRaisesRegex(ValueError, rf"embedding\.{setting}.*string"):
providers.load_config()
def test_cli_reports_configuration_errors_without_traceback(self) -> None:
import contextlib
import io
import tempfile
from scripts import build_vector_index
import build_fts_index
with tempfile.TemporaryDirectory() as tmp:
index_path = Path(tmp) / "search_index.json"
index_path.write_text('{"items": [{"relative_path": "card.md"}]}', encoding="utf-8")
args = ["build_vector_index.py", "--index-file", str(index_path), "--vector-file", str(Path(tmp) / "vectors.json")]
stderr = io.StringIO()
with mock.patch.object(sys, "argv", args), \
mock.patch.object(build_vector_index.xkb_paths, "BOOKMARKS_DIR", Path(tmp)), \
mock.patch.object(build_fts_index, "FTS_DB", Path(tmp) / "fts_index.db"), \
mock.patch.object(build_fts_index, "knowledge_section_docs", return_value=[]), \
mock.patch.object(build_vector_index, "knowledge_section_docs", return_value=[]), \
mock.patch.object(build_vector_index, "extract_card_text", return_value="text"), \
mock.patch.object(build_vector_index, "load_vector_index", return_value={}), \
mock.patch.object(build_vector_index, "get_provider", side_effect=ValueError("embedding.provider must be a string")), \
contextlib.redirect_stderr(stderr):
result = build_vector_index.main()
self.assertEqual(result, 1)
self.assertIn("embedding.provider must be a string", stderr.getvalue())
self.assertNotIn("Traceback", stderr.getvalue())
class ProviderRegistryTests(unittest.TestCase):
def test_active_embedding_entry_points_do_not_read_private_runtime_config(self) -> None:
scripts_dir = Path(__file__).resolve().parents[1] / "scripts"
checked = {
"continuity_recall.py",
"health_check.py",
"build_vector_index.py",
}
forbidden = ('Path.home() / ".openclaw"', "openclaw.json")
for name in checked:
source = (scripts_dir / name).read_text(encoding="utf-8")
for token in forbidden:
self.assertNotIn(token, source, f"{name} must use the shared loader")
def test_registry_contains_supported_providers(self) -> None:
self.assertEqual(set(providers.PROVIDER_REGISTRY), {"gemini", "openai", "ollama"})
def test_gemini_batch_request_remains_mockable(self) -> None:
provider = providers.GeminiProvider("runtime-only")
with mock.patch.object(providers, "_post", return_value={"embeddings": [{"values": [1.0]}]}) as post:
self.assertEqual(provider.embed_batch(["hello"]), [[1.0]])
self.assertIn(":batchEmbedContents", post.call_args.args[0])
def test_gemini_single_request_uses_configured_endpoint_and_redacts_api_error(self) -> None:
provider = providers.GeminiProvider("sanitized-gemini-key", base_url="https://mock.example/v1/models")
response = mock.Mock(ok=False, status_code=401, text="echo sanitized-gemini-key")
with mock.patch.object(providers.requests, "post", return_value=response) as post:
with self.assertRaisesRegex(RuntimeError, r"Embedding API error 401") as raised:
provider.embed("fixture text")
self.assertNotIn("sanitized-gemini-key", str(raised.exception))
self.assertEqual(post.call_args.args[0], "https://mock.example/v1/models/gemini-embedding-2-preview:embedContent?key=sanitized-gemini-key")
def test_configured_workspace_root_is_metadata_only_and_does_not_read_private_path(self) -> None:
malformed = {"embedding": {"workspace_root": "/private/not-a-checkout"}}
with mock.patch.dict(os.environ, {**clean_env(), }, clear=True), \
mock.patch.object(providers, "_load_xkb_config", return_value=malformed):
config = providers.load_config()
self.assertEqual(config.workspace_root, "/private/not-a-checkout")
def test_ollama_provider_is_credential_free_and_mockable(self) -> None:
with mock.patch.dict(os.environ, {**clean_env(), "EMBEDDING_PROVIDER": "ollama"}, clear=True):
provider = providers.get_provider()
with mock.patch.object(providers, "_post", return_value={"embedding": [0.1, 0.2]}):
self.assertEqual(provider.embed("fixture"), [0.1, 0.2])
def test_health_cosine_rejects_mixed_vector_dimensions(self) -> None:
import importlib.util
health_path = Path(__file__).resolve().parents[1] / "scripts" / "health_check.py"
spec = importlib.util.spec_from_file_location("health_check_fixture", health_path)
self.assertIsNotNone(spec)
assert spec is not None and spec.loader is not None
health_check = importlib.util.module_from_spec(spec)
spec.loader.exec_module(health_check)
with self.assertRaisesRegex(ValueError, "dimension mismatch"):
health_check.cosine_similarity([1.0, 0.0], [1.0])
def test_health_embed_passes_env_file_to_shared_loader(self) -> None:
import importlib.util
health_path = Path(__file__).resolve().parents[1] / "scripts" / "health_check.py"
spec = importlib.util.spec_from_file_location("health_check_env_fixture", health_path)
assert spec is not None and spec.loader is not None
health_check = importlib.util.module_from_spec(spec)
spec.loader.exec_module(health_check)
provider = mock.Mock()
provider.embed.return_value = [0.1, 0.2]
with mock.patch.object(health_check, "get_provider", return_value=provider) as loader:
result = health_check._embed("ignored-legacy-key", "fixture", env_file="fixture.env")
self.assertEqual(result, [0.1, 0.2])
loader.assert_called_once_with(env_file="fixture.env")
if __name__ == "__main__":
unittest.main()