cocoindex-io__cocoindex-code/tests/test_e2e.py

"""End-to-end tests exercising the full CLI → daemon → index → search flow.

Each test function represents a complete session: a series of CLI commands
executed in order, verifying compound stateful effects.  Tests use a real
daemon subprocess (via COCOINDEX_CODE_DIR env var) and the actual CLI
commands through typer's CliRunner.
"""

from __future__ import annotations

import json
import os
import tempfile
from collections.abc import Iterator
from pathlib import Path

import pytest
from cocoindex.connectors import sqlite as coco_sqlite
from conftest import make_test_user_settings
from typer.testing import CliRunner

from cocoindex_code.cli import app
from cocoindex_code.client import stop_daemon
from cocoindex_code.settings import (
    _reset_db_path_mapping_cache,
    default_project_settings,
    find_parent_with_marker,
    load_user_settings,
    save_project_settings,
    save_user_settings,
    user_settings_path,
)

runner = CliRunner()

SAMPLE_MAIN_PY = '''\
"""Main application entry point."""

def calculate_fibonacci(n: int) -> int:
    """Calculate the nth Fibonacci number recursively."""
    if n <= 1:
        return n
    return calculate_fibonacci(n - 1) + calculate_fibonacci(n - 2)

def greet_user(name: str) -> str:
    """Return a personalized greeting message."""
    return f"Hello, {name}! Welcome to the application."

if __name__ == "__main__":
    print(greet_user("World"))
    print(calculate_fibonacci(10))
'''

SAMPLE_UTILS_PY = '''\
"""Utility functions for data processing."""

def parse_csv_line(line: str) -> list[str]:
    """Parse a CSV line into a list of values."""
    return line.strip().split(",")

def format_currency(amount: float) -> str:
    """Format a number as USD currency."""
    return f"${amount:,.2f}"

def validate_email(email: str) -> bool:
    """Check if an email address is valid."""
    return "@" in email and "." in email
'''

SAMPLE_DATABASE_PY = '''\
"""Database connection and query utilities."""

class DatabaseConnection:
    """Manages database connections."""

    def __init__(self, host: str, port: int):
        self.host = host
        self.port = port
        self._connected = False

    def connect(self) -> None:
        """Establish connection to the database."""
        self._connected = True

    def execute_query(self, sql: str) -> list[dict]:
        """Execute a SQL query and return results."""
        if not self._connected:
            raise RuntimeError("Not connected to database")
        return []
'''

SAMPLE_APP_JS = """\
/** Express web application server. */

const express = require('express');
const app = express();

function handleRequest(req, res) {
    const name = req.query.name || 'World';
    res.json({ message: `Hello, ${name}!` });
}

module.exports = { handleRequest };
"""


@pytest.fixture()
def e2e_project() -> Iterator[Path]:
    """Set up a temp project dir with sample files.

    Cleans up with ``ccc reset --all -f`` and daemon stop.
    """
    base_dir = Path(tempfile.mkdtemp(prefix="ccc_e2e_"))
    project_dir = base_dir / "project"
    project_dir.mkdir()
    (project_dir / "main.py").write_text(SAMPLE_MAIN_PY)
    (project_dir / "utils.py").write_text(SAMPLE_UTILS_PY)
    lib_dir = project_dir / "lib"
    lib_dir.mkdir()
    (lib_dir / "database.py").write_text(SAMPLE_DATABASE_PY)
    (project_dir / ".git").mkdir()

    old_env = os.environ.get("COCOINDEX_CODE_DIR")
    os.environ["COCOINDEX_CODE_DIR"] = str(base_dir)
    old_cwd = os.getcwd()
    os.chdir(project_dir)

    # Pre-write global settings with the lightweight test model so `ccc init`
    # in these tests skips the new interactive flow and existing assertions
    # continue to exercise the same indexing behavior as before.
    save_user_settings(make_test_user_settings())

    try:
        yield project_dir
    finally:
        os.chdir(project_dir)
        runner.invoke(app, ["reset", "--all", "-f"])
        stop_daemon()
        os.chdir(old_cwd)
        if old_env is None:
            os.environ.pop("COCOINDEX_CODE_DIR", None)
        else:
            os.environ["COCOINDEX_CODE_DIR"] = old_env


# ---------------------------------------------------------------------------
# Session tests — each function is a complete scenario
# ---------------------------------------------------------------------------


