503 lines
18 KiB
Python
503 lines
18 KiB
Python
"""Step definitions for ACMS context add command with --tag and --policy flags.
|
|
|
|
All step names are prefixed with ``acms`` to avoid AmbiguousStep
|
|
conflicts with existing steps.
|
|
|
|
Issue #9982: feat(acms): implement context add CLI command for file and
|
|
directory indexing.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
from typer.testing import CliRunner
|
|
|
|
from cleveragents.acms.index import AcmsIndexEntry, ChunkedFileTraverser
|
|
from cleveragents.cli.commands.context import app as context_app
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("an acms context add test runner")
|
|
def step_acms_runner(context: Context) -> None:
|
|
"""Set up the CLI runner for ACMS context add tests."""
|
|
context.acms_runner = CliRunner()
|
|
context.acms_error: Exception | None = None
|
|
context.acms_entries: list[AcmsIndexEntry] = []
|
|
context.acms_progress_calls: list[tuple[int, int]] = []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ChunkedFileTraverser unit tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('a temporary directory with a single Python file "{filename}"')
|
|
def step_acms_single_file(context: Context, filename: str) -> None:
|
|
"""Create a temp dir with a single Python file."""
|
|
context.acms_tmpdir = tempfile.mkdtemp()
|
|
filepath = Path(context.acms_tmpdir) / filename
|
|
filepath.write_text("# test content\nprint('hello')\n", encoding="utf-8")
|
|
context.acms_target_path = filepath
|
|
|
|
|
|
@given("a temporary directory with 3 Python files")
|
|
def step_acms_three_files(context: Context) -> None:
|
|
"""Create a temp dir with 3 Python files."""
|
|
context.acms_tmpdir = tempfile.mkdtemp()
|
|
tmpdir = Path(context.acms_tmpdir)
|
|
for i in range(3):
|
|
(tmpdir / f"file_{i}.py").write_text(f"# file {i}\n", encoding="utf-8")
|
|
context.acms_target_path = tmpdir
|
|
|
|
|
|
@given("a temporary directory with files in subdirectory")
|
|
def step_acms_files_in_subdir(context: Context) -> None:
|
|
"""Create a temp dir with files at root and in a subdirectory."""
|
|
context.acms_tmpdir = tempfile.mkdtemp()
|
|
tmpdir = Path(context.acms_tmpdir)
|
|
(tmpdir / "root_file.py").write_text("# root\n", encoding="utf-8")
|
|
subdir = tmpdir / "subdir"
|
|
subdir.mkdir()
|
|
(subdir / "sub_file.py").write_text("# sub\n", encoding="utf-8")
|
|
context.acms_target_path = tmpdir
|
|
context.acms_top_level_count = 1 # only root_file.py
|
|
|
|
|
|
@given("a temporary directory with 5 Python files")
|
|
def step_acms_five_files(context: Context) -> None:
|
|
"""Create a temp dir with 5 Python files."""
|
|
context.acms_tmpdir = tempfile.mkdtemp()
|
|
tmpdir = Path(context.acms_tmpdir)
|
|
for i in range(5):
|
|
(tmpdir / f"file_{i}.py").write_text(f"# file {i}\n", encoding="utf-8")
|
|
context.acms_target_path = tmpdir
|
|
|
|
|
|
@given("a temporary directory with a Python file and a .pyc file")
|
|
def step_acms_py_and_pyc(context: Context) -> None:
|
|
"""Create a temp dir with a .py and a .pyc file."""
|
|
context.acms_tmpdir = tempfile.mkdtemp()
|
|
tmpdir = Path(context.acms_tmpdir)
|
|
(tmpdir / "module.py").write_text("# module\n", encoding="utf-8")
|
|
(tmpdir / "module.pyc").write_bytes(b"\x00\x01\x02\x03")
|
|
context.acms_target_path = tmpdir
|
|
|
|
|
|
@given("a temporary directory with a Python file and an ignored __pycache__ file")
|
|
def step_acms_py_and_ignored_cache_dir(context: Context) -> None:
|
|
"""Create a temp dir with a visible file and a file under __pycache__."""
|
|
context.acms_tmpdir = tempfile.mkdtemp()
|
|
tmpdir = Path(context.acms_tmpdir)
|
|
(tmpdir / "module.py").write_text("# module\n", encoding="utf-8")
|
|
cache_dir = tmpdir / "__pycache__"
|
|
cache_dir.mkdir()
|
|
(cache_dir / "cached.py").write_text("# ignored cache module\n", encoding="utf-8")
|
|
context.acms_target_path = tmpdir
|
|
|
|
|
|
@when("I traverse the file with ChunkedFileTraverser")
|
|
def step_acms_traverse_file(context: Context) -> None:
|
|
"""Traverse a single file with ChunkedFileTraverser."""
|
|
traverser = ChunkedFileTraverser()
|
|
context.acms_entries = list(traverser.traverse(context.acms_target_path))
|
|
|
|
|
|
@when("I traverse the directory recursively with ChunkedFileTraverser")
|
|
def step_acms_traverse_dir_recursive(context: Context) -> None:
|
|
"""Traverse a directory recursively with ChunkedFileTraverser."""
|
|
traverser = ChunkedFileTraverser()
|
|
context.acms_entries = list(
|
|
traverser.traverse(context.acms_target_path, recursive=True)
|
|
)
|
|
|
|
|
|
@when("I traverse the directory non-recursively with ChunkedFileTraverser")
|
|
def step_acms_traverse_dir_nonrecursive(context: Context) -> None:
|
|
"""Traverse a directory non-recursively with ChunkedFileTraverser."""
|
|
traverser = ChunkedFileTraverser()
|
|
context.acms_entries = list(
|
|
traverser.traverse(context.acms_target_path, recursive=False)
|
|
)
|
|
|
|
|
|
@when('I traverse with tags "{tag1}" and "{tag2}"')
|
|
def step_acms_traverse_with_tags(context: Context, tag1: str, tag2: str) -> None:
|
|
"""Traverse with multiple tags."""
|
|
traverser = ChunkedFileTraverser(tags=[tag1, tag2])
|
|
context.acms_entries = list(traverser.traverse(context.acms_target_path))
|
|
context.acms_expected_tags = [tag1, tag2]
|
|
|
|
|
|
@when('I traverse with policy "{policy}"')
|
|
def step_acms_traverse_with_policy(context: Context, policy: str) -> None:
|
|
"""Traverse with a policy."""
|
|
traverser = ChunkedFileTraverser(policy=policy)
|
|
context.acms_entries = list(traverser.traverse(context.acms_target_path))
|
|
context.acms_expected_policy = policy
|
|
|
|
|
|
@when("I traverse with a progress callback and chunk_size {chunk_size:d}")
|
|
def step_acms_traverse_with_progress(context: Context, chunk_size: int) -> None:
|
|
"""Traverse with a progress callback."""
|
|
calls: list[tuple[int, int]] = []
|
|
|
|
def on_progress(done: int, total: int) -> None:
|
|
calls.append((done, total))
|
|
|
|
traverser = ChunkedFileTraverser(
|
|
chunk_size=chunk_size,
|
|
on_progress=on_progress,
|
|
)
|
|
context.acms_entries = list(
|
|
traverser.traverse(context.acms_target_path, recursive=True)
|
|
)
|
|
context.acms_progress_calls = calls
|
|
|
|
|
|
@when("I traverse a non-existent path with ChunkedFileTraverser")
|
|
def step_acms_traverse_nonexistent(context: Context) -> None:
|
|
"""Traverse a non-existent path."""
|
|
traverser = ChunkedFileTraverser()
|
|
try:
|
|
list(traverser.traverse(Path("/nonexistent/path/xyz")))
|
|
context.acms_error = None
|
|
except FileNotFoundError as exc:
|
|
context.acms_error = exc
|
|
|
|
|
|
@when("I create a ChunkedFileTraverser with chunk_size 0")
|
|
def step_acms_invalid_chunk_size(context: Context) -> None:
|
|
"""Create a ChunkedFileTraverser with invalid chunk_size."""
|
|
try:
|
|
ChunkedFileTraverser(chunk_size=0)
|
|
context.acms_error = None
|
|
except ValueError as exc:
|
|
context.acms_error = exc
|
|
|
|
|
|
@when("I create an AcmsIndexEntry with a relative path")
|
|
def step_acms_relative_path_entry(context: Context) -> None:
|
|
"""Create an AcmsIndexEntry with a relative path."""
|
|
try:
|
|
AcmsIndexEntry(
|
|
path=Path("relative/path.py"),
|
|
content="# test",
|
|
content_hash="abc123",
|
|
size_bytes=10,
|
|
)
|
|
context.acms_error = None
|
|
except ValueError as exc:
|
|
context.acms_error = exc
|
|
|
|
|
|
@when("I create an AcmsIndexEntry with size_bytes -1")
|
|
def step_acms_negative_size_entry(context: Context) -> None:
|
|
"""Create an AcmsIndexEntry with negative size_bytes."""
|
|
try:
|
|
AcmsIndexEntry(
|
|
path=Path("/tmp/test.py"),
|
|
content="# test",
|
|
content_hash="abc123",
|
|
size_bytes=-1,
|
|
)
|
|
context.acms_error = None
|
|
except ValueError as exc:
|
|
context.acms_error = exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Assertions for traversal
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the acms traversal should yield {count:d} entry")
|
|
def step_acms_yield_one_entry(context: Context, count: int) -> None:
|
|
assert len(context.acms_entries) == count, (
|
|
f"Expected {count} entries, got {len(context.acms_entries)}"
|
|
)
|
|
|
|
|
|
@then("the acms traversal should yield {count:d} entries")
|
|
def step_acms_yield_n_entries(context: Context, count: int) -> None:
|
|
assert len(context.acms_entries) == count, (
|
|
f"Expected {count} entries, got {len(context.acms_entries)}"
|
|
)
|
|
|
|
|
|
@then("the acms entry path should be absolute")
|
|
def step_acms_entry_path_absolute(context: Context) -> None:
|
|
for entry in context.acms_entries:
|
|
assert entry.path.is_absolute(), f"Path not absolute: {entry.path}"
|
|
|
|
|
|
@then("all acms entry paths should be absolute")
|
|
def step_acms_all_paths_absolute(context: Context) -> None:
|
|
for entry in context.acms_entries:
|
|
assert entry.path.is_absolute(), f"Path not absolute: {entry.path}"
|
|
|
|
|
|
@then("the acms entry content should be non-empty")
|
|
def step_acms_entry_content_nonempty(context: Context) -> None:
|
|
for entry in context.acms_entries:
|
|
assert entry.content, f"Empty content for {entry.path}"
|
|
|
|
|
|
@then("the acms entry content_hash should be a valid SHA-256 hex string")
|
|
def step_acms_entry_hash_valid(context: Context) -> None:
|
|
for entry in context.acms_entries:
|
|
assert len(entry.content_hash) == 64, (
|
|
f"Invalid hash length: {len(entry.content_hash)}"
|
|
)
|
|
int(entry.content_hash, 16) # raises ValueError if not hex
|
|
|
|
|
|
@then("the acms traversal should yield only top-level files")
|
|
def step_acms_only_top_level(context: Context) -> None:
|
|
assert len(context.acms_entries) == context.acms_top_level_count, (
|
|
f"Expected {context.acms_top_level_count} top-level files, "
|
|
f"got {len(context.acms_entries)}"
|
|
)
|
|
|
|
|
|
@then("all acms entries should have tags {tags_repr}")
|
|
def step_acms_entries_have_tags(context: Context, tags_repr: str) -> None:
|
|
expected_tags = ast.literal_eval(tags_repr)
|
|
for entry in context.acms_entries:
|
|
assert entry.tags == expected_tags, (
|
|
f"Expected tags {expected_tags}, got {entry.tags}"
|
|
)
|
|
|
|
|
|
@then('all acms entries should have policy "{policy}"')
|
|
def step_acms_entries_have_policy(context: Context, policy: str) -> None:
|
|
for entry in context.acms_entries:
|
|
assert entry.policy == policy, (
|
|
f"Expected policy {policy!r}, got {entry.policy!r}"
|
|
)
|
|
|
|
|
|
@then("the acms progress callback should have been called at least once")
|
|
def step_acms_progress_called(context: Context) -> None:
|
|
assert len(context.acms_progress_calls) > 0, "Progress callback was never called"
|
|
|
|
|
|
@then("the acms final progress call should report total {total:d}")
|
|
def step_acms_final_progress_total(context: Context, total: int) -> None:
|
|
assert context.acms_progress_calls, "No progress calls recorded"
|
|
_last_done, last_total = context.acms_progress_calls[-1]
|
|
assert last_total == total, f"Expected final total={total}, got {last_total}"
|
|
|
|
|
|
@then("the acms traversal should yield only the Python file")
|
|
def step_acms_only_py_file(context: Context) -> None:
|
|
assert len(context.acms_entries) == 1, (
|
|
f"Expected 1 entry (Python file only), got {len(context.acms_entries)}"
|
|
)
|
|
assert context.acms_entries[0].path.suffix == ".py", (
|
|
f"Expected .py file, got {context.acms_entries[0].path}"
|
|
)
|
|
|
|
|
|
@then("the acms traversal should yield only the visible Python file")
|
|
def step_acms_only_visible_py_file(context: Context) -> None:
|
|
assert len(context.acms_entries) == 1, (
|
|
f"Expected 1 visible Python file, got {len(context.acms_entries)}"
|
|
)
|
|
assert context.acms_entries[0].path.name == "module.py", (
|
|
f"Expected module.py, got {context.acms_entries[0].path}"
|
|
)
|
|
|
|
|
|
@then("an acms FileNotFoundError should be raised")
|
|
def step_acms_file_not_found(context: Context) -> None:
|
|
assert isinstance(context.acms_error, FileNotFoundError), (
|
|
f"Expected FileNotFoundError, got {type(context.acms_error)}"
|
|
)
|
|
|
|
|
|
@then('an acms ValueError should be raised mentioning "{text}"')
|
|
def step_acms_value_error_mentioning(context: Context, text: str) -> None:
|
|
assert isinstance(context.acms_error, ValueError), (
|
|
f"Expected ValueError, got {type(context.acms_error)}"
|
|
)
|
|
assert text in str(context.acms_error), (
|
|
f"Expected '{text}' in error message: {context.acms_error}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI integration tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _mock_acms_container(
|
|
added_files: list[Path] | None = None,
|
|
) -> MagicMock:
|
|
"""Build a mock DI container for ACMS context add CLI tests."""
|
|
container = MagicMock()
|
|
mock_context_service = MagicMock()
|
|
mock_context_service.add_to_context.return_value = (
|
|
added_files or [],
|
|
[],
|
|
)
|
|
container.context_service.return_value = mock_context_service
|
|
mock_project_service = MagicMock()
|
|
mock_project_service.get_current_project.return_value = MagicMock()
|
|
container.project_service.return_value = mock_project_service
|
|
return container
|
|
|
|
|
|
@given("a mocked context service for acms add")
|
|
def step_acms_mocked_service(context: Context) -> None:
|
|
"""Set up a mocked context service."""
|
|
context.acms_tmpdir = tempfile.mkdtemp()
|
|
|
|
|
|
@given('a temporary file "{filename}" exists')
|
|
def step_acms_temp_file_exists(context: Context, filename: str) -> None:
|
|
"""Create a temporary file."""
|
|
tmpdir = Path(context.acms_tmpdir)
|
|
filepath = tmpdir / filename
|
|
filepath.parent.mkdir(parents=True, exist_ok=True)
|
|
filepath.write_text("# test content\n", encoding="utf-8")
|
|
context.acms_temp_file = filepath
|
|
|
|
|
|
@given('a temporary directory "{dirname}" with files')
|
|
def step_acms_temp_dir_with_files(context: Context, dirname: str) -> None:
|
|
"""Create a temporary directory with files."""
|
|
tmpdir = Path(context.acms_tmpdir)
|
|
dirpath = tmpdir / dirname.rstrip("/")
|
|
dirpath.mkdir(parents=True, exist_ok=True)
|
|
(dirpath / "file1.py").write_text("# file1\n", encoding="utf-8")
|
|
(dirpath / "file2.py").write_text("# file2\n", encoding="utf-8")
|
|
context.acms_temp_dir = dirpath
|
|
|
|
|
|
@given('a temporary directory "src/" with 3 files')
|
|
def step_acms_temp_dir_3_files(context: Context) -> None:
|
|
"""Create a temporary directory with 3 files."""
|
|
tmpdir = Path(context.acms_tmpdir)
|
|
dirpath = tmpdir / "src"
|
|
dirpath.mkdir(parents=True, exist_ok=True)
|
|
for i in range(3):
|
|
(dirpath / f"file{i}.py").write_text(f"# file{i}\n", encoding="utf-8")
|
|
context.acms_temp_dir = dirpath
|
|
|
|
|
|
@when('I invoke context add with path "{filename}" and tag "{tag}"')
|
|
def step_acms_invoke_add_with_tag(context: Context, filename: str, tag: str) -> None:
|
|
"""Invoke context add with a single tag."""
|
|
filepath = context.acms_temp_file
|
|
container = _mock_acms_container(added_files=[filepath])
|
|
with patch(
|
|
"cleveragents.application.container.get_container",
|
|
return_value=container,
|
|
):
|
|
result = context.acms_runner.invoke(
|
|
context_app, ["add", str(filepath), "--tag", tag]
|
|
)
|
|
context.acms_result = result
|
|
|
|
|
|
@when('I invoke context add with path "{filename}" and tags "{tag1}" and "{tag2}"')
|
|
def step_acms_invoke_add_with_tags(
|
|
context: Context, filename: str, tag1: str, tag2: str
|
|
) -> None:
|
|
"""Invoke context add with multiple tags."""
|
|
filepath = context.acms_temp_file
|
|
container = _mock_acms_container(added_files=[filepath])
|
|
with patch(
|
|
"cleveragents.application.container.get_container",
|
|
return_value=container,
|
|
):
|
|
result = context.acms_runner.invoke(
|
|
context_app,
|
|
["add", str(filepath), "--tag", tag1, "--tag", tag2],
|
|
)
|
|
context.acms_result = result
|
|
|
|
|
|
@when('I invoke context add with path "{filename}" and policy "{policy}"')
|
|
def step_acms_invoke_add_with_policy(
|
|
context: Context, filename: str, policy: str
|
|
) -> None:
|
|
"""Invoke context add with a policy."""
|
|
filepath = context.acms_temp_file
|
|
container = _mock_acms_container(added_files=[filepath])
|
|
with patch(
|
|
"cleveragents.application.container.get_container",
|
|
return_value=container,
|
|
):
|
|
result = context.acms_runner.invoke(
|
|
context_app, ["add", str(filepath), "--policy", policy]
|
|
)
|
|
context.acms_result = result
|
|
|
|
|
|
@when('I invoke context add with directory "{dirname}" and --no-recursive')
|
|
def step_acms_invoke_add_no_recursive(context: Context, dirname: str) -> None:
|
|
"""Invoke context add with --no-recursive."""
|
|
dirpath = context.acms_temp_dir
|
|
container = _mock_acms_container(added_files=[dirpath / "file1.py"])
|
|
with patch(
|
|
"cleveragents.application.container.get_container",
|
|
return_value=container,
|
|
):
|
|
result = context.acms_runner.invoke(
|
|
context_app, ["add", str(dirpath), "--no-recursive"]
|
|
)
|
|
context.acms_result = result
|
|
|
|
|
|
@when('I invoke context add with directory "src/"')
|
|
def step_acms_invoke_add_directory(context: Context) -> None:
|
|
"""Invoke context add with a directory."""
|
|
dirpath = context.acms_temp_dir
|
|
container = _mock_acms_container(
|
|
added_files=[dirpath / f"file{i}.py" for i in range(3)]
|
|
)
|
|
with patch(
|
|
"cleveragents.application.container.get_container",
|
|
return_value=container,
|
|
):
|
|
result = context.acms_runner.invoke(context_app, ["add", str(dirpath)])
|
|
context.acms_result = result
|
|
|
|
|
|
@when("I invoke context add --help")
|
|
def step_acms_invoke_help(context: Context) -> None:
|
|
"""Invoke context add --help."""
|
|
result = context.acms_runner.invoke(context_app, ["add", "--help"])
|
|
context.acms_result = result
|
|
|
|
|
|
@then("the acms context add command should succeed")
|
|
def step_acms_add_success(context: Context) -> None:
|
|
assert context.acms_result.exit_code == 0, (
|
|
f"Exit code: {context.acms_result.exit_code}\n{context.acms_result.output}"
|
|
)
|
|
|
|
|
|
@then('the acms output should mention "{text}"')
|
|
def step_acms_output_mentions(context: Context, text: str) -> None:
|
|
assert text in context.acms_result.output, (
|
|
f"Expected '{text}' in output:\n{context.acms_result.output}"
|
|
)
|
|
|
|
|
|
@then('the acms help output should contain "{text}"')
|
|
def step_acms_help_contains(context: Context, text: str) -> None:
|
|
assert text in context.acms_result.output, (
|
|
f"Expected '{text}' in help output:\n{context.acms_result.output}"
|
|
)
|