feat: Add cli (#1197)
<!-- .github/pull_request_template.md --> ## Description <!-- Provide a clear description of the changes in this PR --> ## DCO Affirmation I affirm that all code in every commit of this pull request conforms to the terms of the Topoteretes Developer Certificate of Origin.
This commit is contained in:
Vasilije
GitHub
commit
d6e6e874eb
32 changed files with 6969 additions and 3813 deletions
10
.github/workflows/basic_tests.yml
vendored
10
.github/workflows/basic_tests.yml
vendored
|
|
@ -103,6 +103,16 @@ jobs:
|
|||
integration-tests:
|
||||
name: Run Integration Tests
|
||||
runs-on: ubuntu-22.04
|
||||
env:
|
||||
LLM_PROVIDER: openai
|
||||
LLM_MODEL: ${{ secrets.LLM_MODEL }}
|
||||
LLM_ENDPOINT: ${{ secrets.LLM_ENDPOINT }}
|
||||
LLM_API_KEY: ${{ secrets.LLM_API_KEY }}
|
||||
LLM_API_VERSION: ${{ secrets.LLM_API_VERSION }}
|
||||
EMBEDDING_MODEL: ${{ secrets.EMBEDDING_MODEL }}
|
||||
EMBEDDING_ENDPOINT: ${{ secrets.EMBEDDING_ENDPOINT }}
|
||||
EMBEDDING_API_KEY: ${{ secrets.EMBEDDING_API_KEY }}
|
||||
EMBEDDING_API_VERSION: ${{ secrets.EMBEDDING_API_VERSION }}
|
||||
steps:
|
||||
- name: Check out repository
|
||||
uses: actions/checkout@v4
|
||||
|
|
|
|||
146
.github/workflows/cli_tests.yml
vendored
Normal file
146
.github/workflows/cli_tests.yml
vendored
Normal file
|
|
@ -0,0 +1,146 @@
|
|||
name: CLI Tests
|
||||
|
||||
on:
|
||||
workflow_call:
|
||||
inputs:
|
||||
python-version:
|
||||
required: false
|
||||
type: string
|
||||
default: '3.11.x'
|
||||
secrets:
|
||||
LLM_PROVIDER:
|
||||
required: true
|
||||
LLM_MODEL:
|
||||
required: true
|
||||
LLM_ENDPOINT:
|
||||
required: true
|
||||
LLM_API_KEY:
|
||||
required: true
|
||||
LLM_API_VERSION:
|
||||
required: true
|
||||
EMBEDDING_PROVIDER:
|
||||
required: true
|
||||
EMBEDDING_MODEL:
|
||||
required: true
|
||||
EMBEDDING_ENDPOINT:
|
||||
required: true
|
||||
EMBEDDING_API_KEY:
|
||||
required: true
|
||||
EMBEDDING_API_VERSION:
|
||||
required: true
|
||||
|
||||
|
||||
env:
|
||||
RUNTIME__LOG_LEVEL: ERROR
|
||||
ENV: 'dev'
|
||||
|
||||
jobs:
|
||||
cli-unit-tests:
|
||||
name: CLI Unit Tests
|
||||
runs-on: ubuntu-22.04
|
||||
env:
|
||||
LLM_PROVIDER: openai
|
||||
LLM_MODEL: ${{ secrets.LLM_MODEL }}
|
||||
LLM_ENDPOINT: ${{ secrets.LLM_ENDPOINT }}
|
||||
LLM_API_KEY: ${{ secrets.LLM_API_KEY }}
|
||||
LLM_API_VERSION: ${{ secrets.LLM_API_VERSION }}
|
||||
EMBEDDING_MODEL: ${{ secrets.EMBEDDING_MODEL }}
|
||||
EMBEDDING_ENDPOINT: ${{ secrets.EMBEDDING_ENDPOINT }}
|
||||
EMBEDDING_API_KEY: ${{ secrets.EMBEDDING_API_KEY }}
|
||||
EMBEDDING_API_VERSION: ${{ secrets.EMBEDDING_API_VERSION }}
|
||||
steps:
|
||||
- name: Check out repository
|
||||
uses: actions/checkout@v4
|
||||
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
|
||||
env:
|
||||
LLM_PROVIDER: openai
|
||||
LLM_MODEL: ${{ secrets.LLM_MODEL }}
|
||||
LLM_ENDPOINT: ${{ secrets.LLM_ENDPOINT }}
|
||||
LLM_API_KEY: ${{ secrets.LLM_API_KEY }}
|
||||
LLM_API_VERSION: ${{ secrets.LLM_API_VERSION }}
|
||||
EMBEDDING_MODEL: ${{ secrets.EMBEDDING_MODEL }}
|
||||
EMBEDDING_ENDPOINT: ${{ secrets.EMBEDDING_ENDPOINT }}
|
||||
EMBEDDING_API_KEY: ${{ secrets.EMBEDDING_API_KEY }}
|
||||
EMBEDDING_API_VERSION: ${{ secrets.EMBEDDING_API_VERSION }}
|
||||
steps:
|
||||
- name: Check out repository
|
||||
uses: actions/checkout@v4
|
||||
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
|
||||
env:
|
||||
LLM_PROVIDER: openai
|
||||
LLM_MODEL: ${{ secrets.LLM_MODEL }}
|
||||
LLM_ENDPOINT: ${{ secrets.LLM_ENDPOINT }}
|
||||
LLM_API_KEY: ${{ secrets.LLM_API_KEY }}
|
||||
LLM_API_VERSION: ${{ secrets.LLM_API_VERSION }}
|
||||
EMBEDDING_MODEL: ${{ secrets.EMBEDDING_MODEL }}
|
||||
EMBEDDING_ENDPOINT: ${{ secrets.EMBEDDING_ENDPOINT }}
|
||||
EMBEDDING_API_KEY: ${{ secrets.EMBEDDING_API_KEY }}
|
||||
EMBEDDING_API_VERSION: ${{ secrets.EMBEDDING_API_VERSION }}
|
||||
steps:
|
||||
- name: Check out repository
|
||||
uses: actions/checkout@v4
|
||||
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
34
.github/workflows/test_suites.yml
vendored
|
|
@ -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" &&
|
||||
|
|
|
|||
4
cognee/__main__.py
Normal file
4
cognee/__main__.py
Normal file
|
|
@ -0,0 +1,4 @@
|
|||
from cognee.cli._cognee import main
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
10
cognee/cli/__init__.py
Normal file
10
cognee/cli/__init__.py
Normal file
|
|
@ -0,0 +1,10 @@
|
|||
from cognee.cli.reference import SupportsCliCommand
|
||||
from cognee.cli.exceptions import CliCommandException
|
||||
|
||||
DEFAULT_DOCS_URL = "https://docs.cognee.ai"
|
||||
|
||||
__all__ = [
|
||||
"SupportsCliCommand",
|
||||
"CliCommandException",
|
||||
"DEFAULT_DOCS_URL",
|
||||
]
|
||||
180
cognee/cli/_cognee.py
Normal file
180
cognee/cli/_cognee.py
Normal file
|
|
@ -0,0 +1,180 @@
|
|||
import sys
|
||||
import os
|
||||
import argparse
|
||||
from typing import Any, Sequence, Dict, Type, cast, List
|
||||
import click
|
||||
|
||||
try:
|
||||
import rich_argparse
|
||||
from rich.markdown import Markdown
|
||||
|
||||
HAS_RICH = True
|
||||
except ImportError:
|
||||
HAS_RICH = False
|
||||
|
||||
from cognee.cli import SupportsCliCommand, DEFAULT_DOCS_URL
|
||||
from cognee.cli.config import CLI_DESCRIPTION
|
||||
from cognee.cli import debug
|
||||
import cognee.cli.echo as fmt
|
||||
from cognee.cli.exceptions import CliCommandException
|
||||
|
||||
|
||||
ACTION_EXECUTED = False
|
||||
|
||||
|
||||
def print_help(parser: argparse.ArgumentParser) -> None:
|
||||
if not ACTION_EXECUTED:
|
||||
parser.print_help()
|
||||
|
||||
|
||||
class DebugAction(argparse.Action):
|
||||
def __init__(
|
||||
self,
|
||||
option_strings: Sequence[str],
|
||||
dest: Any = argparse.SUPPRESS,
|
||||
default: Any = argparse.SUPPRESS,
|
||||
help: str = None,
|
||||
) -> None:
|
||||
super(DebugAction, self).__init__(
|
||||
option_strings=option_strings, dest=dest, default=default, nargs=0, help=help
|
||||
)
|
||||
|
||||
def __call__(
|
||||
self,
|
||||
parser: argparse.ArgumentParser,
|
||||
namespace: argparse.Namespace,
|
||||
values: Any,
|
||||
option_string: str = None,
|
||||
) -> None:
|
||||
# Enable debug mode for stack traces
|
||||
debug.enable_debug()
|
||||
fmt.note("Debug mode enabled. Full stack traces will be shown.")
|
||||
|
||||
|
||||
# Debug functionality is now in cognee.cli.debug module
|
||||
|
||||
|
||||
def _discover_commands() -> List[Type[SupportsCliCommand]]:
|
||||
"""Discover all available CLI commands"""
|
||||
# Import commands dynamically to avoid early cognee initialization
|
||||
commands = []
|
||||
|
||||
command_modules = [
|
||||
("cognee.cli.commands.add_command", "AddCommand"),
|
||||
("cognee.cli.commands.search_command", "SearchCommand"),
|
||||
("cognee.cli.commands.cognify_command", "CognifyCommand"),
|
||||
("cognee.cli.commands.delete_command", "DeleteCommand"),
|
||||
("cognee.cli.commands.config_command", "ConfigCommand"),
|
||||
]
|
||||
|
||||
for module_path, class_name in command_modules:
|
||||
try:
|
||||
module = __import__(module_path, fromlist=[class_name])
|
||||
command_class = getattr(module, class_name)
|
||||
commands.append(command_class)
|
||||
except (ImportError, AttributeError) as e:
|
||||
fmt.warning(f"Failed to load command {class_name}: {e}")
|
||||
|
||||
return commands
|
||||
|
||||
|
||||
def _create_parser() -> tuple[argparse.ArgumentParser, Dict[str, SupportsCliCommand]]:
|
||||
parser = argparse.ArgumentParser(
|
||||
description=f"{CLI_DESCRIPTION} Further help is available at {DEFAULT_DOCS_URL}."
|
||||
)
|
||||
|
||||
# Get version dynamically
|
||||
try:
|
||||
from cognee.version import get_cognee_version
|
||||
|
||||
version = get_cognee_version()
|
||||
except ImportError:
|
||||
version = "unknown"
|
||||
|
||||
parser.add_argument("--version", action="version", version=f"cognee {version}")
|
||||
parser.add_argument(
|
||||
"--debug",
|
||||
action=DebugAction,
|
||||
help="Enable debug mode to show full stack traces on exceptions",
|
||||
)
|
||||
|
||||
subparsers = parser.add_subparsers(title="Available commands", dest="command")
|
||||
|
||||
# Discover and install commands
|
||||
command_classes = _discover_commands()
|
||||
installed_commands: Dict[str, SupportsCliCommand] = {}
|
||||
|
||||
for command_class in command_classes:
|
||||
command = command_class()
|
||||
if command.command_string in installed_commands:
|
||||
continue
|
||||
|
||||
command_parser = subparsers.add_parser(
|
||||
command.command_string,
|
||||
help=command.help_string,
|
||||
description=command.description if hasattr(command, "description") else None,
|
||||
)
|
||||
command.configure_parser(command_parser)
|
||||
installed_commands[command.command_string] = command
|
||||
|
||||
# Add rich formatting if available
|
||||
if HAS_RICH:
|
||||
|
||||
def add_formatter_class(parser: argparse.ArgumentParser) -> None:
|
||||
parser.formatter_class = rich_argparse.RichHelpFormatter
|
||||
|
||||
if parser.description:
|
||||
parser.description = Markdown(parser.description, style="argparse.text")
|
||||
for action in parser._actions:
|
||||
if isinstance(action, argparse._SubParsersAction):
|
||||
for _subcmd, subparser in action.choices.items():
|
||||
add_formatter_class(subparser)
|
||||
|
||||
add_formatter_class(parser)
|
||||
|
||||
return parser, installed_commands
|
||||
|
||||
|
||||
def main() -> int:
|
||||
"""Main CLI entry point"""
|
||||
parser, installed_commands = _create_parser()
|
||||
args = parser.parse_args()
|
||||
|
||||
if cmd := installed_commands.get(args.command):
|
||||
try:
|
||||
cmd.execute(args)
|
||||
except Exception as ex:
|
||||
docs_url = cmd.docs_url if hasattr(cmd, "docs_url") else DEFAULT_DOCS_URL
|
||||
error_code = -1
|
||||
raiseable_exception = ex
|
||||
|
||||
# Handle CLI-specific exceptions
|
||||
if isinstance(ex, CliCommandException):
|
||||
error_code = ex.error_code
|
||||
docs_url = ex.docs_url or docs_url
|
||||
raiseable_exception = ex.raiseable_exception
|
||||
|
||||
# Print exception
|
||||
if raiseable_exception:
|
||||
fmt.error(str(ex))
|
||||
|
||||
fmt.note(f"Please refer to our docs at '{docs_url}' for further assistance.")
|
||||
|
||||
if debug.is_debug_enabled() and raiseable_exception:
|
||||
raise raiseable_exception
|
||||
|
||||
return error_code
|
||||
else:
|
||||
print_help(parser)
|
||||
return -1
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
def _main() -> None:
|
||||
"""Script entry point"""
|
||||
sys.exit(main())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
1
cognee/cli/commands/__init__.py
Normal file
1
cognee/cli/commands/__init__.py
Normal file
|
|
@ -0,0 +1 @@
|
|||
# CLI Commands package
|
||||
80
cognee/cli/commands/add_command.py
Normal file
80
cognee/cli/commands/add_command.py
Normal file
|
|
@ -0,0 +1,80 @@
|
|||
import argparse
|
||||
import asyncio
|
||||
from typing import Optional
|
||||
|
||||
from cognee.cli.reference import SupportsCliCommand
|
||||
from cognee.cli import DEFAULT_DOCS_URL
|
||||
import cognee.cli.echo as fmt
|
||||
from cognee.cli.exceptions import CliCommandException, CliCommandInnerException
|
||||
|
||||
|
||||
class AddCommand(SupportsCliCommand):
|
||||
command_string = "add"
|
||||
help_string = "Add data to Cognee for knowledge graph processing"
|
||||
docs_url = DEFAULT_DOCS_URL
|
||||
description = """
|
||||
Add data to Cognee for knowledge graph processing.
|
||||
|
||||
This is the first step in the Cognee workflow - it ingests raw data and prepares it
|
||||
for processing. The function accepts various data formats including text, files, and
|
||||
binary streams, then stores them in a specified dataset for further processing.
|
||||
|
||||
Supported Input Types:
|
||||
- **Text strings**: Direct text content
|
||||
- **File paths**: Local file paths (absolute paths starting with "/")
|
||||
- **File URLs**: "file:///absolute/path" or "file://relative/path"
|
||||
- **S3 paths**: "s3://bucket-name/path/to/file"
|
||||
- **Lists**: Multiple files or text strings in a single call
|
||||
|
||||
Supported File Formats:
|
||||
- Text files (.txt, .md, .csv)
|
||||
- PDFs (.pdf)
|
||||
- Images (.png, .jpg, .jpeg) - extracted via OCR/vision models
|
||||
- Audio files (.mp3, .wav) - transcribed to text
|
||||
- Code files (.py, .js, .ts, etc.) - parsed for structure and content
|
||||
- Office documents (.docx, .pptx)
|
||||
|
||||
After adding data, use `cognee cognify` to process it into knowledge graphs.
|
||||
"""
|
||||
|
||||
def configure_parser(self, parser: argparse.ArgumentParser) -> None:
|
||||
parser.add_argument(
|
||||
"data",
|
||||
nargs="+",
|
||||
help="Data to add: text content, file paths (/path/to/file), file URLs (file://path), S3 paths (s3://bucket/file), or mix of these",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--dataset-name",
|
||||
"-d",
|
||||
default="main_dataset",
|
||||
help="Dataset name to organize your data (default: main_dataset)",
|
||||
)
|
||||
|
||||
def execute(self, args: argparse.Namespace) -> None:
|
||||
try:
|
||||
# Import cognee here to avoid circular imports
|
||||
import cognee
|
||||
|
||||
fmt.echo(f"Adding {len(args.data)} item(s) to dataset '{args.dataset_name}'...")
|
||||
|
||||
# Run the async add function
|
||||
async def run_add():
|
||||
try:
|
||||
# Pass all data items as a list to cognee.add if multiple items
|
||||
if len(args.data) == 1:
|
||||
data_to_add = args.data[0]
|
||||
else:
|
||||
data_to_add = args.data
|
||||
|
||||
fmt.echo("Processing data...")
|
||||
await cognee.add(data=data_to_add, dataset_name=args.dataset_name)
|
||||
fmt.success(f"Successfully added data to dataset '{args.dataset_name}'")
|
||||
except Exception as e:
|
||||
raise CliCommandInnerException(f"Failed to add data: {str(e)}")
|
||||
|
||||
asyncio.run(run_add())
|
||||
|
||||
except Exception as e:
|
||||
if isinstance(e, CliCommandInnerException):
|
||||
raise CliCommandException(str(e), error_code=1)
|
||||
raise CliCommandException(f"Error adding data: {str(e)}", error_code=1)
|
||||
128
cognee/cli/commands/cognify_command.py
Normal file
128
cognee/cli/commands/cognify_command.py
Normal file
|
|
@ -0,0 +1,128 @@
|
|||
import argparse
|
||||
import asyncio
|
||||
from typing import Optional
|
||||
|
||||
from cognee.cli.reference import SupportsCliCommand
|
||||
from cognee.cli import DEFAULT_DOCS_URL
|
||||
from cognee.cli.config import CHUNKER_CHOICES
|
||||
import cognee.cli.echo as fmt
|
||||
from cognee.cli.exceptions import CliCommandException, CliCommandInnerException
|
||||
|
||||
|
||||
class CognifyCommand(SupportsCliCommand):
|
||||
command_string = "cognify"
|
||||
help_string = "Transform ingested data into a structured knowledge graph"
|
||||
docs_url = DEFAULT_DOCS_URL
|
||||
description = """
|
||||
Transform ingested data into a structured knowledge graph.
|
||||
|
||||
This is the core processing step in Cognee that converts raw text and documents
|
||||
into an intelligent knowledge graph. It analyzes content, extracts entities and
|
||||
relationships, and creates semantic connections for enhanced search and reasoning.
|
||||
|
||||
Processing Pipeline:
|
||||
1. **Document Classification**: Identifies document types and structures
|
||||
2. **Permission Validation**: Ensures user has processing rights
|
||||
3. **Text Chunking**: Breaks content into semantically meaningful segments
|
||||
4. **Entity Extraction**: Identifies key concepts, people, places, organizations
|
||||
5. **Relationship Detection**: Discovers connections between entities
|
||||
6. **Graph Construction**: Builds semantic knowledge graph with embeddings
|
||||
7. **Content Summarization**: Creates hierarchical summaries for navigation
|
||||
|
||||
After successful cognify processing, use `cognee search` to query the knowledge graph.
|
||||
"""
|
||||
|
||||
def configure_parser(self, parser: argparse.ArgumentParser) -> None:
|
||||
parser.add_argument(
|
||||
"--datasets",
|
||||
"-d",
|
||||
nargs="*",
|
||||
help="Dataset name(s) to process. Processes all available data if not specified. Can be multiple: --datasets dataset1 dataset2",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--chunk-size",
|
||||
type=int,
|
||||
help="Maximum tokens per chunk. Auto-calculated based on LLM if not specified (~512-8192 tokens)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--ontology-file", help="Path to RDF/OWL ontology file for domain-specific entity types"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--chunker",
|
||||
choices=CHUNKER_CHOICES,
|
||||
default="TextChunker",
|
||||
help="Text chunking strategy (default: TextChunker)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--background",
|
||||
"-b",
|
||||
action="store_true",
|
||||
help="Run processing in background and return immediately (recommended for large datasets)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--verbose", "-v", action="store_true", help="Show detailed progress information"
|
||||
)
|
||||
|
||||
def execute(self, args: argparse.Namespace) -> None:
|
||||
try:
|
||||
# Import cognee here to avoid circular imports
|
||||
import cognee
|
||||
|
||||
# Prepare datasets parameter
|
||||
datasets = args.datasets if args.datasets else None
|
||||
dataset_msg = f" for datasets {datasets}" if datasets else " for all available data"
|
||||
fmt.echo(f"Starting cognification{dataset_msg}...")
|
||||
|
||||
if args.verbose:
|
||||
fmt.note("This process will analyze your data and build knowledge graphs.")
|
||||
fmt.note("Depending on data size, this may take several minutes.")
|
||||
if args.background:
|
||||
fmt.note(
|
||||
"Running in background mode - the process will continue after this command exits."
|
||||
)
|
||||
|
||||
# Prepare chunker parameter - will be handled in the async function
|
||||
|
||||
# Run the async cognify function
|
||||
async def run_cognify():
|
||||
try:
|
||||
# Import chunker classes here
|
||||
from cognee.modules.chunking import TextChunker
|
||||
|
||||
chunker_class = TextChunker # Default
|
||||
if args.chunker == "LangchainChunker":
|
||||
try:
|
||||
from cognee.modules.chunking import LangchainChunker
|
||||
|
||||
chunker_class = LangchainChunker
|
||||
except ImportError:
|
||||
fmt.warning("LangchainChunker not available, using TextChunker")
|
||||
|
||||
result = await cognee.cognify(
|
||||
datasets=datasets,
|
||||
chunker=chunker_class,
|
||||
chunk_size=args.chunk_size,
|
||||
ontology_file_path=args.ontology_file,
|
||||
run_in_background=args.background,
|
||||
)
|
||||
return result
|
||||
except Exception as e:
|
||||
raise CliCommandInnerException(f"Failed to cognify: {str(e)}")
|
||||
|
||||
result = asyncio.run(run_cognify())
|
||||
|
||||
if args.background:
|
||||
fmt.success("Cognification started in background!")
|
||||
if args.verbose and result:
|
||||
fmt.echo(
|
||||
"Background processing initiated. Use pipeline monitoring to track progress."
|
||||
)
|
||||
else:
|
||||
fmt.success("Cognification completed successfully!")
|
||||
if args.verbose and result:
|
||||
fmt.echo(f"Processing results: {result}")
|
||||
|
||||
except Exception as e:
|
||||
if isinstance(e, CliCommandInnerException):
|
||||
raise CliCommandException(str(e), error_code=1)
|
||||
raise CliCommandException(f"Error during cognification: {str(e)}", error_code=1)
|
||||
225
cognee/cli/commands/config_command.py
Normal file
225
cognee/cli/commands/config_command.py
Normal file
|
|
@ -0,0 +1,225 @@
|
|||
import argparse
|
||||
import json
|
||||
from typing import Optional, Any
|
||||
|
||||
from cognee.cli.reference import SupportsCliCommand
|
||||
from cognee.cli import DEFAULT_DOCS_URL
|
||||
import cognee.cli.echo as fmt
|
||||
from cognee.cli.exceptions import CliCommandException, CliCommandInnerException
|
||||
|
||||
|
||||
class ConfigCommand(SupportsCliCommand):
|
||||
command_string = "config"
|
||||
help_string = "Manage cognee configuration settings"
|
||||
docs_url = DEFAULT_DOCS_URL
|
||||
description = """
|
||||
The `cognee config` command allows you to view and modify configuration settings.
|
||||
|
||||
You can:
|
||||
- View all current configuration settings
|
||||
- Get specific configuration values
|
||||
- Set configuration values
|
||||
- Unset (reset to default) specific configuration values
|
||||
- Reset all configuration to defaults
|
||||
|
||||
Configuration changes will affect how cognee processes and stores data.
|
||||
"""
|
||||
|
||||
def configure_parser(self, parser: argparse.ArgumentParser) -> None:
|
||||
subparsers = parser.add_subparsers(dest="config_action", help="Configuration actions")
|
||||
|
||||
# Get command
|
||||
get_parser = subparsers.add_parser("get", help="Get configuration value(s)")
|
||||
get_parser.add_argument(
|
||||
"key", nargs="?", help="Configuration key to retrieve (shows all if not specified)"
|
||||
)
|
||||
|
||||
# Set command
|
||||
set_parser = subparsers.add_parser("set", help="Set configuration value")
|
||||
set_parser.add_argument("key", help="Configuration key to set")
|
||||
set_parser.add_argument("value", help="Configuration value to set")
|
||||
|
||||
# List command
|
||||
subparsers.add_parser("list", help="List all configuration keys")
|
||||
|
||||
# Unset command
|
||||
unset_parser = subparsers.add_parser("unset", help="Remove/unset a configuration value")
|
||||
unset_parser.add_argument("key", help="Configuration key to unset")
|
||||
unset_parser.add_argument(
|
||||
"--force", "-f", action="store_true", help="Skip confirmation prompt"
|
||||
)
|
||||
|
||||
# Reset command
|
||||
reset_parser = subparsers.add_parser("reset", help="Reset configuration to defaults")
|
||||
reset_parser.add_argument(
|
||||
"--force", "-f", action="store_true", help="Skip confirmation prompt"
|
||||
)
|
||||
|
||||
def execute(self, args: argparse.Namespace) -> None:
|
||||
try:
|
||||
# Import cognee here to avoid circular imports
|
||||
import cognee
|
||||
|
||||
if not hasattr(args, "config_action") or args.config_action is None:
|
||||
fmt.error("Please specify a config action: get, set, unset, list, or reset")
|
||||
return
|
||||
|
||||
if args.config_action == "get":
|
||||
self._handle_get(args)
|
||||
elif args.config_action == "set":
|
||||
self._handle_set(args)
|
||||
elif args.config_action == "unset":
|
||||
self._handle_unset(args)
|
||||
elif args.config_action == "list":
|
||||
self._handle_list(args)
|
||||
elif args.config_action == "reset":
|
||||
self._handle_reset(args)
|
||||
else:
|
||||
fmt.error(f"Unknown config action: {args.config_action}")
|
||||
|
||||
except Exception as e:
|
||||
if isinstance(e, CliCommandInnerException):
|
||||
raise CliCommandException(str(e), error_code=1)
|
||||
raise CliCommandException(f"Error managing configuration: {str(e)}", error_code=1)
|
||||
|
||||
def _handle_get(self, args: argparse.Namespace) -> None:
|
||||
try:
|
||||
import cognee
|
||||
|
||||
if args.key:
|
||||
# Get specific key
|
||||
try:
|
||||
if hasattr(cognee.config, "get"):
|
||||
value = cognee.config.get(args.key)
|
||||
fmt.echo(f"{args.key}: {value}")
|
||||
else:
|
||||
fmt.error("Configuration retrieval not implemented yet")
|
||||
fmt.note(
|
||||
"The config system currently only supports setting values, not retrieving them"
|
||||
)
|
||||
fmt.note(f"To set this value: 'cognee config set {args.key} <value>'")
|
||||
except Exception:
|
||||
fmt.error(f"Configuration key '{args.key}' not found or retrieval failed")
|
||||
else:
|
||||
# Get all configuration
|
||||
try:
|
||||
if hasattr(cognee.config, "get_all"):
|
||||
config_dict = cognee.config.get_all()
|
||||
if config_dict:
|
||||
fmt.echo("Current configuration:")
|
||||
for key, value in config_dict.items():
|
||||
fmt.echo(f" {key}: {value}")
|
||||
else:
|
||||
fmt.echo("No configuration settings found")
|
||||
else:
|
||||
fmt.error("Configuration viewing not implemented yet")
|
||||
fmt.note(
|
||||
"The config system currently only supports setting values, not retrieving them"
|
||||
)
|
||||
fmt.note("Available commands: 'cognee config set <key> <value>'")
|
||||
except Exception:
|
||||
fmt.error("Failed to retrieve configuration")
|
||||
fmt.note("Configuration viewing not fully implemented yet")
|
||||
|
||||
except Exception as e:
|
||||
raise CliCommandInnerException(f"Failed to get configuration: {str(e)}")
|
||||
|
||||
def _handle_set(self, args: argparse.Namespace) -> None:
|
||||
try:
|
||||
import cognee
|
||||
|
||||
# Try to parse value as JSON, otherwise treat as string
|
||||
try:
|
||||
value = json.loads(args.value)
|
||||
except json.JSONDecodeError:
|
||||
value = args.value
|
||||
|
||||
try:
|
||||
cognee.config.set(args.key, value)
|
||||
fmt.success(f"Set {args.key} = {value}")
|
||||
except Exception:
|
||||
fmt.error(f"Failed to set configuration key '{args.key}'")
|
||||
|
||||
except Exception as e:
|
||||
raise CliCommandInnerException(f"Failed to set configuration: {str(e)}")
|
||||
|
||||
def _handle_unset(self, args: argparse.Namespace) -> None:
|
||||
try:
|
||||
import cognee
|
||||
|
||||
# Confirm unset unless forced
|
||||
if not args.force:
|
||||
if not fmt.confirm(f"Unset configuration key '{args.key}'?"):
|
||||
fmt.echo("Unset cancelled.")
|
||||
return
|
||||
|
||||
# Since the config system doesn't have explicit unset methods,
|
||||
# we need to map config keys to their reset/default behaviors
|
||||
config_key_mappings = {
|
||||
# LLM configuration
|
||||
"llm_provider": ("set_llm_provider", "openai"),
|
||||
"llm_model": ("set_llm_model", "gpt-5-mini"),
|
||||
"llm_api_key": ("set_llm_api_key", ""),
|
||||
"llm_endpoint": ("set_llm_endpoint", ""),
|
||||
# Database configuration
|
||||
"graph_database_provider": ("set_graph_database_provider", "kuzu"),
|
||||
"vector_db_provider": ("set_vector_db_provider", "lancedb"),
|
||||
"vector_db_url": ("set_vector_db_url", ""),
|
||||
"vector_db_key": ("set_vector_db_key", ""),
|
||||
# Chunking configuration
|
||||
"chunk_size": ("set_chunk_size", 1500),
|
||||
"chunk_overlap": ("set_chunk_overlap", 10),
|
||||
}
|
||||
|
||||
if args.key in config_key_mappings:
|
||||
method_name, default_value = config_key_mappings[args.key]
|
||||
|
||||
try:
|
||||
# Get the method and call it with the default value
|
||||
method = getattr(cognee.config, method_name)
|
||||
method(default_value)
|
||||
fmt.success(f"Unset {args.key} (reset to default: {default_value})")
|
||||
except AttributeError:
|
||||
fmt.error(f"Configuration method '{method_name}' not found")
|
||||
except Exception as e:
|
||||
fmt.error(f"Failed to unset '{args.key}': {str(e)}")
|
||||
else:
|
||||
fmt.error(f"Unknown configuration key '{args.key}'")
|
||||
fmt.note("Available keys: " + ", ".join(config_key_mappings.keys()))
|
||||
fmt.note("Use 'cognee config list' to see all available configuration options")
|
||||
|
||||
except Exception as e:
|
||||
raise CliCommandInnerException(f"Failed to unset configuration: {str(e)}")
|
||||
|
||||
def _handle_list(self, args: argparse.Namespace) -> None:
|
||||
try:
|
||||
import cognee
|
||||
|
||||
# This would need to be implemented in cognee.config
|
||||
fmt.note("Available configuration keys:")
|
||||
fmt.echo(" llm_provider, llm_model, llm_api_key, llm_endpoint")
|
||||
fmt.echo(" graph_database_provider, vector_db_provider")
|
||||
fmt.echo(" vector_db_url, vector_db_key")
|
||||
fmt.echo(" chunk_size, chunk_overlap")
|
||||
fmt.echo("")
|
||||
fmt.echo("Commands:")
|
||||
fmt.echo(" cognee config get [key] - View configuration")
|
||||
fmt.echo(" cognee config set <key> <value> - Set configuration")
|
||||
fmt.echo(" cognee config unset <key> - Reset key to default")
|
||||
fmt.echo(" cognee config reset - Reset all to defaults")
|
||||
|
||||
except Exception as e:
|
||||
raise CliCommandInnerException(f"Failed to list configuration: {str(e)}")
|
||||
|
||||
def _handle_reset(self, args: argparse.Namespace) -> None:
|
||||
try:
|
||||
if not args.force:
|
||||
if not fmt.confirm("Reset all configuration to defaults?"):
|
||||
fmt.echo("Reset cancelled.")
|
||||
return
|
||||
|
||||
fmt.note("Configuration reset not fully implemented yet")
|
||||
fmt.echo("This would reset all settings to their default values")
|
||||
|
||||
except Exception as e:
|
||||
raise CliCommandInnerException(f"Failed to reset configuration: {str(e)}")
|
||||
80
cognee/cli/commands/delete_command.py
Normal file
80
cognee/cli/commands/delete_command.py
Normal file
|
|
@ -0,0 +1,80 @@
|
|||
import argparse
|
||||
import asyncio
|
||||
from typing import Optional
|
||||
|
||||
from cognee.cli.reference import SupportsCliCommand
|
||||
from cognee.cli import DEFAULT_DOCS_URL
|
||||
import cognee.cli.echo as fmt
|
||||
from cognee.cli.exceptions import CliCommandException, CliCommandInnerException
|
||||
|
||||
|
||||
class DeleteCommand(SupportsCliCommand):
|
||||
command_string = "delete"
|
||||
help_string = "Delete data from cognee knowledge base"
|
||||
docs_url = DEFAULT_DOCS_URL
|
||||
description = """
|
||||
The `cognee delete` command removes data from your knowledge base.
|
||||
|
||||
You can delete:
|
||||
- Specific datasets by name
|
||||
- All data (with confirmation)
|
||||
- Data for specific users
|
||||
|
||||
Be careful with deletion operations as they are irreversible.
|
||||
"""
|
||||
|
||||
def configure_parser(self, parser: argparse.ArgumentParser) -> None:
|
||||
parser.add_argument("--dataset-name", "-d", help="Specific dataset to delete")
|
||||
parser.add_argument("--user-id", "-u", help="User ID to delete data for")
|
||||
parser.add_argument(
|
||||
"--all", action="store_true", help="Delete all data (requires confirmation)"
|
||||
)
|
||||
parser.add_argument("--force", "-f", action="store_true", help="Skip confirmation prompts")
|
||||
|
||||
def execute(self, args: argparse.Namespace) -> None:
|
||||
try:
|
||||
# Import cognee here to avoid circular imports
|
||||
import cognee
|
||||
|
||||
# Validate arguments
|
||||
if not any([args.dataset_name, args.user_id, args.all]):
|
||||
fmt.error("Please specify what to delete: --dataset-name, --user-id, or --all")
|
||||
return
|
||||
|
||||
# Build confirmation message
|
||||
if args.all:
|
||||
confirm_msg = "Delete ALL data from cognee?"
|
||||
operation = "all data"
|
||||
elif args.dataset_name:
|
||||
confirm_msg = f"Delete dataset '{args.dataset_name}'?"
|
||||
operation = f"dataset '{args.dataset_name}'"
|
||||
elif args.user_id:
|
||||
confirm_msg = f"Delete all data for user '{args.user_id}'?"
|
||||
operation = f"data for user '{args.user_id}'"
|
||||
|
||||
# Confirm deletion unless forced
|
||||
if not args.force:
|
||||
fmt.warning("This operation is irreversible!")
|
||||
if not fmt.confirm(confirm_msg):
|
||||
fmt.echo("Deletion cancelled.")
|
||||
return
|
||||
|
||||
fmt.echo(f"Deleting {operation}...")
|
||||
|
||||
# Run the async delete function
|
||||
async def run_delete():
|
||||
try:
|
||||
if args.all:
|
||||
await cognee.delete(dataset_name=None, user_id=args.user_id)
|
||||
else:
|
||||
await cognee.delete(dataset_name=args.dataset_name, user_id=args.user_id)
|
||||
except Exception as e:
|
||||
raise CliCommandInnerException(f"Failed to delete: {str(e)}")
|
||||
|
||||
asyncio.run(run_delete())
|
||||
fmt.success(f"Successfully deleted {operation}")
|
||||
|
||||
except Exception as e:
|
||||
if isinstance(e, CliCommandInnerException):
|
||||
raise CliCommandException(str(e), error_code=1)
|
||||
raise CliCommandException(f"Error deleting data: {str(e)}", error_code=1)
|
||||
149
cognee/cli/commands/search_command.py
Normal file
149
cognee/cli/commands/search_command.py
Normal file
|
|
@ -0,0 +1,149 @@
|
|||
import argparse
|
||||
import asyncio
|
||||
import json
|
||||
from typing import Optional
|
||||
|
||||
from cognee.cli.reference import SupportsCliCommand
|
||||
from cognee.cli import DEFAULT_DOCS_URL
|
||||
from cognee.cli.config import SEARCH_TYPE_CHOICES, OUTPUT_FORMAT_CHOICES
|
||||
import cognee.cli.echo as fmt
|
||||
from cognee.cli.exceptions import CliCommandException, CliCommandInnerException
|
||||
|
||||
|
||||
class SearchCommand(SupportsCliCommand):
|
||||
command_string = "search"
|
||||
help_string = "Search and query the knowledge graph for insights, information, and connections"
|
||||
docs_url = DEFAULT_DOCS_URL
|
||||
description = """
|
||||
Search and query the knowledge graph for insights, information, and connections.
|
||||
|
||||
This is the final step in the Cognee workflow that retrieves information from the
|
||||
processed knowledge graph. It supports multiple search modes optimized for different
|
||||
use cases - from simple fact retrieval to complex reasoning and code analysis.
|
||||
|
||||
Search Types & Use Cases:
|
||||
|
||||
**GRAPH_COMPLETION** (Default - Recommended):
|
||||
Natural language Q&A using full graph context and LLM reasoning.
|
||||
Best for: Complex questions, analysis, summaries, insights.
|
||||
|
||||
**RAG_COMPLETION**:
|
||||
Traditional RAG using document chunks without graph structure.
|
||||
Best for: Direct document retrieval, specific fact-finding.
|
||||
|
||||
**INSIGHTS**:
|
||||
Structured entity relationships and semantic connections.
|
||||
Best for: Understanding concept relationships, knowledge mapping.
|
||||
|
||||
**CHUNKS**:
|
||||
Raw text segments that match the query semantically.
|
||||
Best for: Finding specific passages, citations, exact content.
|
||||
|
||||
**SUMMARIES**:
|
||||
Pre-generated hierarchical summaries of content.
|
||||
Best for: Quick overviews, document abstracts, topic summaries.
|
||||
|
||||
**CODE**:
|
||||
Code-specific search with syntax and semantic understanding.
|
||||
Best for: Finding functions, classes, implementation patterns.
|
||||
"""
|
||||
|
||||
def configure_parser(self, parser: argparse.ArgumentParser) -> None:
|
||||
parser.add_argument("query_text", help="Your question or search query in natural language")
|
||||
parser.add_argument(
|
||||
"--query-type",
|
||||
"-t",
|
||||
choices=SEARCH_TYPE_CHOICES,
|
||||
default="GRAPH_COMPLETION",
|
||||
help="Search mode (default: GRAPH_COMPLETION for conversational AI responses)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--datasets",
|
||||
"-d",
|
||||
nargs="*",
|
||||
help="Dataset name(s) to search within. Searches all accessible datasets if not specified",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--top-k",
|
||||
"-k",
|
||||
type=int,
|
||||
default=10,
|
||||
help="Maximum number of results to return (default: 10, max: 100)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--system-prompt",
|
||||
help="Custom system prompt file for LLM-based search types (default: answer_simple_question.txt)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--output-format",
|
||||
"-f",
|
||||
choices=OUTPUT_FORMAT_CHOICES,
|
||||
default="pretty",
|
||||
help="Output format (default: pretty)",
|
||||
)
|
||||
|
||||
def execute(self, args: argparse.Namespace) -> None:
|
||||
try:
|
||||
# Import cognee here to avoid circular imports
|
||||
import cognee
|
||||
from cognee.modules.search.types import SearchType
|
||||
|
||||
# Convert string to SearchType enum
|
||||
query_type = SearchType[args.query_type]
|
||||
|
||||
datasets_msg = (
|
||||
f" in datasets {args.datasets}" if args.datasets else " across all datasets"
|
||||
)
|
||||
fmt.echo(f"Searching for: '{args.query_text}' (type: {args.query_type}){datasets_msg}")
|
||||
|
||||
# Run the async search function
|
||||
async def run_search():
|
||||
try:
|
||||
results = await cognee.search(
|
||||
query_text=args.query_text,
|
||||
query_type=query_type,
|
||||
datasets=args.datasets,
|
||||
system_prompt_path=args.system_prompt or "answer_simple_question.txt",
|
||||
top_k=args.top_k,
|
||||
)
|
||||
return results
|
||||
except Exception as e:
|
||||
raise CliCommandInnerException(f"Failed to search: {str(e)}")
|
||||
|
||||
results = asyncio.run(run_search())
|
||||
|
||||
# Format and display results
|
||||
if args.output_format == "json":
|
||||
fmt.echo(json.dumps(results, indent=2, default=str))
|
||||
elif args.output_format == "simple":
|
||||
for i, result in enumerate(results, 1):
|
||||
fmt.echo(f"{i}. {result}")
|
||||
else: # pretty format
|
||||
if not results:
|
||||
fmt.warning("No results found for your query.")
|
||||
return
|
||||
|
||||
fmt.echo(f"\nFound {len(results)} result(s) using {args.query_type}:")
|
||||
fmt.echo("=" * 60)
|
||||
|
||||
if args.query_type in ["GRAPH_COMPLETION", "RAG_COMPLETION"]:
|
||||
# These return conversational responses
|
||||
for i, result in enumerate(results, 1):
|
||||
fmt.echo(f"{fmt.bold('Response:')} {result}")
|
||||
if i < len(results):
|
||||
fmt.echo("-" * 40)
|
||||
elif args.query_type == "CHUNKS":
|
||||
# These return text chunks
|
||||
for i, result in enumerate(results, 1):
|
||||
fmt.echo(f"{fmt.bold(f'Chunk {i}:')} {result}")
|
||||
fmt.echo()
|
||||
else:
|
||||
# Generic formatting for other types
|
||||
for i, result in enumerate(results, 1):
|
||||
fmt.echo(f"{fmt.bold(f'Result {i}:')} {result}")
|
||||
fmt.echo()
|
||||
|
||||
except Exception as e:
|
||||
if isinstance(e, CliCommandInnerException):
|
||||
raise CliCommandException(str(e), error_code=1)
|
||||
raise CliCommandException(f"Error searching: {str(e)}", error_code=1)
|
||||
33
cognee/cli/config.py
Normal file
33
cognee/cli/config.py
Normal file
|
|
@ -0,0 +1,33 @@
|
|||
"""
|
||||
CLI configuration and constants to avoid hardcoded values
|
||||
"""
|
||||
|
||||
# CLI Constants
|
||||
CLI_DESCRIPTION = "Cognee CLI - Manage your knowledge graphs and cognitive processing pipelines."
|
||||
DEFAULT_DOCS_URL = "https://docs.cognee.ai"
|
||||
|
||||
# Command descriptions - these should match the actual command implementations
|
||||
COMMAND_DESCRIPTIONS = {
|
||||
"add": "Add data to Cognee for knowledge graph processing",
|
||||
"search": "Search and query the knowledge graph for insights, information, and connections",
|
||||
"cognify": "Transform ingested data into a structured knowledge graph",
|
||||
"delete": "Delete data from cognee knowledge base",
|
||||
"config": "Manage cognee configuration settings",
|
||||
}
|
||||
|
||||
# Search type choices
|
||||
SEARCH_TYPE_CHOICES = [
|
||||
"GRAPH_COMPLETION",
|
||||
"RAG_COMPLETION",
|
||||
"INSIGHTS",
|
||||
"CHUNKS",
|
||||
"SUMMARIES",
|
||||
"CODE",
|
||||
"CYPHER",
|
||||
]
|
||||
|
||||
# Chunker choices
|
||||
CHUNKER_CHOICES = ["TextChunker", "LangchainChunker"]
|
||||
|
||||
# Output format choices
|
||||
OUTPUT_FORMAT_CHOICES = ["json", "pretty", "simple"]
|
||||
21
cognee/cli/debug.py
Normal file
21
cognee/cli/debug.py
Normal file
|
|
@ -0,0 +1,21 @@
|
|||
"""Provides a global debug setting for the CLI - following dlt patterns"""
|
||||
|
||||
_DEBUG_FLAG = False
|
||||
|
||||
|
||||
def enable_debug() -> None:
|
||||
"""Enable debug mode for CLI"""
|
||||
global _DEBUG_FLAG
|
||||
_DEBUG_FLAG = True
|
||||
|
||||
|
||||
def disable_debug() -> None:
|
||||
"""Disable debug mode for CLI"""
|
||||
global _DEBUG_FLAG
|
||||
_DEBUG_FLAG = False
|
||||
|
||||
|
||||
def is_debug_enabled() -> bool:
|
||||
"""Check if debug mode is enabled"""
|
||||
global _DEBUG_FLAG
|
||||
return _DEBUG_FLAG
|
||||
45
cognee/cli/echo.py
Normal file
45
cognee/cli/echo.py
Normal file
|
|
@ -0,0 +1,45 @@
|
|||
"""CLI output formatting utilities"""
|
||||
|
||||
import sys
|
||||
import click
|
||||
from typing import Any
|
||||
|
||||
|
||||
def echo(message: str = "", color: str = None, err: bool = False) -> None:
|
||||
"""Echo a message to stdout or stderr with optional color"""
|
||||
click.secho(message, fg=color, err=err)
|
||||
|
||||
|
||||
def note(message: str) -> None:
|
||||
"""Print a note in blue"""
|
||||
echo(f"Note: {message}", color="blue")
|
||||
|
||||
|
||||
def warning(message: str) -> None:
|
||||
"""Print a warning in yellow"""
|
||||
echo(f"Warning: {message}", color="yellow")
|
||||
|
||||
|
||||
def error(message: str) -> None:
|
||||
"""Print an error in red"""
|
||||
echo(f"Error: {message}", color="red", err=True)
|
||||
|
||||
|
||||
def success(message: str) -> None:
|
||||
"""Print a success message in green"""
|
||||
echo(f"Success: {message}", color="green")
|
||||
|
||||
|
||||
def bold(text: str) -> str:
|
||||
"""Make text bold"""
|
||||
return click.style(text, bold=True)
|
||||
|
||||
|
||||
def confirm(message: str, default: bool = False) -> bool:
|
||||
"""Ask for user confirmation"""
|
||||
return click.confirm(message, default=default)
|
||||
|
||||
|
||||
def prompt(message: str, default: Any = None) -> str:
|
||||
"""Prompt user for input"""
|
||||
return click.prompt(message, default=default)
|
||||
23
cognee/cli/exceptions.py
Normal file
23
cognee/cli/exceptions.py
Normal file
|
|
@ -0,0 +1,23 @@
|
|||
from typing import Optional
|
||||
|
||||
|
||||
class CliCommandException(Exception):
|
||||
"""Exception raised by CLI commands with additional context"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
message: str,
|
||||
error_code: int = -1,
|
||||
docs_url: Optional[str] = None,
|
||||
raiseable_exception: Optional[Exception] = None,
|
||||
) -> None:
|
||||
super().__init__(message)
|
||||
self.error_code = error_code
|
||||
self.docs_url = docs_url
|
||||
self.raiseable_exception = raiseable_exception
|
||||
|
||||
|
||||
class CliCommandInnerException(Exception):
|
||||
"""Inner exception for wrapping other exceptions in CLI context"""
|
||||
|
||||
pass
|
||||
97
cognee/cli/minimal_cli.py
Normal file
97
cognee/cli/minimal_cli.py
Normal file
|
|
@ -0,0 +1,97 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Minimal CLI entry point for cognee that avoids early initialization
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
from typing import Any, Sequence
|
||||
|
||||
# CRITICAL: Prevent verbose logging initialization for CLI-only usage
|
||||
# This must be set before any cognee imports to be effective
|
||||
os.environ["COGNEE_MINIMAL_LOGGING"] = "true"
|
||||
os.environ["COGNEE_CLI_MODE"] = "true"
|
||||
|
||||
|
||||
def get_version() -> str:
|
||||
"""Get cognee version without importing the main package"""
|
||||
try:
|
||||
# Try to get version from pyproject.toml first (for development)
|
||||
from pathlib import Path
|
||||
|
||||
pyproject_path = Path(__file__).parent.parent.parent / "pyproject.toml"
|
||||
if pyproject_path.exists():
|
||||
with open(pyproject_path, encoding="utf-8") as f:
|
||||
for line in f:
|
||||
if line.startswith("version"):
|
||||
version = line.split("=")[1].strip("'\"\n ")
|
||||
return f"{version}-local"
|
||||
|
||||
# Fallback to installed package version
|
||||
import importlib.metadata
|
||||
|
||||
return importlib.metadata.version("cognee")
|
||||
except Exception:
|
||||
return "unknown"
|
||||
|
||||
|
||||
def get_command_info() -> dict:
|
||||
"""Get command information without importing cognee"""
|
||||
return {
|
||||
"add": "Add data to Cognee for knowledge graph processing",
|
||||
"search": "Search and query the knowledge graph for insights, information, and connections",
|
||||
"cognify": "Transform ingested data into a structured knowledge graph",
|
||||
"delete": "Delete data from cognee knowledge base",
|
||||
"config": "Manage cognee configuration settings",
|
||||
}
|
||||
|
||||
|
||||
def print_help() -> None:
|
||||
"""Print help message with dynamic command descriptions"""
|
||||
commands = get_command_info()
|
||||
command_list = "\n".join(f" {cmd:<12} {desc}" for cmd, desc in commands.items())
|
||||
|
||||
print(f"""
|
||||
usage: cognee [-h] [--version] [--debug] {{{"|".join(commands.keys())}}} ...
|
||||
|
||||
Cognee CLI - Manage your knowledge graphs and cognitive processing pipelines.
|
||||
|
||||
options:
|
||||
-h, --help show this help message and exit
|
||||
--version show program's version number and exit
|
||||
--debug Enable debug mode to show full stack traces on exceptions
|
||||
|
||||
Available commands:
|
||||
{{{",".join(commands.keys())}}}
|
||||
{command_list}
|
||||
|
||||
For more information on each command, use: cognee <command> --help
|
||||
""")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
"""Minimal CLI main function"""
|
||||
# Handle help and version without any imports - purely static
|
||||
if len(sys.argv) == 1 or (len(sys.argv) == 2 and sys.argv[1] in ["-h", "--help"]):
|
||||
print_help()
|
||||
return 0
|
||||
|
||||
if len(sys.argv) == 2 and sys.argv[1] == "--version":
|
||||
print(f"cognee {get_version()}")
|
||||
return 0
|
||||
|
||||
# For actual commands, import the full CLI with minimal logging
|
||||
try:
|
||||
from cognee.cli._cognee import main as full_main
|
||||
|
||||
return full_main()
|
||||
except Exception as e:
|
||||
if "--debug" in sys.argv:
|
||||
raise
|
||||
print(f"Error: {e}")
|
||||
print("Use --debug for full stack trace")
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
26
cognee/cli/reference.py
Normal file
26
cognee/cli/reference.py
Normal file
|
|
@ -0,0 +1,26 @@
|
|||
from abc import abstractmethod
|
||||
from typing import Protocol, Optional
|
||||
import argparse
|
||||
|
||||
|
||||
class SupportsCliCommand(Protocol):
|
||||
"""Protocol for defining one cognee cli command"""
|
||||
|
||||
command_string: str
|
||||
"""name of the command"""
|
||||
help_string: str
|
||||
"""the help string for argparse"""
|
||||
description: Optional[str]
|
||||
"""the more detailed description for argparse, may include markdown for the docs"""
|
||||
docs_url: Optional[str]
|
||||
"""the default docs url to be printed in case of an exception"""
|
||||
|
||||
@abstractmethod
|
||||
def configure_parser(self, parser: argparse.ArgumentParser) -> None:
|
||||
"""Configures the parser for the given argument"""
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def execute(self, args: argparse.Namespace) -> None:
|
||||
"""Executes the command with the given arguments"""
|
||||
...
|
||||
12
cognee/cli/suppress_logging.py
Normal file
12
cognee/cli/suppress_logging.py
Normal file
|
|
@ -0,0 +1,12 @@
|
|||
"""
|
||||
Module to suppress verbose logging before any cognee imports.
|
||||
This must be imported before any other cognee modules.
|
||||
"""
|
||||
|
||||
import os
|
||||
|
||||
# Set CLI mode to suppress verbose logging
|
||||
os.environ["COGNEE_CLI_MODE"] = "true"
|
||||
|
||||
# Also set log level to ERROR for extra safety
|
||||
os.environ["LOG_LEVEL"] = "ERROR"
|
||||
|
|
@ -173,29 +173,17 @@ def log_database_configuration(logger):
|
|||
from cognee.infrastructure.databases.graph.config import get_graph_config
|
||||
|
||||
try:
|
||||
# Log relational database configuration
|
||||
relational_config = get_relational_config()
|
||||
if relational_config.db_provider == "postgres":
|
||||
logger.info(f"Postgres host: {relational_config.db_host}:{relational_config.db_port}")
|
||||
elif relational_config.db_provider == "sqlite":
|
||||
logger.info(f"SQLite path: {relational_config.db_path}")
|
||||
# Get base database directory path
|
||||
from cognee.base_config import get_base_config
|
||||
|
||||
# Log vector database configuration
|
||||
vector_config = get_vectordb_config()
|
||||
if vector_config.vector_db_provider == "lancedb":
|
||||
logger.info(f"Vector database path: {vector_config.vector_db_url}")
|
||||
else:
|
||||
logger.info(f"Vector database URL: {vector_config.vector_db_url}")
|
||||
base_config = get_base_config()
|
||||
databases_path = os.path.join(base_config.system_root_directory, "databases")
|
||||
|
||||
# Log graph database configuration
|
||||
graph_config = get_graph_config()
|
||||
if graph_config.graph_database_provider == "kuzu":
|
||||
logger.info(f"Graph database path: {graph_config.graph_file_path}")
|
||||
else:
|
||||
logger.info(f"Graph database URL: {graph_config.graph_database_url}")
|
||||
# Log concise database info
|
||||
logger.info(f"Database storage: {databases_path}")
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"Could not retrieve database configuration: {str(e)}")
|
||||
logger.debug(f"Could not retrieve database configuration: {str(e)}")
|
||||
|
||||
|
||||
def cleanup_old_logs(logs_dir, max_files):
|
||||
|
|
@ -216,13 +204,22 @@ def cleanup_old_logs(logs_dir, max_files):
|
|||
|
||||
# Remove old files that exceed the maximum
|
||||
if len(log_files) > max_files:
|
||||
deleted_count = 0
|
||||
for old_file in log_files[max_files:]:
|
||||
try:
|
||||
old_file.unlink()
|
||||
logger.info(f"Deleted old log file: {old_file}")
|
||||
deleted_count += 1
|
||||
# Only log individual files in non-CLI mode
|
||||
if os.getenv("COGNEE_CLI_MODE") != "true":
|
||||
logger.info(f"Deleted old log file: {old_file}")
|
||||
except Exception as e:
|
||||
# Always log errors
|
||||
logger.error(f"Failed to delete old log file {old_file}: {e}")
|
||||
|
||||
# In CLI mode, show compact summary
|
||||
if os.getenv("COGNEE_CLI_MODE") == "true" and deleted_count > 0:
|
||||
logger.info(f"Cleaned up {deleted_count} old log files")
|
||||
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"Error cleaning up log files: {e}")
|
||||
|
|
@ -241,6 +238,7 @@ def setup_logging(log_level=None, name=None):
|
|||
"""
|
||||
global _is_structlog_configured
|
||||
|
||||
# Regular detailed logging for non-CLI usage
|
||||
log_level = log_level if log_level else log_levels[os.getenv("LOG_LEVEL", "INFO")]
|
||||
|
||||
# Configure external library logging early to suppress verbose output
|
||||
|
|
@ -298,11 +296,6 @@ def setup_logging(log_level=None, name=None):
|
|||
# Hand back to the original hook → prints traceback and exits
|
||||
sys.__excepthook__(exc_type, exc_value, traceback)
|
||||
|
||||
logger.info("Want to learn more? Visit the Cognee documentation: https://docs.cognee.ai")
|
||||
logger.info(
|
||||
"Need help? Reach out to us on our Discord server: https://discord.gg/NQPKmU5CCg"
|
||||
)
|
||||
|
||||
# Install exception handlers
|
||||
sys.excepthook = handle_exception
|
||||
|
||||
|
|
@ -380,18 +373,38 @@ def setup_logging(log_level=None, name=None):
|
|||
# Mark logging as configured
|
||||
_is_structlog_configured = True
|
||||
|
||||
from cognee.infrastructure.databases.relational.config import get_relational_config
|
||||
from cognee.infrastructure.databases.vector.config import get_vectordb_config
|
||||
from cognee.infrastructure.databases.graph.config import get_graph_config
|
||||
|
||||
graph_config = get_graph_config()
|
||||
vector_config = get_vectordb_config()
|
||||
relational_config = get_relational_config()
|
||||
|
||||
try:
|
||||
# Get base database directory path
|
||||
from cognee.base_config import get_base_config
|
||||
|
||||
base_config = get_base_config()
|
||||
databases_path = os.path.join(base_config.system_root_directory, "databases")
|
||||
except Exception as e:
|
||||
raise ValueError from e
|
||||
|
||||
# Get a configured logger and log system information
|
||||
logger = structlog.get_logger(name if name else __name__)
|
||||
# Detailed initialization for regular usage
|
||||
logger.info(
|
||||
"Logging initialized",
|
||||
python_version=PYTHON_VERSION,
|
||||
structlog_version=STRUCTLOG_VERSION,
|
||||
cognee_version=COGNEE_VERSION,
|
||||
os_info=OS_INFO,
|
||||
database_path=databases_path,
|
||||
graph_database_name=graph_config.graph_database_name,
|
||||
vector_config=vector_config.vector_db_provider,
|
||||
relational_config=relational_config.db_name,
|
||||
)
|
||||
|
||||
logger.info("Want to learn more? Visit the Cognee documentation: https://docs.cognee.ai")
|
||||
|
||||
# Log database configuration
|
||||
log_database_configuration(logger)
|
||||
|
||||
|
|
|
|||
3
cognee/tests/integration/cli/__init__.py
Normal file
3
cognee/tests/integration/cli/__init__.py
Normal file
|
|
@ -0,0 +1,3 @@
|
|||
"""
|
||||
CLI integration tests package.
|
||||
"""
|
||||
326
cognee/tests/integration/cli/test_cli_integration.py
Normal file
326
cognee/tests/integration/cli/test_cli_integration.py
Normal file
|
|
@ -0,0 +1,326 @@
|
|||
"""
|
||||
Integration tests for CLI commands that test end-to-end functionality.
|
||||
"""
|
||||
|
||||
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()
|
||||
3
cognee/tests/unit/cli/__init__.py
Normal file
3
cognee/tests/unit/cli/__init__.py
Normal file
|
|
@ -0,0 +1,3 @@
|
|||
"""
|
||||
CLI unit tests package.
|
||||
"""
|
||||
483
cognee/tests/unit/cli/test_cli_commands.py
Normal file
483
cognee/tests/unit/cli/test_cli_commands.py
Normal file
|
|
@ -0,0 +1,483 @@
|
|||
"""
|
||||
Tests for individual CLI commands with proper mocking and coroutine handling.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import sys
|
||||
import argparse
|
||||
import asyncio
|
||||
from unittest.mock import patch, MagicMock, AsyncMock, ANY
|
||||
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
|
||||
|
||||
|
||||
# Mock asyncio.run to properly handle coroutines
|
||||
def _mock_run(coro):
|
||||
# Create an event loop and run the coroutine
|
||||
loop = asyncio.new_event_loop()
|
||||
try:
|
||||
return loop.run_until_complete(coro)
|
||||
finally:
|
||||
loop.close()
|
||||
|
||||
|
||||
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", side_effect=_mock_run)
|
||||
def test_execute_single_item(self, mock_asyncio_run):
|
||||
"""Test execute with single data item"""
|
||||
# Mock the cognee module
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.add = AsyncMock()
|
||||
|
||||
with patch.dict(sys.modules, {"cognee": mock_cognee}):
|
||||
command = AddCommand()
|
||||
args = argparse.Namespace(data=["test.txt"], dataset_name="test_dataset")
|
||||
command.execute(args)
|
||||
|
||||
mock_asyncio_run.assert_called_once()
|
||||
assert asyncio.iscoroutine(mock_asyncio_run.call_args[0][0])
|
||||
mock_cognee.add.assert_awaited_once_with(data="test.txt", dataset_name="test_dataset")
|
||||
|
||||
@patch("cognee.cli.commands.add_command.asyncio.run", side_effect=_mock_run)
|
||||
def test_execute_multiple_items(self, mock_asyncio_run):
|
||||
"""Test execute with multiple data items"""
|
||||
# Mock the cognee module
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.add = AsyncMock()
|
||||
|
||||
with patch.dict(sys.modules, {"cognee": mock_cognee}):
|
||||
command = AddCommand()
|
||||
args = argparse.Namespace(data=["test1.txt", "test2.txt"], dataset_name="test_dataset")
|
||||
command.execute(args)
|
||||
|
||||
mock_asyncio_run.assert_called_once()
|
||||
assert asyncio.iscoroutine(mock_asyncio_run.call_args[0][0])
|
||||
mock_cognee.add.assert_awaited_once_with(
|
||||
data=["test1.txt", "test2.txt"], dataset_name="test_dataset"
|
||||
)
|
||||
|
||||
@patch("cognee.cli.commands.add_command.asyncio.run")
|
||||
def test_execute_with_exception(self, 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", side_effect=_mock_run)
|
||||
def test_execute_basic_search(self, mock_asyncio_run):
|
||||
"""Test execute with basic search"""
|
||||
# Mock the cognee module and SearchType
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.search = AsyncMock(return_value=["result1", "result2"])
|
||||
mock_search_type = MagicMock()
|
||||
mock_search_type.__getitem__.return_value = "GRAPH_COMPLETION"
|
||||
|
||||
with patch.dict(sys.modules, {"cognee": mock_cognee}):
|
||||
command = SearchCommand()
|
||||
args = argparse.Namespace(
|
||||
query_text="test query",
|
||||
query_type="GRAPH_COMPLETION",
|
||||
datasets=None,
|
||||
top_k=10,
|
||||
system_prompt=None,
|
||||
output_format="pretty",
|
||||
)
|
||||
command.execute(args)
|
||||
|
||||
mock_asyncio_run.assert_called_once()
|
||||
assert asyncio.iscoroutine(mock_asyncio_run.call_args[0][0])
|
||||
mock_cognee.search.assert_awaited_once_with(
|
||||
query_text="test query",
|
||||
query_type=ANY,
|
||||
datasets=None,
|
||||
top_k=10,
|
||||
system_prompt_path="answer_simple_question.txt",
|
||||
)
|
||||
# verify the enum’s name separately
|
||||
called_enum = mock_cognee.search.await_args.kwargs["query_type"]
|
||||
assert called_enum.name == "GRAPH_COMPLETION"
|
||||
|
||||
@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", side_effect=_mock_run)
|
||||
def test_execute_basic_cognify(self, mock_asyncio_run):
|
||||
"""Test execute with basic cognify"""
|
||||
# Mock the cognee module
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.cognify = AsyncMock(return_value="success")
|
||||
|
||||
with patch.dict(sys.modules, {"cognee": mock_cognee}):
|
||||
command = CognifyCommand()
|
||||
args = argparse.Namespace(
|
||||
datasets=None,
|
||||
chunk_size=None,
|
||||
ontology_file=None,
|
||||
chunker="TextChunker",
|
||||
background=False,
|
||||
verbose=False,
|
||||
)
|
||||
command.execute(args)
|
||||
|
||||
mock_asyncio_run.assert_called_once()
|
||||
assert asyncio.iscoroutine(mock_asyncio_run.call_args[0][0])
|
||||
from cognee.modules.chunking import TextChunker
|
||||
|
||||
mock_cognee.cognify.assert_awaited_once_with(
|
||||
datasets=None,
|
||||
chunk_size=None,
|
||||
ontology_file_path=None,
|
||||
chunker=TextChunker,
|
||||
run_in_background=False,
|
||||
)
|
||||
|
||||
@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", side_effect=_mock_run)
|
||||
def test_execute_delete_dataset_with_confirmation(self, mock_asyncio_run, mock_confirm):
|
||||
"""Test execute delete dataset with user confirmation"""
|
||||
# Mock the cognee module
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.delete = AsyncMock()
|
||||
|
||||
with patch.dict(sys.modules, {"cognee": mock_cognee}):
|
||||
command = DeleteCommand()
|
||||
args = argparse.Namespace(
|
||||
dataset_name="test_dataset", user_id=None, all=False, force=False
|
||||
)
|
||||
|
||||
mock_confirm.return_value = True
|
||||
|
||||
command.execute(args)
|
||||
|
||||
mock_confirm.assert_called_once_with(f"Delete dataset '{args.dataset_name}'?")
|
||||
mock_asyncio_run.assert_called_once()
|
||||
assert asyncio.iscoroutine(mock_asyncio_run.call_args[0][0])
|
||||
mock_cognee.delete.assert_awaited_once_with(dataset_name="test_dataset", user_id=None)
|
||||
|
||||
@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_with(f"Delete dataset '{args.dataset_name}'?")
|
||||
|
||||
@patch("cognee.cli.commands.delete_command.asyncio.run", side_effect=_mock_run)
|
||||
def test_execute_delete_forced(self, mock_asyncio_run):
|
||||
"""Test execute delete with force flag"""
|
||||
# Mock the cognee module
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.delete = AsyncMock()
|
||||
|
||||
with patch.dict(sys.modules, {"cognee": mock_cognee}):
|
||||
command = DeleteCommand()
|
||||
args = argparse.Namespace(
|
||||
dataset_name="test_dataset", user_id=None, all=False, force=True
|
||||
)
|
||||
|
||||
command.execute(args)
|
||||
|
||||
mock_asyncio_run.assert_called_once()
|
||||
assert asyncio.iscoroutine(mock_asyncio_run.call_args[0][0])
|
||||
mock_cognee.delete.assert_awaited_once_with(dataset_name="test_dataset", user_id=None)
|
||||
|
||||
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("builtins.__import__")
|
||||
def test_execute_get_action(self, mock_import):
|
||||
"""Test execute get action"""
|
||||
# Mock the cognee module
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.config.get = MagicMock(return_value="openai")
|
||||
mock_import.return_value = mock_cognee
|
||||
|
||||
command = ConfigCommand()
|
||||
args = argparse.Namespace(config_action="get", key="llm_provider")
|
||||
|
||||
command.execute(args)
|
||||
|
||||
@patch("builtins.__import__")
|
||||
def test_execute_set_action(self, mock_import):
|
||||
"""Test execute set action"""
|
||||
# Mock the cognee module
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.config.set = MagicMock()
|
||||
mock_import.return_value = mock_cognee
|
||||
|
||||
command = ConfigCommand()
|
||||
args = argparse.Namespace(config_action="set", key="llm_provider", value="anthropic")
|
||||
|
||||
command.execute(args)
|
||||
|
||||
@patch("builtins.__import__")
|
||||
def test_execute_set_action_json_value(self, mock_import):
|
||||
"""Test execute set action with JSON value"""
|
||||
# Mock the cognee module
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.config.set = MagicMock()
|
||||
mock_import.return_value = mock_cognee
|
||||
|
||||
command = ConfigCommand()
|
||||
args = argparse.Namespace(config_action="set", key="chunk_size", value="1024")
|
||||
|
||||
command.execute(args)
|
||||
|
||||
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")
|
||||
def test_execute_unset_action(self, mock_confirm):
|
||||
"""Test execute unset action"""
|
||||
# Mock the cognee module
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.config.set_llm_provider = MagicMock()
|
||||
|
||||
with patch.dict(sys.modules, {"cognee": mock_cognee}):
|
||||
command = ConfigCommand()
|
||||
args = argparse.Namespace(config_action="unset", key="llm_provider", force=False)
|
||||
|
||||
mock_confirm.return_value = True
|
||||
|
||||
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"""
|
||||
# Test with an invalid action that will cause an exception in the main execute method
|
||||
command = ConfigCommand()
|
||||
args = argparse.Namespace(config_action="invalid_action")
|
||||
|
||||
# This should not raise CliCommandException, just handle it gracefully
|
||||
# The config command handles unknown actions by showing an error message
|
||||
command.execute(args)
|
||||
617
cognee/tests/unit/cli/test_cli_edge_cases.py
Normal file
617
cognee/tests/unit/cli/test_cli_edge_cases.py
Normal file
|
|
@ -0,0 +1,617 @@
|
|||
"""
|
||||
Tests for CLI edge cases and error scenarios with proper mocking.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import sys
|
||||
import asyncio
|
||||
import argparse
|
||||
from unittest.mock import patch, MagicMock, AsyncMock, ANY, call
|
||||
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
|
||||
|
||||
|
||||
# Mock asyncio.run to properly handle coroutines
|
||||
def _mock_run(coro):
|
||||
# Create an event loop and run the coroutine
|
||||
loop = asyncio.new_event_loop()
|
||||
try:
|
||||
return loop.run_until_complete(coro)
|
||||
finally:
|
||||
loop.close()
|
||||
|
||||
|
||||
class TestAddCommandEdgeCases:
|
||||
"""Test edge cases for AddCommand"""
|
||||
|
||||
@patch("cognee.cli.commands.add_command.asyncio.run", side_effect=_mock_run)
|
||||
def test_add_empty_data_list(self, mock_asyncio_run):
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.add = AsyncMock()
|
||||
|
||||
with patch.dict(sys.modules, {"cognee": mock_cognee}):
|
||||
command = AddCommand()
|
||||
args = argparse.Namespace(data=[], dataset_name="test_dataset")
|
||||
command.execute(args)
|
||||
|
||||
mock_asyncio_run.assert_called_once()
|
||||
assert asyncio.iscoroutine(mock_asyncio_run.call_args[0][0])
|
||||
mock_cognee.add.assert_awaited_once_with(data=[], dataset_name="test_dataset")
|
||||
|
||||
@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", side_effect=_mock_run)
|
||||
def test_search_empty_results(self, mock_asyncio_run):
|
||||
"""Test search command with empty results"""
|
||||
# Mock the cognee module and SearchType
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.search = AsyncMock(return_value=[])
|
||||
mock_search_type = MagicMock()
|
||||
mock_search_type.__getitem__.return_value = "GRAPH_COMPLETION"
|
||||
|
||||
with patch.dict(sys.modules, {"cognee": mock_cognee}):
|
||||
command = SearchCommand()
|
||||
args = argparse.Namespace(
|
||||
query_text="nonexistent query",
|
||||
query_type="GRAPH_COMPLETION",
|
||||
datasets=None,
|
||||
top_k=10,
|
||||
system_prompt=None,
|
||||
output_format="pretty",
|
||||
)
|
||||
|
||||
# Should handle empty results gracefully
|
||||
command.execute(args)
|
||||
|
||||
mock_asyncio_run.return_value = []
|
||||
mock_asyncio_run.assert_called_once()
|
||||
assert asyncio.iscoroutine(mock_asyncio_run.call_args[0][0])
|
||||
mock_cognee.search.assert_awaited_once_with(
|
||||
query_text="nonexistent query",
|
||||
query_type=ANY,
|
||||
datasets=None,
|
||||
top_k=10,
|
||||
system_prompt_path="answer_simple_question.txt",
|
||||
)
|
||||
# verify the enum’s name separately
|
||||
called_enum = mock_cognee.search.await_args.kwargs["query_type"]
|
||||
assert called_enum.name == "GRAPH_COMPLETION"
|
||||
|
||||
@patch("cognee.cli.commands.search_command.asyncio.run", side_effect=_mock_run)
|
||||
def test_search_very_large_top_k(self, mock_asyncio_run):
|
||||
"""Test search command with very large top-k value"""
|
||||
# Mock the cognee module and SearchType
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.search = AsyncMock(return_value=["result1"])
|
||||
mock_search_type = MagicMock()
|
||||
mock_search_type.__getitem__.return_value = "CHUNKS"
|
||||
|
||||
mock_asyncio_run.return_value = ["result1"]
|
||||
|
||||
with patch.dict(sys.modules, {"cognee": mock_cognee}):
|
||||
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",
|
||||
)
|
||||
|
||||
command.execute(args)
|
||||
|
||||
mock_asyncio_run.assert_called_once()
|
||||
assert asyncio.iscoroutine(mock_asyncio_run.call_args[0][0])
|
||||
mock_cognee.search.assert_awaited_once_with(
|
||||
query_text="test query",
|
||||
query_type=ANY,
|
||||
datasets=None,
|
||||
top_k=999999,
|
||||
system_prompt_path="answer_simple_question.txt",
|
||||
)
|
||||
# verify the enum’s name separately
|
||||
called_enum = mock_cognee.search.await_args.kwargs["query_type"]
|
||||
assert called_enum.name == "CHUNKS"
|
||||
|
||||
@patch("builtins.__import__")
|
||||
def test_search_invalid_search_type_enum(self, mock_import):
|
||||
"""Test search command with invalid SearchType enum conversion"""
|
||||
# Mock SearchType to raise KeyError
|
||||
mock_search_type = MagicMock()
|
||||
mock_search_type.__getitem__.side_effect = KeyError("INVALID_TYPE")
|
||||
|
||||
def mock_import_func(name, fromlist=None, *args, **kwargs):
|
||||
if name == "cognee.modules.search.types":
|
||||
module = MagicMock()
|
||||
module.SearchType = mock_search_type
|
||||
return module
|
||||
return MagicMock()
|
||||
|
||||
mock_import.side_effect = mock_import_func
|
||||
|
||||
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",
|
||||
)
|
||||
|
||||
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", side_effect=_mock_run)
|
||||
def test_search_results_with_none_values(self, mock_asyncio_run):
|
||||
"""Test search command when results contain None values"""
|
||||
# Mock the cognee module and SearchType
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.search = AsyncMock(return_value=[None, "valid result", None])
|
||||
mock_search_type = MagicMock()
|
||||
mock_search_type.__getitem__.return_value = "CHUNKS"
|
||||
|
||||
with patch.dict(sys.modules, {"cognee": mock_cognee}):
|
||||
command = SearchCommand()
|
||||
args = argparse.Namespace(
|
||||
query_text="test query",
|
||||
query_type="CHUNKS",
|
||||
datasets=None,
|
||||
top_k=10,
|
||||
system_prompt=None,
|
||||
output_format="pretty",
|
||||
)
|
||||
|
||||
# Should handle None values gracefully
|
||||
command.execute(args)
|
||||
|
||||
mock_asyncio_run.assert_called_once()
|
||||
assert asyncio.iscoroutine(mock_asyncio_run.call_args[0][0])
|
||||
mock_cognee.search.assert_awaited_once_with(
|
||||
query_text="test query",
|
||||
query_type=ANY,
|
||||
datasets=None,
|
||||
top_k=10,
|
||||
system_prompt_path="answer_simple_question.txt",
|
||||
)
|
||||
# verify the enum’s name separately
|
||||
called_enum = mock_cognee.search.await_args.kwargs["query_type"]
|
||||
assert called_enum.name == "CHUNKS"
|
||||
|
||||
|
||||
class TestCognifyCommandEdgeCases:
|
||||
"""Test edge cases for CognifyCommand"""
|
||||
|
||||
@patch("cognee.cli.commands.cognify_command.asyncio.run", side_effect=_mock_run)
|
||||
def test_cognify_invalid_chunk_size(self, mock_asyncio_run):
|
||||
"""Test cognify command with invalid chunk size"""
|
||||
# Mock the cognee module
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.cognify = AsyncMock()
|
||||
|
||||
with patch.dict(sys.modules, {"cognee": mock_cognee}):
|
||||
command = CognifyCommand()
|
||||
args = argparse.Namespace(
|
||||
datasets=None,
|
||||
chunk_size=-100, # Invalid negative chunk size
|
||||
ontology_file=None,
|
||||
chunker="TextChunker",
|
||||
background=False,
|
||||
verbose=False,
|
||||
)
|
||||
|
||||
# Should pass the invalid value to cognify and let it handle the validation
|
||||
command.execute(args)
|
||||
|
||||
mock_asyncio_run.assert_called_once()
|
||||
assert asyncio.iscoroutine(mock_asyncio_run.call_args[0][0])
|
||||
from cognee.modules.chunking import TextChunker
|
||||
|
||||
mock_cognee.cognify.assert_awaited_once_with(
|
||||
datasets=None,
|
||||
chunk_size=-100,
|
||||
ontology_file_path=None,
|
||||
chunker=TextChunker,
|
||||
run_in_background=False,
|
||||
)
|
||||
|
||||
@patch("cognee.cli.commands.cognify_command.asyncio.run", side_effect=_mock_run)
|
||||
def test_cognify_nonexistent_ontology_file(self, mock_asyncio_run):
|
||||
"""Test cognify command with nonexistent ontology file"""
|
||||
# Mock the cognee module
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.cognify = AsyncMock()
|
||||
|
||||
with patch.dict(sys.modules, {"cognee": mock_cognee}):
|
||||
command = CognifyCommand()
|
||||
args = argparse.Namespace(
|
||||
datasets=None,
|
||||
chunk_size=None,
|
||||
ontology_file="/nonexistent/path/ontology.owl",
|
||||
chunker="TextChunker",
|
||||
background=False,
|
||||
verbose=False,
|
||||
)
|
||||
|
||||
# Should pass the path to cognify and let it handle file validation
|
||||
command.execute(args)
|
||||
|
||||
mock_asyncio_run.assert_called_once()
|
||||
assert asyncio.iscoroutine(mock_asyncio_run.call_args[0][0])
|
||||
from cognee.modules.chunking import TextChunker
|
||||
|
||||
mock_cognee.cognify.assert_awaited_once_with(
|
||||
datasets=None,
|
||||
chunk_size=None,
|
||||
ontology_file_path="/nonexistent/path/ontology.owl",
|
||||
chunker=TextChunker,
|
||||
run_in_background=False,
|
||||
)
|
||||
|
||||
@patch("cognee.cli.commands.cognify_command.asyncio.run")
|
||||
def test_cognify_langchain_chunker_import_error(self, mock_asyncio_run):
|
||||
"""Test cognify command when LangchainChunker import fails"""
|
||||
# Mock the cognee module
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.cognify = AsyncMock()
|
||||
|
||||
def mock_import_func(name, fromlist=None, *args, **kwargs):
|
||||
if name == "cognee":
|
||||
return mock_cognee
|
||||
elif name == "cognee.modules.chunking" and fromlist and "LangchainChunker" in fromlist:
|
||||
raise ImportError("LangchainChunker not available")
|
||||
elif name == "cognee.modules.chunking":
|
||||
module = MagicMock()
|
||||
module.TextChunker = MagicMock()
|
||||
return module
|
||||
return MagicMock()
|
||||
|
||||
with (
|
||||
patch("builtins.__import__", side_effect=mock_import_func),
|
||||
patch.dict(sys.modules, {"cognee": mock_cognee}),
|
||||
):
|
||||
command = CognifyCommand()
|
||||
args = argparse.Namespace(
|
||||
datasets=None,
|
||||
chunk_size=None,
|
||||
ontology_file=None,
|
||||
chunker="LangchainChunker",
|
||||
background=False,
|
||||
verbose=True,
|
||||
)
|
||||
|
||||
# Should fall back to TextChunker and show warning
|
||||
command.execute(args)
|
||||
|
||||
mock_asyncio_run.assert_called_once()
|
||||
assert asyncio.iscoroutine(mock_asyncio_run.call_args[0][0])
|
||||
|
||||
@patch("cognee.cli.commands.cognify_command.asyncio.run", side_effect=_mock_run)
|
||||
def test_cognify_empty_datasets_list(self, mock_asyncio_run):
|
||||
"""Test cognify command with nonexistent ontology file"""
|
||||
# Mock the cognee module
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.cognify = AsyncMock()
|
||||
|
||||
with patch.dict(sys.modules, {"cognee": mock_cognee}):
|
||||
command = CognifyCommand()
|
||||
args = argparse.Namespace(
|
||||
datasets=[],
|
||||
chunk_size=None,
|
||||
ontology_file=None,
|
||||
chunker="TextChunker",
|
||||
background=False,
|
||||
verbose=False,
|
||||
)
|
||||
|
||||
command.execute(args)
|
||||
|
||||
mock_asyncio_run.assert_called_once()
|
||||
assert asyncio.iscoroutine(mock_asyncio_run.call_args[0][0])
|
||||
from cognee.modules.chunking import TextChunker
|
||||
|
||||
mock_cognee.cognify.assert_awaited_once_with(
|
||||
datasets=None,
|
||||
chunk_size=None,
|
||||
ontology_file_path=None,
|
||||
chunker=TextChunker,
|
||||
run_in_background=False,
|
||||
)
|
||||
|
||||
|
||||
class TestDeleteCommandEdgeCases:
|
||||
"""Test edge cases for DeleteCommand"""
|
||||
|
||||
@patch("cognee.cli.commands.delete_command.fmt.confirm")
|
||||
@patch("cognee.cli.commands.delete_command.asyncio.run", side_effect=_mock_run)
|
||||
def test_delete_all_with_user_id(self, mock_asyncio_run, mock_confirm):
|
||||
"""Test delete command with both --all and --user-id"""
|
||||
# Mock the cognee module
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.delete = AsyncMock()
|
||||
|
||||
with patch.dict(sys.modules, {"cognee": mock_cognee}):
|
||||
command = DeleteCommand()
|
||||
args = argparse.Namespace(dataset_name=None, user_id="test_user", all=True, force=False)
|
||||
|
||||
mock_confirm.return_value = True
|
||||
|
||||
# Should handle both flags being set
|
||||
command.execute(args)
|
||||
|
||||
mock_confirm.assert_called_once_with("Delete ALL data from cognee?")
|
||||
mock_asyncio_run.assert_called_once()
|
||||
assert asyncio.iscoroutine(mock_asyncio_run.call_args[0][0])
|
||||
mock_cognee.delete.assert_awaited_once_with(dataset_name=None, user_id="test_user")
|
||||
|
||||
@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")
|
||||
def test_delete_async_exception_handling(self, 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 asyncio.run to raise exception directly
|
||||
mock_asyncio_run.side_effect = ValueError("Database connection failed")
|
||||
|
||||
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("builtins.__import__")
|
||||
def test_config_get_nonexistent_key(self, mock_import):
|
||||
"""Test config get with nonexistent key"""
|
||||
# Mock config.get to raise exception for nonexistent key
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.config.get = MagicMock(side_effect=KeyError("Key not found"))
|
||||
mock_import.return_value = mock_cognee
|
||||
|
||||
command = ConfigCommand()
|
||||
args = argparse.Namespace(config_action="get", key="nonexistent_key")
|
||||
|
||||
# Should handle the exception gracefully
|
||||
command.execute(args)
|
||||
mock_cognee.config.get.assert_called_once_with("nonexistent_key")
|
||||
|
||||
@patch("builtins.__import__")
|
||||
def test_config_set_complex_json_value(self, mock_import):
|
||||
"""Test config set with complex JSON value"""
|
||||
# Mock the cognee module
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.config.set = MagicMock()
|
||||
mock_import.return_value = mock_cognee
|
||||
|
||||
command = ConfigCommand()
|
||||
complex_json = '{"nested": {"key": "value"}, "array": [1, 2, 3]}'
|
||||
complex_json_expected_value = {"nested": {"key": "value"}, "array": [1, 2, 3]}
|
||||
args = argparse.Namespace(config_action="set", key="complex_config", value=complex_json)
|
||||
|
||||
command.execute(args)
|
||||
mock_cognee.config.set.assert_called_once_with(
|
||||
"complex_config", complex_json_expected_value
|
||||
)
|
||||
|
||||
@patch("builtins.__import__")
|
||||
def test_config_set_invalid_json_value(self, mock_import):
|
||||
"""Test config set with invalid JSON value"""
|
||||
# Mock the cognee module
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.config.set = MagicMock()
|
||||
mock_import.return_value = mock_cognee
|
||||
|
||||
command = ConfigCommand()
|
||||
invalid_json = '{"invalid": json}'
|
||||
args = argparse.Namespace(config_action="set", key="test_key", value=invalid_json)
|
||||
|
||||
command.execute(args)
|
||||
mock_cognee.config.set.assert_called_once_with("test_key", invalid_json)
|
||||
|
||||
@patch("cognee.cli.commands.config_command.fmt.confirm")
|
||||
def test_config_unset_unknown_key(self, mock_confirm):
|
||||
"""Test config unset with unknown key"""
|
||||
# Mock the cognee module
|
||||
mock_cognee = MagicMock()
|
||||
|
||||
with patch.dict(sys.modules, {"cognee": mock_cognee}):
|
||||
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("builtins.__import__")
|
||||
def test_config_unset_method_not_found(self, mock_import):
|
||||
"""Test config unset when method doesn't exist on config object"""
|
||||
# Mock config object without the expected method
|
||||
mock_cognee = MagicMock()
|
||||
mock_cognee.config = MagicMock()
|
||||
# Don't set the set_llm_provider method
|
||||
mock_import.return_value = mock_cognee
|
||||
|
||||
command = ConfigCommand()
|
||||
args = argparse.Namespace(config_action="unset", key="llm_provider", force=True)
|
||||
|
||||
# Should handle AttributeError gracefully
|
||||
command.execute(args)
|
||||
mock_cognee.config.unset.assert_not_called()
|
||||
|
||||
def test_config_invalid_subcommand(self):
|
||||
"""Test config command with invalid subcommand"""
|
||||
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
173
cognee/tests/unit/cli/test_cli_main.py
Normal 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.")
|
||||
62
cognee/tests/unit/cli/test_cli_runner.py
Normal file
62
cognee/tests/unit/cli/test_cli_runner.py
Normal file
|
|
@ -0,0 +1,62 @@
|
|||
"""
|
||||
Test runner and utilities for CLI tests.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import sys
|
||||
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)
|
||||
127
cognee/tests/unit/cli/test_cli_utils.py
Normal file
127
cognee/tests/unit/cli/test_cli_utils.py
Normal file
|
|
@ -0,0 +1,127 @@
|
|||
"""
|
||||
Tests for CLI utility functions and helper modules.
|
||||
"""
|
||||
|
||||
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,
|
||||
)
|
||||
from cognee.cli._cognee import _discover_commands
|
||||
|
||||
|
||||
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://")
|
||||
assert "cognee.ai" in DEFAULT_DOCS_URL
|
||||
|
||||
def test_command_descriptions_complete(self):
|
||||
"""Test that all expected commands have descriptions"""
|
||||
commands = _discover_commands()
|
||||
assert len(commands) > 0
|
||||
|
||||
expected_commands = []
|
||||
for command_class in commands:
|
||||
command = command_class()
|
||||
expected_commands.append(command.command_string)
|
||||
|
||||
for command in expected_commands:
|
||||
assert command in COMMAND_DESCRIPTIONS
|
||||
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_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() is True
|
||||
|
||||
# Reset for other tests
|
||||
debug._debug_enabled = False
|
||||
78
github-secure-open-source-graduation.md
Normal file
78
github-secure-open-source-graduation.md
Normal file
|
|
@ -0,0 +1,78 @@
|
|||
# cognee Graduates from GitHub Secure Open Source Program
|
||||
|
||||
*Building Trust and Security in AI Memory Systems*
|
||||
|
||||
We're excited to announce that **cognee** has successfully graduated from the GitHub Secure Open Source Program! This milestone reflects our commitment to maintaining the highest standards of security and reliability in open source AI infrastructure.
|
||||
|
||||
## What is cognee?
|
||||
|
||||
cognee is an open source library that provides **memory for AI agents in just 5 lines of code**. It transforms raw data into structured knowledge graphs through our innovative ECL (Extract, Cognify, Load) pipeline, enabling AI systems to build dynamic memory that goes far beyond traditional RAG systems.
|
||||
|
||||
### Key Features:
|
||||
- **Interconnected Knowledge**: Links conversations, documents, images, and audio transcriptions
|
||||
- **Scalable Architecture**: Loads data to graph and vector databases using only Pydantic
|
||||
- **30+ Data Sources**: Manipulates data while ingesting from diverse sources
|
||||
- **Developer-Friendly**: Reduces complexity and cost compared to traditional RAG implementations
|
||||
|
||||
## GitHub Secure Open Source Program Achievement
|
||||
|
||||
The GitHub Secure Open Source Program helps maintainers adopt security best practices and ensures that critical open source projects meet enterprise-grade security standards. Our graduation demonstrates that cognee has successfully implemented:
|
||||
|
||||
- **Security-first development practices**
|
||||
- **Comprehensive vulnerability management**
|
||||
- **Secure dependency management**
|
||||
- **Code quality and review processes**
|
||||
- **Community safety guidelines**
|
||||
|
||||
## Why This Matters for AI Development
|
||||
|
||||
As AI systems become more prevalent in production environments, security becomes paramount. cognee's graduation from this program means developers can confidently build AI memory systems knowing they're using infrastructure that meets rigorous security standards.
|
||||
|
||||
### Benefits for Our Community:
|
||||
- **Enterprise Adoption**: Companies can deploy cognee with confidence in security-sensitive environments
|
||||
- **Vulnerability Response**: Our security practices ensure rapid identification and resolution of potential issues
|
||||
- **Supply Chain Security**: Dependencies are carefully managed and regularly audited
|
||||
- **Trust & Transparency**: Open source development with security-first principles
|
||||
|
||||
## What's Next?
|
||||
|
||||
With over **5,000 GitHub stars** and a growing community of developers, cognee continues to evolve. We recently launched **Cogwit beta** - our fully-hosted AI Memory platform, and our [research paper](https://arxiv.org/abs/2505.24478) demonstrates the effectiveness of our approach.
|
||||
|
||||
Our commitment to security doesn't end with graduation. We'll continue following best practices and contributing to the broader conversation about secure AI infrastructure.
|
||||
|
||||
## Get Started Today
|
||||
|
||||
Ready to add intelligent memory to your AI applications? Get started with cognee:
|
||||
|
||||
```python
|
||||
import cognee
|
||||
import asyncio
|
||||
|
||||
async def main():
|
||||
# Add your data
|
||||
await cognee.add("Your document content here")
|
||||
|
||||
# Transform into knowledge graph
|
||||
await cognee.cognify()
|
||||
|
||||
# Query intelligently
|
||||
results = await cognee.search("What insights can you find?")
|
||||
|
||||
for result in results:
|
||||
print(result)
|
||||
|
||||
asyncio.run(main())
|
||||
```
|
||||
|
||||
## Join Our Community
|
||||
|
||||
- 🌟 [Star us on GitHub](https://github.com/topoteretes/cognee)
|
||||
- 💬 [Join our Discord](https://discord.gg/NQPKmU5CCg)
|
||||
- 📖 [Read our documentation](https://docs.cognee.ai/)
|
||||
- 🚀 [Try Cogwit beta](https://platform.cognee.ai/)
|
||||
|
||||
The future of AI memory is secure, scalable, and open source. We're grateful for the GitHub team's support and excited to continue building the infrastructure that powers the next generation of intelligent applications.
|
||||
|
||||
---
|
||||
|
||||
*About cognee: We're building the memory layer for AI agents, enabling them to learn, remember, and reason across conversations and data sources. Our open source approach ensures that advanced AI memory capabilities are accessible to developers worldwide.*
|
||||
7
poetry.lock
generated
7
poetry.lock
generated
|
|
@ -1,4 +1,4 @@
|
|||
# This file is automatically @generated by Poetry 2.1.3 and should not be changed by hand.
|
||||
# This file is automatically @generated by Poetry 2.1.2 and should not be changed by hand.
|
||||
|
||||
[[package]]
|
||||
name = "aiobotocore"
|
||||
|
|
@ -4698,12 +4698,10 @@ files = [
|
|||
{file = "lxml-6.0.0-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:219e0431ea8006e15005767f0351e3f7f9143e793e58519dc97fe9e07fae5563"},
|
||||
{file = "lxml-6.0.0-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:bd5913b4972681ffc9718bc2d4c53cde39ef81415e1671ff93e9aa30b46595e7"},
|
||||
{file = "lxml-6.0.0-cp312-cp312-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:390240baeb9f415a82eefc2e13285016f9c8b5ad71ec80574ae8fa9605093cd7"},
|
||||
{file = "lxml-6.0.0-cp312-cp312-manylinux_2_27_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:d6e200909a119626744dd81bae409fc44134389e03fbf1d68ed2a55a2fb10991"},
|
||||
{file = "lxml-6.0.0-cp312-cp312-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:ca50bd612438258a91b5b3788c6621c1f05c8c478e7951899f492be42defc0da"},
|
||||
{file = "lxml-6.0.0-cp312-cp312-manylinux_2_31_armv7l.whl", hash = "sha256:c24b8efd9c0f62bad0439283c2c795ef916c5a6b75f03c17799775c7ae3c0c9e"},
|
||||
{file = "lxml-6.0.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:afd27d8629ae94c5d863e32ab0e1d5590371d296b87dae0a751fb22bf3685741"},
|
||||
{file = "lxml-6.0.0-cp312-cp312-musllinux_1_2_armv7l.whl", hash = "sha256:54c4855eabd9fc29707d30141be99e5cd1102e7d2258d2892314cf4c110726c3"},
|
||||
{file = "lxml-6.0.0-cp312-cp312-musllinux_1_2_ppc64le.whl", hash = "sha256:c907516d49f77f6cd8ead1322198bdfd902003c3c330c77a1c5f3cc32a0e4d16"},
|
||||
{file = "lxml-6.0.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:36531f81c8214e293097cd2b7873f178997dae33d3667caaae8bdfb9666b76c0"},
|
||||
{file = "lxml-6.0.0-cp312-cp312-win32.whl", hash = "sha256:690b20e3388a7ec98e899fd54c924e50ba6693874aa65ef9cb53de7f7de9d64a"},
|
||||
{file = "lxml-6.0.0-cp312-cp312-win_amd64.whl", hash = "sha256:310b719b695b3dd442cdfbbe64936b2f2e231bb91d998e99e6f0daf991a3eba3"},
|
||||
|
|
@ -4714,12 +4712,10 @@ files = [
|
|||
{file = "lxml-6.0.0-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:d18a25b19ca7307045581b18b3ec9ead2b1db5ccd8719c291f0cd0a5cec6cb81"},
|
||||
{file = "lxml-6.0.0-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:d4f0c66df4386b75d2ab1e20a489f30dc7fd9a06a896d64980541506086be1f1"},
|
||||
{file = "lxml-6.0.0-cp313-cp313-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:9f4b481b6cc3a897adb4279216695150bbe7a44c03daba3c894f49d2037e0a24"},
|
||||
{file = "lxml-6.0.0-cp313-cp313-manylinux_2_27_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:8a78d6c9168f5bcb20971bf3329c2b83078611fbe1f807baadc64afc70523b3a"},
|
||||
{file = "lxml-6.0.0-cp313-cp313-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:2ae06fbab4f1bb7db4f7c8ca9897dc8db4447d1a2b9bee78474ad403437bcc29"},
|
||||
{file = "lxml-6.0.0-cp313-cp313-manylinux_2_31_armv7l.whl", hash = "sha256:1fa377b827ca2023244a06554c6e7dc6828a10aaf74ca41965c5d8a4925aebb4"},
|
||||
{file = "lxml-6.0.0-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:1676b56d48048a62ef77a250428d1f31f610763636e0784ba67a9740823988ca"},
|
||||
{file = "lxml-6.0.0-cp313-cp313-musllinux_1_2_armv7l.whl", hash = "sha256:0e32698462aacc5c1cf6bdfebc9c781821b7e74c79f13e5ffc8bfe27c42b1abf"},
|
||||
{file = "lxml-6.0.0-cp313-cp313-musllinux_1_2_ppc64le.whl", hash = "sha256:4d6036c3a296707357efb375cfc24bb64cd955b9ec731abf11ebb1e40063949f"},
|
||||
{file = "lxml-6.0.0-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:7488a43033c958637b1a08cddc9188eb06d3ad36582cebc7d4815980b47e27ef"},
|
||||
{file = "lxml-6.0.0-cp313-cp313-win32.whl", hash = "sha256:5fcd7d3b1d8ecb91445bd71b9c88bdbeae528fefee4f379895becfc72298d181"},
|
||||
{file = "lxml-6.0.0-cp313-cp313-win_amd64.whl", hash = "sha256:2f34687222b78fff795feeb799a7d44eca2477c3d9d3a46ce17d51a4f383e32e"},
|
||||
|
|
@ -7302,7 +7298,6 @@ files = [
|
|||
{file = "psycopg2-2.9.10-cp311-cp311-win_amd64.whl", hash = "sha256:0435034157049f6846e95103bd8f5a668788dd913a7c30162ca9503fdf542cb4"},
|
||||
{file = "psycopg2-2.9.10-cp312-cp312-win32.whl", hash = "sha256:65a63d7ab0e067e2cdb3cf266de39663203d38d6a8ed97f5ca0cb315c73fe067"},
|
||||
{file = "psycopg2-2.9.10-cp312-cp312-win_amd64.whl", hash = "sha256:4a579d6243da40a7b3182e0430493dbd55950c493d8c68f4eec0b302f6bbf20e"},
|
||||
{file = "psycopg2-2.9.10-cp313-cp313-win_amd64.whl", hash = "sha256:91fd603a2155da8d0cfcdbf8ab24a2d54bca72795b90d2a3ed2b6da8d979dee2"},
|
||||
{file = "psycopg2-2.9.10-cp39-cp39-win32.whl", hash = "sha256:9d5b3b94b79a844a986d029eee38998232451119ad653aea42bb9220a8c5066b"},
|
||||
{file = "psycopg2-2.9.10-cp39-cp39-win_amd64.whl", hash = "sha256:88138c8dedcbfa96408023ea2b0c369eda40fe5d75002c0964c78f46f11fa442"},
|
||||
{file = "psycopg2-2.9.10.tar.gz", hash = "sha256:12ec0b40b0273f95296233e8750441339298e6a572f7039da5b260e3c8b60e11"},
|
||||
|
|
|
|||
|
|
@ -186,4 +186,4 @@ exclude = [
|
|||
]
|
||||
|
||||
[tool.ruff.lint]
|
||||
ignore = ["F401"]
|
||||
ignore = ["F401"]
|
||||
Loading…
Add table
Reference in a new issue