def test_session_happy_path(e2e_project: Path) -> None:
    """Init → init (idempotent) → index → status → search variants → daemon status."""
    # Init
    result = runner.invoke(app, ["init"], catch_exceptions=False)
    assert result.exit_code == 0, result.output
    assert (e2e_project / ".cocoindex_code" / "settings.yml").exists()
    assert "Created project settings" in result.output or "settings" in result.output

    # Init again — already initialized
    result = runner.invoke(app, ["init"], catch_exceptions=False)
    assert result.exit_code == 0
    assert "already initialized" in result.output

    # Index
    result = runner.invoke(app, ["index"], catch_exceptions=False)
    assert result.exit_code == 0, result.output
    assert "Chunks:" in result.output
    assert "Files:" in result.output

    # Status
    result = runner.invoke(app, ["status"], catch_exceptions=False)
    assert result.exit_code == 0, result.output
    assert "Chunks:" in result.output

    # Search — fibonacci
    result = runner.invoke(app, ["search", "fibonacci", "calculation"], catch_exceptions=False)
    assert result.exit_code == 0, result.output
    assert "main.py" in result.output

    # Search — database
    result = runner.invoke(app, ["search", "database", "connection"], catch_exceptions=False)
    assert result.exit_code == 0, result.output
    assert "database.py" in result.output

    # Search — --lang filter
    result = runner.invoke(app, ["search", "function", "--lang", "python"], catch_exceptions=False)
    assert result.exit_code == 0, result.output
    assert "python" in result.output.lower()

    # Search — --path filter
    result = runner.invoke(app, ["search", "function", "--path", "lib/*"], catch_exceptions=False)
    assert result.exit_code == 0, result.output
    assert "lib/" in result.output

    # Search — no results
    result = runner.invoke(
        app, ["search", "xyzzy_nonexistent_symbol_12345"], catch_exceptions=False
    )
    assert result.exit_code == 0

    # Daemon status
    result = runner.invoke(app, ["daemon", "status"], catch_exceptions=False)
    assert result.exit_code == 0, result.output
    assert "Daemon version:" in result.output


def test_session_incremental_index(e2e_project: Path) -> None:
    """Init → index → add new file → re-index → search finds new content."""
    runner.invoke(app, ["init"], catch_exceptions=False)
    result = runner.invoke(app, ["index"], catch_exceptions=False)
    assert result.exit_code == 0, result.output

    # Add a new file
    (e2e_project / "app.js").write_text(SAMPLE_APP_JS)

    # Re-index
    result = runner.invoke(app, ["index"], catch_exceptions=False)
    assert result.exit_code == 0, result.output

    # Search should find the new file
    result = runner.invoke(app, ["search", "handleRequest"], catch_exceptions=False)
    assert result.exit_code == 0, result.output
    assert "app.js" in result.output


def test_session_reset_databases(e2e_project: Path) -> None:
    """Init → index → search → reset (dbs only) → re-index → search works again."""
    runner.invoke(app, ["init"], catch_exceptions=False)
    runner.invoke(app, ["index"], catch_exceptions=False)

    # Search works before reset
    result = runner.invoke(app, ["search", "fibonacci"], catch_exceptions=False)
    assert result.exit_code == 0
    assert "main.py" in result.output

    # Reset databases only
    result = runner.invoke(app, ["reset", "-f"], catch_exceptions=False)
    assert result.exit_code == 0
    assert "Databases deleted" in result.output

    # Settings should still exist
    assert (e2e_project / ".cocoindex_code" / "settings.yml").exists()

    # DB files should be gone
    assert not (e2e_project / ".cocoindex_code" / "cocoindex.db").exists()
    assert not (e2e_project / ".cocoindex_code" / "target_sqlite.db").exists()

    # Restart daemon to fully release LMDB handles.
    # On free-threaded Python (3.14t), deferred refcounting in the daemon
    # process prevents the Rust LMDB environment from being freed promptly
    # after remove_project; restarting is the reliable way to ensure cleanup.
    runner.invoke(app, ["daemon", "restart"], catch_exceptions=False)

    # Re-index — project is still initialized, just databases gone
    result = runner.invoke(app, ["index"], catch_exceptions=False)
    assert result.exit_code == 0, result.output

    # Search works again
    result = runner.invoke(app, ["search", "fibonacci"], catch_exceptions=False)
    assert result.exit_code == 0
    assert "main.py" in result.output


def test_session_reset_all(e2e_project: Path) -> None:
    """Init → index → reset --all → verify full cleanup → search errors."""
    runner.invoke(app, ["init"], catch_exceptions=False)
    runner.invoke(app, ["index"], catch_exceptions=False)

    # .gitignore should have the entry (project has .git dir)
    gitignore = e2e_project / ".gitignore"
    assert gitignore.is_file()
    assert "/.cocoindex_code/" in gitignore.read_text()

    # Reset --all
    result = runner.invoke(app, ["reset", "--all", "-f"], catch_exceptions=False)
    assert result.exit_code == 0
    assert "fully reset" in result.output

    # Settings should be gone
    assert not (e2e_project / ".cocoindex_code" / "settings.yml").exists()

    # .gitignore entry should be removed
    assert "/.cocoindex_code/" not in gitignore.read_text()

    # Search should fail — not initialized
    result = runner.invoke(app, ["search", "fibonacci"])
    assert result.exit_code != 0
    assert "ccc init" in result.output


