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

"""
Tests for embeddings_config module.

These tests verify the get_embedding_function and related configuration functions.
"""

import pytest
from unittest.mock import Mock, patch, MagicMock


class TestGetEmbeddingFunction:
    """Tests for get_embedding_function."""

    def test_get_embedding_function_exists(self):
        """Verify get_embedding_function can be imported."""
        from local_deep_research.embeddings.embeddings_config import (
            get_embedding_function,
        )

        assert callable(get_embedding_function)

    def test_get_embedding_function_returns_callable(self):
        """get_embedding_function should return a callable embed_documents method."""
        from local_deep_research.embeddings.embeddings_config import (
            get_embedding_function,
        )

        # Mock the get_embeddings function
        mock_embeddings = MagicMock()
        mock_embeddings.embed_documents = Mock(return_value=[[0.1, 0.2, 0.3]])

        with patch(
            "local_deep_research.embeddings.embeddings_config.get_embeddings",
            return_value=mock_embeddings,
        ):
            func = get_embedding_function(
                provider="sentence_transformers", model_name="test-model"
            )

            # Should return the embed_documents method
            assert callable(func)

            # Should be the embed_documents method specifically
            result = func(["test text"])
            assert result == [[0.1, 0.2, 0.3]]
            mock_embeddings.embed_documents.assert_called_once_with(
                ["test text"]
            )

    def test_get_embedding_function_passes_parameters(self):
        """get_embedding_function should pass all parameters to get_embeddings."""
        from local_deep_research.embeddings.embeddings_config import (
            get_embedding_function,
        )

        mock_embeddings = MagicMock()
        mock_embeddings.embed_documents = Mock()

        with patch(
            "local_deep_research.embeddings.embeddings_config.get_embeddings",
            return_value=mock_embeddings,
        ) as mock_get_embeddings:
            settings = {"key": "value"}
            get_embedding_function(
                provider="ollama",
                model_name="nomic-embed-text",
                settings_snapshot=settings,
                extra_param="extra",
            )

            mock_get_embeddings.assert_called_once_with(
                provider="ollama",
                model="nomic-embed-text",
                settings_snapshot=settings,
                extra_param="extra",
            )


class TestGetEmbeddings:
    """Tests for get_embeddings function."""

    def test_get_embeddings_exists(self):
        """Verify get_embeddings can be imported."""
        from local_deep_research.embeddings.embeddings_config import (
            get_embeddings,
        )

        assert callable(get_embeddings)

    def test_get_embeddings_validates_provider(self):
        """get_embeddings should raise ValueError for invalid provider."""
        from local_deep_research.embeddings.embeddings_config import (
            get_embeddings,
        )

        with pytest.raises(ValueError, match="Invalid embedding provider"):
            get_embeddings(provider="invalid_provider")


class TestAvailableProviders:
    """Tests for provider availability functions."""

    def test_valid_embedding_providers_list(self):
        """VALID_EMBEDDING_PROVIDERS should contain expected providers."""
        from local_deep_research.embeddings.embeddings_config import (
            VALID_EMBEDDING_PROVIDERS,
        )

        assert "sentence_transformers" in VALID_EMBEDDING_PROVIDERS
        assert "ollama" in VALID_EMBEDDING_PROVIDERS
        assert "openai" in VALID_EMBEDDING_PROVIDERS

    def test_get_available_embedding_providers_exists(self):
        """Verify get_available_embedding_providers can be imported."""
        from local_deep_research.embeddings.embeddings_config import (
            get_available_embedding_providers,
        )

        assert callable(get_available_embedding_providers)


class TestSnapshotlessEgressGate:
    """get_embeddings() with no settings_snapshot can't read the
    embeddings.require_local toggle, so it fails closed: only the
    localhost-default providers may proceed. A snapshot-less
    get_embeddings(provider="openai") must NOT silently ship the local
    corpus to a cloud embedder.

    Regression (PR #4300 review): the gate was `if settings_snapshot is
    not None:` — the entire check was skipped when snapshot was None,
    so a snapshot-less cloud-provider call proceeded unchecked.

    Source: src/local_deep_research/embeddings/embeddings_config.py
    """

    def test_openai_without_snapshot_is_denied(self):
        from local_deep_research.embeddings.embeddings_config import (
            get_embeddings,
        )
        from local_deep_research.security.egress.policy import (
            PolicyDeniedError,
        )

        with pytest.raises(PolicyDeniedError) as excinfo:
            get_embeddings(provider="openai", settings_snapshot=None)
        assert excinfo.value.decision.reason == "no_snapshot_for_provider"

    def test_local_default_providers_are_allowlisted(self):
        """sentence_transformers and ollama are the localhost-default
        embedding providers — they must pass the snapshot-less gate
        (i.e. NOT raise the no_snapshot_for_provider denial)."""
        from local_deep_research.security.egress.policy import (
            _LOCAL_DEFAULT_EMBEDDING_PROVIDERS,
        )

        assert "sentence_transformers" in _LOCAL_DEFAULT_EMBEDDING_PROVIDERS
        assert "ollama" in _LOCAL_DEFAULT_EMBEDDING_PROVIDERS
        # Cloud providers must NOT be on the allow-list.
        assert "openai" not in _LOCAL_DEFAULT_EMBEDDING_PROVIDERS

    def test_sentence_transformers_without_snapshot_passes_gate(self):
        """The local default must get past the egress gate without a
        snapshot. We patch the provider class so the test asserts the
        POLICY decision, not the heavy model load."""
        from local_deep_research.embeddings.embeddings_config import (
            get_embeddings,
        )
        from local_deep_research.security.egress.policy import (
            PolicyDeniedError,
        )

        sentinel = object()
        fake_provider = Mock()
        fake_provider.return_value.create_embeddings.return_value = sentinel

        with patch(
            "local_deep_research.embeddings.embeddings_config._get_provider_classes",
            return_value={"sentence_transformers": fake_provider},
        ):
            try:
                get_embeddings(
                    provider="sentence_transformers",
                    settings_snapshot=None,
                )
            except PolicyDeniedError:
                pytest.fail(
                    "sentence_transformers must pass the snapshot-less gate"
                )
            except Exception:
                # Any non-policy error means we got PAST the gate, which is
                # all this test asserts.
                pass