Remove duplicate test files.

This commit is contained in:
Andrej Milicevic 2025-08-25 16:29:31 +02:00
parent a805c640ec
commit 573e24fb50
3 changed files with 0 additions and 710 deletions

View file

@ -1,484 +0,0 @@
"""
Tests for CLI edge cases and error scenarios.
"""
import pytest
import argparse
import tempfile
import os
import asyncio
from unittest.mock import patch, MagicMock, AsyncMock
from cognee.cli.commands.add_command import AddCommand
from cognee.cli.commands.search_command import SearchCommand
from cognee.cli.commands.cognify_command import CognifyCommand
from cognee.cli.commands.delete_command import DeleteCommand
from cognee.cli.commands.config_command import ConfigCommand
from cognee.cli.exceptions import CliCommandException, CliCommandInnerException
class TestAddCommandEdgeCases:
"""Test edge cases for AddCommand"""
@patch("cognee.cli.commands.add_command.asyncio.run")
@patch("cognee.cli.commands.add_command.cognee")
def test_add_empty_data_list(self, mock_cognee, mock_asyncio_run):
"""Test add command with empty data list"""
command = AddCommand()
# This shouldn't happen due to argparse nargs="+", but test defensive coding
args = argparse.Namespace(data=[], dataset_name="test_dataset")
mock_cognee.add = AsyncMock()
command.execute(args)
mock_asyncio_run.assert_called_once()
@patch("cognee.cli.commands.add_command.asyncio.run")
@patch("cognee.cli.commands.add_command.cognee")
def test_add_very_long_dataset_name(self, mock_cognee, mock_asyncio_run):
"""Test add command with very long dataset name"""
command = AddCommand()
long_name = "a" * 1000 # Very long dataset name
args = argparse.Namespace(data=["test.txt"], dataset_name=long_name)
mock_cognee.add = AsyncMock()
command.execute(args)
mock_asyncio_run.assert_called_once()
@patch("cognee.cli.commands.add_command.asyncio.run")
def test_add_asyncio_run_exception(self, mock_asyncio_run):
"""Test add command when asyncio.run itself fails"""
command = AddCommand()
args = argparse.Namespace(data=["test.txt"], dataset_name="test_dataset")
mock_asyncio_run.side_effect = RuntimeError("Event loop error")
with pytest.raises(CliCommandException):
command.execute(args)
def test_add_special_characters_in_data(self):
"""Test add command with special characters in file paths"""
command = AddCommand()
# Create parser to test argument parsing with special characters
parser = argparse.ArgumentParser()
command.configure_parser(parser)
# Test parsing with special characters
special_paths = [
"file with spaces.txt",
"file-with-dashes.txt",
"file_with_underscores.txt",
"file.with.dots.txt",
]
args = parser.parse_args(special_paths + ["--dataset-name", "test"])
assert args.data == special_paths
assert args.dataset_name == "test"
class TestSearchCommandEdgeCases:
"""Test edge cases for SearchCommand"""
@patch("cognee.cli.commands.search_command.asyncio.run")
@patch("cognee.cli.commands.search_command.cognee")
def test_search_empty_results(self, mock_cognee, mock_asyncio_run):
"""Test search command with empty results"""
command = SearchCommand()
args = argparse.Namespace(
query_text="nonexistent query",
query_type="GRAPH_COMPLETION",
datasets=None,
top_k=10,
system_prompt=None,
output_format="pretty",
)
mock_cognee.search = AsyncMock(return_value=[])
mock_asyncio_run.return_value = []
# Should handle empty results gracefully
command.execute(args)
mock_asyncio_run.assert_called_once()
@patch("cognee.cli.commands.search_command.asyncio.run")
@patch("cognee.cli.commands.search_command.cognee")
def test_search_very_large_top_k(self, mock_cognee, mock_asyncio_run):
"""Test search command with very large top-k value"""
command = SearchCommand()
args = argparse.Namespace(
query_text="test query",
query_type="CHUNKS",
datasets=None,
top_k=999999, # Very large value
system_prompt=None,
output_format="json",
)
mock_cognee.search = AsyncMock(return_value=["result1"])
mock_asyncio_run.return_value = ["result1"]
command.execute(args)
mock_asyncio_run.assert_called_once()
@patch("cognee.cli.commands.search_command.asyncio.run")
@patch("cognee.cli.commands.search_command.cognee")
def test_search_invalid_search_type_enum(self, mock_cognee, mock_asyncio_run):
"""Test search command with invalid SearchType enum conversion"""
command = SearchCommand()
args = argparse.Namespace(
query_text="test query",
query_type="INVALID_TYPE", # This would fail enum conversion
datasets=None,
top_k=10,
system_prompt=None,
output_format="pretty",
)
# Mock SearchType to raise KeyError
with patch("cognee.cli.commands.search_command.SearchType") as mock_search_type:
mock_search_type.__getitem__.side_effect = KeyError("INVALID_TYPE")
with pytest.raises(CliCommandException):
command.execute(args)
def test_search_unicode_query(self):
"""Test search command with unicode characters in query"""
command = SearchCommand()
parser = argparse.ArgumentParser()
command.configure_parser(parser)
unicode_query = "测试查询 🔍 émojis and spéciál chars"
args = parser.parse_args([unicode_query])
assert args.query_text == unicode_query
@patch("cognee.cli.commands.search_command.asyncio.run")
@patch("cognee.cli.commands.search_command.cognee")
def test_search_results_with_none_values(self, mock_cognee, mock_asyncio_run):
"""Test search command when results contain None values"""
command = SearchCommand()
args = argparse.Namespace(
query_text="test query",
query_type="CHUNKS",
datasets=None,
top_k=10,
system_prompt=None,
output_format="pretty",
)
# Results with None values
mock_cognee.search = AsyncMock(return_value=[None, "valid result", None])
mock_asyncio_run.return_value = [None, "valid result", None]
# Should handle None values gracefully
command.execute(args)
mock_asyncio_run.assert_called_once()
class TestCognifyCommandEdgeCases:
"""Test edge cases for CognifyCommand"""
@patch("cognee.cli.commands.cognify_command.asyncio.run")
@patch("cognee.cli.commands.cognify_command.cognee")
def test_cognify_invalid_chunk_size(self, mock_cognee, mock_asyncio_run):
"""Test cognify command with invalid chunk size"""
command = CognifyCommand()
args = argparse.Namespace(
datasets=None,
chunk_size=-100, # Invalid negative chunk size
ontology_file=None,
chunker="TextChunker",
background=False,
verbose=False,
)
mock_cognee.cognify = AsyncMock()
# Should pass the invalid value to cognify and let it handle the validation
command.execute(args)
mock_asyncio_run.assert_called_once()
@patch("cognee.cli.commands.cognify_command.asyncio.run")
@patch("cognee.cli.commands.cognify_command.cognee")
def test_cognify_nonexistent_ontology_file(self, mock_cognee, mock_asyncio_run):
"""Test cognify command with nonexistent ontology file"""
command = CognifyCommand()
args = argparse.Namespace(
datasets=None,
chunk_size=None,
ontology_file="/nonexistent/path/ontology.owl",
chunker="TextChunker",
background=False,
verbose=False,
)
mock_cognee.cognify = AsyncMock()
# Should pass the path to cognify and let it handle file validation
command.execute(args)
mock_asyncio_run.assert_called_once()
@patch("cognee.cli.commands.cognify_command.asyncio.run")
@patch("cognee.cli.commands.cognify_command.cognee")
def test_cognify_langchain_chunker_import_error(self, mock_cognee, mock_asyncio_run):
"""Test cognify command when LangchainChunker import fails"""
command = CognifyCommand()
args = argparse.Namespace(
datasets=None,
chunk_size=None,
ontology_file=None,
chunker="LangchainChunker",
background=False,
verbose=True,
)
mock_cognee.cognify = AsyncMock()
# Mock import error for LangchainChunker
with patch("cognee.cli.commands.cognify_command.LangchainChunker", side_effect=ImportError):
# Should fall back to TextChunker and show warning
command.execute(args)
mock_asyncio_run.assert_called_once()
def test_cognify_empty_datasets_list(self):
"""Test cognify command with empty datasets list"""
command = CognifyCommand()
parser = argparse.ArgumentParser()
command.configure_parser(parser)
# Empty datasets list should be handled
args = parser.parse_args(["--datasets"])
assert args.datasets == []
class TestDeleteCommandEdgeCases:
"""Test edge cases for DeleteCommand"""
@patch("cognee.cli.commands.delete_command.fmt.confirm")
@patch("cognee.cli.commands.delete_command.asyncio.run")
@patch("cognee.cli.commands.delete_command.cognee")
def test_delete_all_with_user_id(self, mock_cognee, mock_asyncio_run, mock_confirm):
"""Test delete command with both --all and --user-id"""
command = DeleteCommand()
args = argparse.Namespace(dataset_name=None, user_id="test_user", all=True, force=False)
mock_confirm.return_value = True
mock_cognee.delete = AsyncMock()
# Should handle both flags being set
command.execute(args)
mock_asyncio_run.assert_called_once()
@patch("cognee.cli.commands.delete_command.fmt.confirm")
def test_delete_confirmation_keyboard_interrupt(self, mock_confirm):
"""Test delete command when user interrupts confirmation"""
command = DeleteCommand()
args = argparse.Namespace(dataset_name="test_dataset", user_id=None, all=False, force=False)
mock_confirm.side_effect = KeyboardInterrupt()
# Should handle KeyboardInterrupt gracefully
with pytest.raises(KeyboardInterrupt):
command.execute(args)
@patch("cognee.cli.commands.delete_command.asyncio.run")
@patch("cognee.cli.commands.delete_command.cognee")
def test_delete_async_exception_handling(self, mock_cognee, mock_asyncio_run):
"""Test delete command async exception handling"""
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())
with pytest.raises(CliCommandException):
command.execute(args)
def test_delete_special_characters_in_dataset_name(self):
"""Test delete command with special characters in dataset name"""
command = DeleteCommand()
parser = argparse.ArgumentParser()
command.configure_parser(parser)
special_names = [
"dataset with spaces",
"dataset-with-dashes",
"dataset_with_underscores",
"dataset.with.dots",
"dataset/with/slashes",
]
for name in special_names:
args = parser.parse_args(["--dataset-name", name])
assert args.dataset_name == name
class TestConfigCommandEdgeCases:
"""Test edge cases for ConfigCommand"""
def test_config_no_subcommand_specified(self):
"""Test config command when no subcommand is specified"""
command = ConfigCommand()
parser = argparse.ArgumentParser()
command.configure_parser(parser)
# Parse with no subcommand - should set config_action to None
args = parser.parse_args([])
assert not hasattr(args, "config_action") or args.config_action is None
@patch("cognee.cli.commands.config_command.cognee")
def test_config_get_nonexistent_key(self, mock_cognee):
"""Test config get with nonexistent key"""
command = ConfigCommand()
args = argparse.Namespace(config_action="get", key="nonexistent_key")
# Mock config.get to raise exception for nonexistent key
mock_cognee.config.get = MagicMock(side_effect=KeyError("Key not found"))
# Should handle the exception gracefully
command.execute(args)
@patch("cognee.cli.commands.config_command.cognee")
def test_config_set_complex_json_value(self, mock_cognee):
"""Test config set with complex JSON value"""
command = ConfigCommand()
complex_json = '{"nested": {"key": "value"}, "array": [1, 2, 3]}'
args = argparse.Namespace(config_action="set", key="complex_config", value=complex_json)
mock_cognee.config.set = MagicMock()
command.execute(args)
# Should parse JSON and pass parsed object
expected_value = {"nested": {"key": "value"}, "array": [1, 2, 3]}
mock_cognee.config.set.assert_called_with("complex_config", expected_value)
@patch("cognee.cli.commands.config_command.cognee")
def test_config_set_invalid_json_value(self, mock_cognee):
"""Test config set with invalid JSON value"""
command = ConfigCommand()
invalid_json = '{"invalid": json}'
args = argparse.Namespace(config_action="set", key="test_key", value=invalid_json)
mock_cognee.config.set = MagicMock()
command.execute(args)
# Should treat as string when JSON parsing fails
mock_cognee.config.set.assert_called_with("test_key", invalid_json)
@patch("cognee.cli.commands.config_command.fmt.confirm")
@patch("cognee.cli.commands.config_command.cognee")
def test_config_unset_unknown_key(self, mock_cognee, mock_confirm):
"""Test config unset with unknown key"""
command = ConfigCommand()
args = argparse.Namespace(config_action="unset", key="unknown_key", force=False)
mock_confirm.return_value = True
# Should show error for unknown key
command.execute(args)
mock_confirm.assert_called_once()
@patch("cognee.cli.commands.config_command.cognee")
def test_config_unset_method_not_found(self, mock_cognee):
"""Test config unset when method doesn't exist on config object"""
command = ConfigCommand()
args = argparse.Namespace(config_action="unset", key="llm_provider", force=True)
# Mock config object without the expected method
mock_cognee.config = MagicMock()
del mock_cognee.config.set_llm_provider # Remove the method
# Should handle AttributeError gracefully
command.execute(args)
def test_config_invalid_subcommand(self):
"""Test config command with invalid subcommand"""
command = ConfigCommand()
args = argparse.Namespace(config_action="invalid_action")
# Should handle unknown subcommand gracefully
command.execute(args)
class TestGeneralEdgeCases:
"""Test general edge cases that apply to multiple commands"""
def test_command_with_none_args(self):
"""Test command execution with None args"""
commands = [
AddCommand(),
SearchCommand(),
CognifyCommand(),
DeleteCommand(),
ConfigCommand(),
]
for command in commands:
# Should not crash with None args, though it might raise exceptions
try:
command.execute(None)
except (AttributeError, CliCommandException):
# Expected behavior for None args
pass
def test_parser_configuration_with_none_parser(self):
"""Test parser configuration with None parser"""
commands = [
AddCommand(),
SearchCommand(),
CognifyCommand(),
DeleteCommand(),
ConfigCommand(),
]
for command in commands:
# Should not crash, though it might raise AttributeError
try:
command.configure_parser(None)
except AttributeError:
# Expected behavior for None parser
pass
def test_command_properties_are_strings(self):
"""Test that all command properties are proper strings"""
commands = [
AddCommand(),
SearchCommand(),
CognifyCommand(),
DeleteCommand(),
ConfigCommand(),
]
for command in commands:
assert isinstance(command.command_string, str)
assert len(command.command_string) > 0
assert isinstance(command.help_string, str)
assert len(command.help_string) > 0
if hasattr(command, "description") and command.description:
assert isinstance(command.description, str)
if hasattr(command, "docs_url") and command.docs_url:
assert isinstance(command.docs_url, str)
@patch("tempfile.NamedTemporaryFile")
def test_commands_with_temp_files(self, mock_temp_file):
"""Test commands that might work with temporary files"""
# Mock a temporary file
mock_file = MagicMock()
mock_file.name = "/tmp/test_file.txt"
mock_temp_file.return_value.__enter__.return_value = mock_file
# Test AddCommand with temp file
command = AddCommand()
parser = argparse.ArgumentParser()
command.configure_parser(parser)
args = parser.parse_args([mock_file.name])
assert args.data == [mock_file.name]