def test_session_reset_then_full_reinit(e2e_project: Path) -> None:
    """Init → index → reset --all → re-init → re-index → search works again."""
    runner.invoke(app, ["init"], catch_exceptions=False)
    runner.invoke(app, ["index"], catch_exceptions=False)

    # Reset everything
    runner.invoke(app, ["reset", "--all", "-f"], catch_exceptions=False)

    # Restart daemon to fully release LMDB handles (see test_session_reset_databases).
    runner.invoke(app, ["daemon", "restart"], catch_exceptions=False)

    # Re-init from scratch
    result = runner.invoke(app, ["init"], catch_exceptions=False)
    assert result.exit_code == 0
    assert (e2e_project / ".cocoindex_code" / "settings.yml").exists()

    # Re-index
    result = runner.invoke(app, ["index"], catch_exceptions=False)
    assert result.exit_code == 0, result.output

    # Search works again
    result = runner.invoke(app, ["search", "fibonacci"], catch_exceptions=False)
    assert result.exit_code == 0
    assert "main.py" in result.output


def test_session_respects_gitignore(e2e_project: Path) -> None:
    """Indexing should skip files ignored by .gitignore while honoring negations."""
    gitignore_path = e2e_project / ".gitignore"
    gitignore_path.write_text("ignored.py\nignored_dir/\n!important.py\n")

    (e2e_project / "ignored.py").write_text("IGNORED_TOKEN = True\n")
    ignored_dir = e2e_project / "ignored_dir"
    ignored_dir.mkdir()
    (ignored_dir / "nested.py").write_text("NESTED_IGNORED = True\n")
    (e2e_project / "important.py").write_text("IMPORTANT_TOKEN = True\n")

    runner.invoke(app, ["init"], catch_exceptions=False)
    result = runner.invoke(app, ["index"], catch_exceptions=False)
    assert result.exit_code == 0, result.output

    db_path = e2e_project / ".cocoindex_code" / "target_sqlite.db"
    conn = coco_sqlite.connect(str(db_path), load_vec=True)
    try:
        with conn.readonly() as db:
            file_paths = {
                row[0] for row in db.execute("SELECT DISTINCT file_path FROM code_chunks_vec")
            }
    finally:
        conn.close()

    assert "ignored.py" not in file_paths
    assert "ignored_dir/nested.py" not in file_paths
    assert "important.py" in file_paths


@pytest.mark.usefixtures("e2e_project")
def test_session_daemon_stop_and_auto_start() -> None:
    """Init → index → daemon stop → index auto-starts daemon → search works."""
    runner.invoke(app, ["init"], catch_exceptions=False)
    runner.invoke(app, ["index"], catch_exceptions=False)

    # Stop daemon
    result = runner.invoke(app, ["daemon", "stop"], catch_exceptions=False)
    assert result.exit_code == 0

    # Index should auto-start daemon via ensure_daemon()
    result = runner.invoke(app, ["index"], catch_exceptions=False)
    assert result.exit_code == 0, result.output

    # Search should work with the new daemon
    result = runner.invoke(app, ["search", "fibonacci"], catch_exceptions=False)
    assert result.exit_code == 0
    assert "main.py" in result.output


@pytest.mark.usefixtures("e2e_project")
def test_session_daemon_restart() -> None:
    """Init → index → daemon restart → re-index → search works."""
    runner.invoke(app, ["init"], catch_exceptions=False)
    runner.invoke(app, ["index"], catch_exceptions=False)

    # Restart daemon
    result = runner.invoke(app, ["daemon", "restart"], catch_exceptions=False)
    assert result.exit_code == 0, result.output
    assert "restarted" in result.output.lower()

    # Re-index in the new daemon
    result = runner.invoke(app, ["index"], catch_exceptions=False)
    assert result.exit_code == 0, result.output

    # Search should work
    result = runner.invoke(app, ["search", "fibonacci"], catch_exceptions=False)
    assert result.exit_code == 0
    assert "main.py" in result.output


