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

added tests

Browse source
This commit is contained in:
vasilije 2025-08-18 22:58:14 +02:00
parent cae173b106
commit d084d00a4d
13 changed files with 2651 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

94
.github/workflows/cli_tests.yml vendored Normal file
View file

@ -0,0 +1,94 @@
name: CLI Tests
on:
workflow_call:
inputs:
python-version:
required: false
type: string
default: '3.11.x'
env:
RUNTIME__LOG_LEVEL: ERROR
ENV: 'dev'
jobs:
cli-unit-tests:
name: CLI Unit Tests
runs-on: ubuntu-22.04
steps:
- name: Check out repository
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Cognee Setup
uses: ./.github/actions/cognee_setup
with:
python-version: ${{ inputs.python-version }}
- name: Run CLI Unit Tests
run: uv run pytest cognee/tests/unit/cli/ -v
cli-integration-tests:
name: CLI Integration Tests
runs-on: ubuntu-22.04
steps:
- name: Check out repository
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Cognee Setup
uses: ./.github/actions/cognee_setup
with:
python-version: ${{ inputs.python-version }}
- name: Run CLI Integration Tests
run: uv run pytest cognee/tests/integration/cli/ -v
cli-functionality-tests:
name: CLI Functionality Tests
runs-on: ubuntu-22.04
steps:
- name: Check out repository
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Cognee Setup
uses: ./.github/actions/cognee_setup
with:
python-version: ${{ inputs.python-version }}
- name: Test CLI Help Commands
run: |
uv run python -m cognee.cli._cognee --help
uv run python -m cognee.cli._cognee --version
uv run python -m cognee.cli._cognee add --help
uv run python -m cognee.cli._cognee search --help
uv run python -m cognee.cli._cognee cognify --help
uv run python -m cognee.cli._cognee delete --help
uv run python -m cognee.cli._cognee config --help
- name: Test CLI Config Subcommands
run: |
uv run python -m cognee.cli._cognee config get --help
uv run python -m cognee.cli._cognee config set --help
uv run python -m cognee.cli._cognee config list --help
uv run python -m cognee.cli._cognee config unset --help
uv run python -m cognee.cli._cognee config reset --help
- name: Test CLI Error Handling
run: |
# Test invalid command (should fail gracefully)
! uv run python -m cognee.cli._cognee invalid_command
# Test missing required arguments (should fail gracefully)
! uv run python -m cognee.cli._cognee search
# Test invalid search type (should fail gracefully)
! uv run python -m cognee.cli._cognee search "test" --query-type INVALID_TYPE
# Test invalid chunker (should fail gracefully)
! uv run python -m cognee.cli._cognee cognify --chunker InvalidChunker

34
.github/workflows/test_suites.yml vendored
View file

