brent.edwards
321ab18a37
CI / lint (pull_request) Successful in 16s
CI / typecheck (pull_request) Successful in 27s
CI / security (pull_request) Failing after 15s
CI / quality (pull_request) Successful in 15s
CI / integration_tests (pull_request) Failing after 2m42s
CI / build (pull_request) Successful in 15s
CI / unit_tests (pull_request) Successful in 6m27s
CI / docker (pull_request) Has been skipped
CI / coverage (pull_request) Successful in 5m11s
- Fix Rich Console line-wrapping breaking assertions in
context_unit_tests_steps.py: collapse newlines before checking for
filenames and overflow summaries (CI temp paths exceed 80 columns)
- Fix features.mocks import failure in database_integration.robot:
replace hardcoded sys.path '/app' with portable ${CURDIR}/..
- Fix --load-context help text assertions in load_context_test.robot:
merge stderr into stdout via stderr=STDOUT and remove duplicate test
- Add standalone dead_code nox session running vulture directly
- Rename security nox session to security_scan to match CI references
- Restore --exclude discovery to integration_tests nox session (lost
during merge conflict resolution)
1814 lines
68 KiB
Python
1814 lines
68 KiB
Python
"""Step definitions for context unit tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import tempfile
|
|
from contextlib import contextmanager
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import given, then, when
|
|
from typer.testing import CliRunner
|
|
|
|
from cleveragents.cli.commands import context as context_app
|
|
from cleveragents.core.exceptions import CleverAgentsError, FileSystemError
|
|
from cleveragents.domain.models.core import Context, ContextType
|
|
|
|
|
|
def strip_ansi_codes(text):
|
|
"""Remove ANSI escape sequences from text."""
|
|
# This regex matches ANSI escape sequences
|
|
ansi_escape = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
|
|
return ansi_escape.sub("", text)
|
|
|
|
|
|
@contextmanager
|
|
def patched_container(context, project):
|
|
"""Patch the CleverAgents container to provide mocked services."""
|
|
with patch(
|
|
"cleveragents.application.container.get_container"
|
|
) as mock_get_container:
|
|
container = MagicMock()
|
|
container.context_service.return_value = context.mock_service
|
|
project_service = MagicMock()
|
|
project_service.get_current_project.return_value = project
|
|
container.project_service.return_value = project_service
|
|
mock_get_container.return_value = container
|
|
yield container
|
|
|
|
|
|
@given("a mock context service is configured")
|
|
def step_mock_context_service(context):
|
|
"""Set up a mock context service."""
|
|
context.mock_service = MagicMock()
|
|
context.temp_dir = tempfile.mkdtemp()
|
|
|
|
|
|
@given("a mock context service that raises FileSystemError")
|
|
def step_mock_filesystem_error_service(context):
|
|
"""Set up mock that raises FileSystemError."""
|
|
context.mock_service = MagicMock()
|
|
context.mock_error = FileSystemError("Filesystem access denied")
|
|
context.temp_dir = tempfile.mkdtemp()
|
|
|
|
|
|
@given("a mock context service that raises CleverAgentsError")
|
|
def step_mock_cleveragents_error_service(context):
|
|
"""Set up mock that raises CleverAgentsError."""
|
|
context.mock_service = MagicMock()
|
|
context.mock_error = CleverAgentsError("Context operation failed")
|
|
context.temp_dir = tempfile.mkdtemp()
|
|
|
|
|
|
@given("a mock context service with remove capability")
|
|
def step_mock_remove_service(context):
|
|
"""Set up mock for remove operations."""
|
|
context.mock_service = MagicMock()
|
|
# The remove function is called twice (once for each file)
|
|
# Each call returns 1, so total removed is 2
|
|
context.mock_service.remove_from_context.side_effect = [1, 1]
|
|
|
|
|
|
@given("a mock context service returning zero removals")
|
|
def step_mock_zero_removals(context):
|
|
"""Set up mock that removes nothing."""
|
|
context.mock_service = MagicMock()
|
|
# Called twice, each returns 0
|
|
context.mock_service.remove_from_context.side_effect = [0, 0]
|
|
|
|
|
|
@given("a mock context service with already tracked files")
|
|
def step_mock_already_tracked(context):
|
|
"""Set up mock that reports already tracked files."""
|
|
context.mock_service = MagicMock()
|
|
context.temp_dir = tempfile.mkdtemp()
|
|
context.already_paths = [
|
|
Path(context.temp_dir) / f"existing_{i}.txt" for i in range(12)
|
|
]
|
|
context.mock_service.add_to_context.return_value = ([], context.already_paths)
|
|
|
|
|
|
@given("a mock context service with a few already tracked files")
|
|
def step_mock_already_tracked_few(context):
|
|
"""Set up mock with a small number of already tracked files."""
|
|
context.mock_service = MagicMock()
|
|
context.temp_dir = tempfile.mkdtemp()
|
|
context.already_paths = [
|
|
Path(context.temp_dir) / f"existing_{i}.txt" for i in range(3)
|
|
]
|
|
context.mock_service.add_to_context.return_value = ([], context.already_paths)
|
|
|
|
|
|
@given("a mock context service with partial removals")
|
|
def step_mock_partial_removals(context):
|
|
"""Set up mock that removes some but not all files."""
|
|
context.mock_service = MagicMock()
|
|
context.mock_service.remove_from_context.side_effect = [1, 0]
|
|
|
|
|
|
@given("a mock context service with files")
|
|
def step_mock_with_files(context):
|
|
"""Set up mock with file list."""
|
|
context.mock_service = MagicMock()
|
|
context.mock_files = [
|
|
{"path": "/src/main.py", "size": "1.2 KB", "added_at": "2024-01-01"},
|
|
{"path": "/src/test.py", "size": "0.8 KB", "added_at": "2024-01-02"},
|
|
]
|
|
context.mock_service.list_context.return_value = context.mock_files
|
|
|
|
|
|
@given("a mock context service with numeric file sizes")
|
|
def step_mock_numeric_sizes(context):
|
|
"""Set up mock with numeric size values."""
|
|
context.mock_service = MagicMock()
|
|
context.mock_files = [
|
|
{"path": "/src/data.bin", "size": 0, "added_at": "2024-02-01"},
|
|
{"path": "/src/large.bin", "size": 2048, "added_at": "2024-02-02"},
|
|
]
|
|
context.mock_service.list_context.return_value = context.mock_files
|
|
|
|
|
|
@given("a mock context service with no files")
|
|
def step_mock_no_files(context):
|
|
"""Set up mock with empty context."""
|
|
context.mock_service = MagicMock()
|
|
context.mock_service.list_context.return_value = []
|
|
|
|
|
|
@given("a mock context service with file content")
|
|
def step_mock_with_content(context):
|
|
"""Set up mock with file content."""
|
|
context.mock_service = MagicMock()
|
|
context.mock_content = "File content here\nLine 2"
|
|
context.mock_service.get_context_content.return_value = context.mock_content
|
|
|
|
|
|
@given("a mock context service returning no content")
|
|
def step_mock_no_content(context):
|
|
"""Set up mock returning no content."""
|
|
context.mock_service = MagicMock()
|
|
context.mock_service.get_context_content.return_value = None
|
|
|
|
|
|
@given("a mock context service with multiple files")
|
|
def step_mock_multiple_files(context):
|
|
"""Set up mock with multiple files."""
|
|
context.mock_service = MagicMock()
|
|
context.mock_files = [
|
|
{"path": f"/file{i}.txt", "size_bytes": i * 1000} for i in range(1, 6)
|
|
]
|
|
context.mock_service.list_context.return_value = context.mock_files
|
|
|
|
|
|
@given("a mock context service with single file content")
|
|
def step_mock_single_file_content(context):
|
|
"""Set up mock with a single file containing content."""
|
|
context.mock_service = MagicMock()
|
|
context.single_file_entry = {
|
|
"path": "docs/report.txt",
|
|
"content": "Generated summary",
|
|
"size": 512,
|
|
"added_at": "2024-03-01",
|
|
}
|
|
context.mock_service.list_context.return_value = [context.single_file_entry]
|
|
|
|
|
|
@given("a mock context service with single file lacking content")
|
|
def step_mock_single_file_blank(context):
|
|
"""Set up mock with a single file but no stored content."""
|
|
context.mock_service = MagicMock()
|
|
context.single_file_entry = {
|
|
"path": "docs/empty.txt",
|
|
"content": None,
|
|
"size": 128,
|
|
"added_at": "2024-03-02",
|
|
}
|
|
context.mock_service.list_context.return_value = [context.single_file_entry]
|
|
|
|
|
|
@given("a mock context service with clearable files")
|
|
def step_mock_clearable(context):
|
|
"""Set up mock for clear operations."""
|
|
context.mock_service = MagicMock()
|
|
context.mock_files = [{"path": f"/file{i}.txt"} for i in range(3)]
|
|
context.mock_service.list_context.return_value = context.mock_files
|
|
|
|
|
|
@when("the context add command is executed with valid paths")
|
|
def step_exec_context_add_valid(context):
|
|
"""Execute context add with valid paths."""
|
|
runner = CliRunner()
|
|
|
|
# Create temp files
|
|
test_file = Path(context.temp_dir) / "test.txt"
|
|
test_file.write_text("content")
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
# add_to_context should return a tuple: (added_files, already_in_context)
|
|
context.mock_service.add_to_context.return_value = ([test_file], [])
|
|
mock_container.return_value.context_service.return_value = context.mock_service
|
|
|
|
result = runner.invoke(context_app.app, ["add", str(test_file)])
|
|
context.result = result
|
|
|
|
|
|
@when("the context add command is executed with non-existent path")
|
|
def step_exec_context_add_nonexistent(context):
|
|
"""Execute context add with missing path."""
|
|
runner = CliRunner()
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_container.return_value.context_service.return_value = context.mock_service
|
|
|
|
result = runner.invoke(context_app.app, ["add", "/nonexistent/path.txt"])
|
|
context.result = result
|
|
|
|
|
|
@when("the context add command is executed without an active project")
|
|
def step_exec_context_add_no_project(context):
|
|
"""Execute context add when no project is configured."""
|
|
runner = CliRunner()
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
container = mock_container.return_value
|
|
container.context_service.return_value = context.mock_service
|
|
project_service = MagicMock()
|
|
project_service.get_current_project.return_value = None
|
|
container.project_service.return_value = project_service
|
|
|
|
result = runner.invoke(context_app.app, ["add", "placeholder.txt"])
|
|
context.result = result
|
|
|
|
|
|
@when("the context add command processes already tracked paths")
|
|
def step_exec_context_add_already(context):
|
|
"""Execute context add when files are already tracked."""
|
|
runner = CliRunner()
|
|
|
|
test_file = Path(context.temp_dir) / "tracked.txt"
|
|
test_file.write_text("content")
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
container = mock_container.return_value
|
|
container.context_service.return_value = context.mock_service
|
|
project_service = MagicMock()
|
|
project_service.get_current_project.return_value = object()
|
|
container.project_service.return_value = project_service
|
|
|
|
result = runner.invoke(context_app.app, ["add", str(test_file)])
|
|
context.result = result
|
|
|
|
|
|
@when("the context add command is executed and fails")
|
|
def step_exec_context_add_fails(context):
|
|
"""Execute context add that fails."""
|
|
runner = CliRunner()
|
|
|
|
test_file = Path(context.temp_dir) / "test.txt"
|
|
test_file.write_text("content")
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
context.mock_service.add_to_context.side_effect = context.mock_error
|
|
mock_container.return_value.context_service.return_value = context.mock_service
|
|
|
|
result = runner.invoke(context_app.app, ["add", str(test_file)])
|
|
context.result = result
|
|
|
|
|
|
@when("the context load alias is executed")
|
|
def step_exec_context_load(context):
|
|
"""Execute context load."""
|
|
runner = CliRunner()
|
|
|
|
test_file = Path(context.temp_dir) / "test.txt"
|
|
test_file.write_text("content")
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
# add_to_context should return a tuple: (added_files, already_in_context)
|
|
context.mock_service.add_to_context.return_value = ([test_file], [])
|
|
mock_container.return_value.context_service.return_value = context.mock_service
|
|
|
|
result = runner.invoke(context_app.app, ["load", str(test_file)])
|
|
context.result = result
|
|
|
|
|
|
@when("the context remove command is executed")
|
|
def step_exec_context_remove(context):
|
|
"""Execute context remove."""
|
|
runner = CliRunner()
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
container = mock_container.return_value
|
|
container.context_service.return_value = context.mock_service
|
|
project_service = MagicMock()
|
|
project_service.get_current_project.return_value = object()
|
|
container.project_service.return_value = project_service
|
|
|
|
result = runner.invoke(context_app.app, ["rm", "/file1.txt", "/file2.txt"])
|
|
context.result = result
|
|
|
|
|
|
@when("the context remove command is executed without an active project")
|
|
def step_exec_context_remove_no_project(context):
|
|
"""Execute context remove when no project is configured."""
|
|
runner = CliRunner()
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
container = mock_container.return_value
|
|
container.context_service.return_value = context.mock_service
|
|
project_service = MagicMock()
|
|
project_service.get_current_project.return_value = None
|
|
container.project_service.return_value = project_service
|
|
|
|
result = runner.invoke(context_app.app, ["rm", "/file1.txt"])
|
|
context.result = result
|
|
|
|
|
|
@when("the context remove command is executed for relative paths")
|
|
def step_exec_context_remove_relative(context):
|
|
"""Execute context remove for relative file names."""
|
|
runner = CliRunner()
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
container = mock_container.return_value
|
|
container.context_service.return_value = context.mock_service
|
|
project_service = MagicMock()
|
|
project_service.get_current_project.return_value = object()
|
|
container.project_service.return_value = project_service
|
|
|
|
result = runner.invoke(context_app.app, ["rm", "file1.txt", "file2.txt"])
|
|
context.result = result
|
|
|
|
|
|
@when("the context remove command is executed with service failure")
|
|
def step_exec_context_remove_failure(context):
|
|
"""Execute context remove that raises CleverAgentsError."""
|
|
runner = CliRunner()
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
container = mock_container.return_value
|
|
container.context_service.return_value = context.mock_service
|
|
context.mock_service.remove_from_context.side_effect = context.mock_error
|
|
project_service = MagicMock()
|
|
project_service.get_current_project.return_value = object()
|
|
container.project_service.return_value = project_service
|
|
|
|
result = runner.invoke(context_app.app, ["rm", "/file1.txt"])
|
|
context.result = result
|
|
|
|
|
|
@when("the context list command is executed")
|
|
def step_exec_context_list(context):
|
|
"""Execute context list."""
|
|
runner = CliRunner()
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
container = mock_container.return_value
|
|
container.context_service.return_value = context.mock_service
|
|
project_service = MagicMock()
|
|
project_service.get_current_project.return_value = object()
|
|
container.project_service.return_value = project_service
|
|
|
|
result = runner.invoke(context_app.app, ["list"])
|
|
context.result = result
|
|
|
|
|
|
@when("the context list command is executed without an active project")
|
|
def step_exec_context_list_no_project(context):
|
|
"""Execute context list when no project is configured."""
|
|
runner = CliRunner()
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
container = mock_container.return_value
|
|
container.context_service.return_value = context.mock_service
|
|
project_service = MagicMock()
|
|
project_service.get_current_project.return_value = None
|
|
container.project_service.return_value = project_service
|
|
|
|
result = runner.invoke(context_app.app, ["list"])
|
|
context.result = result
|
|
|
|
|
|
@when("the context list command is executed with service failure")
|
|
def step_exec_context_list_failure(context):
|
|
"""Execute context list when the service raises an error."""
|
|
runner = CliRunner()
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
container = mock_container.return_value
|
|
container.context_service.return_value = context.mock_service
|
|
context.mock_service.list_context.side_effect = context.mock_error
|
|
project_service = MagicMock()
|
|
project_service.get_current_project.return_value = object()
|
|
container.project_service.return_value = project_service
|
|
|
|
result = runner.invoke(context_app.app, ["list"])
|
|
context.result = result
|
|
|
|
|
|
@when("the context show command is executed with path")
|
|
def step_exec_context_show_with_path(context):
|
|
"""Execute context show with path."""
|
|
runner = CliRunner()
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_container.return_value.context_service.return_value = context.mock_service
|
|
|
|
result = runner.invoke(context_app.app, ["show", "/test.txt"])
|
|
context.result = result
|
|
|
|
|
|
@when("the context show command is executed without path")
|
|
def step_exec_context_show_without_path(context):
|
|
"""Execute context show without path."""
|
|
runner = CliRunner()
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_container.return_value.context_service.return_value = context.mock_service
|
|
|
|
result = runner.invoke(context_app.app, ["show"])
|
|
context.result = result
|
|
|
|
|
|
@when("the context show command is executed without an active project")
|
|
def step_exec_context_show_no_project(context):
|
|
"""Execute context show when no project is configured."""
|
|
runner = CliRunner()
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
container = mock_container.return_value
|
|
container.context_service.return_value = context.mock_service
|
|
project_service = MagicMock()
|
|
project_service.get_current_project.return_value = None
|
|
container.project_service.return_value = project_service
|
|
|
|
result = runner.invoke(context_app.app, ["show"])
|
|
context.result = result
|
|
|
|
|
|
@when("the context show command is executed with service failure")
|
|
def step_exec_context_show_failure(context):
|
|
"""Execute context show when the service raises an error."""
|
|
runner = CliRunner()
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
container = mock_container.return_value
|
|
container.context_service.return_value = context.mock_service
|
|
context.mock_service.get_context_content.side_effect = context.mock_error
|
|
project_service = MagicMock()
|
|
project_service.get_current_project.return_value = object()
|
|
container.project_service.return_value = project_service
|
|
|
|
result = runner.invoke(context_app.app, ["show", "/test.txt"])
|
|
context.result = result
|
|
|
|
|
|
@when("the context clear command is executed with yes flag")
|
|
def step_exec_context_clear_yes(context):
|
|
"""Execute context clear with --yes."""
|
|
runner = CliRunner()
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
container = mock_container.return_value
|
|
container.context_service.return_value = context.mock_service
|
|
project_service = MagicMock()
|
|
project_service.get_current_project.return_value = object()
|
|
container.project_service.return_value = project_service
|
|
|
|
result = runner.invoke(context_app.app, ["clear", "--yes"])
|
|
context.result = result
|
|
|
|
|
|
@when("the context clear command is executed and user cancels")
|
|
def step_exec_context_clear_cancel(context):
|
|
"""Execute context clear with cancel."""
|
|
runner = CliRunner()
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
container = mock_container.return_value
|
|
container.context_service.return_value = context.mock_service
|
|
project_service = MagicMock()
|
|
project_service.get_current_project.return_value = object()
|
|
container.project_service.return_value = project_service
|
|
|
|
result = runner.invoke(context_app.app, ["clear"], input="n\n")
|
|
context.result = result
|
|
|
|
|
|
@when("the context clear command is executed")
|
|
def step_exec_context_clear(context):
|
|
"""Execute context clear."""
|
|
runner = CliRunner()
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
container = mock_container.return_value
|
|
container.context_service.return_value = context.mock_service
|
|
project_service = MagicMock()
|
|
project_service.get_current_project.return_value = object()
|
|
container.project_service.return_value = project_service
|
|
|
|
result = runner.invoke(context_app.app, ["clear"])
|
|
context.result = result
|
|
|
|
|
|
@when("the context clear command is executed without an active project")
|
|
def step_exec_context_clear_no_project(context):
|
|
"""Execute context clear when no project is configured."""
|
|
runner = CliRunner()
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
container = mock_container.return_value
|
|
container.context_service.return_value = context.mock_service
|
|
project_service = MagicMock()
|
|
project_service.get_current_project.return_value = None
|
|
container.project_service.return_value = project_service
|
|
|
|
result = runner.invoke(context_app.app, ["clear"])
|
|
context.result = result
|
|
|
|
|
|
@when("the context clear command is executed with service failure")
|
|
def step_exec_context_clear_failure(context):
|
|
"""Execute context clear when the service raises an error."""
|
|
runner = CliRunner()
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
container = mock_container.return_value
|
|
container.context_service.return_value = context.mock_service
|
|
context.mock_service.list_context.return_value = [{"path": "pending.txt"}]
|
|
context.mock_service.clear_context.side_effect = context.mock_error
|
|
project_service = MagicMock()
|
|
project_service.get_current_project.return_value = object()
|
|
container.project_service.return_value = project_service
|
|
|
|
result = runner.invoke(context_app.app, ["clear", "--yes"])
|
|
context.result = result
|
|
|
|
|
|
@when("context command is invoked without subcommand")
|
|
def step_exec_context_no_subcommand(context):
|
|
"""Execute context without subcommand."""
|
|
runner = CliRunner()
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_container.return_value.context_service.return_value = context.mock_service
|
|
|
|
result = runner.invoke(context_app.app, [])
|
|
context.result = result
|
|
|
|
|
|
@then("files should be added successfully with output")
|
|
def step_check_add_success(context):
|
|
"""Check add success."""
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}\n"
|
|
f"Exception: {context.result.exception}"
|
|
)
|
|
|
|
# Strip ANSI color codes before checking
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
clean_output_lower = clean_output.lower()
|
|
|
|
# Check for success message - actual output is "✓ Added 1 file(s) to context:" (with colon)
|
|
assert (
|
|
"added" in clean_output_lower and "file(s) to context" in clean_output_lower
|
|
), (
|
|
f"Expected message about adding files to context.\n"
|
|
f"Raw output: '{context.result.output}'\n"
|
|
f"Clean output (no color codes): '{clean_output}'\n"
|
|
f"Looking for: 'added' and 'file(s) to context' (case-insensitive)\n"
|
|
f"Note: Checking after stripping ANSI color codes"
|
|
)
|
|
|
|
|
|
@then("already tracked file details should be reported")
|
|
def step_check_already_tracked(context):
|
|
"""Verify output when files are already tracked."""
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
# Collapse Rich line-wraps so long paths broken across lines are searchable
|
|
unwrapped_output = clean_output.replace("\n", "")
|
|
assert "File(s) already in context" in clean_output, (
|
|
f"Expected warning about already tracked files.\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
assert "... and 2 more files" in unwrapped_output, (
|
|
f"Expected summary of remaining already tracked files.\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("already tracked file details should be reported without overflow summary")
|
|
def step_check_already_tracked_without_summary(context):
|
|
"""Verify output when tracked files do not exceed the display limit."""
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
# Collapse Rich line-wraps so long paths broken across lines are searchable
|
|
unwrapped_output = clean_output.replace("\n", "")
|
|
assert "File(s) already in context" in clean_output, (
|
|
f"Expected warning about already tracked files.\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
assert "... and" not in unwrapped_output, (
|
|
f"Did not expect overflow summary for fewer than 11 files.\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
if context.already_paths:
|
|
expected_name = context.already_paths[0].name
|
|
assert expected_name in unwrapped_output, (
|
|
f"Expected tracked file name '{expected_name}' in output.\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("path does not exist error should be shown")
|
|
def step_check_path_error(context):
|
|
"""Check path error."""
|
|
# Strip ANSI color codes before checking
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
|
|
assert "Path does not exist" in clean_output, (
|
|
f"Expected 'Path does not exist' error message.\n"
|
|
f"Raw output: '{context.result.output}'\n"
|
|
f"Clean output (no color codes): '{clean_output}'\n"
|
|
f"Exit code: {context.result.exit_code}"
|
|
)
|
|
|
|
|
|
@then("a no active project error should be shown")
|
|
def step_check_no_project_error(context):
|
|
"""Ensure commands abort when no project is active."""
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
assert context.result.exit_code != 0, (
|
|
f"Expected non-zero exit code when project is missing.\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
assert "No active project" in clean_output, (
|
|
f"Expected 'No active project' message.\nOutput: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("FileSystemError should be displayed and command aborted")
|
|
def step_check_filesystem_error(context):
|
|
"""Check filesystem error."""
|
|
assert context.result.exit_code != 0, (
|
|
f"Expected non-zero exit code for error, got {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
|
|
# Strip ANSI color codes before checking
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
|
|
assert "File System Error" in clean_output, (
|
|
f"Expected 'File System Error' in output.\n"
|
|
f"Raw output: '{context.result.output}'\n"
|
|
f"Clean output (no color codes): '{clean_output}'"
|
|
)
|
|
|
|
|
|
@then("CleverAgentsError should be displayed and command aborted")
|
|
def step_check_cleveragents_error(context):
|
|
"""Check CleverAgentsError."""
|
|
assert context.result.exit_code != 0, (
|
|
f"Expected non-zero exit code for error, got {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
|
|
# Strip ANSI color codes before checking
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
|
|
assert "Error:" in clean_output, (
|
|
f"Expected 'Error:' in output for CleverAgentsError.\n"
|
|
f"Raw output: '{context.result.output}'\n"
|
|
f"Clean output (no color codes): '{clean_output}'"
|
|
)
|
|
|
|
|
|
@then("it should delegate to context add function")
|
|
def step_check_load_delegation(context):
|
|
"""Check load delegates to add."""
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}\n"
|
|
f"Exception: {context.result.exception}"
|
|
)
|
|
|
|
assert context.mock_service.add_to_context.called, (
|
|
f"Expected add_to_context to be called (load should delegate to add).\n"
|
|
f"Mock called: {context.mock_service.add_to_context.called}\n"
|
|
f"Call count: {context.mock_service.add_to_context.call_count}"
|
|
)
|
|
|
|
|
|
@then("files should be removed with confirmation message")
|
|
def step_check_remove_success(context):
|
|
"""Check remove success."""
|
|
# First check exit code
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}\n"
|
|
f"Exception: {context.result.exception}"
|
|
)
|
|
|
|
# Strip ANSI color codes before checking
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
clean_output_lower = clean_output.lower()
|
|
|
|
# Check for the success message - the actual output is "✓ Removed 2 file(s) from context."
|
|
assert (
|
|
"removed" in clean_output_lower and "file(s) from context" in clean_output_lower
|
|
), (
|
|
f"Expected message about removing files from context.\n"
|
|
f"Raw output: '{context.result.output}'\n"
|
|
f"Clean output (no color codes): '{clean_output}'\n"
|
|
f"Looking for: 'Removed' and 'file(s) from context' (case-insensitive)"
|
|
)
|
|
|
|
|
|
@then("no files removed message should be shown")
|
|
def step_check_no_remove(context):
|
|
"""Check no files removed."""
|
|
# First check if command succeeded
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}\n"
|
|
f"Exception: {context.result.exception}"
|
|
)
|
|
|
|
# Strip ANSI color codes before checking
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
|
|
# Check for the exact message
|
|
assert "No files were removed from context" in clean_output, (
|
|
f"Expected 'No files were removed from context' message.\n"
|
|
f"Raw output: '{context.result.output}'\n"
|
|
f"Clean output (no color codes): '{clean_output}'\n"
|
|
f"Note: This message appears when remove_from_context returns 0 for all paths"
|
|
)
|
|
|
|
|
|
@then("missing context files should trigger an abort")
|
|
def step_check_missing_context_abort(context):
|
|
"""Ensure missing files cause the command to abort."""
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
assert context.result.exit_code != 0, (
|
|
f"Expected abort when files are missing from context.\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
assert "File(s) not in context" in clean_output, (
|
|
f"Expected message about missing context files.\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("a formatted table of files should be displayed")
|
|
def step_check_table_display(context):
|
|
"""Check table display."""
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}\n"
|
|
f"Exception: {context.result.exception}"
|
|
)
|
|
|
|
# Strip ANSI color codes before checking
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
|
|
# Check for table headers or file content
|
|
assert "File Path" in clean_output or "main.py" in clean_output, (
|
|
f"Expected table headers ('File Path') or file names in output.\n"
|
|
f"Raw output: '{context.result.output}'\n"
|
|
f"Clean output (no color codes): '{clean_output}'"
|
|
)
|
|
|
|
|
|
@then("numeric sizes should be rendered as bytes")
|
|
def step_check_numeric_sizes(context):
|
|
"""Ensure numeric sizes are humanized as byte strings."""
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
assert "bytes" in clean_output, (
|
|
f"Expected sizes to be formatted in bytes.\nOutput: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("empty context message should be shown with help")
|
|
def step_check_empty_message(context):
|
|
"""Check empty message."""
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}\n"
|
|
f"Exception: {context.result.exception}"
|
|
)
|
|
|
|
# Strip ANSI color codes before checking
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
|
|
assert "No files in context" in clean_output, (
|
|
f"Expected 'No files in context' message.\n"
|
|
f"Raw output: '{context.result.output}'\n"
|
|
f"Clean output (no color codes): '{clean_output}'"
|
|
)
|
|
|
|
assert "agents context add" in clean_output, (
|
|
f"Expected help text 'agents context add' in output.\n"
|
|
f"Raw output: '{context.result.output}'\n"
|
|
f"Clean output (no color codes): '{clean_output}'"
|
|
)
|
|
|
|
|
|
@then("file content should be displayed in panel")
|
|
def step_check_content_panel(context):
|
|
"""Check content panel."""
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}\n"
|
|
f"Exception: {context.result.exception}"
|
|
)
|
|
|
|
# Strip ANSI color codes before checking
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
|
|
# Check for either the content or the panel structure
|
|
assert (
|
|
"File content" in clean_output
|
|
or "Content:" in clean_output
|
|
or "Panel" in clean_output
|
|
), (
|
|
f"Expected content display with 'File content', 'Content:', or 'Panel'.\n"
|
|
f"Raw output: '{context.result.output}'\n"
|
|
f"Clean output (no color codes): '{clean_output}'"
|
|
)
|
|
|
|
|
|
@then("file not found message should be displayed")
|
|
def step_check_not_found(context):
|
|
"""Check not found message."""
|
|
# Strip ANSI color codes before checking
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
|
|
assert "File not found in context" in clean_output, (
|
|
f"Expected 'File not found in context' message.\n"
|
|
f"Raw output: '{context.result.output}'\n"
|
|
f"Clean output (no color codes): '{clean_output}'"
|
|
)
|
|
|
|
|
|
@then("context summary with stats should be displayed")
|
|
def step_check_summary(context):
|
|
"""Check summary display."""
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}\n"
|
|
f"Exception: {context.result.exception}"
|
|
)
|
|
|
|
# Strip ANSI color codes before checking
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
|
|
assert "Total files" in clean_output or "Context Summary" in clean_output, (
|
|
f"Expected summary with 'Total files' or 'Context Summary'.\n"
|
|
f"Raw output: '{context.result.output}'\n"
|
|
f"Clean output (no color codes): '{clean_output}'"
|
|
)
|
|
|
|
|
|
@then("the show command should report an empty context")
|
|
def step_check_show_empty_context(context):
|
|
"""Verify show command informs when no files exist."""
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
assert context.result.exit_code == 0, (
|
|
f"Expected successful exit for empty context message.\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
assert "No files in context" in clean_output, (
|
|
f"Expected 'No files in context' message.\nOutput: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("the show command should report missing content")
|
|
def step_check_show_missing_content(context):
|
|
"""Ensure show command highlights missing file content."""
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
assert context.result.exit_code == 0, (
|
|
f"Expected successful exit when content is missing.\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
assert "No content available for" in clean_output, (
|
|
f"Expected message indicating missing content.\nOutput: {context.result.output}"
|
|
)
|
|
|
|
assert "No content available" in clean_output, (
|
|
f"Expected message indicating missing content.\nOutput: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("context should be cleared with success message")
|
|
def step_check_clear_success(context):
|
|
"""Check clear success."""
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}\n"
|
|
f"Exception: {context.result.exception}"
|
|
)
|
|
|
|
# Strip ANSI color codes before checking
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
|
|
assert "Cleared all files from context" in clean_output, (
|
|
f"Expected 'Cleared all files from context' message.\n"
|
|
f"Raw output: '{context.result.output}'\n"
|
|
f"Clean output (no color codes): '{clean_output}'"
|
|
)
|
|
|
|
|
|
@then("cancelled message should appear and context unchanged")
|
|
def step_check_cancelled(context):
|
|
"""Check cancelled."""
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}\n"
|
|
f"Exception: {context.result.exception}"
|
|
)
|
|
|
|
# Strip ANSI color codes before checking
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
|
|
assert "Cancelled" in clean_output, (
|
|
f"Expected 'Cancelled' message when user cancels clear operation.\n"
|
|
f"Raw output: '{context.result.output}'\n"
|
|
f"Clean output (no color codes): '{clean_output}'"
|
|
)
|
|
|
|
|
|
@then("already empty message should be shown")
|
|
def step_check_already_empty(context):
|
|
"""Check already empty."""
|
|
# Strip ANSI color codes before checking
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
|
|
assert "Context is already empty" in clean_output, (
|
|
f"Expected 'Context is already empty' message.\n"
|
|
f"Raw output: '{context.result.output}'\n"
|
|
f"Clean output (no color codes): '{clean_output}'"
|
|
)
|
|
|
|
|
|
@then("the list function should be called by default")
|
|
def step_check_default_list(context):
|
|
"""Check default list behavior."""
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}\n"
|
|
f"Exception: {context.result.exception}"
|
|
)
|
|
|
|
# Strip ANSI color codes before checking
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
|
|
# Default behavior shows the list
|
|
assert "File Path" in clean_output or "main.py" in clean_output, (
|
|
f"Expected default list output with table headers or file names.\n"
|
|
f"Raw output: '{context.result.output}'\n"
|
|
f"Clean output (no color codes): '{clean_output}'"
|
|
)
|
|
|
|
|
|
@given("a dictionary-style context entry with missing metadata")
|
|
def step_given_dict_entry(context):
|
|
"""Prepare dictionary entry missing optional fields."""
|
|
context.dict_entry = {
|
|
"path": "relative/file.txt",
|
|
"type": "NOTE",
|
|
"size": "5 KB",
|
|
}
|
|
|
|
|
|
@given("a domain Context model entry with metadata")
|
|
def step_given_context_model_entry(context):
|
|
"""Prepare a Context model instance for normalization."""
|
|
context.dict_entry = Context(
|
|
id=7,
|
|
plan_id=42,
|
|
type=ContextType.NOTE,
|
|
path="/project/docs/readme.md",
|
|
content="Example context body",
|
|
file_hash="abc123",
|
|
size=2048,
|
|
added_at=datetime(2024, 1, 5, 15, 30, 45),
|
|
)
|
|
|
|
|
|
@when("the normalization helper processes the context entry")
|
|
def step_when_normalize_entry(context):
|
|
"""Normalize dictionary-based context entry."""
|
|
context.normalized_entry = context_app._normalize_context_entry(context.dict_entry)
|
|
|
|
|
|
@then("the normalized context tuple should include placeholder defaults")
|
|
def step_then_normalized_defaults(context):
|
|
"""Verify normalization uses defaults for missing fields."""
|
|
path_text, type_label, size, added_at, content_text = context.normalized_entry
|
|
assert path_text == "relative/file.txt"
|
|
assert type_label == "NOTE"
|
|
assert size == "5 KB"
|
|
assert added_at == "Unknown"
|
|
assert content_text == ""
|
|
|
|
|
|
@then("the normalized context tuple should match the model attributes")
|
|
def step_then_normalized_model(context):
|
|
"""Verify normalization preserves Context model details."""
|
|
path_text, type_label, size, added_at, content_text = context.normalized_entry
|
|
assert path_text == "/project/docs/readme.md"
|
|
assert type_label == "note"
|
|
assert size == 2048
|
|
assert added_at == "2024-01-05 15:30:45"
|
|
assert content_text == "Example context body"
|
|
|
|
|
|
@when("the programmatic add command runs with a missing file path")
|
|
def step_when_programmatic_add_missing(context):
|
|
"""Run add_command with a path that does not exist."""
|
|
missing_path = Path(context.temp_dir) / "missing.txt"
|
|
context.programmatic_exception = None
|
|
context.mock_service.add_to_context.reset_mock()
|
|
with patched_container(context, project=object()):
|
|
try:
|
|
context_app.add_command([str(missing_path)])
|
|
except Exception as exc: # pragma: no cover - captured for assertions
|
|
context.programmatic_exception = exc
|
|
context.programmatic_result = None
|
|
|
|
|
|
@then("no files are sent to the context service")
|
|
def step_then_no_files_sent(context):
|
|
"""Ensure add_to_context is not called for missing files."""
|
|
assert context.programmatic_exception is None
|
|
context.mock_service.add_to_context.assert_not_called()
|
|
|
|
|
|
@when("the programmatic add command runs without an active project")
|
|
def step_when_programmatic_add_no_project(context):
|
|
"""Run add_command without an active project."""
|
|
test_file = Path(context.temp_dir) / "active.txt"
|
|
test_file.write_text("data")
|
|
context.programmatic_exception = None
|
|
with patched_container(context, project=None):
|
|
try:
|
|
context_app.add_command([str(test_file)])
|
|
except Exception as exc:
|
|
context.programmatic_exception = exc
|
|
|
|
|
|
@then("a CleverAgentsError is raised for programmatic add")
|
|
def step_then_add_requires_project(context):
|
|
"""Ensure add_command raises when project is missing."""
|
|
assert isinstance(context.programmatic_exception, CleverAgentsError)
|
|
|
|
|
|
@when("the programmatic list command is executed")
|
|
def step_when_programmatic_list(context):
|
|
"""Execute list_command with active project."""
|
|
context.programmatic_exception = None
|
|
with patched_container(context, project=object()):
|
|
context.programmatic_result = context_app.list_command()
|
|
|
|
|
|
@then("the programmatic list command should return those files")
|
|
def step_then_list_returns_files(context):
|
|
"""Verify list_command returns mock files."""
|
|
assert context.programmatic_result == context.mock_files
|
|
|
|
|
|
@when("the programmatic list command runs without an active project")
|
|
def step_when_programmatic_list_no_project(context):
|
|
"""Execute list_command without active project."""
|
|
context.programmatic_exception = None
|
|
with patched_container(context, project=None):
|
|
try:
|
|
context_app.list_command()
|
|
except Exception as exc:
|
|
context.programmatic_exception = exc
|
|
|
|
|
|
@then("a CleverAgentsError is raised for programmatic list")
|
|
def step_then_list_requires_project(context):
|
|
"""Ensure list_command raises when project is missing."""
|
|
assert isinstance(context.programmatic_exception, CleverAgentsError)
|
|
|
|
|
|
@when("the programmatic show command runs without an active project")
|
|
def step_when_programmatic_show_no_project(context):
|
|
"""Execute show_command without an active project configured."""
|
|
context.programmatic_exception = None
|
|
context.programmatic_result = None
|
|
with patched_container(context, project=None):
|
|
try:
|
|
context_app.show_command()
|
|
except Exception as exc:
|
|
context.programmatic_exception = exc
|
|
|
|
|
|
@when("the programmatic show command runs without a path")
|
|
def step_when_programmatic_show_summary(context):
|
|
"""Execute show_command when listing context entries."""
|
|
context.programmatic_exception = None
|
|
with patched_container(context, project=object()):
|
|
context.programmatic_result = context_app.show_command()
|
|
|
|
|
|
@then("the programmatic show command should return a summary string")
|
|
def step_then_show_returns_summary(context):
|
|
"""Verify show_command returns summary when no path provided."""
|
|
expected = f"Total files: {len(context.mock_files)}"
|
|
assert context.programmatic_result == expected
|
|
|
|
|
|
@then("a CleverAgentsError is raised for programmatic show")
|
|
def step_then_show_requires_project(context):
|
|
"""Ensure show_command raises when no project is active."""
|
|
assert isinstance(context.programmatic_exception, CleverAgentsError)
|
|
|
|
|
|
@then("the programmatic show command should return no content summary")
|
|
def step_then_show_returns_none(context):
|
|
"""Ensure show_command returns None when context has no files."""
|
|
assert context.programmatic_exception is None
|
|
assert context.programmatic_result is None
|
|
|
|
|
|
@when("the programmatic show command runs for a specific path")
|
|
def step_when_programmatic_show_specific(context):
|
|
"""Execute show_command for a single file path."""
|
|
context.programmatic_exception = None
|
|
with patched_container(context, project=object()):
|
|
context.programmatic_result = context_app.show_command("/test.txt")
|
|
|
|
|
|
@then("the programmatic show command should return that content")
|
|
def step_then_show_returns_content(context):
|
|
"""Ensure show_command returns content for requested file."""
|
|
assert context.programmatic_result == context.mock_content
|
|
|
|
|
|
@when("the programmatic remove command runs without an active project")
|
|
def step_when_programmatic_remove_no_project(context):
|
|
"""Execute rm_command without an active project configured."""
|
|
context.programmatic_exception = None
|
|
with patched_container(context, project=None):
|
|
try:
|
|
context_app.rm_command(["missing.txt"])
|
|
except Exception as exc:
|
|
context.programmatic_exception = exc
|
|
|
|
|
|
@when("the programmatic remove command runs for files")
|
|
def step_when_programmatic_remove(context):
|
|
"""Execute rm_command where files are missing."""
|
|
context.programmatic_exception = None
|
|
with patched_container(context, project=object()):
|
|
try:
|
|
context_app.rm_command(["missing.txt", "other.txt"])
|
|
except Exception as exc:
|
|
context.programmatic_exception = exc
|
|
|
|
|
|
@when("the programmatic remove command runs with removable files")
|
|
def step_when_programmatic_remove_success(context):
|
|
"""Execute rm_command when files are present in context."""
|
|
context.programmatic_exception = None
|
|
context.mock_service.remove_from_context.reset_mock()
|
|
with patched_container(context, project=object()):
|
|
try:
|
|
context_app.rm_command(["/file1.txt", "/file2.txt"])
|
|
except Exception as exc:
|
|
context.programmatic_exception = exc
|
|
|
|
|
|
@then("a CleverAgentsError is raised for programmatic remove")
|
|
def step_then_remove_requires_presence(context):
|
|
"""Ensure rm_command raises when files are not in context."""
|
|
assert isinstance(context.programmatic_exception, CleverAgentsError)
|
|
|
|
|
|
@then("the programmatic remove command should complete successfully")
|
|
def step_then_remove_success(context):
|
|
"""Ensure rm_command succeeds when files are removed."""
|
|
assert context.programmatic_exception is None
|
|
assert context.mock_service.remove_from_context.call_count == 2
|
|
|
|
|
|
@when("the programmatic clear command runs without an active project")
|
|
def step_when_programmatic_clear_no_project(context):
|
|
"""Execute clear_command without an active project."""
|
|
context.programmatic_exception = None
|
|
with patched_container(context, project=None):
|
|
try:
|
|
context_app.clear_command()
|
|
except Exception as exc:
|
|
context.programmatic_exception = exc
|
|
|
|
|
|
@then("a CleverAgentsError is raised for programmatic clear")
|
|
def step_then_clear_requires_project(context):
|
|
"""Ensure clear_command raises when project is missing."""
|
|
assert isinstance(context.programmatic_exception, CleverAgentsError)
|
|
|
|
|
|
@when("the programmatic clear command runs with an active project")
|
|
def step_when_programmatic_clear_active(context):
|
|
"""Execute clear_command with active project."""
|
|
context.programmatic_exception = None
|
|
context.mock_service.clear_context.reset_mock()
|
|
with patched_container(context, project=object()):
|
|
try:
|
|
context_app.clear_command()
|
|
except Exception as exc:
|
|
context.programmatic_exception = exc
|
|
|
|
|
|
@then("the context service clear function should be called")
|
|
def step_then_clear_calls_service(context):
|
|
"""Verify clear_command calls the context service clear method."""
|
|
assert context.programmatic_exception is None
|
|
context.mock_service.clear_context.assert_called_once()
|
|
|
|
|
|
@when("the context add command is executed with more than ten added files")
|
|
def step_exec_context_add_many_files(context):
|
|
"""Execute context add with more than ten returned files."""
|
|
runner = CliRunner()
|
|
base_dir = Path(context.temp_dir) / "bulk_add"
|
|
base_dir.mkdir(parents=True, exist_ok=True)
|
|
added_files = []
|
|
for index in range(11):
|
|
file_path = base_dir / f"file_{index}.txt"
|
|
file_path.write_text(f"data {index}")
|
|
added_files.append(file_path)
|
|
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
container = mock_container.return_value
|
|
container.context_service.return_value = context.mock_service
|
|
project_service = MagicMock()
|
|
project_service.get_current_project.return_value = object()
|
|
container.project_service.return_value = project_service
|
|
context.mock_service.add_to_context.return_value = (added_files, [])
|
|
|
|
result = runner.invoke(context_app.app, ["add", str(base_dir)])
|
|
context.result = result
|
|
context.added_files = added_files
|
|
|
|
|
|
@then("remaining added files should be summarized")
|
|
def step_then_added_files_summary(context):
|
|
"""Ensure the add command summarizes remaining files."""
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}\n"
|
|
f"Exception: {context.result.exception}"
|
|
)
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
remaining = len(context.added_files) - 10
|
|
expected = f"... and {remaining} more files"
|
|
assert expected in clean_output, (
|
|
f"Expected overflow summary message.\nOutput: {context.result.output}"
|
|
)
|
|
|
|
|
|
@given("a missing context directory for listing")
|
|
def step_given_missing_context_dir_listing(context):
|
|
"""Prepare a non-existent context directory path."""
|
|
context.context_dir = Path(tempfile.mkdtemp()) / "missing_contexts"
|
|
|
|
|
|
@given("an empty context directory for listing")
|
|
def step_given_empty_context_dir_listing(context):
|
|
"""Prepare an empty context directory."""
|
|
context.context_dir = Path(tempfile.mkdtemp())
|
|
|
|
|
|
@given("a context directory with named contexts")
|
|
def step_given_context_dir_named(context):
|
|
"""Prepare a context directory with subdirectories."""
|
|
base_dir = Path(tempfile.mkdtemp())
|
|
context.context_names = ["alpha", "beta"]
|
|
for name in context.context_names:
|
|
(base_dir / name).mkdir(parents=True, exist_ok=True)
|
|
context.context_dir = base_dir
|
|
|
|
|
|
@when("the context list command is executed with the context directory")
|
|
def step_when_list_with_context_dir(context):
|
|
"""Execute list command with a provided context directory."""
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
context_app.app, ["list", "--context-dir", str(context.context_dir)]
|
|
)
|
|
context.result = result
|
|
|
|
|
|
@then("no contexts found message should be shown for list")
|
|
def step_then_list_no_contexts_found(context):
|
|
"""Ensure list reports no contexts found."""
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
assert "No contexts found." in clean_output, (
|
|
f"Expected no contexts found message.\nOutput: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("the context names should be listed")
|
|
def step_then_context_names_listed(context):
|
|
"""Ensure context names are listed in output."""
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
for name in context.context_names:
|
|
assert name in clean_output, (
|
|
f"Expected context name '{name}' in output.\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
|
|
|
|
@given("a missing named context for export")
|
|
def step_given_missing_context_export(context):
|
|
"""Prepare export inputs for a missing context."""
|
|
base_dir = Path(tempfile.mkdtemp())
|
|
context.context_dir = base_dir
|
|
context.context_name = "missing_ctx"
|
|
context.export_file = base_dir / "export.json"
|
|
|
|
|
|
@given("a named context exists for export")
|
|
def step_given_context_exists_export(context):
|
|
"""Prepare export inputs for an existing context."""
|
|
base_dir = Path(tempfile.mkdtemp())
|
|
context.context_dir = base_dir
|
|
context.context_name = "export_ctx"
|
|
context_dir = base_dir / context.context_name
|
|
context_dir.mkdir(parents=True, exist_ok=True)
|
|
(context_dir / "messages.json").write_text("[]", encoding="utf-8")
|
|
context.export_file = base_dir / "export.json"
|
|
|
|
|
|
@when("the context export command is executed")
|
|
def step_when_context_export(context):
|
|
"""Execute the export command."""
|
|
runner = CliRunner()
|
|
args = [
|
|
"export",
|
|
context.context_name,
|
|
str(context.export_file),
|
|
"--context-dir",
|
|
str(context.context_dir),
|
|
]
|
|
context.result = runner.invoke(context_app.app, args)
|
|
|
|
|
|
@then("the context export should fail for missing context")
|
|
def step_then_export_missing(context):
|
|
"""Ensure export fails when context is missing."""
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
assert context.result.exit_code != 0, (
|
|
f"Expected non-zero exit code for missing context.\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
assert "does not exist" in clean_output, (
|
|
f"Expected missing context error message.\nOutput: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("the context export should succeed")
|
|
def step_then_export_success(context):
|
|
"""Ensure export writes the JSON file."""
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
assert "exported to" in clean_output, (
|
|
f"Expected export confirmation message.\nOutput: {context.result.output}"
|
|
)
|
|
assert context.export_file.exists(), "Expected export file to be created."
|
|
|
|
|
|
@given("a context import file is available")
|
|
def step_given_import_file(context):
|
|
"""Prepare an import JSON file for context import."""
|
|
base_dir = Path(tempfile.mkdtemp())
|
|
context.context_dir = base_dir
|
|
context.context_name = "import_ctx"
|
|
context.import_file = base_dir / "import.json"
|
|
payload = {
|
|
"messages": [{"role": "user", "content": "hello"}],
|
|
"metadata": {"context_name": context.context_name},
|
|
"state": {"step": 1},
|
|
"global_context": {"topic": "demo"},
|
|
}
|
|
context.import_file.write_text(json.dumps(payload), encoding="utf-8")
|
|
|
|
|
|
@when("the context import command is executed")
|
|
def step_when_context_import(context):
|
|
"""Execute the import command."""
|
|
runner = CliRunner()
|
|
args = [
|
|
"import",
|
|
context.context_name,
|
|
str(context.import_file),
|
|
"--context-dir",
|
|
str(context.context_dir),
|
|
]
|
|
context.result = runner.invoke(context_app.app, args)
|
|
|
|
|
|
@then("the context should be imported successfully")
|
|
def step_then_import_success(context):
|
|
"""Ensure import creates context files."""
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
assert "imported from" in clean_output, (
|
|
f"Expected import confirmation message.\nOutput: {context.result.output}"
|
|
)
|
|
messages_file = context.context_dir / context.context_name / "messages.json"
|
|
assert messages_file.exists(), "Expected messages.json to be created."
|
|
|
|
|
|
@when("the context delete command is executed with both name and all")
|
|
def step_when_delete_name_and_all(context):
|
|
"""Execute delete with a name and --all."""
|
|
runner = CliRunner()
|
|
name = getattr(context, "context_name", "sample_ctx")
|
|
context.result = runner.invoke(context_app.app, ["delete", name, "--all"])
|
|
|
|
|
|
@when("the context delete command is executed without required arguments")
|
|
def step_when_delete_without_required(context):
|
|
"""Execute delete with neither name nor --all."""
|
|
runner = CliRunner()
|
|
context.result = runner.invoke(context_app.app, ["delete"])
|
|
|
|
|
|
@given("a missing context directory for deletion")
|
|
def step_given_missing_context_dir_deletion(context):
|
|
"""Prepare a non-existent context directory for deletion."""
|
|
context.context_dir = Path(tempfile.mkdtemp()) / "missing_contexts"
|
|
|
|
|
|
@given("an empty context directory for deletion")
|
|
def step_given_empty_context_dir_deletion(context):
|
|
"""Prepare an empty context directory for deletion."""
|
|
context.context_dir = Path(tempfile.mkdtemp())
|
|
|
|
|
|
@given("a context directory with multiple contexts")
|
|
def step_given_multiple_contexts_deletion(context):
|
|
"""Prepare a context directory with multiple contexts."""
|
|
base_dir = Path(tempfile.mkdtemp())
|
|
context.context_names = ["alpha", "beta"]
|
|
for name in context.context_names:
|
|
(base_dir / name).mkdir(parents=True, exist_ok=True)
|
|
context.context_dir = base_dir
|
|
|
|
|
|
@given("a context directory without the named context")
|
|
def step_given_missing_named_context_deletion(context):
|
|
"""Prepare a directory that lacks the requested context."""
|
|
base_dir = Path(tempfile.mkdtemp())
|
|
(base_dir / "other").mkdir(parents=True, exist_ok=True)
|
|
context.context_dir = base_dir
|
|
context.context_name = "missing_ctx"
|
|
|
|
|
|
@given("a context directory with a named context")
|
|
def step_given_named_context_deletion(context):
|
|
"""Prepare a directory containing the named context."""
|
|
base_dir = Path(tempfile.mkdtemp())
|
|
context.context_dir = base_dir
|
|
context.context_name = "target_ctx"
|
|
target_dir = base_dir / context.context_name
|
|
target_dir.mkdir(parents=True, exist_ok=True)
|
|
context.context_target = target_dir
|
|
|
|
|
|
@when("the context delete command is executed for all contexts with yes")
|
|
def step_when_delete_all_yes(context):
|
|
"""Execute delete for all contexts with --yes."""
|
|
runner = CliRunner()
|
|
context.result = runner.invoke(
|
|
context_app.app,
|
|
[
|
|
"delete",
|
|
"--all",
|
|
"--yes",
|
|
"--context-dir",
|
|
str(context.context_dir),
|
|
],
|
|
)
|
|
|
|
|
|
@when("the context delete command is executed for all contexts and user cancels")
|
|
def step_when_delete_all_cancel(context):
|
|
"""Execute delete for all contexts and cancel the prompt."""
|
|
runner = CliRunner()
|
|
context.result = runner.invoke(
|
|
context_app.app,
|
|
["delete", "--all", "--context-dir", str(context.context_dir)],
|
|
input="n\n",
|
|
)
|
|
|
|
|
|
@when("the context delete command is executed for a missing named context")
|
|
def step_when_delete_missing_named_context(context):
|
|
"""Execute delete for a missing context name."""
|
|
runner = CliRunner()
|
|
context.result = runner.invoke(
|
|
context_app.app,
|
|
["delete", context.context_name, "--context-dir", str(context.context_dir)],
|
|
)
|
|
|
|
|
|
@when("the context delete command is executed for a named context and user cancels")
|
|
def step_when_delete_named_cancel(context):
|
|
"""Execute delete for a named context and cancel."""
|
|
runner = CliRunner()
|
|
context.result = runner.invoke(
|
|
context_app.app,
|
|
[
|
|
"delete",
|
|
context.context_name,
|
|
"--context-dir",
|
|
str(context.context_dir),
|
|
],
|
|
input="n\n",
|
|
)
|
|
|
|
|
|
@when("the context delete command is executed for a named context with yes")
|
|
def step_when_delete_named_yes(context):
|
|
"""Execute delete for a named context with --yes."""
|
|
runner = CliRunner()
|
|
context.result = runner.invoke(
|
|
context_app.app,
|
|
[
|
|
"delete",
|
|
context.context_name,
|
|
"--yes",
|
|
"--context-dir",
|
|
str(context.context_dir),
|
|
],
|
|
)
|
|
|
|
|
|
@then("the delete command should fail with a mutually exclusive options error")
|
|
def step_then_delete_mutually_exclusive(context):
|
|
"""Ensure delete rejects name with --all."""
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
assert context.result.exit_code != 0, (
|
|
f"Expected non-zero exit code for invalid options.\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
assert "Cannot specify NAME when using --all" in clean_output, (
|
|
f"Expected mutual exclusivity error.\nOutput: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("the delete command should fail with a missing arguments error")
|
|
def step_then_delete_missing_args(context):
|
|
"""Ensure delete requires a name or --all."""
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
assert context.result.exit_code != 0, (
|
|
f"Expected non-zero exit code for missing arguments.\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
assert "Must specify NAME or use --all" in clean_output, (
|
|
f"Expected missing arguments error.\nOutput: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("the delete command should report no contexts to delete")
|
|
def step_then_delete_no_contexts(context):
|
|
"""Ensure delete reports no contexts found."""
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
assert "No contexts found to delete." in clean_output, (
|
|
f"Expected no contexts found message.\nOutput: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("the delete command should list contexts and cancel")
|
|
def step_then_delete_list_cancel(context):
|
|
"""Ensure delete lists contexts before cancellation."""
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
assert "Delete cancelled." in clean_output, (
|
|
f"Expected cancellation message.\nOutput: {context.result.output}"
|
|
)
|
|
for name in context.context_names:
|
|
assert name in clean_output, (
|
|
f"Expected context name '{name}' in output.\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
assert (context.context_dir / name).exists(), (
|
|
f"Expected context '{name}' to remain after cancel."
|
|
)
|
|
|
|
|
|
@then("the delete command should remove all contexts")
|
|
def step_then_delete_all_contexts(context):
|
|
"""Ensure delete removes all contexts."""
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
expected = f"Deleted {len(context.context_names)} context(s)."
|
|
assert expected in clean_output, (
|
|
f"Expected delete confirmation message.\nOutput: {context.result.output}"
|
|
)
|
|
for name in context.context_names:
|
|
assert not (context.context_dir / name).exists(), (
|
|
f"Expected context '{name}' to be deleted."
|
|
)
|
|
|
|
|
|
@then("the delete command should fail for missing named context")
|
|
def step_then_delete_missing_named_context(context):
|
|
"""Ensure delete errors when named context is missing."""
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
assert context.result.exit_code != 0, (
|
|
f"Expected non-zero exit code for missing context.\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
assert "does not exist" in clean_output, (
|
|
f"Expected missing context error.\nOutput: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("the delete command should cancel deleting the named context")
|
|
def step_then_delete_named_cancel(context):
|
|
"""Ensure delete cancellation keeps the named context."""
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
assert "Delete cancelled." in clean_output, (
|
|
f"Expected cancellation message.\nOutput: {context.result.output}"
|
|
)
|
|
assert (context.context_dir / context.context_name).exists(), (
|
|
"Expected named context to remain after cancellation."
|
|
)
|
|
|
|
|
|
@then("the delete command should delete the named context")
|
|
def step_then_delete_named_context(context):
|
|
"""Ensure delete removes the named context."""
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
assert "deleted" in clean_output.lower(), (
|
|
f"Expected deletion confirmation.\nOutput: {context.result.output}"
|
|
)
|
|
assert not (context.context_dir / context.context_name).exists(), (
|
|
"Expected named context to be deleted."
|
|
)
|
|
|
|
|
|
@given("a missing named context for clearing")
|
|
def step_given_missing_named_context_clear(context):
|
|
"""Prepare a missing named context for clear."""
|
|
base_dir = Path(tempfile.mkdtemp())
|
|
context.context_dir = base_dir
|
|
context.context_name = "missing_clear_ctx"
|
|
|
|
|
|
@given("a named context directory with contents")
|
|
def step_given_named_context_with_contents(context):
|
|
"""Prepare a named context directory with files and subdirectories."""
|
|
base_dir = Path(tempfile.mkdtemp())
|
|
context.context_dir = base_dir
|
|
context.context_name = "clear_ctx"
|
|
target_dir = base_dir / context.context_name
|
|
target_dir.mkdir(parents=True, exist_ok=True)
|
|
(target_dir / "file.txt").write_text("data", encoding="utf-8")
|
|
nested_dir = target_dir / "subdir"
|
|
nested_dir.mkdir(parents=True, exist_ok=True)
|
|
(nested_dir / "nested.txt").write_text("data", encoding="utf-8")
|
|
context.named_context_dir = target_dir
|
|
|
|
|
|
@when("the context clear command is executed for a missing named context")
|
|
def step_when_clear_missing_named_context(context):
|
|
"""Execute clear for a missing named context."""
|
|
runner = CliRunner()
|
|
context.result = runner.invoke(
|
|
context_app.app,
|
|
["clear", context.context_name, "--context-dir", str(context.context_dir)],
|
|
)
|
|
|
|
|
|
@when("the context clear command is executed for the named context and user cancels")
|
|
def step_when_clear_named_context_cancel(context):
|
|
"""Execute clear for a named context and cancel."""
|
|
runner = CliRunner()
|
|
context.result = runner.invoke(
|
|
context_app.app,
|
|
["clear", context.context_name, "--context-dir", str(context.context_dir)],
|
|
input="n\n",
|
|
)
|
|
|
|
|
|
@when("the context clear command is executed for the named context with yes")
|
|
def step_when_clear_named_context_yes(context):
|
|
"""Execute clear for a named context with --yes."""
|
|
runner = CliRunner()
|
|
context.result = runner.invoke(
|
|
context_app.app,
|
|
[
|
|
"clear",
|
|
context.context_name,
|
|
"--yes",
|
|
"--context-dir",
|
|
str(context.context_dir),
|
|
],
|
|
)
|
|
|
|
|
|
@then("the clear command should fail for missing named context")
|
|
def step_then_clear_missing_named_context(context):
|
|
"""Ensure clear errors when named context is missing."""
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
assert context.result.exit_code != 0, (
|
|
f"Expected non-zero exit code for missing context.\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
assert "does not exist" in clean_output, (
|
|
f"Expected missing context error message.\nOutput: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("the clear command should report cancellation for named context")
|
|
def step_then_clear_named_cancel(context):
|
|
"""Ensure cancellation keeps named context contents."""
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
assert "Clear cancelled." in clean_output, (
|
|
f"Expected cancellation message.\nOutput: {context.result.output}"
|
|
)
|
|
assert any(context.named_context_dir.iterdir()), (
|
|
"Expected named context contents to remain after cancellation."
|
|
)
|
|
|
|
|
|
@then("the named context should be cleared")
|
|
def step_then_named_context_cleared(context):
|
|
"""Ensure clear removes contents but keeps the directory."""
|
|
clean_output = strip_ansi_codes(context.result.output)
|
|
assert context.result.exit_code == 0, (
|
|
f"Command failed with exit code {context.result.exit_code}\n"
|
|
f"Output: {context.result.output}"
|
|
)
|
|
expected = f"Context '{context.context_name}' cleared."
|
|
assert expected in clean_output, (
|
|
f"Expected clear confirmation message.\nOutput: {context.result.output}"
|
|
)
|
|
assert context.named_context_dir.exists(), (
|
|
"Expected named context directory to remain after clearing."
|
|
)
|
|
assert list(context.named_context_dir.iterdir()) == [], (
|
|
"Expected named context contents to be removed."
|
|
)
|