TestInitValidator Class — langchain Architecture

Source Code

libs/core/tests/unit_tests/load/test_serializable.py lines 595–712

class TestInitValidator:
    """Tests for `init_validator` on `load()` and `loads()`."""

    def test_init_validator_allows_valid_kwargs(self) -> None:
        """Test that `init_validator` returning None allows deserialization."""
        msg = AIMessage(content="Hello")
        serialized = dumpd(msg)

        def allow_all(_class_path: tuple[str, ...], _kwargs: dict[str, Any]) -> None:
            pass  # Allow all by doing nothing

        loaded = load(serialized, allowed_objects=[AIMessage], init_validator=allow_all)
        assert loaded == msg

    def test_init_validator_blocks_deserialization(self) -> None:
        """Test that `init_validator` can block deserialization by raising."""
        doc = Document(page_content="test", metadata={"source": "test.txt"})
        serialized = dumpd(doc)

        def block_metadata(
            _class_path: tuple[str, ...], kwargs: dict[str, Any]
        ) -> None:
            if "metadata" in kwargs:
                msg = "Metadata not allowed"
                raise ValueError(msg)

        with pytest.raises(ValueError, match="Metadata not allowed"):
            load(serialized, allowed_objects=[Document], init_validator=block_metadata)

    def test_init_validator_receives_correct_class_path(self) -> None:
        """Test that `init_validator` receives the correct class path."""
        msg = AIMessage(content="Hello")
        serialized = dumpd(msg)

        received_class_paths: list[tuple[str, ...]] = []

        def capture_class_path(
            class_path: tuple[str, ...], _kwargs: dict[str, Any]
        ) -> None:
            received_class_paths.append(class_path)

        load(serialized, allowed_objects=[AIMessage], init_validator=capture_class_path)

        assert len(received_class_paths) == 1
        assert received_class_paths[0] == (
            "langchain",
            "schema",
            "messages",
            "AIMessage",
        )

    def test_init_validator_receives_correct_kwargs(self) -> None:
        """Test that `init_validator` receives the kwargs dict."""
        msg = AIMessage(content="Hello world", name="test_name")
        serialized = dumpd(msg)

        received_kwargs: list[dict[str, Any]] = []

        def capture_kwargs(
            _class_path: tuple[str, ...], kwargs: dict[str, Any]
        ) -> None:
            received_kwargs.append(kwargs)

        load(serialized, allowed_objects=[AIMessage], init_validator=capture_kwargs)

        assert len(received_kwargs) == 1
        assert "content" in received_kwargs[0]
        assert received_kwargs[0]["content"] == "Hello world"
        assert "name" in received_kwargs[0]
        assert received_kwargs[0]["name"] == "test_name"

    def test_init_validator_with_loads(self) -> None:
        """Test that `init_validator` works with `loads()` function."""
        doc = Document(page_content="test", metadata={"key": "value"})
        json_str = dumps(doc)

        def block_metadata(
            _class_path: tuple[str, ...], kwargs: dict[str, Any]
        ) -> None:
            if "metadata" in kwargs:
                msg = "Metadata not allowed"