Adds tests so that Neo4jChatMessageHistory class now has 100% coverage

libs/neo4j/tests/integration_tests/chat_message_histories/test_neo4j.py

@@ -1,5 +1,7 @@
 import os
+import urllib.parse
 
+import pytest
 from langchain_core.messages import AIMessage, HumanMessage
 
 from langchain_neo4j.chat_message_histories.neo4j import Neo4jChatMessageHistory
@@ -82,3 +84,39 @@ def test_add_messages_graph_object() -> None:
     del os.environ["NEO4J_URI"]
     del os.environ["NEO4J_USERNAME"]
     del os.environ["NEO4J_PASSWORD"]
+
+
+def test_invalid_url() -> None:
+    """Test initializing with invalid credentials raises ValueError."""
+    # Parse the original URL
+    parsed_url = urllib.parse.urlparse(url)
+    # Increment the port number by 1 and wrap around if necessary
+    original_port = parsed_url.port or 7687
+    new_port = (original_port + 1) % 65535 or 1
+    # Reconstruct the netloc (hostname:port)
+    new_netloc = f"{parsed_url.hostname}:{new_port}"
+    # Rebuild the URL with the new netloc
+    new_url = parsed_url._replace(netloc=new_netloc).geturl()
+
+    with pytest.raises(ValueError) as exc_info:
+        Neo4jChatMessageHistory(
+            "test_session",
+            url=new_url,
+            username=username,
+            password=password,
+        )
+    assert "Please ensure that the url is correct" in str(exc_info.value)
+
+
+def test_invalid_credentials() -> None:
+    """Test initializing with invalid credentials raises ValueError."""
+    with pytest.raises(ValueError) as exc_info:
+        Neo4jChatMessageHistory(
+            "test_session",
+            url=url,
+            username="invalid_username",
+            password="invalid_password",
+        )
+    assert "Please ensure that the username and password are correct" in str(
+        exc_info.value
+    )

libs/neo4j/tests/unit_tests/chat_message_histories/test_neo4j_chat_message_history.py

@@ -0,0 +1,69 @@
+import gc
+from types import ModuleType
+from typing import Mapping, Sequence, Union
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from langchain_neo4j.chat_message_histories.neo4j import Neo4jChatMessageHistory
+
+
+def test_init_without_session_id() -> None:
+    """Test initializing without session_id raises ValueError."""
+    with pytest.raises(ValueError) as exc_info:
+        Neo4jChatMessageHistory(None)  # type: ignore[arg-type]
+    assert "Please ensure that the session_id parameter is provided" in str(
+        exc_info.value
+    )
+
+
+def test_messages_setter() -> None:
+    """Test that assigning to messages raises NotImplementedError."""
+    with patch("neo4j.GraphDatabase.driver", autospec=True):
+        message_store = Neo4jChatMessageHistory(
+            session_id="test_session",
+            url="bolt://url",
+            username="username",
+            password="password",
+        )
+
+        with pytest.raises(NotImplementedError) as exc_info:
+            message_store.messages = []
+        assert "Direct assignment to 'messages' is not allowed." in str(exc_info.value)
+
+
+def test_import_error() -> None:
+    """Test that ImportError is raised when neo4j package is not installed."""
+    original_import = __import__
+
+    def mock_import(
+        name: str,
+        globals: Union[Mapping[str, object], None] = None,
+        locals: Union[Mapping[str, object], None] = None,
+        fromlist: Sequence[str] = (),
+        level: int = 0,
+    ) -> ModuleType:
+        if name == "neo4j":
+            raise ImportError()
+        return original_import(name, globals, locals, fromlist, level)
+
+    with patch("builtins.__import__", side_effect=mock_import):
+        with pytest.raises(ImportError) as exc_info:
+            Neo4jChatMessageHistory("test_session")
+        assert "Could not import neo4j python package." in str(exc_info.value)
+
+
+def test_driver_closed_on_delete() -> None:
+    """Test that the driver is closed when the object is deleted."""
+    with patch("neo4j.GraphDatabase.driver", autospec=True):
+        message_store = Neo4jChatMessageHistory(
+            session_id="test_session",
+            url="bolt://url",
+            username="username",
+            password="password",
+        )
+        mock_driver = message_store._driver
+        assert isinstance(mock_driver.close, MagicMock)
+        message_store.__del__()
+        gc.collect()
+        mock_driver.close.assert_called_once()