learningcircuit--local-deep-research/tests/embeddings/test_text_splitter_registry.py

"""
Tests for embeddings/splitters/text_splitter_registry.py

Tests cover:
- get_text_splitter() function with various splitter types
- is_semantic_chunker_available() function
- VALID_SPLITTER_TYPES constant
"""

import os

import pytest
from unittest.mock import patch, MagicMock


def _huggingface_hub_reachable() -> bool:
    """Return True when HuggingFace Hub is reachable (best-effort check)."""
    try:
        import httpx

        resp = httpx.head(
            "https://huggingface.co/sentence-transformers/all-mpnet-base-v2/resolve/main/config.json",
            timeout=5,
            follow_redirects=True,
        )
        return resp.status_code < 400
    except Exception:
        return False


_skip_no_hf = pytest.mark.skipif(
    not _huggingface_hub_reachable(),
    reason="HuggingFace Hub is unreachable (network or auth)",
)


class TestValidSplitterTypes:
    """Tests for VALID_SPLITTER_TYPES constant."""

    def test_valid_splitter_types_contains_recursive(self):
        """Test that recursive splitter is valid."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            VALID_SPLITTER_TYPES,
        )

        assert "recursive" in VALID_SPLITTER_TYPES

    def test_valid_splitter_types_contains_token(self):
        """Test that token splitter is valid."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            VALID_SPLITTER_TYPES,
        )

        assert "token" in VALID_SPLITTER_TYPES

    def test_valid_splitter_types_contains_sentence(self):
        """Test that sentence splitter is valid."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            VALID_SPLITTER_TYPES,
        )

        assert "sentence" in VALID_SPLITTER_TYPES

    def test_valid_splitter_types_contains_semantic(self):
        """Test that semantic splitter is valid."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            VALID_SPLITTER_TYPES,
        )

        assert "semantic" in VALID_SPLITTER_TYPES


class TestGetTextSplitterRecursive:
    """Tests for get_text_splitter with recursive type."""

    def test_get_text_splitter_recursive_default(self):
        """Test getting recursive splitter with defaults."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            get_text_splitter,
        )
        from langchain_text_splitters import RecursiveCharacterTextSplitter

        splitter = get_text_splitter(splitter_type="recursive")

        assert isinstance(splitter, RecursiveCharacterTextSplitter)

    def test_get_text_splitter_recursive_custom_chunk_size(self):
        """Test recursive splitter with custom chunk size."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            get_text_splitter,
        )
        from langchain_text_splitters import RecursiveCharacterTextSplitter

        splitter = get_text_splitter(
            splitter_type="recursive",
            chunk_size=500,
            chunk_overlap=50,
        )

        assert isinstance(splitter, RecursiveCharacterTextSplitter)
        assert splitter._chunk_size == 500
        assert splitter._chunk_overlap == 50

    def test_get_text_splitter_recursive_custom_separators(self):
        """Test recursive splitter with custom separators."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            get_text_splitter,
        )

        custom_separators = ["\n\n", "\n", " "]

        splitter = get_text_splitter(
            splitter_type="recursive",
            text_separators=custom_separators,
        )

        assert splitter._separators == custom_separators

    def test_get_text_splitter_recursive_default_separators(self):
        """Test recursive splitter uses default separators."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            get_text_splitter,
        )

        splitter = get_text_splitter(splitter_type="recursive")

        # Default separators
        assert "\n\n" in splitter._separators
        assert "\n" in splitter._separators


class TestGetTextSplitterToken:
    """Tests for get_text_splitter with token type."""

    def test_get_text_splitter_token(self):
        """Test getting token splitter."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            get_text_splitter,
        )
        from langchain_text_splitters import TokenTextSplitter

        splitter = get_text_splitter(splitter_type="token")

        assert isinstance(splitter, TokenTextSplitter)

    def test_get_text_splitter_token_custom_params(self):
        """Test token splitter with custom parameters."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            get_text_splitter,
        )
        from langchain_text_splitters import TokenTextSplitter

        splitter = get_text_splitter(
            splitter_type="token",
            chunk_size=256,
            chunk_overlap=32,
        )

        assert isinstance(splitter, TokenTextSplitter)


@pytest.mark.skipif(
    os.getenv("CI") is not None,
    reason="Flaky in CI: sentence-transformers model download times out (>60s)",
)
class TestGetTextSplitterSentence:
    """Tests for get_text_splitter with sentence type."""

    @_skip_no_hf
    def test_get_text_splitter_sentence(self):
        """Test getting sentence splitter."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            get_text_splitter,
        )
        from langchain_text_splitters import (
            SentenceTransformersTokenTextSplitter,
        )

        # Use small chunk size within model's token limit (384)
        splitter = get_text_splitter(splitter_type="sentence", chunk_size=256)

        assert isinstance(splitter, SentenceTransformersTokenTextSplitter)

    @_skip_no_hf
    def test_get_text_splitter_sentence_custom_params(self):
        """Test sentence splitter with custom parameters."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            get_text_splitter,
        )
        from langchain_text_splitters import (
            SentenceTransformersTokenTextSplitter,
        )

        # Use chunk size within model's token limit (384)
        splitter = get_text_splitter(
            splitter_type="sentence",
            chunk_size=200,
            chunk_overlap=32,
        )

        assert isinstance(splitter, SentenceTransformersTokenTextSplitter)