@pytest.mark.usefixtures("e2e_project")
def test_session_search_refresh() -> None:
    """Init (no explicit index) → search --refresh indexes then searches."""
    runner.invoke(app, ["init"], catch_exceptions=False)

    # search --refresh without prior explicit index
    result = runner.invoke(app, ["search", "--refresh", "fibonacci"], catch_exceptions=False)
    assert result.exit_code == 0, result.output
    assert "main.py" in result.output


def test_session_index_auto_initializes(e2e_project: Path) -> None:
    """Running ``ccc index`` from an uninitialized dir auto-creates project settings."""
    result = runner.invoke(app, ["index"], catch_exceptions=False)
    assert result.exit_code == 0, result.output
    assert "Created project settings" in result.output
    assert (e2e_project / ".cocoindex_code" / "settings.yml").is_file()

    # The auto-initialized project is fully usable: search works.
    result = runner.invoke(app, ["search", "fibonacci"], catch_exceptions=False)
    assert result.exit_code == 0, result.output
    assert "main.py" in result.output


def test_session_search_json_output(e2e_project: Path) -> None:
    """`ccc search --json` emits the SearchResponse as parseable JSON on stdout."""
    runner.invoke(app, ["init"], catch_exceptions=False)
    runner.invoke(app, ["index"], catch_exceptions=False)

    result = runner.invoke(app, ["search", "fibonacci", "--json"], catch_exceptions=False)
    assert result.exit_code == 0, result.output
    payload = json.loads(result.stdout)
    assert payload["success"] is True
    assert payload["results"], "expected at least one result"
    top = payload["results"][0]
    assert {"file_path", "language", "content", "start_line", "end_line", "score"} <= top.keys()
    assert isinstance(top["score"], float)
    assert any("main.py" in r["file_path"] for r in payload["results"])

    # --json --refresh composes: index progress goes to stderr, stdout stays pure JSON.
    result = runner.invoke(
        app, ["search", "fibonacci", "--json", "--refresh"], catch_exceptions=False
    )
    assert result.exit_code == 0, result.output
    payload = json.loads(result.stdout)
    assert payload["success"] is True


def test_session_subdirectory_path_default(e2e_project: Path) -> None:
    """Search from a subdirectory defaults path filter to that subdirectory."""
    runner.invoke(app, ["init"], catch_exceptions=False)
    runner.invoke(app, ["index"], catch_exceptions=False)

    # Search from project root — should find main.py
    result = runner.invoke(app, ["search", "fibonacci"], catch_exceptions=False)
    assert result.exit_code == 0, result.output
    assert "main.py" in result.output

    # Search from lib/ — default path filter restricts to lib/*
    os.chdir(e2e_project / "lib")
    result = runner.invoke(app, ["search", "database", "connection"], catch_exceptions=False)
    assert result.exit_code == 0, result.output
    assert "database.py" in result.output

    # From lib/, searching for fibonacci should NOT find main.py (outside lib/)
    result = runner.invoke(app, ["search", "fibonacci"], catch_exceptions=False)
    assert result.exit_code == 0
    assert "main.py" not in result.output

    # Back to project root
    os.chdir(e2e_project)


def test_session_not_initialized_errors(e2e_project: Path) -> None:
    """Search and status from uninitialized dir should error with guidance."""
    standalone = Path(tempfile.mkdtemp(prefix="ccc_standalone_"))
    os.chdir(standalone)

    result = runner.invoke(app, ["search", "hello"])
    assert result.exit_code != 0
    assert "ccc init" in result.output

    result = runner.invoke(app, ["status"])
    assert result.exit_code != 0
    assert "ccc init" in result.output

    # Return to project dir so fixture cleanup works
    os.chdir(e2e_project)


def test_session_doctor_happy_path(e2e_project: Path) -> None:
    """Init → index → doctor shows global settings, daemon, model, project, and index info."""
    runner.invoke(app, ["init"], catch_exceptions=False)
    result = runner.invoke(app, ["index"], catch_exceptions=False)
    assert result.exit_code == 0, result.output

    result = runner.invoke(app, ["doctor"], catch_exceptions=False)
    assert result.exit_code == 0, result.output

    # Global settings section
    assert "Global Settings" in result.output
    assert "global_settings.yml" in result.output
    assert "provider=" in result.output
    assert "model=" in result.output

    # Daemon section
    assert "Daemon" in result.output
    assert "Version:" in result.output
    assert "Uptime:" in result.output

    # Model check
    assert "[OK] Model Check" in result.output
    assert "Embedding dimension:" in result.output

    # Project settings section
    assert "Project Settings" in result.output
    assert "settings.yml" in result.output
    assert "Include patterns (" in result.output
    assert "Exclude patterns (" in result.output

    # File walk
    assert "[OK] File Walk" in result.output
    assert "Total matched files:" in result.output
    # Our sample project has .py files
    assert ".py:" in result.output

    # Index status
    assert "[OK] Index Status" in result.output
    assert "Chunks:" in result.output
    assert "Files:" in result.output

    # Log files section
    assert "Log Files" in result.output
    assert "daemon.log" in result.output