@ -27,45 +27,50 @@ jobs:
uses: ./.github/workflows/e2e_tests.yml
secrets: inherit
cli-tests:
name: CLI Tests
uses: ./.github/workflows/cli_tests.yml
secrets: inherit
docker-compose-test:
name: Docker Compose Test
needs: [basic-tests, e2e-tests]
needs: [basic-tests, e2e-tests, cli-tests]
uses: ./.github/workflows/docker_compose.yml
secrets: inherit
docker-ci-test:
name: Docker CI test
needs: [basic-tests, e2e-tests]
needs: [basic-tests, e2e-tests, cli-tests]
uses: ./.github/workflows/backend_docker_build_test.yml
secrets: inherit
graph-db-tests:
name: Graph Database Tests
needs: [basic-tests, e2e-tests]
needs: [basic-tests, e2e-tests, cli-tests]
uses: ./.github/workflows/graph_db_tests.yml
secrets: inherit
search-db-tests:
name: Search Test on Different DBs
needs: [basic-tests, e2e-tests, graph-db-tests]
needs: [basic-tests, e2e-tests, cli-tests, graph-db-tests]
uses: ./.github/workflows/search_db_tests.yml
secrets: inherit
relational-db-migration-tests:
name: Relational DB Migration Tests
needs: [ basic-tests, e2e-tests, graph-db-tests]
needs: [basic-tests, e2e-tests, cli-tests, graph-db-tests]
uses: ./.github/workflows/relational_db_migration_tests.yml
secrets: inherit
notebook-tests:
name: Notebook Tests
needs: [basic-tests, e2e-tests]
needs: [basic-tests, e2e-tests, cli-tests]
uses: ./.github/workflows/notebooks_tests.yml
secrets: inherit
python-version-tests:
name: Python Version Tests
needs: [basic-tests, e2e-tests]
needs: [basic-tests, e2e-tests, cli-tests]
uses: ./.github/workflows/python_version_tests.yml
with:
python-versions: '["3.10.x", "3.11.x", "3.12.x"]'
@ -74,20 +79,20 @@ jobs:
# Matrix-based vector database tests
vector-db-tests:
name: Vector DB Tests
needs: [basic-tests, e2e-tests]
needs: [basic-tests, e2e-tests, cli-tests]
uses: ./.github/workflows/vector_db_tests.yml
secrets: inherit
# Matrix-based example tests
example-tests:
name: Example Tests
needs: [basic-tests, e2e-tests]
needs: [basic-tests, e2e-tests, cli-tests]
uses: ./.github/workflows/examples_tests.yml
secrets: inherit
mcp-test:
name: Example Tests
needs: [ basic-tests, e2e-tests ]
name: MCP Tests
needs: [basic-tests, e2e-tests, cli-tests]
uses: ./.github/workflows/test_mcp.yml
secrets: inherit
@ -99,14 +104,14 @@ jobs:
s3-file-storage-test:
name: S3 File Storage Test
needs: [basic-tests, e2e-tests]
needs: [basic-tests, e2e-tests, cli-tests]
uses: ./.github/workflows/test_s3_file_storage.yml
secrets: inherit
# Additional LLM tests
gemini-tests:
name: Gemini Tests
needs: [basic-tests, e2e-tests]
needs: [basic-tests, e2e-tests, cli-tests]
uses: ./.github/workflows/test_gemini.yml
secrets: inherit
@ -116,6 +121,7 @@ jobs:
needs: [
basic-tests,
e2e-tests,
cli-tests,
graph-db-tests,
notebook-tests,
python-version-tests,
@ -135,6 +141,7 @@ jobs:
needs: [
basic-tests,
e2e-tests,
cli-tests,
graph-db-tests,
notebook-tests,
python-version-tests,
@ -155,6 +162,7 @@ jobs:
run: |
if [[ "${{ needs.basic-tests.result }}" == "success" &&
"${{ needs.e2e-tests.result }}" == "success" &&
"${{ needs.cli-tests.result }}" == "success" &&
"${{ needs.graph-db-tests.result }}" == "success" &&
"${{ needs.notebook-tests.result }}" == "success" &&
"${{ needs.python-version-tests.result }}" == "success" &&

3
cognee/tests/integration/cli/__init__.py Normal file
View file

@ -0,0 +1,3 @@
"""
CLI integration tests package.
"""

327
cognee/tests/integration/cli/test_cli_integration.py Normal file
View file

@ -0,0 +1,327 @@
"""
Integration tests for CLI commands that test end-to-end functionality.
"""
import pytest
import tempfile
import os
import sys
import subprocess
from pathlib import Path
from unittest.mock import patch, MagicMock
class TestCliIntegration:
"""Integration tests for CLI commands"""
def test_cli_help(self):
"""Test that CLI help works"""
result = subprocess.run(
[sys.executable, "-m", "cognee.cli._cognee", "--help"],
capture_output=True,
text=True,
cwd=Path(__file__).parent.parent.parent, # Go to project root
)
assert result.returncode == 0
assert "cognee" in result.stdout.lower()
assert "available commands" in result.stdout.lower()
def test_cli_version(self):
"""Test that CLI version works"""
result = subprocess.run(
[sys.executable, "-m", "cognee.cli._cognee", "--version"],
capture_output=True,
text=True,
cwd=Path(__file__).parent.parent.parent, # Go to project root
)
assert result.returncode == 0
assert "cognee" in result.stdout.lower()
def test_command_help(self):
"""Test that individual command help works"""
commands = ["add", "search", "cognify", "delete", "config"]
for command in commands:
result = subprocess.run(
[sys.executable, "-m", "cognee.cli._cognee", command, "--help"],
capture_output=True,
text=True,
cwd=Path(__file__).parent.parent.parent, # Go to project root
)
assert result.returncode == 0, f"Command {command} help failed"
assert command in result.stdout.lower()
def test_invalid_command(self):
"""Test that invalid commands are handled properly"""
result = subprocess.run(
[sys.executable, "-m", "cognee.cli._cognee", "invalid_command"],
capture_output=True,
text=True,
cwd=Path(__file__).parent.parent.parent, # Go to project root
)
assert result.returncode != 0
@patch("cognee.add")
def test_add_command_integration(self, mock_add):
"""Test add command integration"""
mock_add.return_value = None
with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f:
f.write("Test content for CLI integration")
temp_file = f.name
try:
result = subprocess.run(
[sys.executable, "-m", "cognee.cli._cognee", "add", temp_file],
capture_output=True,
text=True,
cwd=Path(__file__).parent.parent.parent, # Go to project root
)
# Note: This might fail due to dependencies, but we're testing the CLI structure
# The important thing is that it doesn't crash with argument parsing errors
assert (
"error" not in result.stderr.lower()
or "failed to add data" in result.stderr.lower()
)
finally:
os.unlink(temp_file)
def test_config_subcommands(self):
"""Test config subcommands help"""
subcommands = ["get", "set", "list", "unset", "reset"]
for subcommand in subcommands:
result = subprocess.run(
[sys.executable, "-m", "cognee.cli._cognee", "config", subcommand, "--help"],
capture_output=True,
text=True,
cwd=Path(__file__).parent.parent.parent, # Go to project root
)
assert result.returncode == 0, f"Config {subcommand} help failed"
def test_search_command_missing_query(self):
"""Test search command fails when query is missing"""
result = subprocess.run(
[sys.executable, "-m", "cognee.cli._cognee", "search"],
capture_output=True,
text=True,
cwd=Path(__file__).parent.parent.parent, # Go to project root
)
assert result.returncode != 0
assert "required" in result.stderr.lower() or "error" in result.stderr.lower()
def test_delete_command_no_target(self):
"""Test delete command with no target specified"""
result = subprocess.run(
[sys.executable, "-m", "cognee.cli._cognee", "delete"],
capture_output=True,
text=True,
cwd=Path(__file__).parent.parent.parent, # Go to project root
)
# Should run but show error message about missing target
# Return code might be 0 since the command handles this gracefully
assert (
"specify what to delete" in result.stdout.lower()
or "specify what to delete" in result.stderr.lower()
)
class TestCliArgumentParsing:
"""Test CLI argument parsing edge cases"""
def test_add_multiple_files(self):
"""Test add command with multiple file arguments"""
with tempfile.TemporaryDirectory() as temp_dir:
file1 = os.path.join(temp_dir, "file1.txt")
file2 = os.path.join(temp_dir, "file2.txt")
with open(file1, "w") as f:
f.write("Content 1")
with open(file2, "w") as f:
f.write("Content 2")
result = subprocess.run(
[
sys.executable,
"-m",
"cognee.cli._cognee",
"add",
file1,
file2,
"--dataset-name",
"test",
],
capture_output=True,
text=True,
cwd=Path(__file__).parent.parent.parent, # Go to project root
)
# Test that argument parsing works (regardless of actual execution)
assert (
"argument" not in result.stderr.lower() or "failed to add" in result.stderr.lower()
)
def test_search_with_all_options(self):
"""Test search command with all possible options"""
result = subprocess.run(
[
sys.executable,
"-m",
"cognee.cli._cognee",
"search",
"test query",
"--query-type",
"CHUNKS",
"--datasets",
"dataset1",
"dataset2",
"--top-k",
"5",
"--output-format",
"json",
],
capture_output=True,
text=True,
cwd=Path(__file__).parent.parent.parent, # Go to project root
)
# Should not have argument parsing errors
assert "unrecognized arguments" not in result.stderr.lower()
assert "invalid choice" not in result.stderr.lower()
def test_cognify_with_all_options(self):
"""Test cognify command with all possible options"""
result = subprocess.run(
[
sys.executable,
"-m",
"cognee.cli._cognee",
"cognify",
"--datasets",
"dataset1",
"dataset2",
"--chunk-size",
"1024",
"--chunker",
"TextChunker",
"--background",
"--verbose",
],
capture_output=True,
text=True,
cwd=Path(__file__).parent.parent.parent, # Go to project root
)
# Should not have argument parsing errors
assert "unrecognized arguments" not in result.stderr.lower()
assert "invalid choice" not in result.stderr.lower()
def test_config_set_command(self):
"""Test config set command argument parsing"""
result = subprocess.run(
[sys.executable, "-m", "cognee.cli._cognee", "config", "set", "test_key", "test_value"],
capture_output=True,
text=True,
cwd=Path(__file__).parent.parent.parent, # Go to project root
)
# Should not have argument parsing errors
assert "unrecognized arguments" not in result.stderr.lower()
assert "required" not in result.stderr.lower() or "failed to set" in result.stderr.lower()
def test_delete_with_force(self):
"""Test delete command with force flag"""
result = subprocess.run(
[
sys.executable,
"-m",
"cognee.cli._cognee",
"delete",
"--dataset-name",
"test_dataset",
"--force",
],
capture_output=True,
text=True,
cwd=Path(__file__).parent.parent.parent, # Go to project root
)
# Should not have argument parsing errors
assert "unrecognized arguments" not in result.stderr.lower()
class TestCliErrorHandling:
"""Test CLI error handling and edge cases"""
def test_debug_mode_flag(self):
"""Test that debug flag is accepted"""
result = subprocess.run(
[sys.executable, "-m", "cognee.cli._cognee", "--debug", "search", "test query"],
capture_output=True,
text=True,
cwd=Path(__file__).parent.parent.parent, # Go to project root
)
# Should not have argument parsing errors for debug flag
assert "unrecognized arguments" not in result.stderr.lower()
def test_invalid_search_type(self):
"""Test invalid search type handling"""
result = subprocess.run(
[
sys.executable,
"-m",
"cognee.cli._cognee",
"search",
"test query",
"--query-type",
"INVALID_TYPE",
],
capture_output=True,
text=True,
cwd=Path(__file__).parent.parent.parent, # Go to project root
)
assert result.returncode != 0
assert "invalid choice" in result.stderr.lower()
def test_invalid_chunker(self):
"""Test invalid chunker handling"""
result = subprocess.run(
[sys.executable, "-m", "cognee.cli._cognee", "cognify", "--chunker", "InvalidChunker"],
capture_output=True,
text=True,
cwd=Path(__file__).parent.parent.parent, # Go to project root
)
assert result.returncode != 0
assert "invalid choice" in result.stderr.lower()
def test_invalid_output_format(self):
"""Test invalid output format handling"""
result = subprocess.run(
[
sys.executable,
"-m",
"cognee.cli._cognee",
"search",
"test query",
"--output-format",
"invalid",
],
capture_output=True,
text=True,
cwd=Path(__file__).parent.parent.parent, # Go to project root
)
assert result.returncode != 0
assert "invalid choice" in result.stderr.lower()

484
cognee/tests/test_cli_edge_cases.py Normal file
View file

@ -0,0 +1,484 @@
"""
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]

173
cognee/tests/test_cli_main.py Normal file
View file

@ -0,0 +1,173 @@
"""
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.")

53
cognee/tests/test_cli_runner.py Normal file
View file

@ -0,0 +1,53 @@
"""
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)

3
cognee/tests/unit/cli/__init__.py Normal file
View file

@ -0,0 +1,3 @@
"""
CLI unit tests package.
"""

474
cognee/tests/unit/cli/test_cli_commands.py Normal file
View file

@ -0,0 +1,474 @@
"""
Tests for individual CLI commands.
"""
import pytest
import argparse
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 TestAddCommand:
"""Test the AddCommand class"""
def test_command_properties(self):
"""Test basic command properties"""
command = AddCommand()
assert command.command_string == "add"
assert "Add data" in command.help_string
assert command.docs_url is not None
def test_configure_parser(self):
"""Test parser configuration"""
command = AddCommand()
parser = argparse.ArgumentParser()
command.configure_parser(parser)
# Check that required arguments are added
actions = {action.dest: action for action in parser._actions}
assert "data" in actions
assert "dataset_name" in actions
# Check data argument accepts multiple values
assert actions["data"].nargs == "+"
@patch("cognee.cli.commands.add_command.asyncio.run")
@patch("cognee.cli.commands.add_command.cognee")
def test_execute_single_item(self, mock_cognee, mock_asyncio_run):
"""Test execute with single data item"""
command = AddCommand()
args = argparse.Namespace(data=["test.txt"], dataset_name="test_dataset")
mock_cognee.add = AsyncMock()
command.execute(args)
mock_asyncio_run.assert_called_once()
# Check that the async function would be called correctly
assert mock_asyncio_run.call_args[0][0] # async function was passed
@patch("cognee.cli.commands.add_command.asyncio.run")
@patch("cognee.cli.commands.add_command.cognee")
def test_execute_multiple_items(self, mock_cognee, mock_asyncio_run):
"""Test execute with multiple data items"""
command = AddCommand()
args = argparse.Namespace(data=["test1.txt", "test2.txt"], 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_execute_with_exception(self, mock_cognee, mock_asyncio_run):
"""Test execute handles exceptions properly"""
command = AddCommand()
args = argparse.Namespace(data=["test.txt"], dataset_name="test_dataset")
mock_asyncio_run.side_effect = Exception("Test error")
with pytest.raises(CliCommandException):
command.execute(args)
class TestSearchCommand:
"""Test the SearchCommand class"""
def test_command_properties(self):
"""Test basic command properties"""
command = SearchCommand()
assert command.command_string == "search"
assert "Search and query" in command.help_string
assert command.docs_url is not None
def test_configure_parser(self):
"""Test parser configuration"""
command = SearchCommand()
parser = argparse.ArgumentParser()
command.configure_parser(parser)
# Check that required arguments are added
actions = {action.dest: action for action in parser._actions}
assert "query_text" in actions
assert "query_type" in actions
assert "datasets" in actions
assert "top_k" in actions
assert "output_format" in actions
# Check default values
assert actions["query_type"].default == "GRAPH_COMPLETION"
assert actions["top_k"].default == 10
assert actions["output_format"].default == "pretty"
@patch("cognee.cli.commands.search_command.asyncio.run")
@patch("cognee.cli.commands.search_command.cognee")
def test_execute_basic_search(self, mock_cognee, mock_asyncio_run):
"""Test execute with basic search"""
command = SearchCommand()
args = argparse.Namespace(
query_text="test query",
query_type="GRAPH_COMPLETION",
datasets=None,
top_k=10,
system_prompt=None,
output_format="pretty",
)
mock_cognee.search = AsyncMock(return_value=["result1", "result2"])
mock_asyncio_run.return_value = ["result1", "result2"]
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_execute_json_output(self, mock_cognee, mock_asyncio_run):
"""Test execute with JSON output format"""
command = SearchCommand()
args = argparse.Namespace(
query_text="test query",
query_type="CHUNKS",
datasets=["dataset1"],
top_k=5,
system_prompt=None,
output_format="json",
)
mock_cognee.search = AsyncMock(return_value=[{"chunk": "test"}])
mock_asyncio_run.return_value = [{"chunk": "test"}]
command.execute(args)
mock_asyncio_run.assert_called_once()
@patch("cognee.cli.commands.search_command.asyncio.run")
def test_execute_with_exception(self, mock_asyncio_run):
"""Test execute handles exceptions properly"""
command = SearchCommand()
args = argparse.Namespace(
query_text="test query",
query_type="GRAPH_COMPLETION",
datasets=None,
top_k=10,
system_prompt=None,
output_format="pretty",
)
mock_asyncio_run.side_effect = Exception("Search error")
with pytest.raises(CliCommandException):
command.execute(args)
class TestCognifyCommand:
"""Test the CognifyCommand class"""
def test_command_properties(self):
"""Test basic command properties"""
command = CognifyCommand()
assert command.command_string == "cognify"
assert "Transform ingested data" in command.help_string
assert command.docs_url is not None
def test_configure_parser(self):
"""Test parser configuration"""
command = CognifyCommand()
parser = argparse.ArgumentParser()
command.configure_parser(parser)
# Check that arguments are added
actions = {action.dest: action for action in parser._actions}
assert "datasets" in actions
assert "chunk_size" in actions
assert "ontology_file" in actions
assert "chunker" in actions
assert "background" in actions
assert "verbose" in actions
# Check default values
assert actions["chunker"].default == "TextChunker"
@patch("cognee.cli.commands.cognify_command.asyncio.run")
@patch("cognee.cli.commands.cognify_command.cognee")
def test_execute_basic_cognify(self, mock_cognee, mock_asyncio_run):
"""Test execute with basic cognify"""
command = CognifyCommand()
args = argparse.Namespace(
datasets=None,
chunk_size=None,
ontology_file=None,
chunker="TextChunker",
background=False,
verbose=False,
)
mock_cognee.cognify = AsyncMock(return_value="success")
mock_asyncio_run.return_value = "success"
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_execute_background_mode(self, mock_cognee, mock_asyncio_run):
"""Test execute with background mode"""
command = CognifyCommand()
args = argparse.Namespace(
datasets=["dataset1"],
chunk_size=1024,
ontology_file="/path/to/ontology.owl",
chunker="LangchainChunker",
background=True,
verbose=True,
)
mock_cognee.cognify = AsyncMock(return_value="background_started")
mock_asyncio_run.return_value = "background_started"
command.execute(args)
mock_asyncio_run.assert_called_once()
@patch("cognee.cli.commands.cognify_command.asyncio.run")
def test_execute_with_exception(self, mock_asyncio_run):
"""Test execute handles exceptions properly"""
command = CognifyCommand()
args = argparse.Namespace(
datasets=None,
chunk_size=None,
ontology_file=None,
chunker="TextChunker",
background=False,
verbose=False,
)
mock_asyncio_run.side_effect = Exception("Cognify error")
with pytest.raises(CliCommandException):
command.execute(args)
class TestDeleteCommand:
"""Test the DeleteCommand class"""
def test_command_properties(self):
"""Test basic command properties"""
command = DeleteCommand()
assert command.command_string == "delete"
assert "Delete data" in command.help_string
assert command.docs_url is not None
def test_configure_parser(self):
"""Test parser configuration"""
command = DeleteCommand()
parser = argparse.ArgumentParser()
command.configure_parser(parser)
# Check that arguments are added
actions = {action.dest: action for action in parser._actions}
assert "dataset_name" in actions
assert "user_id" in actions
assert "all" in actions
assert "force" in actions
@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_execute_delete_dataset_with_confirmation(
self, mock_cognee, mock_asyncio_run, mock_confirm
):
"""Test execute delete dataset with user confirmation"""
command = DeleteCommand()
args = argparse.Namespace(dataset_name="test_dataset", user_id=None, all=False, force=False)
mock_confirm.return_value = True
mock_cognee.delete = AsyncMock()
command.execute(args)
mock_confirm.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):
"""Test execute when user cancels deletion"""
command = DeleteCommand()
args = argparse.Namespace(dataset_name="test_dataset", user_id=None, all=False, force=False)
mock_confirm.return_value = False
# Should not raise exception, just return
command.execute(args)
mock_confirm.assert_called_once()
@patch("cognee.cli.commands.delete_command.asyncio.run")
@patch("cognee.cli.commands.delete_command.cognee")
def test_execute_delete_forced(self, mock_cognee, mock_asyncio_run):
"""Test execute delete with force flag"""
command = DeleteCommand()
args = argparse.Namespace(dataset_name="test_dataset", user_id=None, all=False, force=True)
mock_cognee.delete = AsyncMock()
command.execute(args)
mock_asyncio_run.assert_called_once()
def test_execute_no_delete_target(self):
"""Test execute when no delete target is specified"""
command = DeleteCommand()
args = argparse.Namespace(dataset_name=None, user_id=None, all=False, force=False)
# Should not raise exception, just return with error message
command.execute(args)
@patch("cognee.cli.commands.delete_command.asyncio.run")
def test_execute_with_exception(self, mock_asyncio_run):
"""Test execute handles exceptions properly"""
command = DeleteCommand()
args = argparse.Namespace(dataset_name="test_dataset", user_id=None, all=False, force=True)
mock_asyncio_run.side_effect = Exception("Delete error")
with pytest.raises(CliCommandException):
command.execute(args)
class TestConfigCommand:
"""Test the ConfigCommand class"""
def test_command_properties(self):
"""Test basic command properties"""
command = ConfigCommand()
assert command.command_string == "config"
assert "Manage cognee configuration" in command.help_string
assert command.docs_url is not None
def test_configure_parser(self):
"""Test parser configuration"""
command = ConfigCommand()
parser = argparse.ArgumentParser()
command.configure_parser(parser)
# Check that subparsers are created
subparsers_actions = [
action for action in parser._actions if isinstance(action, argparse._SubParsersAction)
]
assert len(subparsers_actions) == 1
subparsers = subparsers_actions[0]
assert "get" in subparsers.choices
assert "set" in subparsers.choices
assert "list" in subparsers.choices
assert "unset" in subparsers.choices
assert "reset" in subparsers.choices
def test_execute_no_action(self):
"""Test execute when no config action is provided"""
command = ConfigCommand()
args = argparse.Namespace()
# Should not raise exception, just return with error message
command.execute(args)
@patch("cognee.cli.commands.config_command.cognee")
def test_execute_get_action(self, mock_cognee):
"""Test execute get action"""
command = ConfigCommand()
args = argparse.Namespace(config_action="get", key="llm_provider")
mock_cognee.config.get = MagicMock(return_value="openai")
command.execute(args)
# Should call get method if available
if hasattr(mock_cognee.config, "get"):
mock_cognee.config.get.assert_called_with("llm_provider")
@patch("cognee.cli.commands.config_command.cognee")
def test_execute_set_action(self, mock_cognee):
"""Test execute set action"""
command = ConfigCommand()
args = argparse.Namespace(config_action="set", key="llm_provider", value="anthropic")
mock_cognee.config.set = MagicMock()
command.execute(args)
mock_cognee.config.set.assert_called_with("llm_provider", "anthropic")
@patch("cognee.cli.commands.config_command.cognee")
def test_execute_set_action_json_value(self, mock_cognee):
"""Test execute set action with JSON value"""
command = ConfigCommand()
args = argparse.Namespace(config_action="set", key="chunk_size", value="1024")
mock_cognee.config.set = MagicMock()
command.execute(args)
# Value should be parsed as string since it's not valid JSON
mock_cognee.config.set.assert_called_with("chunk_size", "1024")
def test_execute_list_action(self):
"""Test execute list action"""
command = ConfigCommand()
args = argparse.Namespace(config_action="list")
# Should not raise exception
command.execute(args)
@patch("cognee.cli.commands.config_command.fmt.confirm")
@patch("cognee.cli.commands.config_command.cognee")
def test_execute_unset_action(self, mock_cognee, mock_confirm):
"""Test execute unset action"""
command = ConfigCommand()
args = argparse.Namespace(config_action="unset", key="llm_provider", force=False)
mock_confirm.return_value = True
mock_cognee.config.set_llm_provider = MagicMock()
command.execute(args)
mock_confirm.assert_called_once()
@patch("cognee.cli.commands.config_command.fmt.confirm")
def test_execute_reset_action(self, mock_confirm):
"""Test execute reset action"""
command = ConfigCommand()
args = argparse.Namespace(config_action="reset", force=False)
mock_confirm.return_value = True
# Should not raise exception
command.execute(args)
mock_confirm.assert_called_once()
def test_execute_with_exception(self):
"""Test execute handles exceptions properly"""
command = ConfigCommand()
# Create args that will cause an exception in _handle_set
args = argparse.Namespace(config_action="set", key="invalid_key", value="value")
with patch("cognee.cli.commands.config_command.cognee") as mock_cognee:
mock_cognee.config.set.side_effect = Exception("Config error")
with pytest.raises(CliCommandException):
command.execute(args)

484
cognee/tests/unit/cli/test_cli_edge_cases.py Normal file
View file

@ -0,0 +1,484 @@
"""
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]

173
cognee/tests/unit/cli/test_cli_main.py Normal file
View file

@ -0,0 +1,173 @@
"""
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.")

63
cognee/tests/unit/cli/test_cli_runner.py Normal file
View file

@ -0,0 +1,63 @@
"""
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
integration_dir = test_dir.parent.parent / "integration" / "cli"
cli_unit_test_files = [
"test_cli_main.py",
"test_cli_commands.py",
"test_cli_utils.py",
"test_cli_edge_cases.py",
]
cli_integration_test_files = ["test_cli_integration.py"]
# Run tests with pytest
args = ["-v", "--tb=short"]
# Add unit tests
for test_file in cli_unit_test_files:
test_path = test_dir / test_file
if test_path.exists():
args.append(str(test_path))
# Add integration tests
for test_file in cli_integration_test_files:
test_path = integration_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)

299
cognee/tests/unit/cli/test_cli_utils.py Normal file
View file

@ -0,0 +1,299 @@
"""
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,
DEFAULT_DOCS_URL,
COMMAND_DESCRIPTIONS,
SEARCH_TYPE_CHOICES,
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
class TestCliConfig:
"""Test CLI configuration constants"""
def test_cli_description_exists(self):
"""Test that CLI description is defined"""
assert CLI_DESCRIPTION
assert isinstance(CLI_DESCRIPTION, str)
assert "cognee" in CLI_DESCRIPTION.lower()
def test_default_docs_url_exists(self):
"""Test that default docs URL is defined"""
assert DEFAULT_DOCS_URL
assert isinstance(DEFAULT_DOCS_URL, str)
assert DEFAULT_DOCS_URL.startswith("https://")
def test_command_descriptions_complete(self):
"""Test that all expected commands have descriptions"""
expected_commands = ["add", "search", "cognify", "delete", "config"]
for command in expected_commands:
assert command in COMMAND_DESCRIPTIONS
assert isinstance(COMMAND_DESCRIPTIONS[command], str)
assert len(COMMAND_DESCRIPTIONS[command]) > 0
def test_search_type_choices_valid(self):
"""Test that search type choices are valid"""
assert isinstance(SEARCH_TYPE_CHOICES, list)
assert len(SEARCH_TYPE_CHOICES) > 0
expected_types = [
"GRAPH_COMPLETION",
"RAG_COMPLETION",
"INSIGHTS",
"CHUNKS",
"SUMMARIES",
"CODE",
"CYPHER",
]
for expected_type in expected_types:
assert expected_type in SEARCH_TYPE_CHOICES
def test_chunker_choices_valid(self):
"""Test that chunker choices are valid"""
assert isinstance(CHUNKER_CHOICES, list)
assert len(CHUNKER_CHOICES) > 0
assert "TextChunker" in CHUNKER_CHOICES
assert "LangchainChunker" in CHUNKER_CHOICES
def test_output_format_choices_valid(self):
"""Test that output format choices are valid"""
assert isinstance(OUTPUT_FORMAT_CHOICES, list)
assert len(OUTPUT_FORMAT_CHOICES) > 0
expected_formats = ["json", "pretty", "simple"]
for expected_format in expected_formats:
assert expected_format in OUTPUT_FORMAT_CHOICES
class TestCliReference:
"""Test CLI reference protocol"""
def test_supports_cli_command_protocol(self):
"""Test that SupportsCliCommand protocol is properly defined"""
from cognee.cli.reference import SupportsCliCommand
# Test that it's a protocol
assert hasattr(SupportsCliCommand, "__annotations__")
# Test required attributes
annotations = SupportsCliCommand.__annotations__
assert "command_string" in annotations
assert "help_string" in annotations
assert "description" in annotations
assert "docs_url" in annotations
def test_protocol_methods(self):
"""Test that protocol defines required methods"""
from cognee.cli.reference import SupportsCliCommand
import inspect
# Get abstract methods
abstract_methods = []
for name, method in inspect.getmembers(SupportsCliCommand, predicate=inspect.ismethod):
if getattr(method, "__isabstractmethod__", False):
abstract_methods.append(name)
# Should have abstract methods for configure_parser and execute
method_names = [name for name, _ in inspect.getmembers(SupportsCliCommand)]
assert "configure_parser" in method_names
assert "execute" in method_names
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()
# 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