class TestGetTextSplitterSemantic:
    """Tests for get_text_splitter with semantic type."""

    def test_get_text_splitter_semantic_without_embeddings_raises(self):
        """Test that semantic splitter without embeddings raises ValueError."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            get_text_splitter,
        )

        with pytest.raises(ValueError, match="requires 'embeddings' parameter"):
            get_text_splitter(splitter_type="semantic")

    def test_get_text_splitter_semantic_with_embeddings(self):
        """Test semantic splitter with embeddings."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            get_text_splitter,
        )

        mock_embeddings = MagicMock()
        mock_chunker = MagicMock()

        with patch(
            "langchain_experimental.text_splitter.SemanticChunker",
            return_value=mock_chunker,
        ) as mock_class:
            splitter = get_text_splitter(
                splitter_type="semantic",
                embeddings=mock_embeddings,
            )

            assert splitter is mock_chunker
            mock_class.assert_called_once()
            call_kwargs = mock_class.call_args[1]
            assert call_kwargs["embeddings"] is mock_embeddings

    def test_get_text_splitter_semantic_with_threshold_type(self):
        """Test semantic splitter with custom threshold type."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            get_text_splitter,
        )

        mock_embeddings = MagicMock()
        mock_chunker = MagicMock()

        with patch(
            "langchain_experimental.text_splitter.SemanticChunker",
            return_value=mock_chunker,
        ) as mock_class:
            get_text_splitter(
                splitter_type="semantic",
                embeddings=mock_embeddings,
                breakpoint_threshold_type="standard_deviation",
            )

            call_kwargs = mock_class.call_args[1]
            assert (
                call_kwargs["breakpoint_threshold_type"] == "standard_deviation"
            )

    def test_get_text_splitter_semantic_with_threshold_amount(self):
        """Test semantic splitter with custom threshold amount."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            get_text_splitter,
        )

        mock_embeddings = MagicMock()
        mock_chunker = MagicMock()

        with patch(
            "langchain_experimental.text_splitter.SemanticChunker",
            return_value=mock_chunker,
        ) as mock_class:
            get_text_splitter(
                splitter_type="semantic",
                embeddings=mock_embeddings,
                breakpoint_threshold_amount=0.5,
            )

            call_kwargs = mock_class.call_args[1]
            assert call_kwargs["breakpoint_threshold_amount"] == 0.5

    def test_get_text_splitter_semantic_import_error(self):
        """Test semantic splitter raises ImportError if experimental not installed."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            get_text_splitter,
        )

        mock_embeddings = MagicMock()

        with patch(
            "langchain_experimental.text_splitter.SemanticChunker",
            side_effect=ImportError("No module"),
        ):
            with pytest.raises(ImportError, match="langchain-experimental"):
                get_text_splitter(
                    splitter_type="semantic",
                    embeddings=mock_embeddings,
                )


class TestGetTextSplitterInvalid:
    """Tests for get_text_splitter with invalid types."""

    def test_get_text_splitter_invalid_type_raises(self):
        """Test that invalid splitter type raises ValueError."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            get_text_splitter,
        )

        with pytest.raises(ValueError, match="Invalid splitter type"):
            get_text_splitter(splitter_type="invalid_type")

    def test_get_text_splitter_invalid_type_shows_valid_options(self):
        """Test that error message shows valid options."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            get_text_splitter,
        )

        with pytest.raises(ValueError) as exc_info:
            get_text_splitter(splitter_type="unknown")

        error_msg = str(exc_info.value)
        assert "recursive" in error_msg
        assert "token" in error_msg
        assert "sentence" in error_msg
        assert "semantic" in error_msg


class TestGetTextSplitterNormalization:
    """Tests for splitter type normalization."""

    def test_get_text_splitter_normalizes_case(self):
        """Test that splitter type is case insensitive."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            get_text_splitter,
        )
        from langchain_text_splitters import RecursiveCharacterTextSplitter

        splitter = get_text_splitter(splitter_type="RECURSIVE")
        assert isinstance(splitter, RecursiveCharacterTextSplitter)

        splitter = get_text_splitter(splitter_type="Recursive")
        assert isinstance(splitter, RecursiveCharacterTextSplitter)

    def test_get_text_splitter_strips_whitespace(self):
        """Test that splitter type whitespace is stripped."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            get_text_splitter,
        )
        from langchain_text_splitters import RecursiveCharacterTextSplitter

        splitter = get_text_splitter(splitter_type="  recursive  ")
        assert isinstance(splitter, RecursiveCharacterTextSplitter)


class TestIsSemanticChunkerAvailable:
    """Tests for is_semantic_chunker_available function."""

    def test_is_semantic_chunker_available_when_installed(self):
        """Test returns True when langchain_experimental is installed."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            is_semantic_chunker_available,
        )

        mock_spec = MagicMock()

        with patch("importlib.util.find_spec", return_value=mock_spec):
            assert is_semantic_chunker_available() is True

    def test_is_semantic_chunker_available_when_not_installed(self):
        """Test returns False when langchain_experimental is not installed."""
        from local_deep_research.embeddings.splitters.text_splitter_registry import (
            is_semantic_chunker_available,
        )

        with patch("importlib.util.find_spec", return_value=None):
            assert is_semantic_chunker_available() is False