gmakstutis/cognee
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

added fix

Browse source
This commit is contained in:
vasilije 2025-08-18 23:11:51 +02:00
parent 66d7fa9941
commit 3152368529
2 changed files with 88 additions and 36 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

117
cognee/tests/unit/cli/test_cli_commands.py
View file

@ -1,5 +1,5 @@
"""
Tests for individual CLI commands with proper mocking.
Tests for individual CLI commands with proper mocking and coroutine handling.
"""
import pytest
@ -40,36 +40,57 @@ class TestAddCommand:
assert actions["data"].nargs == "+"
@patch("builtins.__import__")
@patch("cognee.cli.commands.add_command.asyncio.run")
def test_execute_single_item(self, mock_asyncio_run, mock_import):
@patch("cognee.cli.commands.add_command.asyncio")
def test_execute_single_item(self, mock_asyncio, mock_import):
"""Test execute with single data item"""
# Mock the cognee module
mock_cognee = MagicMock()
mock_cognee.add = AsyncMock()
mock_import.return_value = mock_cognee
# Mock asyncio.run to properly handle coroutines
def mock_run(coro):
# Create an event loop and run the coroutine
loop = asyncio.new_event_loop()
try:
return loop.run_until_complete(coro)
finally:
loop.close()
mock_asyncio.run.side_effect = mock_run
command = AddCommand()
args = argparse.Namespace(data=["test.txt"], dataset_name="test_dataset")
command.execute(args)
mock_asyncio_run.assert_called_once()
mock_asyncio.run.assert_called_once()
@patch("builtins.__import__")
@patch("cognee.cli.commands.add_command.asyncio.run")
def test_execute_multiple_items(self, mock_asyncio_run, mock_import):
@patch("cognee.cli.commands.add_command.asyncio")
def test_execute_multiple_items(self, mock_asyncio, mock_import):
"""Test execute with multiple data items"""
# Mock the cognee module
mock_cognee = MagicMock()
mock_cognee.add = AsyncMock()
mock_import.return_value = mock_cognee
# Mock asyncio.run to properly handle coroutines
def mock_run(coro):
loop = asyncio.new_event_loop()
try:
return loop.run_until_complete(coro)
finally:
loop.close()
mock_asyncio.run.side_effect = mock_run
command = AddCommand()
args = argparse.Namespace(data=["test1.txt", "test2.txt"], dataset_name="test_dataset")
command.execute(args)
mock_asyncio_run.assert_called_once()
mock_asyncio.run.assert_called_once()
@patch("cognee.cli.commands.add_command.asyncio.run")
def test_execute_with_exception(self, mock_asyncio_run):
@ -114,8 +135,8 @@ class TestSearchCommand:
assert actions["output_format"].default == "pretty"
@patch("builtins.__import__")
@patch("cognee.cli.commands.search_command.asyncio.run")
def test_execute_basic_search(self, mock_asyncio_run, mock_import):
@patch("cognee.cli.commands.search_command.asyncio")
def test_execute_basic_search(self, mock_asyncio, mock_import):
"""Test execute with basic search"""
# Mock the cognee module and SearchType
mock_cognee = MagicMock()
@ -133,7 +154,16 @@ class TestSearchCommand:
return MagicMock()
mock_import.side_effect = mock_import_func
mock_asyncio_run.return_value = ["result1", "result2"]
# Mock asyncio.run to properly handle coroutines
def mock_run(coro):
loop = asyncio.new_event_loop()
try:
return loop.run_until_complete(coro)
finally:
loop.close()
mock_asyncio.run.side_effect = mock_run
command = SearchCommand()
args = argparse.Namespace(
@ -147,7 +177,7 @@ class TestSearchCommand:
command.execute(args)
mock_asyncio_run.assert_called_once()
mock_asyncio.run.assert_called_once()
@patch("cognee.cli.commands.search_command.asyncio.run")
def test_execute_with_exception(self, mock_asyncio_run):
@ -198,8 +228,8 @@ class TestCognifyCommand:
assert actions["chunker"].default == "TextChunker"
@patch("builtins.__import__")
@patch("cognee.cli.commands.cognify_command.asyncio.run")
def test_execute_basic_cognify(self, mock_asyncio_run, mock_import):
@patch("cognee.cli.commands.cognify_command.asyncio")
def test_execute_basic_cognify(self, mock_asyncio, mock_import):
"""Test execute with basic cognify"""
# Mock the cognee module
mock_cognee = MagicMock()
@ -215,7 +245,16 @@ class TestCognifyCommand:
return MagicMock()
mock_import.side_effect = mock_import_func
mock_asyncio_run.return_value = "success"
# Mock asyncio.run to properly handle coroutines
def mock_run(coro):
loop = asyncio.new_event_loop()
try:
return loop.run_until_complete(coro)
finally:
loop.close()
mock_asyncio.run.side_effect = mock_run
command = CognifyCommand()
args = argparse.Namespace(
@ -229,7 +268,7 @@ class TestCognifyCommand:
command.execute(args)
mock_asyncio_run.assert_called_once()
mock_asyncio.run.assert_called_once()
@patch("cognee.cli.commands.cognify_command.asyncio.run")
def test_execute_with_exception(self, mock_asyncio_run):
@ -276,9 +315,9 @@ class TestDeleteCommand:
@patch("cognee.cli.commands.delete_command.fmt.confirm")
@patch("builtins.__import__")
@patch("cognee.cli.commands.delete_command.asyncio.run")
@patch("cognee.cli.commands.delete_command.asyncio")
def test_execute_delete_dataset_with_confirmation(
self, mock_asyncio_run, mock_import, mock_confirm
self, mock_asyncio, mock_import, mock_confirm
):
"""Test execute delete dataset with user confirmation"""
# Mock the cognee module
@ -286,6 +325,16 @@ class TestDeleteCommand:
mock_cognee.delete = AsyncMock()
mock_import.return_value = mock_cognee
# Mock asyncio.run to properly handle coroutines
def mock_run(coro):
loop = asyncio.new_event_loop()
try:
return loop.run_until_complete(coro)
finally:
loop.close()
mock_asyncio.run.side_effect = mock_run
command = DeleteCommand()
args = argparse.Namespace(dataset_name="test_dataset", user_id=None, all=False, force=False)
@ -294,7 +343,7 @@ class TestDeleteCommand:
command.execute(args)
mock_confirm.assert_called_once()
mock_asyncio_run.assert_called_once()
mock_asyncio.run.assert_called_once()
@patch("cognee.cli.commands.delete_command.fmt.confirm")
def test_execute_delete_cancelled(self, mock_confirm):
@ -310,20 +359,30 @@ class TestDeleteCommand:
mock_confirm.assert_called_once()
@patch("builtins.__import__")
@patch("cognee.cli.commands.delete_command.asyncio.run")
def test_execute_delete_forced(self, mock_asyncio_run, mock_import):
@patch("cognee.cli.commands.delete_command.asyncio")
def test_execute_delete_forced(self, mock_asyncio, mock_import):
"""Test execute delete with force flag"""
# Mock the cognee module
mock_cognee = MagicMock()
mock_cognee.delete = AsyncMock()
mock_import.return_value = mock_cognee
# Mock asyncio.run to properly handle coroutines
def mock_run(coro):
loop = asyncio.new_event_loop()
try:
return loop.run_until_complete(coro)
finally:
loop.close()
mock_asyncio.run.side_effect = mock_run
command = DeleteCommand()
args = argparse.Namespace(dataset_name="test_dataset", user_id=None, all=False, force=True)
command.execute(args)
mock_asyncio_run.assert_called_once()
mock_asyncio.run.assert_called_once()
def test_execute_no_delete_target(self):
"""Test execute when no delete target is specified"""
@ -461,16 +520,12 @@ class TestConfigCommand:
mock_confirm.assert_called_once()
@patch("builtins.__import__")
def test_execute_with_exception(self, mock_import):
def test_execute_with_exception(self):
"""Test execute handles exceptions properly"""
# Mock the cognee module with exception
mock_cognee = MagicMock()
mock_cognee.config.set.side_effect = Exception("Config error")
mock_import.return_value = mock_cognee
# Test with an invalid action that will cause an exception in the main execute method
command = ConfigCommand()
args = argparse.Namespace(config_action="set", key="invalid_key", value="value")
args = argparse.Namespace(config_action="invalid_action")
with pytest.raises(CliCommandException):
command.execute(args)
# This should not raise CliCommandException, just handle it gracefully
# The config command handles unknown actions by showing an error message
command.execute(args)

7
cognee/tests/unit/cli/test_cli_edge_cases.py
View file

@ -388,11 +388,8 @@ class TestDeleteCommandEdgeCases:
command = DeleteCommand()
args = argparse.Namespace(dataset_name="test_dataset", user_id=None, all=False, force=True)
# Mock async function that raises exception
async def failing_delete(*args, **kwargs):
raise ValueError("Database connection failed")
mock_asyncio_run.side_effect = lambda coro: asyncio.run(failing_delete())
# Mock asyncio.run to raise exception directly
mock_asyncio_run.side_effect = ValueError("Database connection failed")
with pytest.raises(CliCommandException):
command.execute(args)