diff --git a/libs/neo4j/tests/integration_tests/chat_message_histories/test_neo4j.py b/libs/neo4j/tests/integration_tests/chat_message_histories/test_neo4j.py index ebc7f53..8343da2 100644 --- a/libs/neo4j/tests/integration_tests/chat_message_histories/test_neo4j.py +++ b/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 + ) diff --git a/libs/neo4j/tests/unit_tests/chat_message_histories/__init__.py b/libs/neo4j/tests/unit_tests/chat_message_histories/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/libs/neo4j/tests/unit_tests/chat_message_histories/test_neo4j_chat_message_history.py b/libs/neo4j/tests/unit_tests/chat_message_histories/test_neo4j_chat_message_history.py new file mode 100644 index 0000000..45e84b0 --- /dev/null +++ b/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()