forked from langchain-ai/langchain-neo4j
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Adds tests so that Neo4jChatMessageHistory class now has 100% coverage (
langchain-ai#20) * Adds tests so that Neo4jChatMessageHistory class now has 100% coverage * Fixed 3.10 issue
- Loading branch information
1 parent
199d500
commit 6b31b1f
Showing
3 changed files
with
107 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Empty file.
69 changes: 69 additions & 0 deletions
69
libs/neo4j/tests/unit_tests/chat_message_histories/test_neo4j_chat_message_history.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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() |