def test_session_doctor_no_index(e2e_project: Path) -> None:
    """Doctor before indexing should show index not created yet."""
    runner.invoke(app, ["init"], catch_exceptions=False)

    result = runner.invoke(app, ["doctor"], catch_exceptions=False)
    assert result.exit_code == 0, result.output

    assert "[OK] Model Check" in result.output
    assert "Index not created yet" in result.output


def test_session_doctor_no_project(e2e_project: Path) -> None:
    """Doctor outside a project should still show global + daemon checks."""
    # Init to create global settings and start daemon
    runner.invoke(app, ["init"], catch_exceptions=False)

    # Move to a standalone directory (not a project)
    standalone = Path(tempfile.mkdtemp(prefix="ccc_standalone_"))
    old_cwd = os.getcwd()
    os.chdir(standalone)
    try:
        result = runner.invoke(app, ["doctor"], catch_exceptions=False)
        assert result.exit_code == 0, result.output

        # Global + daemon checks should be present
        assert "Global Settings" in result.output
        assert "Daemon" in result.output
        assert "[OK] Model Check" in result.output

        # Project-specific sections should NOT be present
        assert "Project Settings" not in result.output
        assert "File Walk" not in result.output
        assert "Index Status" not in result.output

        # Log files always present
        assert "Log Files" in result.output
    finally:
        os.chdir(old_cwd)


# ---------------------------------------------------------------------------
# Daemon startup failure tests
# ---------------------------------------------------------------------------


@pytest.fixture()
def e2e_project_no_global_settings() -> Iterator[Path]:
    """Set up a project with project settings but NO global_settings.yml.

    This reproduces the scenario from issue #113 where a user creates project
    settings manually but hasn't run ``ccc init`` (which creates global settings).
    """
    base_dir = Path(tempfile.mkdtemp(prefix="ccc_e2e_"))
    project_dir = base_dir / "project"
    project_dir.mkdir()
    (project_dir / "main.py").write_text(SAMPLE_MAIN_PY)
    (project_dir / ".git").mkdir()

    old_env = os.environ.get("COCOINDEX_CODE_DIR")
    os.environ["COCOINDEX_CODE_DIR"] = str(base_dir)
    old_cwd = os.getcwd()
    os.chdir(project_dir)

    # Create project settings but NOT global settings — this is the bug scenario
    save_project_settings(project_dir, default_project_settings())

    try:
        yield project_dir
    finally:
        os.chdir(old_cwd)
        stop_daemon()
        if old_env is None:
            os.environ.pop("COCOINDEX_CODE_DIR", None)
        else:
            os.environ["COCOINDEX_CODE_DIR"] = old_env


@pytest.mark.usefixtures("e2e_project_no_global_settings")
def test_session_missing_global_settings_early_error() -> None:
    """When global_settings.yml is missing, project commands should fail early with guidance."""
    # `ccc status` should detect missing global settings before even starting the daemon.
    result = runner.invoke(app, ["status"])
    assert result.exit_code != 0, f"Expected failure but got: {result.output}"
    assert "Global settings not found" in result.output
    assert "global_settings.yml" in result.output
    assert "ccc init" in result.output


@pytest.mark.usefixtures("e2e_project_no_global_settings")
def test_session_daemon_restart_with_no_global_settings() -> None:
    """``ccc daemon restart`` without ``global_settings.yml`` starts the daemon in
    no-settings mode rather than failing hard.

    The daemon comes up, accepts handshakes, but rejects project requests until
    ``ccc init`` writes the file. The handshake mtime mismatch then drives the
    restart that loads real settings — here we just verify the restart itself
    succeeds and the file stays absent (no silent auto-create).
    """
    from cocoindex_code.settings import user_settings_path

    result = runner.invoke(app, ["daemon", "restart"])
    assert result.exit_code == 0, f"Expected success but got: {result.output}"
    assert "Daemon restarted." in result.output

    # No auto-creation — user still needs to run `ccc init` to pick a model
    # (and trigger the supervised respawn with real settings).
    assert not user_settings_path().is_file()


# ---------------------------------------------------------------------------
# Interactive init flow tests
# ---------------------------------------------------------------------------


