diff --git a/.github/workflows/cli_tests.yml b/.github/workflows/cli_tests.yml new file mode 100644 index 000000000..258e7ad76 --- /dev/null +++ b/.github/workflows/cli_tests.yml @@ -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 diff --git a/.github/workflows/test_suites.yml b/.github/workflows/test_suites.yml index d34523ce1..ec56012ac 100644 --- a/.github/workflows/test_suites.yml +++ b/.github/workflows/test_suites.yml @@ -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" && diff --git a/cognee/tests/integration/cli/__init__.py b/cognee/tests/integration/cli/__init__.py new file mode 100644 index 000000000..8d086bbca --- /dev/null +++ b/cognee/tests/integration/cli/__init__.py @@ -0,0 +1,3 @@ +""" +CLI integration tests package. +""" diff --git a/cognee/tests/integration/cli/test_cli_integration.py b/cognee/tests/integration/cli/test_cli_integration.py new file mode 100644 index 000000000..8817f283a --- /dev/null +++ b/cognee/tests/integration/cli/test_cli_integration.py @@ -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() diff --git a/cognee/tests/test_cli_edge_cases.py b/cognee/tests/test_cli_edge_cases.py new file mode 100644 index 000000000..1140b3544 --- /dev/null +++ b/cognee/tests/test_cli_edge_cases.py @@ -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] diff --git a/cognee/tests/test_cli_main.py b/cognee/tests/test_cli_main.py new file mode 100644 index 000000000..44f23ea5c --- /dev/null +++ b/cognee/tests/test_cli_main.py @@ -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.") diff --git a/cognee/tests/test_cli_runner.py b/cognee/tests/test_cli_runner.py new file mode 100644 index 000000000..ef20724e4 --- /dev/null +++ b/cognee/tests/test_cli_runner.py @@ -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) diff --git a/cognee/tests/unit/cli/__init__.py b/cognee/tests/unit/cli/__init__.py new file mode 100644 index 000000000..288b8e055 --- /dev/null +++ b/cognee/tests/unit/cli/__init__.py @@ -0,0 +1,3 @@ +""" +CLI unit tests package. +""" diff --git a/cognee/tests/unit/cli/test_cli_commands.py b/cognee/tests/unit/cli/test_cli_commands.py new file mode 100644 index 000000000..6375a5141 --- /dev/null +++ b/cognee/tests/unit/cli/test_cli_commands.py @@ -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) diff --git a/cognee/tests/unit/cli/test_cli_edge_cases.py b/cognee/tests/unit/cli/test_cli_edge_cases.py new file mode 100644 index 000000000..1140b3544 --- /dev/null +++ b/cognee/tests/unit/cli/test_cli_edge_cases.py @@ -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] diff --git a/cognee/tests/unit/cli/test_cli_main.py b/cognee/tests/unit/cli/test_cli_main.py new file mode 100644 index 000000000..44f23ea5c --- /dev/null +++ b/cognee/tests/unit/cli/test_cli_main.py @@ -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.") diff --git a/cognee/tests/unit/cli/test_cli_runner.py b/cognee/tests/unit/cli/test_cli_runner.py new file mode 100644 index 000000000..bc0357abb --- /dev/null +++ b/cognee/tests/unit/cli/test_cli_runner.py @@ -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) diff --git a/cognee/tests/unit/cli/test_cli_utils.py b/cognee/tests/unit/cli/test_cli_utils.py new file mode 100644 index 000000000..318c04089 --- /dev/null +++ b/cognee/tests/unit/cli/test_cli_utils.py @@ -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