mirror of
https://github.com/cocoindex-io/cocoindex-code.git
synced 2026-09-14 16:39:38 +08:00
870 lines
33 KiB
Python
870 lines
33 KiB
Python
"""Unit tests for the settings module."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
from collections.abc import Iterator
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
# _resolve_chunker_registry is private to daemon.py (single call site), but its
|
|
# error paths (bad format, non-callable) are not exercised by integration tests.
|
|
from cocoindex_code.daemon import _resolve_chunker_registry
|
|
from cocoindex_code.settings import (
|
|
DEFAULT_EXCLUDED_PATTERNS,
|
|
DEFAULT_INCLUDED_PATTERNS,
|
|
ChunkerMapping,
|
|
DaemonSettings,
|
|
EmbeddingSettings,
|
|
LanguageOverride,
|
|
ProjectSettings,
|
|
UserSettings,
|
|
_reset_db_path_mapping_cache,
|
|
_reset_host_path_mapping_cache,
|
|
_user_settings_from_dict,
|
|
default_project_settings,
|
|
default_user_settings,
|
|
find_parent_with_marker,
|
|
find_project_root,
|
|
format_path_for_display,
|
|
get_host_path_mappings,
|
|
load_project_settings,
|
|
load_user_settings,
|
|
normalize_input_path,
|
|
parse_file_size,
|
|
resolve_db_dir,
|
|
save_project_settings,
|
|
save_user_settings,
|
|
)
|
|
|
|
|
|
@pytest.fixture()
|
|
def _patch_user_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Redirect user_settings_dir() to a temp directory."""
|
|
monkeypatch.setattr(
|
|
"cocoindex_code.settings.user_settings_dir",
|
|
lambda: tmp_path / ".cocoindex_code",
|
|
)
|
|
monkeypatch.setattr(
|
|
"cocoindex_code.settings.user_settings_path",
|
|
lambda: tmp_path / ".cocoindex_code" / "global_settings.yml",
|
|
)
|
|
|
|
|
|
def test_default_user_settings() -> None:
|
|
s = default_user_settings()
|
|
assert s.embedding.provider == "sentence-transformers"
|
|
assert s.embedding.model == "Snowflake/snowflake-arctic-embed-xs"
|
|
assert s.embedding.device is None
|
|
assert s.embedding.min_interval_ms is None
|
|
assert s.embedding.mps_low_watermark_ratio == 0.4
|
|
assert s.embedding.mps_high_watermark_ratio == 0.5
|
|
assert s.envs == {}
|
|
|
|
|
|
def test_default_project_settings() -> None:
|
|
s = default_project_settings()
|
|
assert s.include_patterns == DEFAULT_INCLUDED_PATTERNS
|
|
assert s.exclude_patterns == DEFAULT_EXCLUDED_PATTERNS
|
|
assert s.language_overrides == []
|
|
|
|
|
|
def test_default_included_patterns_cover_dart() -> None:
|
|
assert "**/*.dart" in DEFAULT_INCLUDED_PATTERNS
|
|
|
|
|
|
def test_default_included_patterns_cover_elixir() -> None:
|
|
assert "**/*.ex" in DEFAULT_INCLUDED_PATTERNS
|
|
assert "**/*.exs" in DEFAULT_INCLUDED_PATTERNS
|
|
|
|
|
|
@pytest.mark.usefixtures("_patch_user_dir")
|
|
def test_save_and_load_user_settings(tmp_path: Path) -> None:
|
|
settings = UserSettings(
|
|
embedding=EmbeddingSettings(
|
|
provider="litellm",
|
|
model="gemini/text-embedding-004",
|
|
device="cpu",
|
|
min_interval_ms=300,
|
|
mps_low_watermark_ratio=0.35,
|
|
mps_high_watermark_ratio=0.45,
|
|
),
|
|
envs={"GEMINI_API_KEY": "test-key"},
|
|
)
|
|
save_user_settings(settings)
|
|
loaded = load_user_settings()
|
|
assert loaded.embedding.provider == settings.embedding.provider
|
|
assert loaded.embedding.model == settings.embedding.model
|
|
assert loaded.embedding.device == settings.embedding.device
|
|
assert loaded.embedding.min_interval_ms == settings.embedding.min_interval_ms
|
|
assert loaded.embedding.mps_low_watermark_ratio == settings.embedding.mps_low_watermark_ratio
|
|
assert loaded.embedding.mps_high_watermark_ratio == settings.embedding.mps_high_watermark_ratio
|
|
assert loaded.envs == settings.envs
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "value", "message"),
|
|
[
|
|
("mps_low_watermark_ratio", 0, "mps_low_watermark_ratio"),
|
|
("mps_high_watermark_ratio", 1.1, "mps_high_watermark_ratio"),
|
|
],
|
|
)
|
|
def test_embedding_safety_settings_reject_invalid_values(
|
|
field: str,
|
|
value: int | float,
|
|
message: str,
|
|
) -> None:
|
|
with pytest.raises(ValueError, match=message):
|
|
EmbeddingSettings(model="model", **{field: value})
|
|
|
|
|
|
def test_embedding_safety_settings_require_ordered_mps_limits() -> None:
|
|
with pytest.raises(ValueError, match="mps_low_watermark_ratio"):
|
|
EmbeddingSettings(
|
|
model="model",
|
|
mps_low_watermark_ratio=0.6,
|
|
mps_high_watermark_ratio=0.5,
|
|
)
|
|
|
|
|
|
def test_removed_custom_mps_worker_settings_are_ignored() -> None:
|
|
settings = _user_settings_from_dict(
|
|
{
|
|
"embedding": {
|
|
"provider": "sentence-transformers",
|
|
"model": "model",
|
|
"batch_size": 8,
|
|
"mps_memory_limit_ratio": 0.35,
|
|
"worker_timeout_seconds": 300,
|
|
}
|
|
}
|
|
)
|
|
|
|
assert settings.embedding.mps_low_watermark_ratio == 0.4
|
|
assert settings.embedding.mps_high_watermark_ratio == 0.5
|
|
|
|
|
|
def test_save_and_load_project_settings(tmp_path: Path) -> None:
|
|
settings = ProjectSettings(
|
|
include_patterns=["**/*.py", "**/*.rs"],
|
|
exclude_patterns=["**/target"],
|
|
language_overrides=[LanguageOverride(ext="inc", lang="php")],
|
|
)
|
|
save_project_settings(tmp_path, settings)
|
|
loaded = load_project_settings(tmp_path)
|
|
assert loaded.include_patterns == settings.include_patterns
|
|
assert loaded.exclude_patterns == settings.exclude_patterns
|
|
assert len(loaded.language_overrides) == 1
|
|
assert loaded.language_overrides[0].ext == "inc"
|
|
assert loaded.language_overrides[0].lang == "php"
|
|
|
|
|
|
@pytest.mark.usefixtures("_patch_user_dir")
|
|
def test_load_user_settings_missing_file_raises() -> None:
|
|
with pytest.raises(FileNotFoundError):
|
|
load_user_settings()
|
|
|
|
|
|
@pytest.mark.usefixtures("_patch_user_dir")
|
|
def test_load_user_settings_empty_file_raises(tmp_path: Path) -> None:
|
|
path = tmp_path / ".cocoindex_code" / "global_settings.yml"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("{}\n")
|
|
with pytest.raises(ValueError):
|
|
load_user_settings()
|
|
|
|
|
|
@pytest.mark.usefixtures("_patch_user_dir")
|
|
def test_load_user_settings_missing_model_raises(tmp_path: Path) -> None:
|
|
path = tmp_path / ".cocoindex_code" / "global_settings.yml"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("embedding:\n provider: litellm\n")
|
|
with pytest.raises(ValueError):
|
|
load_user_settings()
|
|
|
|
|
|
@pytest.mark.usefixtures("_patch_user_dir")
|
|
def test_from_dict_missing_provider_defaults_to_litellm() -> None:
|
|
from cocoindex_code.settings import _user_settings_from_dict
|
|
|
|
settings = _user_settings_from_dict({"embedding": {"model": "some/model"}})
|
|
assert settings.embedding.provider == "litellm"
|
|
assert settings.embedding.model == "some/model"
|
|
assert settings.embedding.min_interval_ms is None
|
|
|
|
|
|
@pytest.mark.usefixtures("_patch_user_dir")
|
|
def test_save_default_settings_writes_explicit_embedding() -> None:
|
|
from cocoindex_code.settings import user_settings_path
|
|
|
|
save_user_settings(default_user_settings())
|
|
content = user_settings_path().read_text()
|
|
assert "provider:" in content
|
|
assert "model:" in content
|
|
assert "Snowflake/snowflake-arctic-embed-xs" in content
|
|
|
|
|
|
def test_load_project_settings_missing_file_raises(tmp_path: Path) -> None:
|
|
with pytest.raises(FileNotFoundError):
|
|
load_project_settings(tmp_path)
|
|
|
|
|
|
def test_find_project_root_from_subdirectory(tmp_path: Path) -> None:
|
|
project = tmp_path / "project"
|
|
(project / ".cocoindex_code").mkdir(parents=True)
|
|
(project / ".cocoindex_code" / "settings.yml").write_text("include_patterns: []")
|
|
subdir = project / "src" / "lib"
|
|
subdir.mkdir(parents=True)
|
|
assert find_project_root(subdir) == project
|
|
|
|
|
|
def test_find_project_root_from_project_root(tmp_path: Path) -> None:
|
|
project = tmp_path / "project"
|
|
(project / ".cocoindex_code").mkdir(parents=True)
|
|
(project / ".cocoindex_code" / "settings.yml").write_text("include_patterns: []")
|
|
assert find_project_root(project) == project
|
|
|
|
|
|
def test_find_project_root_returns_none_when_not_initialized(tmp_path: Path) -> None:
|
|
standalone = tmp_path / "standalone"
|
|
standalone.mkdir()
|
|
assert find_project_root(standalone) is None
|
|
|
|
|
|
def test_find_parent_with_marker_finds_git(tmp_path: Path) -> None:
|
|
repo = tmp_path / "repo"
|
|
(repo / ".git").mkdir(parents=True)
|
|
subdir = repo / "src"
|
|
subdir.mkdir()
|
|
assert find_parent_with_marker(subdir) == repo
|
|
|
|
|
|
def test_find_parent_with_marker_prefers_cocoindex_code(tmp_path: Path) -> None:
|
|
repo = tmp_path / "repo"
|
|
(repo / ".git").mkdir(parents=True)
|
|
(repo / ".cocoindex_code").mkdir(parents=True)
|
|
subdir = repo / "src"
|
|
subdir.mkdir()
|
|
assert find_parent_with_marker(subdir) == repo
|
|
|
|
|
|
@pytest.mark.usefixtures("_patch_user_dir")
|
|
def test_user_settings_litellm_round_trip() -> None:
|
|
settings = UserSettings(
|
|
embedding=EmbeddingSettings(
|
|
provider="litellm",
|
|
model="gemini/text-embedding-004",
|
|
min_interval_ms=250,
|
|
),
|
|
envs={"GEMINI_API_KEY": "test"},
|
|
)
|
|
save_user_settings(settings)
|
|
loaded = load_user_settings()
|
|
assert loaded.embedding.provider == "litellm"
|
|
assert loaded.embedding.model == "gemini/text-embedding-004"
|
|
assert loaded.embedding.min_interval_ms == 250
|
|
assert loaded.envs == {"GEMINI_API_KEY": "test"}
|
|
|
|
|
|
@pytest.mark.usefixtures("_patch_user_dir")
|
|
def test_load_user_settings_with_min_interval_ms(tmp_path: Path) -> None:
|
|
path = tmp_path / ".cocoindex_code" / "global_settings.yml"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(
|
|
"embedding:\n provider: litellm\n model: text-embedding-3-small\n min_interval_ms: 300\n"
|
|
)
|
|
loaded = load_user_settings()
|
|
assert loaded.embedding.provider == "litellm"
|
|
assert loaded.embedding.model == "text-embedding-3-small"
|
|
assert loaded.embedding.min_interval_ms == 300
|
|
|
|
|
|
def test_project_settings_with_language_overrides(tmp_path: Path) -> None:
|
|
settings = ProjectSettings(
|
|
language_overrides=[LanguageOverride(ext="inc", lang="php")],
|
|
)
|
|
save_project_settings(tmp_path, settings)
|
|
loaded = load_project_settings(tmp_path)
|
|
assert len(loaded.language_overrides) == 1
|
|
assert loaded.language_overrides[0].ext == "inc"
|
|
assert loaded.language_overrides[0].lang == "php"
|
|
|
|
|
|
class TestResolveDbDir:
|
|
"""Tests for COCOINDEX_CODE_DB_PATH_MAPPING and resolve_db_dir()."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear_cache(self, monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
|
|
"""Reset cached mapping before each test."""
|
|
_reset_db_path_mapping_cache()
|
|
monkeypatch.delenv("COCOINDEX_CODE_DB_PATH_MAPPING", raising=False)
|
|
yield
|
|
_reset_db_path_mapping_cache()
|
|
|
|
def test_no_mapping(self, tmp_path: Path) -> None:
|
|
project = tmp_path / "myproject"
|
|
assert resolve_db_dir(project) == project / ".cocoindex_code"
|
|
|
|
def test_single_mapping_match(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
src = tmp_path / "workspace"
|
|
dst = tmp_path / "db-files"
|
|
monkeypatch.setenv("COCOINDEX_CODE_DB_PATH_MAPPING", f"{src}={dst}")
|
|
assert resolve_db_dir(src / "myproject") == dst / "myproject"
|
|
|
|
def test_exact_root_match(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
src = tmp_path / "workspace"
|
|
dst = tmp_path / "db-files"
|
|
monkeypatch.setenv("COCOINDEX_CODE_DB_PATH_MAPPING", f"{src}={dst}")
|
|
assert resolve_db_dir(src) == dst
|
|
|
|
def test_no_match_falls_back(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
src = tmp_path / "workspace"
|
|
dst = tmp_path / "db-files"
|
|
other = tmp_path / "other" / "myproject"
|
|
monkeypatch.setenv("COCOINDEX_CODE_DB_PATH_MAPPING", f"{src}={dst}")
|
|
assert resolve_db_dir(other) == other / ".cocoindex_code"
|
|
|
|
def test_multiple_mappings_first_wins(
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
src = tmp_path / "workspace"
|
|
dst1 = tmp_path / "db1"
|
|
dst2 = tmp_path / "db2"
|
|
monkeypatch.setenv("COCOINDEX_CODE_DB_PATH_MAPPING", f"{src}={dst1},{src / 'sub'}={dst2}")
|
|
assert resolve_db_dir(src / "sub" / "proj") == dst1 / "sub" / "proj"
|
|
|
|
def test_multiple_mappings_second_matches(
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
src1 = tmp_path / "workspace"
|
|
src2 = tmp_path / "other"
|
|
dst1 = tmp_path / "db1"
|
|
dst2 = tmp_path / "db2"
|
|
monkeypatch.setenv("COCOINDEX_CODE_DB_PATH_MAPPING", f"{src1}={dst1},{src2}={dst2}")
|
|
assert resolve_db_dir(src2 / "proj") == dst2 / "proj"
|
|
|
|
def test_no_partial_component_match(
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
src = tmp_path / "workspace"
|
|
dst = tmp_path / "db-files"
|
|
other = tmp_path / "workspace2" / "proj"
|
|
monkeypatch.setenv("COCOINDEX_CODE_DB_PATH_MAPPING", f"{src}={dst}")
|
|
assert resolve_db_dir(other) == other / ".cocoindex_code"
|
|
|
|
def test_rejects_relative_source(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("COCOINDEX_CODE_DB_PATH_MAPPING", "relative/path=/db-files")
|
|
with pytest.raises(ValueError, match="source path must be absolute"):
|
|
resolve_db_dir(Path("/anything"))
|
|
|
|
def test_rejects_relative_target(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
src = tmp_path / "workspace"
|
|
monkeypatch.setenv("COCOINDEX_CODE_DB_PATH_MAPPING", f"{src}=relative/path")
|
|
with pytest.raises(ValueError, match="target path must be absolute"):
|
|
resolve_db_dir(tmp_path / "anything")
|
|
|
|
def test_skips_empty_entries(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
src1 = tmp_path / "workspace"
|
|
src2 = tmp_path / "other"
|
|
dst1 = tmp_path / "db-files"
|
|
dst2 = tmp_path / "db2"
|
|
monkeypatch.setenv("COCOINDEX_CODE_DB_PATH_MAPPING", f"{src1}={dst1},,{src2}={dst2},")
|
|
assert resolve_db_dir(src2 / "proj") == dst2 / "proj"
|
|
|
|
def test_nested_project(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
src = tmp_path / "workspace"
|
|
dst = tmp_path / "db-files"
|
|
monkeypatch.setenv("COCOINDEX_CODE_DB_PATH_MAPPING", f"{src}={dst}")
|
|
assert resolve_db_dir(src / "org" / "repo" / "subdir") == dst / "org" / "repo" / "subdir"
|
|
|
|
|
|
def test_project_settings_with_chunkers(tmp_path: Path) -> None:
|
|
settings = ProjectSettings(
|
|
chunkers=[ChunkerMapping(ext="toml", module="example_toml_chunker:toml_chunker")],
|
|
)
|
|
save_project_settings(tmp_path, settings)
|
|
loaded = load_project_settings(tmp_path)
|
|
assert len(loaded.chunkers) == 1
|
|
assert loaded.chunkers[0].ext == "toml"
|
|
assert loaded.chunkers[0].module == "example_toml_chunker:toml_chunker"
|
|
|
|
|
|
def test_resolve_chunker_registry_missing_colon() -> None:
|
|
with pytest.raises(ValueError, match="module.path:callable"):
|
|
_resolve_chunker_registry([ChunkerMapping(ext="toml", module="no_colon_here")])
|
|
|
|
|
|
def test_resolve_chunker_registry_not_callable() -> None:
|
|
# os.path is a module attribute that is a string — not callable.
|
|
with pytest.raises(ValueError, match="not callable"):
|
|
_resolve_chunker_registry([ChunkerMapping(ext="toml", module="os:sep")])
|
|
|
|
|
|
@pytest.mark.usefixtures("_patch_user_dir")
|
|
def test_save_initial_user_settings_round_trip() -> None:
|
|
from cocoindex_code.settings import (
|
|
save_initial_user_settings,
|
|
user_settings_path,
|
|
)
|
|
|
|
emb = EmbeddingSettings(
|
|
provider="sentence-transformers",
|
|
model="Snowflake/snowflake-arctic-embed-xs",
|
|
)
|
|
path = save_initial_user_settings(emb, defaults_applied=False)
|
|
content = path.read_text()
|
|
|
|
# Hint comment, MPS allocator defaults, and env-var examples.
|
|
assert "ccc doctor" in content
|
|
assert "# mps_low_watermark_ratio: 0.4" in content
|
|
assert "# mps_high_watermark_ratio: 0.5" in content
|
|
assert "CocoIndex's GPU subprocess" in content
|
|
assert "# envs:" in content
|
|
for key in ("OPENAI_API_KEY", "GEMINI_API_KEY", "ANTHROPIC_API_KEY", "VOYAGE_API_KEY"):
|
|
assert f"# {key}:" in content
|
|
|
|
# Must round-trip through the normal loader.
|
|
loaded = load_user_settings()
|
|
assert loaded.embedding.provider == "sentence-transformers"
|
|
assert loaded.embedding.model == "Snowflake/snowflake-arctic-embed-xs"
|
|
|
|
# user_settings_path() is the same path returned by save_initial_user_settings.
|
|
assert path == user_settings_path()
|
|
|
|
|
|
@pytest.mark.usefixtures("_patch_user_dir")
|
|
def test_save_initial_user_settings_model_with_colon() -> None:
|
|
"""Regression: LiteLLM model names can contain `:`; must stay parseable."""
|
|
from cocoindex_code.settings import save_initial_user_settings
|
|
|
|
emb = EmbeddingSettings(
|
|
provider="litellm",
|
|
model="ollama_chat/llama3:latest",
|
|
)
|
|
save_initial_user_settings(emb, defaults_applied=False)
|
|
|
|
loaded = load_user_settings()
|
|
assert loaded.embedding.provider == "litellm"
|
|
assert loaded.embedding.model == "ollama_chat/llama3:latest"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Host path mapping (COCOINDEX_CODE_HOST_PATH_MAPPING)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestHostPathMapping:
|
|
"""Tests for format_path_for_display / normalize_input_path and the shared parser."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear_cache(self, monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
|
|
_reset_host_path_mapping_cache()
|
|
monkeypatch.delenv("COCOINDEX_CODE_HOST_PATH_MAPPING", raising=False)
|
|
yield
|
|
_reset_host_path_mapping_cache()
|
|
|
|
def test_translates_display(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
container = tmp_path / "workspace"
|
|
host = tmp_path / "alice"
|
|
container.mkdir()
|
|
host.mkdir()
|
|
monkeypatch.setenv("COCOINDEX_CODE_HOST_PATH_MAPPING", f"{container}={host}")
|
|
assert format_path_for_display(container / "proj" / "app.py") == str(
|
|
host / "proj" / "app.py"
|
|
)
|
|
|
|
def test_translates_input(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
container = tmp_path / "workspace"
|
|
host = tmp_path / "alice"
|
|
container.mkdir()
|
|
host.mkdir()
|
|
monkeypatch.setenv("COCOINDEX_CODE_HOST_PATH_MAPPING", f"{container}={host}")
|
|
assert normalize_input_path(host / "proj") == str(container / "proj")
|
|
|
|
def test_unmatched_absolute_passes_through(
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
container = tmp_path / "workspace"
|
|
host = tmp_path / "alice"
|
|
container.mkdir()
|
|
host.mkdir()
|
|
monkeypatch.setenv("COCOINDEX_CODE_HOST_PATH_MAPPING", f"{container}={host}")
|
|
unrelated = "/etc/hosts"
|
|
assert format_path_for_display(unrelated) == unrelated
|
|
assert normalize_input_path(unrelated) == unrelated
|
|
|
|
def test_relative_passes_through(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
container = tmp_path / "workspace"
|
|
host = tmp_path / "alice"
|
|
container.mkdir()
|
|
host.mkdir()
|
|
monkeypatch.setenv("COCOINDEX_CODE_HOST_PATH_MAPPING", f"{container}={host}")
|
|
assert format_path_for_display("src/app.py") == "src/app.py"
|
|
assert normalize_input_path("src/app.py") == "src/app.py"
|
|
|
|
def test_first_match_wins(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
ws = tmp_path / "workspace"
|
|
shared = ws / "shared"
|
|
host_ws = tmp_path / "alice"
|
|
host_shared = tmp_path / "mnt-shared"
|
|
ws.mkdir()
|
|
shared.mkdir(parents=True)
|
|
host_ws.mkdir()
|
|
host_shared.mkdir()
|
|
monkeypatch.setenv(
|
|
"COCOINDEX_CODE_HOST_PATH_MAPPING",
|
|
f"{ws}={host_ws},{shared}={host_shared}",
|
|
)
|
|
# Path under shared — first mapping wins, not the more-specific one.
|
|
assert format_path_for_display(shared / "docs" / "x") == str(
|
|
host_ws / "shared" / "docs" / "x"
|
|
)
|
|
|
|
def test_env_unset_is_noop(self) -> None:
|
|
# Fixture already clears env var.
|
|
assert format_path_for_display("/workspace/x") == "/workspace/x"
|
|
assert normalize_input_path("/workspace/x") == "/workspace/x"
|
|
assert get_host_path_mappings() == []
|
|
|
|
def test_invalid_env_raises(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("COCOINDEX_CODE_HOST_PATH_MAPPING", "relative=/abs")
|
|
with pytest.raises(ValueError, match="source path must be absolute"):
|
|
get_host_path_mappings()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# daemon settings (idle timeout)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.usefixtures("_patch_user_dir")
|
|
def test_daemon_settings_absent_section_uses_default(tmp_path: Path) -> None:
|
|
path = tmp_path / ".cocoindex_code" / "global_settings.yml"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("embedding:\n provider: litellm\n model: m\n")
|
|
loaded = load_user_settings()
|
|
assert loaded.daemon.idle_timeout_minutes == 180
|
|
assert loaded.daemon.keep_alive_with_mcp is True
|
|
|
|
|
|
@pytest.mark.usefixtures("_patch_user_dir")
|
|
def test_daemon_settings_parses_idle_timeout(tmp_path: Path) -> None:
|
|
path = tmp_path / ".cocoindex_code" / "global_settings.yml"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(
|
|
"embedding:\n provider: litellm\n model: m\ndaemon:\n idle_timeout_minutes: 30\n"
|
|
)
|
|
loaded = load_user_settings()
|
|
assert loaded.daemon.idle_timeout_minutes == 30
|
|
|
|
|
|
@pytest.mark.usefixtures("_patch_user_dir")
|
|
def test_daemon_settings_can_disable_mcp_keep_alive(tmp_path: Path) -> None:
|
|
path = tmp_path / ".cocoindex_code" / "global_settings.yml"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(
|
|
"embedding:\n provider: litellm\n model: m\ndaemon:\n keep_alive_with_mcp: false\n"
|
|
)
|
|
loaded = load_user_settings()
|
|
assert loaded.daemon.keep_alive_with_mcp is False
|
|
|
|
|
|
@pytest.mark.parametrize("value", ["false", 0, None])
|
|
def test_daemon_settings_rejects_non_boolean_mcp_keep_alive(value: object) -> None:
|
|
with pytest.raises(ValueError, match="keep_alive_with_mcp must be a boolean"):
|
|
_user_settings_from_dict(
|
|
{
|
|
"embedding": {"provider": "litellm", "model": "m"},
|
|
"daemon": {"keep_alive_with_mcp": value},
|
|
}
|
|
)
|
|
|
|
|
|
@pytest.mark.usefixtures("_patch_user_dir")
|
|
def test_daemon_settings_explicit_zero_means_never(tmp_path: Path) -> None:
|
|
path = tmp_path / ".cocoindex_code" / "global_settings.yml"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(
|
|
"embedding:\n provider: litellm\n model: m\ndaemon:\n idle_timeout_minutes: 0\n"
|
|
)
|
|
loaded = load_user_settings()
|
|
assert loaded.daemon.idle_timeout_minutes == 0
|
|
|
|
|
|
@pytest.mark.usefixtures("_patch_user_dir")
|
|
def test_daemon_settings_round_trip() -> None:
|
|
settings = UserSettings(
|
|
embedding=EmbeddingSettings(provider="litellm", model="m"),
|
|
daemon=DaemonSettings(idle_timeout_minutes=45, keep_alive_with_mcp=False),
|
|
)
|
|
save_user_settings(settings)
|
|
loaded = load_user_settings()
|
|
assert loaded.daemon.idle_timeout_minutes == 45
|
|
assert loaded.daemon.keep_alive_with_mcp is False
|
|
|
|
|
|
@pytest.mark.usefixtures("_patch_user_dir")
|
|
def test_daemon_settings_default_omitted_from_yaml() -> None:
|
|
from cocoindex_code.settings import user_settings_path
|
|
|
|
settings = UserSettings(embedding=EmbeddingSettings(provider="litellm", model="m"))
|
|
save_user_settings(settings)
|
|
assert "daemon" not in user_settings_path().read_text()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# find_parent_with_marker — global-only should not match
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_find_parent_with_marker_skips_global_only(tmp_path: Path) -> None:
|
|
"""A workspace-root ``.cocoindex_code/`` holding only ``global_settings.yml``
|
|
should NOT trigger the parent-marker check (it's not a project).
|
|
"""
|
|
ws = tmp_path / "ws"
|
|
(ws / ".cocoindex_code").mkdir(parents=True)
|
|
(ws / ".cocoindex_code" / "global_settings.yml").write_text("embedding: {model: x}\n")
|
|
subdir = ws / "myproject"
|
|
subdir.mkdir()
|
|
assert find_parent_with_marker(subdir) is None
|
|
|
|
|
|
def test_find_parent_with_marker_detects_project_settings(tmp_path: Path) -> None:
|
|
"""``.cocoindex_code/settings.yml`` at a parent is a real project marker."""
|
|
repo = tmp_path / "repo"
|
|
(repo / ".cocoindex_code").mkdir(parents=True)
|
|
(repo / ".cocoindex_code" / "settings.yml").write_text("include_patterns: []\n")
|
|
subdir = repo / "src"
|
|
subdir.mkdir()
|
|
assert find_parent_with_marker(subdir) == repo
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# daemon_runtime_dir
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_daemon_runtime_dir_uses_env_var(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from cocoindex_code._daemon_paths import daemon_runtime_dir
|
|
|
|
target = tmp_path / "runtime"
|
|
monkeypatch.setenv("COCOINDEX_CODE_RUNTIME_DIR", str(target))
|
|
assert daemon_runtime_dir() == target
|
|
|
|
|
|
def test_daemon_runtime_dir_falls_back_to_user_settings_dir(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""When COCOINDEX_CODE_RUNTIME_DIR is unset, falls back to user_settings_dir()."""
|
|
from cocoindex_code._daemon_paths import daemon_runtime_dir
|
|
|
|
settings_dir = tmp_path / "settings"
|
|
monkeypatch.delenv("COCOINDEX_CODE_RUNTIME_DIR", raising=False)
|
|
monkeypatch.setenv("COCOINDEX_CODE_DIR", str(settings_dir))
|
|
assert daemon_runtime_dir() == settings_dir
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# daemon_socket_path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.skipif(sys.platform == "win32", reason="named pipes have no length limit")
|
|
def test_daemon_socket_path_uses_runtime_dir_when_short(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""The common case is unchanged: socket sits in the runtime dir.
|
|
|
|
Deliberately not pytest's ``tmp_path`` — on macOS that is ~118 bytes, past
|
|
sun_path already, which is how routine this overflow is.
|
|
"""
|
|
from cocoindex_code._daemon_paths import daemon_socket_path
|
|
|
|
short_dir = tempfile.mkdtemp(prefix="ccc", dir=tempfile.gettempdir())
|
|
try:
|
|
monkeypatch.setenv("COCOINDEX_CODE_RUNTIME_DIR", short_dir)
|
|
assert daemon_socket_path() == str(Path(short_dir) / "daemon.sock")
|
|
finally:
|
|
shutil.rmtree(short_dir, ignore_errors=True)
|
|
|
|
|
|
@pytest.mark.skipif(sys.platform == "win32", reason="named pipes have no length limit")
|
|
def test_daemon_socket_path_falls_back_when_over_sun_path_limit(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""A deep runtime dir must not produce an unbindable socket address.
|
|
|
|
bind() fails with "AF_UNIX path too long" past sun_path (104 bytes on
|
|
macOS), which surfaces as a generic daemon-startup failure. Seen in the
|
|
wild under sandboxes and containers with long $HOME paths.
|
|
"""
|
|
from cocoindex_code._daemon_paths import _SUN_PATH_MAX, daemon_socket_path
|
|
|
|
deep = tmp_path / ("d" * 80) / ("e" * 80)
|
|
monkeypatch.setenv("COCOINDEX_CODE_RUNTIME_DIR", str(deep))
|
|
|
|
path = daemon_socket_path()
|
|
assert not path.startswith(str(deep))
|
|
assert len(path.encode()) < _SUN_PATH_MAX
|
|
|
|
|
|
@pytest.mark.skipif(sys.platform == "win32", reason="named pipes have no length limit")
|
|
def test_daemon_socket_path_fallback_is_unique_per_runtime_dir(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Two over-long runtime dirs must not collide on one socket address."""
|
|
from cocoindex_code._daemon_paths import daemon_socket_path
|
|
|
|
long_a = tmp_path / ("a" * 80) / ("x" * 80)
|
|
long_b = tmp_path / ("b" * 80) / ("y" * 80)
|
|
|
|
monkeypatch.setenv("COCOINDEX_CODE_RUNTIME_DIR", str(long_a))
|
|
first = daemon_socket_path()
|
|
monkeypatch.setenv("COCOINDEX_CODE_RUNTIME_DIR", str(long_b))
|
|
second = daemon_socket_path()
|
|
|
|
assert first != second
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# indexing_params / query_params round-trip and templates
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.usefixtures("_patch_user_dir")
|
|
def test_embedding_params_missing_load_as_none(tmp_path: Path) -> None:
|
|
path = tmp_path / ".cocoindex_code" / "global_settings.yml"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("embedding:\n provider: litellm\n model: m\n")
|
|
loaded = load_user_settings()
|
|
assert loaded.embedding.indexing_params is None
|
|
assert loaded.embedding.query_params is None
|
|
|
|
|
|
@pytest.mark.usefixtures("_patch_user_dir")
|
|
def test_embedding_params_roundtrip_preserves_empty_dict() -> None:
|
|
settings = UserSettings(
|
|
embedding=EmbeddingSettings(
|
|
provider="sentence-transformers",
|
|
model="x/y",
|
|
indexing_params={"prompt_name": "passage"},
|
|
query_params={}, # explicit empty — must round-trip as {} not None
|
|
),
|
|
)
|
|
save_user_settings(settings)
|
|
loaded = load_user_settings()
|
|
assert loaded.embedding.indexing_params == {"prompt_name": "passage"}
|
|
assert loaded.embedding.query_params == {}
|
|
|
|
|
|
@pytest.mark.usefixtures("_patch_user_dir")
|
|
def test_embedding_params_omit_when_none() -> None:
|
|
settings = UserSettings(
|
|
embedding=EmbeddingSettings(
|
|
provider="litellm",
|
|
model="m",
|
|
),
|
|
)
|
|
save_user_settings(settings)
|
|
from cocoindex_code.settings import user_settings_path
|
|
|
|
content = user_settings_path().read_text()
|
|
assert "indexing_params" not in content
|
|
assert "query_params" not in content
|
|
|
|
|
|
@pytest.mark.usefixtures("_patch_user_dir")
|
|
def test_save_initial_writes_populated_defaults_no_template() -> None:
|
|
from cocoindex_code.settings import save_initial_user_settings, user_settings_path
|
|
|
|
emb = EmbeddingSettings(
|
|
provider="sentence-transformers",
|
|
model="nomic-ai/CodeRankEmbed",
|
|
query_params={"prompt_name": "query"},
|
|
)
|
|
save_initial_user_settings(emb, defaults_applied=True)
|
|
content = user_settings_path().read_text()
|
|
|
|
# Populated as real YAML keys
|
|
assert "query_params:" in content
|
|
assert "prompt_name: query" in content
|
|
# No commented-out template hint
|
|
assert "# indexing_params: {}" not in content
|
|
assert "# query_params: {}" not in content
|
|
|
|
|
|
@pytest.mark.usefixtures("_patch_user_dir")
|
|
def test_save_initial_writes_comment_template_for_unknown_sentence_transformers() -> None:
|
|
from cocoindex_code.settings import save_initial_user_settings, user_settings_path
|
|
|
|
emb = EmbeddingSettings(
|
|
provider="sentence-transformers",
|
|
model="unknown/model",
|
|
)
|
|
save_initial_user_settings(emb, defaults_applied=False)
|
|
content = user_settings_path().read_text()
|
|
|
|
assert "# indexing_params: {}" in content
|
|
assert "# query_params: {}" in content
|
|
assert "prompt_name" in content
|
|
# litellm-only keys should not appear
|
|
assert "input_type" not in content
|
|
|
|
|
|
@pytest.mark.usefixtures("_patch_user_dir")
|
|
def test_save_initial_writes_comment_template_for_unknown_litellm() -> None:
|
|
from cocoindex_code.settings import save_initial_user_settings, user_settings_path
|
|
|
|
emb = EmbeddingSettings(
|
|
provider="litellm",
|
|
model="someprovider/unknown",
|
|
)
|
|
save_initial_user_settings(emb, defaults_applied=False)
|
|
content = user_settings_path().read_text()
|
|
|
|
assert "# indexing_params: {}" in content
|
|
assert "# query_params: {}" in content
|
|
assert "input_type" in content
|
|
# `dimensions` is intentionally NOT in the litellm template — it must be
|
|
# the same on both sides, so we don't expose it as a per-side knob.
|
|
assert "dimensions" not in content
|
|
|
|
|
|
def test_parse_file_size_accepts_units_and_plain_bytes() -> None:
|
|
cases = [
|
|
(1048576, 1048576),
|
|
("2048", 2048),
|
|
("500KB", 500 * 1024),
|
|
("500 kb", 500 * 1024),
|
|
("1MB", 1024**2),
|
|
("1.5MB", int(1.5 * 1024**2)),
|
|
("2GB", 2 * 1024**3),
|
|
("512B", 512),
|
|
]
|
|
for raw, expected in cases:
|
|
assert parse_file_size(raw) == expected, raw
|
|
|
|
|
|
def test_parse_file_size_rejects_invalid_values() -> None:
|
|
for raw in ["", " ", "abc", "10XB", 0, -1, True, None, []]:
|
|
with pytest.raises(ValueError):
|
|
parse_file_size(raw)
|
|
|
|
|
|
def test_project_settings_round_trip_max_file_size(tmp_path: Path) -> None:
|
|
save_project_settings(tmp_path, ProjectSettings(max_file_size=500 * 1024))
|
|
assert load_project_settings(tmp_path).max_file_size == 500 * 1024
|
|
|
|
|
|
def test_project_settings_max_file_size_defaults_to_none(tmp_path: Path) -> None:
|
|
"""Omitting the key keeps the previous behavior of indexing every size."""
|
|
save_project_settings(tmp_path, ProjectSettings())
|
|
assert load_project_settings(tmp_path).max_file_size is None
|
|
|
|
|
|
def test_project_settings_parses_human_readable_max_file_size(tmp_path: Path) -> None:
|
|
path = save_project_settings(tmp_path, ProjectSettings())
|
|
path.write_text(path.read_text() + "\nmax_file_size: 500KB\n")
|
|
assert load_project_settings(tmp_path).max_file_size == 500 * 1024
|