def _fake_doctor_ok(
    project_root: str | None = None,
    on_result: object = None,
) -> list[object]:
    """Stand-in for ``client.doctor`` that returns a single OK Model Check."""
    from cocoindex_code.protocol import DoctorCheckResult

    indexing_ok = DoctorCheckResult(
        name="Model Check (indexing)",
        ok=True,
        details=["params: {} (no extra kwargs)", "Embedding dimension: 384"],
        errors=[],
    )
    query_ok = DoctorCheckResult(
        name="Model Check (query)",
        ok=True,
        details=["params: {} (no extra kwargs)", "Embedding dimension: 384"],
        errors=[],
    )
    done = DoctorCheckResult(name="done", ok=True, details=[], errors=[])
    return [indexing_ok, query_ok, done]


@pytest.fixture()
def e2e_fresh_env(monkeypatch: pytest.MonkeyPatch) -> Iterator[Path]:
    """Fresh COCOINDEX_CODE_DIR with NO global settings and NO project settings.

    Distinct from ``e2e_project`` (which pre-writes global settings): tests
    that exercise the interactive init flow need a genuinely empty state.
    ``client.doctor`` is monkeypatched so no real daemon starts and no real
    model is loaded during tests.
    """
    base_dir = Path(tempfile.mkdtemp(prefix="ccc_e2e_fresh_"))
    project_dir = base_dir / "project"
    project_dir.mkdir()
    (project_dir / "main.py").write_text(SAMPLE_MAIN_PY)
    (project_dir / ".git").mkdir()

    old_env = os.environ.get("COCOINDEX_CODE_DIR")
    os.environ["COCOINDEX_CODE_DIR"] = str(base_dir)
    old_cwd = os.getcwd()
    os.chdir(project_dir)

    monkeypatch.setattr("cocoindex_code.client.doctor", _fake_doctor_ok)

    try:
        yield project_dir
    finally:
        os.chdir(old_cwd)
        stop_daemon()
        if old_env is None:
            os.environ.pop("COCOINDEX_CODE_DIR", None)
        else:
            os.environ["COCOINDEX_CODE_DIR"] = old_env


def test_resolve_embedding_choice_flag_wins() -> None:
    """--litellm-model flag short-circuits all other logic."""
    from cocoindex_code.cli import _resolve_embedding_choice

    embedding = _resolve_embedding_choice(
        litellm_model_flag="openai/text-embedding-3-small",
        st_installed=True,
        tty=True,
    )
    assert embedding.provider == "litellm"
    assert embedding.model == "openai/text-embedding-3-small"


def test_resolve_embedding_choice_non_tty_defaults_to_snowflake() -> None:
    """Non-TTY + ST installed → sentence-transformers + Snowflake defaults."""
    from cocoindex_code.cli import _resolve_embedding_choice
    from cocoindex_code.settings import DEFAULT_ST_MODEL

    embedding = _resolve_embedding_choice(
        litellm_model_flag=None,
        st_installed=True,
        tty=False,
    )
    assert embedding.provider == "sentence-transformers"
    assert embedding.model == DEFAULT_ST_MODEL


def test_resolve_embedding_choice_non_tty_slim_errors() -> None:
    """Non-TTY + ST NOT installed + no flag → typer.Exit with guidance."""
    import typer

    from cocoindex_code.cli import _resolve_embedding_choice

    with pytest.raises(typer.Exit) as exc_info:
        _resolve_embedding_choice(
            litellm_model_flag=None,
            st_installed=False,
            tty=False,
        )
    assert exc_info.value.exit_code == 1


def test_init_non_tty_with_flag(e2e_fresh_env: Path) -> None:
    """Non-TTY (CliRunner default) + --litellm-model works without prompts."""
    result = runner.invoke(
        app,
        ["init", "--litellm-model", "openai/text-embedding-3-small"],
        catch_exceptions=False,
    )
    assert result.exit_code == 0, result.output

    loaded = load_user_settings()
    assert loaded.embedding.provider == "litellm"
    assert loaded.embedding.model == "openai/text-embedding-3-small"


def test_init_non_tty_no_flag_uses_defaults(e2e_fresh_env: Path) -> None:
    """Non-TTY + ST installed → defaults to sentence-transformers + Snowflake."""
    result = runner.invoke(app, ["init"], catch_exceptions=False)
    assert result.exit_code == 0, result.output

    loaded = load_user_settings()
    assert loaded.embedding.provider == "sentence-transformers"
    assert loaded.embedding.model == "Snowflake/snowflake-arctic-embed-xs"