View file

@ -1,173 +0,0 @@
"""
Tests for the main CLI entry point and command discovery.
"""
import pytest
import argparse
from unittest.mock import patch, MagicMock
from cognee.cli._cognee import main, _discover_commands, _create_parser
from cognee.cli.exceptions import CliCommandException, CliCommandInnerException
class TestCliMain:
"""Test the main CLI functionality"""
def test_discover_commands(self):
"""Test that all expected commands are discovered"""
commands = _discover_commands()
# Check that we get command classes back
assert len(commands) > 0
# Check that we have the expected commands
command_strings = []
for command_class in commands:
command = command_class()
command_strings.append(command.command_string)
expected_commands = ["add", "search", "cognify", "delete", "config"]
for expected_command in expected_commands:
assert expected_command in command_strings
def test_create_parser(self):
"""Test parser creation and command installation"""
parser, installed_commands = _create_parser()
# Check parser is created
assert isinstance(parser, argparse.ArgumentParser)
# Check commands are installed
expected_commands = ["add", "search", "cognify", "delete", "config"]
for expected_command in expected_commands:
assert expected_command in installed_commands
# Check parser has version argument
actions = [action.dest for action in parser._actions]
assert "version" in actions
@patch("cognee.cli._cognee._create_parser")
def test_main_no_command(self, mock_create_parser):
"""Test main function when no command is provided"""
mock_parser = MagicMock()
mock_parser.parse_args.return_value = MagicMock(command=None)
mock_create_parser.return_value = (mock_parser, {})
result = main()
assert result == -1
mock_parser.print_help.assert_called_once()
@patch("cognee.cli._cognee._create_parser")
def test_main_with_valid_command(self, mock_create_parser):
"""Test main function with a valid command"""
mock_command = MagicMock()
mock_command.execute.return_value = None
mock_parser = MagicMock()
mock_args = MagicMock(command="test")
mock_parser.parse_args.return_value = mock_args
mock_create_parser.return_value = (mock_parser, {"test": mock_command})
result = main()
assert result == 0
mock_command.execute.assert_called_once_with(mock_args)
@patch("cognee.cli._cognee._create_parser")
@patch("cognee.cli.debug.is_debug_enabled")
def test_main_with_command_exception(self, mock_debug, mock_create_parser):
"""Test main function when command raises exception"""
mock_debug.return_value = False
mock_command = MagicMock()
mock_command.execute.side_effect = CliCommandException("Test error", error_code=2)
mock_parser = MagicMock()
mock_args = MagicMock(command="test")
mock_parser.parse_args.return_value = mock_args
mock_create_parser.return_value = (mock_parser, {"test": mock_command})
result = main()
assert result == 2
@patch("cognee.cli._cognee._create_parser")
@patch("cognee.cli.debug.is_debug_enabled")
def test_main_with_generic_exception(self, mock_debug, mock_create_parser):
"""Test main function when command raises generic exception"""
mock_debug.return_value = False
mock_command = MagicMock()
mock_command.execute.side_effect = Exception("Generic error")
mock_parser = MagicMock()
mock_args = MagicMock(command="test")
mock_parser.parse_args.return_value = mock_args
mock_create_parser.return_value = (mock_parser, {"test": mock_command})
result = main()
assert result == -1
@patch("cognee.cli._cognee._create_parser")
@patch("cognee.cli.debug.is_debug_enabled")
def test_main_debug_mode_reraises_exception(self, mock_debug, mock_create_parser):
"""Test main function reraises exceptions in debug mode"""
mock_debug.return_value = True
test_exception = CliCommandException(
"Test error", error_code=2, raiseable_exception=ValueError("Inner error")
)
mock_command = MagicMock()
mock_command.execute.side_effect = test_exception
mock_parser = MagicMock()
mock_args = MagicMock(command="test")
mock_parser.parse_args.return_value = mock_args
mock_create_parser.return_value = (mock_parser, {"test": mock_command})
with pytest.raises(ValueError, match="Inner error"):
main()
def test_version_argument(self):
"""Test that version argument is properly configured"""
parser, _ = _create_parser()
# Check that version action exists
version_actions = [action for action in parser._actions if action.dest == "version"]
assert len(version_actions) == 1
version_action = version_actions[0]
assert "cognee" in version_action.version
def test_debug_argument(self):
"""Test that debug argument is properly configured"""
parser, _ = _create_parser()
# Check that debug action exists
debug_actions = [action for action in parser._actions if action.dest == "debug"]
assert len(debug_actions) == 1
class TestDebugAction:
"""Test the DebugAction class"""
@patch("cognee.cli.debug.enable_debug")
@patch("cognee.cli.echo.note")
def test_debug_action_call(self, mock_note, mock_enable_debug):
"""Test that DebugAction enables debug mode"""
from cognee.cli._cognee import DebugAction
action = DebugAction([])
parser = MagicMock()
namespace = MagicMock()
action(parser, namespace, None)
mock_enable_debug.assert_called_once()
mock_note.assert_called_once_with("Debug mode enabled. Full stack traces will be shown.")

View file

@ -1,53 +0,0 @@
"""
Test runner and utilities for CLI tests.
"""
import pytest
import sys
import os
from pathlib import Path
def run_cli_tests():
"""Run all CLI tests"""
test_dir = Path(__file__).parent
cli_test_files = [
"test_cli_main.py",
"test_cli_commands.py",
"test_cli_utils.py",
"test_cli_integration.py",
"test_cli_edge_cases.py",
]
# Run tests with pytest
args = ["-v", "--tb=short"]
for test_file in cli_test_files:
test_path = test_dir / test_file
if test_path.exists():
args.append(str(test_path))
return pytest.main(args)
def run_specific_cli_test(test_file):
"""Run a specific CLI test file"""
test_dir = Path(__file__).parent
test_path = test_dir / test_file
if not test_path.exists():
print(f"Test file {test_file} not found")
return 1
return pytest.main(["-v", "--tb=short", str(test_path)])
if __name__ == "__main__":
if len(sys.argv) > 1:
# Run specific test file
exit_code = run_specific_cli_test(sys.argv[1])
else:
# Run all CLI tests
exit_code = run_cli_tests()
sys.exit(exit_code)