Feature/cog 2768 inspect cli tests fix failing ones (#1286)

<!-- .github/pull_request_template.md -->

## Description
<!-- Provide a clear description of the changes in this PR -->
I went through CLI unit and integration tests, removed ones that
provided little to no value. Improved and fixed some tests, where it was
needed.

## DCO Affirmation
I affirm that all code in every commit of this pull request conforms to
the terms of the Topoteretes Developer Certificate of Origin.
This commit is contained in:
Vasilije 2025-08-26 13:50:53 +02:00 committed by GitHub
commit a6a33e9576
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 86 additions and 922 deletions

View file

@ -103,6 +103,16 @@ jobs:
integration-tests:
name: Run Integration Tests
runs-on: ubuntu-22.04
env:
LLM_PROVIDER: openai
LLM_MODEL: ${{ secrets.LLM_MODEL }}
LLM_ENDPOINT: ${{ secrets.LLM_ENDPOINT }}
LLM_API_KEY: ${{ secrets.LLM_API_KEY }}
LLM_API_VERSION: ${{ secrets.LLM_API_VERSION }}
EMBEDDING_MODEL: ${{ secrets.EMBEDDING_MODEL }}
EMBEDDING_ENDPOINT: ${{ secrets.EMBEDDING_ENDPOINT }}
EMBEDDING_API_KEY: ${{ secrets.EMBEDDING_API_KEY }}
EMBEDDING_API_VERSION: ${{ secrets.EMBEDDING_API_VERSION }}
steps:
- name: Check out repository
uses: actions/checkout@v4

View file

@ -38,6 +38,16 @@ jobs:
cli-unit-tests:
name: CLI Unit Tests
runs-on: ubuntu-22.04
env:
LLM_PROVIDER: openai
LLM_MODEL: ${{ secrets.LLM_MODEL }}
LLM_ENDPOINT: ${{ secrets.LLM_ENDPOINT }}
LLM_API_KEY: ${{ secrets.LLM_API_KEY }}
LLM_API_VERSION: ${{ secrets.LLM_API_VERSION }}
EMBEDDING_MODEL: ${{ secrets.EMBEDDING_MODEL }}
EMBEDDING_ENDPOINT: ${{ secrets.EMBEDDING_ENDPOINT }}
EMBEDDING_API_KEY: ${{ secrets.EMBEDDING_API_KEY }}
EMBEDDING_API_VERSION: ${{ secrets.EMBEDDING_API_VERSION }}
steps:
- name: Check out repository
uses: actions/checkout@v4
@ -61,6 +71,10 @@ jobs:
LLM_ENDPOINT: ${{ secrets.LLM_ENDPOINT }}
LLM_API_KEY: ${{ secrets.LLM_API_KEY }}
LLM_API_VERSION: ${{ secrets.LLM_API_VERSION }}
EMBEDDING_MODEL: ${{ secrets.EMBEDDING_MODEL }}
EMBEDDING_ENDPOINT: ${{ secrets.EMBEDDING_ENDPOINT }}
EMBEDDING_API_KEY: ${{ secrets.EMBEDDING_API_KEY }}
EMBEDDING_API_VERSION: ${{ secrets.EMBEDDING_API_VERSION }}
steps:
- name: Check out repository
uses: actions/checkout@v4
@ -78,6 +92,16 @@ jobs:
cli-functionality-tests:
name: CLI Functionality Tests
runs-on: ubuntu-22.04
env:
LLM_PROVIDER: openai
LLM_MODEL: ${{ secrets.LLM_MODEL }}
LLM_ENDPOINT: ${{ secrets.LLM_ENDPOINT }}
LLM_API_KEY: ${{ secrets.LLM_API_KEY }}
LLM_API_VERSION: ${{ secrets.LLM_API_VERSION }}
EMBEDDING_MODEL: ${{ secrets.EMBEDDING_MODEL }}
EMBEDDING_ENDPOINT: ${{ secrets.EMBEDDING_ENDPOINT }}
EMBEDDING_API_KEY: ${{ secrets.EMBEDDING_API_KEY }}
EMBEDDING_API_VERSION: ${{ secrets.EMBEDDING_API_VERSION }}
steps:
- name: Check out repository
uses: actions/checkout@v4

View file

@ -387,8 +387,8 @@ def setup_logging(log_level=None, name=None):
base_config = get_base_config()
databases_path = os.path.join(base_config.system_root_directory, "databases")
except:
raise ValueError
except Exception as e:
raise ValueError from e
# Get a configured logger and log system information
logger = structlog.get_logger(name if name else __name__)

View file

@ -2,7 +2,6 @@
Integration tests for CLI commands that test end-to-end functionality.
"""
import pytest
import tempfile
import os
import sys

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)

View file

@ -4,9 +4,7 @@ Tests for CLI edge cases and error scenarios with proper mocking.
import pytest
import argparse
import tempfile
import os
import asyncio
import types
from unittest.mock import patch, MagicMock, AsyncMock
from cognee.cli.commands.add_command import AddCommand
from cognee.cli.commands.search_command import SearchCommand
@ -35,21 +33,8 @@ class TestAddCommandEdgeCases:
command.execute(args)
mock_asyncio_run.assert_called_once()
@patch("builtins.__import__")
@patch("cognee.cli.commands.add_command.asyncio.run")
def test_add_very_long_dataset_name(self, mock_asyncio_run, mock_import):
"""Test add command with very long dataset name"""
# Mock the cognee module
mock_cognee = MagicMock()
mock_cognee.add = AsyncMock()
mock_import.return_value = mock_cognee
command = AddCommand()
long_name = "a" * 1000 # Very long dataset name
args = argparse.Namespace(data=["test.txt"], dataset_name=long_name)
command.execute(args)
mock_asyncio_run.assert_called_once()
coro = mock_asyncio_run.call_args[0][0]
assert isinstance(coro, types.CoroutineType)
@patch("cognee.cli.commands.add_command.asyncio.run")
def test_add_asyncio_run_exception(self, mock_asyncio_run):
@ -337,15 +322,37 @@ class TestCognifyCommandEdgeCases:
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)
@patch("builtins.__import__")
@patch("cognee.cli.commands.cognify_command.asyncio.run")
def test_cognify_empty_datasets_list(self, mock_asyncio_run, mock_import):
"""Test cognify command with nonexistent ontology file"""
# Mock the cognee module
mock_cognee = MagicMock()
mock_cognee.cognify = AsyncMock()
# Empty datasets list should be handled
args = parser.parse_args(["--datasets"])
assert args.datasets == []
def mock_import_func(name, fromlist=None, *args, **kwargs):
if name == "cognee":
return mock_cognee
elif name == "cognee.modules.chunking":
module = MagicMock()
module.TextChunker = MagicMock()
return module
return MagicMock()
mock_import.side_effect = mock_import_func
command = CognifyCommand()
args = argparse.Namespace(
datasets=[],
chunk_size=None,
ontology_file=None,
chunker="TextChunker",
background=False,
verbose=False,
)
command.execute(args)
mock_asyncio_run.assert_called_once()
class TestDeleteCommandEdgeCases:
@ -439,6 +446,7 @@ class TestConfigCommandEdgeCases:
# Should handle the exception gracefully
command.execute(args)
mock_cognee.config.get.assert_called_once_with("nonexistent_key")
@patch("builtins.__import__")
def test_config_set_complex_json_value(self, mock_import):
@ -450,9 +458,13 @@ class TestConfigCommandEdgeCases:
command = ConfigCommand()
complex_json = '{"nested": {"key": "value"}, "array": [1, 2, 3]}'
complex_json_expected_value = {"nested": {"key": "value"}, "array": [1, 2, 3]}
args = argparse.Namespace(config_action="set", key="complex_config", value=complex_json)
command.execute(args)
mock_cognee.config.set.assert_called_once_with(
"complex_config", complex_json_expected_value
)
@patch("builtins.__import__")
def test_config_set_invalid_json_value(self, mock_import):
@ -467,6 +479,7 @@ class TestConfigCommandEdgeCases:
args = argparse.Namespace(config_action="set", key="test_key", value=invalid_json)
command.execute(args)
mock_cognee.config.set.assert_called_once_with("test_key", invalid_json)
@patch("cognee.cli.commands.config_command.fmt.confirm")
@patch("builtins.__import__")
@ -499,6 +512,7 @@ class TestConfigCommandEdgeCases:
# Should handle AttributeError gracefully
command.execute(args)
mock_cognee.config.unset.assert_not_called()
def test_config_invalid_subcommand(self):
"""Test config command with invalid subcommand"""

View file

@ -4,7 +4,6 @@ Test runner and utilities for CLI tests.
import pytest
import sys
import os
from pathlib import Path

View file

@ -2,11 +2,6 @@
Tests for CLI utility functions and helper modules.
"""
import pytest
from unittest.mock import patch, MagicMock
import click
from cognee.cli import echo as fmt
from cognee.cli.exceptions import CliCommandException, CliCommandInnerException
from cognee.cli import debug
from cognee.cli.config import (
CLI_DESCRIPTION,
@ -16,154 +11,7 @@ from cognee.cli.config import (
CHUNKER_CHOICES,
OUTPUT_FORMAT_CHOICES,
)
class TestEchoModule:
"""Test the CLI echo/formatting module"""
@patch("click.secho")
def test_echo_basic(self, mock_secho):
"""Test basic echo functionality"""
fmt.echo("test message")
mock_secho.assert_called_once_with("test message", fg=None, err=False)
@patch("click.secho")
def test_echo_with_color(self, mock_secho):
"""Test echo with color"""
fmt.echo("test message", color="red")
mock_secho.assert_called_once_with("test message", fg="red", err=False)
@patch("click.secho")
def test_echo_to_stderr(self, mock_secho):
"""Test echo to stderr"""
fmt.echo("test message", err=True)
mock_secho.assert_called_once_with("test message", fg=None, err=True)
@patch("cognee.cli.echo.echo")
def test_note(self, mock_echo):
"""Test note formatting"""
fmt.note("test note")
mock_echo.assert_called_once_with("Note: test note", color="blue")
@patch("cognee.cli.echo.echo")
def test_warning(self, mock_echo):
"""Test warning formatting"""
fmt.warning("test warning")
mock_echo.assert_called_once_with("Warning: test warning", color="yellow")
@patch("cognee.cli.echo.echo")
def test_error(self, mock_echo):
"""Test error formatting"""
fmt.error("test error")
mock_echo.assert_called_once_with("Error: test error", color="red", err=True)
@patch("cognee.cli.echo.echo")
def test_success(self, mock_echo):
"""Test success formatting"""
fmt.success("test success")
mock_echo.assert_called_once_with("Success: test success", color="green")
@patch("click.style")
def test_bold(self, mock_style):
"""Test bold text formatting"""
mock_style.return_value = "bold text"
result = fmt.bold("test text")
mock_style.assert_called_once_with("test text", bold=True)
assert result == "bold text"
@patch("click.confirm")
def test_confirm(self, mock_confirm):
"""Test confirmation prompt"""
mock_confirm.return_value = True
result = fmt.confirm("Are you sure?")
mock_confirm.assert_called_once_with("Are you sure?", default=False)
assert result is True
@patch("click.confirm")
def test_confirm_with_default(self, mock_confirm):
"""Test confirmation prompt with default"""
mock_confirm.return_value = False
result = fmt.confirm("Are you sure?", default=True)
mock_confirm.assert_called_once_with("Are you sure?", default=True)
assert result is False
@patch("click.prompt")
def test_prompt(self, mock_prompt):
"""Test user input prompt"""
mock_prompt.return_value = "user input"
result = fmt.prompt("Enter value:")
mock_prompt.assert_called_once_with("Enter value:", default=None)
assert result == "user input"
@patch("click.prompt")
def test_prompt_with_default(self, mock_prompt):
"""Test user input prompt with default"""
mock_prompt.return_value = "default value"
result = fmt.prompt("Enter value:", default="default value")
mock_prompt.assert_called_once_with("Enter value:", default="default value")
assert result == "default value"
class TestCliExceptions:
"""Test CLI exception classes"""
def test_cli_command_exception_basic(self):
"""Test basic CliCommandException"""
exc = CliCommandException("Test error")
assert str(exc) == "Test error"
assert exc.error_code == -1
assert exc.docs_url is None
assert exc.raiseable_exception is None
def test_cli_command_exception_full(self):
"""Test CliCommandException with all parameters"""
inner_exc = ValueError("Inner error")
exc = CliCommandException(
"Test error",
error_code=2,
docs_url="https://docs.test.com",
raiseable_exception=inner_exc,
)
assert str(exc) == "Test error"
assert exc.error_code == 2
assert exc.docs_url == "https://docs.test.com"
assert exc.raiseable_exception is inner_exc
def test_cli_command_inner_exception(self):
"""Test CliCommandInnerException"""
exc = CliCommandInnerException("Inner error")
assert str(exc) == "Inner error"
assert isinstance(exc, Exception)
class TestDebugModule:
"""Test CLI debug functionality"""
def test_debug_initially_disabled(self):
"""Test that debug is initially disabled"""
# Reset debug state
debug._debug_enabled = False
assert not debug.is_debug_enabled()
def test_enable_debug(self):
"""Test enabling debug mode"""
debug.enable_debug()
assert debug.is_debug_enabled()
# Reset for other tests
debug._debug_enabled = False
def test_debug_state_persistence(self):
"""Test that debug state persists"""
debug.enable_debug()
assert debug.is_debug_enabled()
# Should still be enabled
assert debug.is_debug_enabled()
# Reset for other tests
debug._debug_enabled = False
from cognee.cli._cognee import _discover_commands
class TestCliConfig:
@ -180,10 +28,17 @@ class TestCliConfig:
assert DEFAULT_DOCS_URL
assert isinstance(DEFAULT_DOCS_URL, str)
assert DEFAULT_DOCS_URL.startswith("https://")
assert "cognee.ai" in DEFAULT_DOCS_URL
def test_command_descriptions_complete(self):
"""Test that all expected commands have descriptions"""
expected_commands = ["add", "search", "cognify", "delete", "config"]
commands = _discover_commands()
assert len(commands) > 0
expected_commands = []
for command_class in commands:
command = command_class()
expected_commands.append(command.command_string)
for command in expected_commands:
assert command in COMMAND_DESCRIPTIONS
@ -262,38 +117,11 @@ class TestCliReference:
class TestCliUtilityFunctions:
"""Test utility functions and edge cases"""
def test_echo_empty_message(self):
"""Test echo with empty message"""
with patch("click.secho") as mock_secho:
fmt.echo()
mock_secho.assert_called_once_with("", fg=None, err=False)
def test_echo_none_message(self):
"""Test echo with None message (should not crash)"""
with patch("click.secho") as mock_secho:
# This might raise an exception, which is expected behavior
try:
fmt.echo(None)
except TypeError:
# Expected for None message
pass
def test_multiple_debug_enable_calls(self):
"""Test multiple calls to enable_debug"""
debug.enable_debug()
debug.enable_debug() # Should not cause issues
assert debug.is_debug_enabled()
assert debug.is_debug_enabled() is True
# Reset for other tests
debug._debug_enabled = False
def test_config_constants_immutability(self):
"""Test that config constants are not accidentally modified"""
original_description = CLI_DESCRIPTION
original_url = DEFAULT_DOCS_URL
original_commands = COMMAND_DESCRIPTIONS.copy()
# These should be the same after any test
assert CLI_DESCRIPTION == original_description
assert DEFAULT_DOCS_URL == original_url
assert COMMAND_DESCRIPTIONS == original_commands