def test_init_non_tty_slim_install_no_flag_errors(
    e2e_fresh_env: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Non-TTY + ST NOT installed + no flag → error before writing settings."""
    monkeypatch.setattr(
        "cocoindex_code.shared.is_sentence_transformers_installed",
        lambda: False,
    )
    result = runner.invoke(app, ["init"])
    assert result.exit_code != 0
    combined = result.output + result.stderr if result.stderr else result.output
    assert "--litellm-model" in combined
    assert "embeddings-local" in combined
    assert not user_settings_path().is_file()


def test_init_slim_install_with_flag(e2e_fresh_env: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    """ST not installed, --litellm-model given → LiteLLM settings written."""
    monkeypatch.setattr(
        "cocoindex_code.shared.is_sentence_transformers_installed",
        lambda: False,
    )
    result = runner.invoke(
        app,
        ["init", "--litellm-model", "openai/text-embedding-3-small"],
        catch_exceptions=False,
    )
    assert result.exit_code == 0, result.output

    loaded = load_user_settings()
    assert loaded.embedding.provider == "litellm"
    assert loaded.embedding.model == "openai/text-embedding-3-small"


def test_init_model_test_failure_is_non_fatal(
    e2e_fresh_env: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Model-test failure does NOT abort init; project settings still written."""
    from cocoindex_code.protocol import DoctorCheckResult

    def _fake_doctor_fail(
        project_root: str | None = None,
        on_result: object = None,
    ) -> list[DoctorCheckResult]:
        fail = DoctorCheckResult(
            name="Model Check (indexing)",
            ok=False,
            details=["params: {} (no extra kwargs)"],
            errors=["AuthenticationError: missing key"],
        )
        query_ok = DoctorCheckResult(
            name="Model Check (query)",
            ok=True,
            details=["params: {} (no extra kwargs)", "Embedding dimension: 384"],
            errors=[],
        )
        done = DoctorCheckResult(name="done", ok=True, details=[], errors=[])
        return [fail, query_ok, done]

    monkeypatch.setattr("cocoindex_code.client.doctor", _fake_doctor_fail)
    result = runner.invoke(
        app,
        ["init", "--litellm-model", "openai/text-embedding-3-small"],
        catch_exceptions=False,
    )
    assert result.exit_code == 0, result.output
    combined = result.output + (result.stderr or "")
    assert "[FAIL]" in combined
    assert "AuthenticationError" in combined
    assert "ccc doctor" in combined
    assert "envs:" in combined

    # Settings file was written (not rolled back) and project was initialized.
    assert user_settings_path().is_file()
    assert (e2e_fresh_env / ".cocoindex_code" / "settings.yml").exists()


def test_init_rejects_litellm_model_when_settings_exist(e2e_project: Path) -> None:
    """With global settings pre-written by e2e_project, --litellm-model is rejected."""
    result = runner.invoke(app, ["init", "--litellm-model", "openai/foo"])
    assert result.exit_code != 0
    combined = result.output + (result.stderr or "")
    assert "already exist" in combined


def test_init_force_does_not_suppress_prompts(e2e_fresh_env: Path) -> None:
    """`-f` only affects the parent-marker warning, not the interactive-flow gate."""
    result = runner.invoke(app, ["init", "-f"], catch_exceptions=False)
    # Non-TTY + ST installed → non-TTY defaults, same as without -f.
    assert result.exit_code == 0, result.output

    loaded = load_user_settings()
    assert loaded.embedding.provider == "sentence-transformers"
    assert loaded.embedding.model == "Snowflake/snowflake-arctic-embed-xs"


# ---------------------------------------------------------------------------
# Doctor model-check failure path
# ---------------------------------------------------------------------------


async def test_daemon_check_model_maps_failure_to_doctor_result() -> None:
    """daemon._check_model delegates to check_embedding and maps failures correctly."""
    from cocoindex_code.daemon import _check_model

    class _BoomEmbedder:
        async def embed(self, text: str, **kwargs: object) -> object:  # noqa: ARG002
            raise RuntimeError("boom")

    from typing import cast

    from cocoindex_code.shared import Embedder

    result = await _check_model(cast(Embedder, _BoomEmbedder()), label="indexing", params={})
    assert result.name == "Model Check (indexing)"
    assert result.ok is False
    assert len(result.errors) == 1
    assert result.errors[0].startswith("RuntimeError:")
    assert "boom" in result.errors[0]
    # The full traceback is carried through so `ccc doctor` can display it.
    assert result.traceback is not None
    assert "Traceback (most recent call last):" in result.traceback
    assert "boom" in result.traceback


# ---------------------------------------------------------------------------
# Dockerfile packaging regression guard
# ---------------------------------------------------------------------------


def test_dockerfile_install_line_uses_full_extra() -> None:
    """Dockerfile should install via `cocoindex-code[full]` (not the old
    `[default]` alias) and should not hard-pin sentence-transformers.
    """
    repo_root = Path(__file__).resolve().parent.parent
    content = (repo_root / "docker" / "Dockerfile").read_text()
    assert "cocoindex-code[full]" in content
    assert "cocoindex-code[default]" not in content
    assert "sentence-transformers>=" not in content
    assert "sentence-transformers==" not in content


# ---------------------------------------------------------------------------
# DB path mapping tests
# ---------------------------------------------------------------------------


@pytest.fixture()
def e2e_project_with_db_mapping() -> Iterator[tuple[Path, Path]]:
    """Set up a project with COCOINDEX_CODE_DB_PATH_MAPPING pointing to a separate db dir.

    Yields (project_dir, db_base_dir).
    """
    base_dir = Path(tempfile.mkdtemp(prefix="ccc_e2e_"))
    project_dir = base_dir / "workspace" / "myproject"
    project_dir.mkdir(parents=True)
    db_base_dir = base_dir / "db-files"
    db_base_dir.mkdir()

    (project_dir / "main.py").write_text(SAMPLE_MAIN_PY)
    (project_dir / ".git").mkdir()

    old_env = {
        k: os.environ.get(k) for k in ("COCOINDEX_CODE_DIR", "COCOINDEX_CODE_DB_PATH_MAPPING")
    }
    os.environ["COCOINDEX_CODE_DIR"] = str(base_dir)
    workspace = str(base_dir / "workspace")
    os.environ["COCOINDEX_CODE_DB_PATH_MAPPING"] = f"{workspace}={db_base_dir}"
    _reset_db_path_mapping_cache()
    old_cwd = os.getcwd()
    os.chdir(project_dir)

    try:
        yield project_dir, db_base_dir
    finally:
        os.chdir(project_dir)
        runner.invoke(app, ["reset", "--all", "-f"])
        stop_daemon()
        os.chdir(old_cwd)
        _reset_db_path_mapping_cache()
        for k, v in old_env.items():
            if v is None:
                os.environ.pop(k, None)
            else:
                os.environ[k] = v


def test_session_db_path_mapping(
    e2e_project_with_db_mapping: tuple[Path, Path],
) -> None:
    """Init → index → verify databases are in the mapped directory → search works."""
    project_dir, db_base_dir = e2e_project_with_db_mapping
    mapped_db_dir = db_base_dir / "myproject"

    # Init
    result = runner.invoke(app, ["init"], catch_exceptions=False)
    assert result.exit_code == 0, result.output

    # Settings should be in the project dir, NOT the mapped dir
    assert (project_dir / ".cocoindex_code" / "settings.yml").exists()

    # Index
    result = runner.invoke(app, ["index"], catch_exceptions=False)
    assert result.exit_code == 0, result.output

    # Databases should be in the mapped directory
    assert (mapped_db_dir / "target_sqlite.db").exists()
    # Databases should NOT be in the project's .cocoindex_code dir
    assert not (project_dir / ".cocoindex_code" / "target_sqlite.db").exists()

    # Search should work
    result = runner.invoke(app, ["search", "fibonacci"], catch_exceptions=False)
    assert result.exit_code == 0, result.output
    assert "main.py" in result.output

    # Reset should clean databases from the mapped dir
    result = runner.invoke(app, ["reset", "-f"], catch_exceptions=False)
    assert result.exit_code == 0
    assert not (mapped_db_dir / "target_sqlite.db").exists()
    # Settings still in place
    assert (project_dir / ".cocoindex_code" / "settings.yml").exists()


# ---------------------------------------------------------------------------
# Unit tests (not session-based)
# ---------------------------------------------------------------------------


class TestCodebaseRootDiscovery:
    """Tests for find_parent_with_marker helper."""

    def test_prefers_cocoindex_code_over_git(self, tmp_path: Path) -> None:
        parent = tmp_path / "project"
        parent.mkdir()
        (parent / ".cocoindex_code").mkdir()
        (parent / ".git").mkdir()
        subdir = parent / "src" / "lib"
        subdir.mkdir(parents=True)
        assert find_parent_with_marker(subdir) == parent

    def test_finds_git_in_parent_hierarchy(self, tmp_path: Path) -> None:
        (tmp_path / ".git").mkdir()
        deep_dir = tmp_path / "a" / "b" / "c" / "d" / "e"
        deep_dir.mkdir(parents=True)
        assert find_parent_with_marker(deep_dir) == tmp_path

    def test_falls_back_to_none_when_no_markers(self, tmp_path: Path) -> None:
        empty_dir = tmp_path / "standalone"
        empty_dir.mkdir()
        assert find_parent_with_marker(empty_dir) is None